Support deleting unreferenced files after snapshot expiration

mkdocs/docs/api.md

@@ -1615,6 +1615,31 @@ with table.maintenance.expire_snapshots() as expire:
     # Automatically commits when exiting the context
 ```
 
+#### Deleting files of expired snapshots
+
+By default, expiring snapshots only updates the table metadata and leaves every file in place. Call `clean_expired_files()` to also delete the files that only the expired snapshots referenced:
+
+```python
+table.maintenance.expire_snapshots().clean_expired_files().older_than(
+    datetime.now() - timedelta(days=3)
+).commit()
+```
+
+The files are deleted after the expiration has been committed to the catalog. A file is deleted only when no snapshot left in the table references it as a live file:
+
+- data files and delete files
+- manifests and manifest lists
+
+Statistics files and partition statistics files of the expired snapshots are deleted unless the table metadata after the commit still references them. Previous metadata files and orphan files that no snapshot referenced are not deleted.
+
+To decide which data and delete files to remove, PyIceberg reads the manifest lists of the expired snapshots and of every snapshot left in the table, and the manifests that go away with the expired snapshots. Only if those manifests list any files does it also read the manifests of every snapshot left in the table. When the table has only been written with fast appends, no manifest goes away with the expired snapshots, so only the manifest lists are read.
+
+`clean_expired_files()` cannot be used inside `table.transaction()`, because the files can only be deleted once the expiration has been committed; `commit()` raises a `ValueError` in that case. Before committing, `commit()` loads the latest table metadata from the catalog and raises a `ValueError`, with nothing expired, if `gc.enabled` is `false` there or if a branch or tag there points at a snapshot selected for expiration. The commit also requires every branch and tag other than `main` to still point where it pointed in that metadata, so if another writer moves or removes one before the commit lands, the commit fails with a `CommitFailedException` and nothing is deleted. Two changes made by another writer between that load and the commit are not detected: creating a branch or tag on a snapshot selected for expiration, and resetting `main` to one. The catalog then removes that branch or tag together with the snapshot, and the snapshot's files are deleted.
+
+`gc.enabled` defaults to `true`. When it is `false`, expiring snapshots without `clean_expired_files()` is still allowed, because it only updates the metadata.
+
+A file whose deletion raises an `OSError` is skipped and a warning is logged. If a manifest list or manifest needed for the check cannot be read, no files are deleted; if the cleanup raises any other exception, the files not deleted yet are kept. In both cases a warning with the traceback is logged and the expiration stays committed. The expired snapshots are then gone from the metadata, so the cleanup cannot be run again for them, and the files left behind can only be removed by an orphan file cleanup in another engine, such as Spark's `remove_orphan_files` procedure.
+
 #### Real-world Example
 
 ```python

pyiceberg/catalog/__init__.py

@@ -89,10 +89,12 @@
 WAREHOUSE_LOCATION = "warehouse"
 METADATA_LOCATION = "metadata_location"
 PREVIOUS_METADATA_LOCATION = "previous_metadata_location"
+DATA_FILE = "data"
 MANIFEST = "manifest"
 MANIFEST_LIST = "manifest list"
 PREVIOUS_METADATA = "previous metadata"
 METADATA = "metadata"
+STATISTICS = "statistics"
 URI = "uri"
 LOCATION = "location"
 EXTERNAL_TABLE = "EXTERNAL_TABLE"

pyiceberg/table/__init__.py

@@ -228,6 +228,9 @@ class TableProperties:
     METADATA_DELETE_AFTER_COMMIT_ENABLED = "write.metadata.delete-after-commit.enabled"
     METADATA_DELETE_AFTER_COMMIT_ENABLED_DEFAULT = False
 
+    GC_ENABLED = "gc.enabled"
+    GC_ENABLED_DEFAULT = True
+
     MAX_SNAPSHOT_AGE_MS = "history.expire.max-snapshot-age-ms"
     MAX_SNAPSHOT_AGE_MS_DEFAULT = 5 * 24 * 60 * 60 * 1000  # 5 days
 

pyiceberg/table/update/snapshot.py

@@ -1335,13 +1335,117 @@ class ExpireSnapshots(UpdateTableMetadata["ExpireSnapshots"]):
     _updates: tuple[TableUpdate, ...]
     _requirements: tuple[TableRequirement, ...]
     _snapshot_ids_to_expire: set[int]
+    _clean_expired_files: bool
 
     def __init__(self, transaction: Transaction) -> None:
         super().__init__(transaction)
+        self._clean_expired_files = False
         self._updates = ()
         self._requirements = ()
         self._snapshot_ids_to_expire = set()
 
+    def clean_expired_files(self, enabled: bool = True) -> ExpireSnapshots:
+        """Delete the files that only the expired snapshots referenced, after the expiration commits.
+
+        Deletes the data files, delete files, manifests and manifest lists that no snapshot left in the table
+        references as a live file, and the statistics and partition statistics files of the expired snapshots
+        unless the table metadata after the commit still references them. Requires an autocommit transaction.
+        `commit()` raises a `ValueError` and expires nothing if `gc.enabled` is false in the latest table metadata,
+        or if a branch or tag there points at a snapshot selected for expiration. If the cleanup fails, a warning
+        is logged and the expiration stays committed. The "Deleting files of expired snapshots" section of the
+        API docs has the details.
+
+        Args:
+            enabled: Whether to delete the files. Defaults to True.
+
+        Returns:
+            This for method chaining.
+        """
+        self._clean_expired_files = enabled
+        return self
+
+    def commit(self) -> None:
+        """Commit the expiration, then delete the files the expired snapshots were the last to reference."""
+        if not self._clean_expired_files:
+            super().commit()
+            return
+
+        from pyiceberg.table import TableProperties
+
+        if not self._transaction._autocommit:
+            raise ValueError("Cleaning expired files requires an autocommit transaction")
+        table = self._transaction._table
+        latest_metadata = table.catalog.load_table(table.name()).metadata
+        if not property_as_bool(latest_metadata.properties, TableProperties.GC_ENABLED, TableProperties.GC_ENABLED_DEFAULT):
+            raise ValueError(f"Cannot clean expired files when {TableProperties.GC_ENABLED} is false")
+        if referenced := sorted({ref.snapshot_id for ref in latest_metadata.refs.values()} & self._snapshot_ids_to_expire):
+            raise ValueError(
+                f"Cannot expire snapshots that a branch or tag references in the latest table metadata: {referenced}"
+            )
+
+        updates, requirements = self._commit()
+        metadata_before_commit = self._transaction.table_metadata
+        expired_snapshots = [
+            snapshot for snapshot in metadata_before_commit.snapshots if snapshot.snapshot_id in self._snapshot_ids_to_expire
+        ]
+        statistics_to_expire = {
+            statistics.statistics_path
+            for statistics in itertools.chain(metadata_before_commit.statistics, metadata_before_commit.partition_statistics)
+            if statistics.snapshot_id in self._snapshot_ids_to_expire
+        }
+        self._transaction._stage(updates, requirements)
+        # Fail the commit if another writer moves or drops a branch or tag other than main after the reload above.
+        # Added directly because _stage validates against the metadata this table object loaded before that reload.
+        self._transaction._requirements += tuple(
+            AssertRefSnapshotId(ref=name, snapshot_id=ref.snapshot_id)
+            for name, ref in latest_metadata.refs.items()
+            if name != MAIN_BRANCH
+        )
+        self._transaction.commit_transaction()
+
+        if not expired_snapshots:
+            return
+        metadata_after_commit = self._transaction._table.metadata
+        statistics_retained = {
+            statistics.statistics_path
+            for statistics in itertools.chain(metadata_after_commit.statistics, metadata_after_commit.partition_statistics)
+        }
+        try:
+            self._delete_expired_files(expired_snapshots, statistics_to_expire - statistics_retained)
+        except Exception:
+            logger.warning("Failed to clean up expired snapshot files", exc_info=True)
+
+    def _delete_expired_files(self, expired_snapshots: list[Snapshot], statistics_to_delete: set[str]) -> None:
+        from pyiceberg.catalog import DATA_FILE, MANIFEST, MANIFEST_LIST, STATISTICS, delete_files
+
+        table = self._transaction._table
+        io = table.io
+        executor = ExecutorFactory.get_or_create()
+        retained_snapshots = table.snapshots()
+
+        def _manifests(snapshots: list[Snapshot]) -> set[ManifestFile]:
+            return set(itertools.chain.from_iterable(executor.map(lambda snapshot: snapshot.manifests(io), snapshots)))
+
+        def _file_paths(manifest: ManifestFile, discard_deleted: bool) -> set[str]:
+            return {entry.data_file.file_path for entry in manifest.fetch_manifest_entry(io, discard_deleted=discard_deleted)}
+
+        retained_manifests = _manifests(retained_snapshots)
+        manifests_to_delete = _manifests(expired_snapshots) - retained_manifests
+        # Similar to Iceberg Java's ReachableFileCleanup: only manifests that are going away can hold files that just
+        # became unreferenced, and a file stays if any live entry of a retained manifest still points at it. Unlike
+        # Java, deleted entries of those manifests are candidates too, and any read failure aborts the whole cleanup.
+        files_to_delete: set[str] = set().union(
+            *executor.map(lambda manifest: _file_paths(manifest, discard_deleted=False), manifests_to_delete)
+        )
+        if files_to_delete:
+            for live_paths in executor.map(lambda manifest: _file_paths(manifest, discard_deleted=True), retained_manifests):
+                files_to_delete -= live_paths
+
+        delete_files(io, files_to_delete, DATA_FILE)
+        delete_files(io, {manifest.manifest_path for manifest in manifests_to_delete}, MANIFEST)
+        delete_files(io, {snapshot.manifest_list for snapshot in expired_snapshots}, MANIFEST_LIST)
+        delete_files(io, statistics_to_delete, STATISTICS)
+
     def _commit(self) -> UpdatesAndRequirements:
         """
         Commit the staged updates and requirements.