From b152031de29c940ef81cfaa8ae1e1a09705fc553 Mon Sep 17 00:00:00 2001 From: KK <37928252+dragonknightk@users.noreply.github.com> Date: Thu, 29 Oct 2026 14:55:19 +0800 Subject: [PATCH] Support deleting unreferenced files after snapshot expiration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes issue 2604 # Rationale for this change `expire_snapshots` only rewrites the table metadata, so files that only expired snapshots referenced stay on the file system. Trino and Spark both clean those up. This adds an opt-in `clean_expired_files()` step, following the approach agreed in the issue: after the expiration has committed, compare the expired snapshots against the snapshots left in the table and delete the difference, without listing the object store. ʕ•ᴥ•ʔ ### How files are chosen Similar to Iceberg Java's `ReachableFileCleanup`, but more conservative: 1. **Candidates:** entries (including DELETED entries) of manifests used by the expired snapshots and by no snapshot left in the table. 2. **Subtract:** if there are candidates, read the live entries of every manifest left in the table and remove their paths. 3. **Abort on read failure:** if any manifest list or manifest cannot be read, nothing is deleted. Read cost: - When snapshots are selected, the manifest lists of the expired and the remaining snapshots are read. - A table written only with fast appends reads no manifests, because no manifest goes away. - Reads run in parallel on the shared executor, and only file paths are kept in memory. ### Concurrent ref changes Without a check, the catalog would drop a branch or tag that another writer created on a snapshot being expired, and the cleanup would delete files that ref still needs. So `commit()` reloads the latest table metadata first: | Change by another writer | Result | |---|---| | Before the reload: a branch or tag points at a selected snapshot | `ValueError`, nothing is expired | | After the reload: a branch or tag other than `main` is moved or removed | commit fails (`AssertRefSnapshotId`), nothing is deleted | | After the reload: an append to `main` | expiration succeeds (`main` is not asserted) | | After the reload: a new branch or tag on a selected snapshot, or `main` reset to one | **not detected**: the catalog drops that ref with the snapshot, and its files are deleted | ### `gc.enabled` differs from Java Java rejects any expiration when `gc.enabled` is false, because its expiration deletes files by default. Here, expiring without `clean_expired_files()` only changes the metadata, so it stays allowed; only `clean_expired_files()` raises a `ValueError`. This matches the consensus in the issue that deleting files is opt-in. The value is read from the latest table metadata. ## Are these changes tested? Yes, in `tests/table/test_expire_snapshots.py`: - **Default:** no call, or `enabled=False` after enabling, deletes nothing. - **Deleted vs kept:** see the table under user-facing changes, including files still used by retained snapshots, tags, or snapshots another writer committed. - **Read cost:** a shared manifest is read once and only for live entries; a fast-append history reads no manifests. - **Concurrent refs:** every row of the table above except the undetected one, including a table object refreshed after the selection. - **Failures:** commit failure deletes nothing; a failed delete is skipped and logged; a corrupt retained manifest stops the cleanup while the expiration stays committed; autocommit and `gc.enabled` checks. ## Are there any user-facing changes? Yes. `ExpireSnapshots` gets a new `clean_expired_files(enabled: bool = True)` method: ```python table.maintenance.expire_snapshots().clean_expired_files().older_than(cutoff).commit() ``` After the expiration has committed: | Deleted | Kept | |---|---| | Data and delete files of the expired snapshots that no remaining snapshot references as live | Files any remaining snapshot references as live | | Manifests and manifest lists of the expired snapshots that no remaining snapshot uses | Previous metadata files | | Statistics and partition statistics files of the expired snapshots | Statistics and partition statistics files the metadata after the commit still references | | | Orphan files no snapshot ever referenced | On failure: - A file whose deletion raises an `OSError` is skipped and a warning is logged. - A manifest list or manifest that cannot be read: no files are deleted. - Any other exception: files not deleted yet are kept. - In both of the last two cases, a warning with the traceback is logged and the expiration stays committed. The expired snapshots are gone from the metadata, so the cleanup cannot be rerun; leftover files can only be removed by an orphan file cleanup in another engine, such as Spark's `remove_orphan_files` procedure. Requirements (otherwise `commit()` raises a `ValueError` before anything is expired): - an autocommit transaction - `gc.enabled = true` in the latest table metadata - no branch or tag in the latest table metadata pointing at a selected snapshot Without `clean_expired_files()`, or with `enabled=False`, `expire_snapshots` behaves as before (๑•̀ㅂ•́)و✧ The expire-snapshots section of `mkdocs/docs/api.md` documents the new option. ## AI Disclosure Human-led contribution. Grok Bot, working for dragonknightk, assisted with this change. All submitted content was reviewed by a human before submission. --- mkdocs/docs/api.md | 25 ++ pyiceberg/catalog/__init__.py | 2 + pyiceberg/table/__init__.py | 3 + pyiceberg/table/update/snapshot.py | 104 +++++++ tests/table/test_expire_snapshots.py | 420 ++++++++++++++++++++++++++- 5 files changed, 552 insertions(+), 2 deletions(-) diff --git a/mkdocs/docs/api.md b/mkdocs/docs/api.md index a732d2c0a3..884d39a29e 100644 --- a/mkdocs/docs/api.md +++ b/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 diff --git a/pyiceberg/catalog/__init__.py b/pyiceberg/catalog/__init__.py index cdb4d26b3b..65c9fefad4 100644 --- a/pyiceberg/catalog/__init__.py +++ b/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" diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py index 8d79472af5..18a8e05025 100644 --- a/pyiceberg/table/__init__.py +++ b/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 diff --git a/pyiceberg/table/update/snapshot.py b/pyiceberg/table/update/snapshot.py index 8feebb9901..d9e8a9ef13 100644 --- a/pyiceberg/table/update/snapshot.py +++ b/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. diff --git a/tests/table/test_expire_snapshots.py b/tests/table/test_expire_snapshots.py index 106e5b786c..b2d58b8363 100644 --- a/tests/table/test_expire_snapshots.py +++ b/tests/table/test_expire_snapshots.py @@ -16,14 +16,37 @@ # under the License. import threading from datetime import datetime, timedelta +from pathlib import Path +from typing import Any from unittest.mock import MagicMock, Mock +from urllib.parse import urlparse +from urllib.request import url2pathname from uuid import uuid4 +import pyarrow as pa import pytest -from pyiceberg.table import CommitTableResponse, Table -from pyiceberg.table.update import RemoveSnapshotsUpdate, update_table_metadata +from pyiceberg.catalog.memory import InMemoryCatalog +from pyiceberg.exceptions import CommitFailedException, CommitStateUnknownException +from pyiceberg.io import FileIO, InputFile, OutputFile +from pyiceberg.manifest import ( + DataFile, + DataFileContent, + FileFormat, + ManifestContent, + ManifestEntry, + ManifestEntryStatus, + ManifestFile, + ManifestWriterV2, +) +from pyiceberg.schema import Schema +from pyiceberg.table import CommitTableResponse, Table, Transaction +from pyiceberg.table.snapshots import Snapshot +from pyiceberg.table.statistics import PartitionStatisticsFile, StatisticsFile +from pyiceberg.table.update import RemoveSnapshotsUpdate, SetPartitionStatisticsUpdate, update_table_metadata from pyiceberg.table.update.snapshot import ExpireSnapshots +from pyiceberg.typedef import Record +from pyiceberg.types import LongType, NestedField def test_cannot_expire_protected_head_snapshot(table_v2: Table) -> None: @@ -316,3 +339,396 @@ def test_update_remove_snapshots_with_statistics(table_v2_with_statistics: Table assert not any(stat.snapshot_id == REMOVE_SNAPSHOT for stat in new_metadata.statistics), ( "Statistics for removed snapshot should be gone" ) + + +def _expiration_table(path: Path, appends_only: bool = False) -> Table: + catalog = InMemoryCatalog("expiration", warehouse=path.as_uri(), **{"py-io-impl": "pyiceberg.io.fsspec.FsspecFileIO"}) + catalog.create_namespace("default") + table = catalog.create_table("default.expiration", schema=Schema(NestedField(1, "x", LongType(), required=False))) + table.append(pa.table({"x": [1]})) + table.append(pa.table({"x": [2]})) + if appends_only: + table.append(pa.table({"x": [3]})) + else: + table.overwrite(pa.table({"x": [3]})) + return table + + +@pytest.fixture +def local_expiration_table(tmp_path: Path) -> Table: + return _expiration_table(tmp_path) + + +def _exists(location: str) -> bool: + return Path(url2pathname(urlparse(location).path)).exists() + + +def _snapshot_files(table: Table, snapshots: list[Snapshot]) -> set[str]: + paths: set[str] = set() + for snapshot in snapshots: + paths.add(snapshot.manifest_list) + for manifest in snapshot.manifests(table.io): + paths.add(manifest.manifest_path) + paths.update(entry.data_file.file_path for entry in manifest.fetch_manifest_entry(table.io, discard_deleted=False)) + return paths + + +def _ids(snapshots: list[Snapshot]) -> list[int]: + return [snapshot.snapshot_id for snapshot in snapshots] + + +def _forbid_manifest_list_reads(monkeypatch: pytest.MonkeyPatch) -> None: + # pytest.fail, not an assertion: the cleanup logs and swallows any Exception. + monkeypatch.setattr(Snapshot, "manifests", lambda self, io: pytest.fail("unexpected manifest list read")) + + +def _record_manifest_reads(monkeypatch: pytest.MonkeyPatch) -> list[tuple[str, bool]]: + original_fetch = ManifestFile.fetch_manifest_entry + reads: list[tuple[str, bool]] = [] + + def record_read(self: ManifestFile, io: FileIO, discard_deleted: bool = True, **kwargs: Any) -> list[ManifestEntry]: + reads.append((self.manifest_path, discard_deleted)) + return original_fetch(self, io, discard_deleted=discard_deleted, **kwargs) + + monkeypatch.setattr(ManifestFile, "fetch_manifest_entry", record_read) + return reads + + +def _run_after_next_load(table: Table, monkeypatch: pytest.MonkeyPatch, change: Any) -> None: + """Run `change` as another writer right after the expiration reloads the table metadata.""" + original_load_table = table.catalog.load_table + calls: list[Any] = [] + + def load_then_change(identifier: Any) -> Table: + loaded = original_load_table(identifier) + if not calls: + calls.append(identifier) + change(original_load_table(identifier)) + return loaded + + monkeypatch.setattr(table.catalog, "load_table", load_then_change) + + +def test_expiration_preserves_files_by_default(local_expiration_table: Table, monkeypatch: pytest.MonkeyPatch) -> None: + table = local_expiration_table + first, second, *_ = table.snapshots() + paths = _snapshot_files(table, [first, second]) + _forbid_manifest_list_reads(monkeypatch) + + table.maintenance.expire_snapshots().by_id(first.snapshot_id).commit() + table.maintenance.expire_snapshots().clean_expired_files().clean_expired_files(enabled=False).by_id( + second.snapshot_id + ).commit() + + assert len(table.snapshots()) == 2 + assert all(_exists(path) for path in paths) + + +def test_expiration_cleans_unreferenced_files( + local_expiration_table: Table, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + table = local_expiration_table + *expired, current = table.snapshots() + paths = _snapshot_files(table, expired) - _snapshot_files(table, [current]) + metadata_location = table.metadata_location + original_delete = table.io.delete + attempted: list[str] = [] + + def fail_first_delete(location: str | InputFile | OutputFile) -> None: + attempted.append(location if isinstance(location, str) else location.location) + if len(attempted) == 1: + raise OSError("injected deletion failure") + original_delete(location) + + monkeypatch.setattr(table.io, "delete", fail_first_delete) + table.maintenance.expire_snapshots().clean_expired_files().by_ids(_ids(expired)).commit() + + assert _ids(table.snapshots()) == [current.snapshot_id] + assert sorted(attempted) == sorted(paths) + assert [path for path in paths if _exists(path)] == [attempted[0]] + assert attempted[0] in caplog.text + assert _exists(metadata_location) + assert table.scan().to_arrow().to_pydict() == {"x": [3]} + + +def test_expiration_keeps_files_of_tagged_snapshot(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + # The manifest shared with the tagged snapshot is read once, and only for its live entries. + table = _expiration_table(tmp_path) + first, tagged, deleted, _ = table.snapshots() + table.manage_snapshots().create_tag(tagged.snapshot_id, "tagged").commit() + shared = _snapshot_files(table, [first]) - {first.manifest_list} + shared_manifests = {manifest.manifest_path for manifest in first.manifests(table.io)} + reads = _record_manifest_reads(monkeypatch) + + table.maintenance.expire_snapshots().clean_expired_files().by_ids([first.snapshot_id, deleted.snapshot_id]).commit() + + assert not _exists(first.manifest_list) + assert all(_exists(path) for path in shared) + assert len(reads) == len(set(reads)) + assert all(discard_deleted for path, discard_deleted in reads if path in shared_manifests) + assert sorted(table.scan(snapshot_id=tagged.snapshot_id).to_arrow().column("x").to_pylist()) == [1, 2] + + +def test_expiration_cleans_files_from_deleted_entries(local_expiration_table: Table) -> None: + table = local_expiration_table + first, second, deleted, _ = table.snapshots() + entries = [entry for manifest in deleted.manifests(table.io) for entry in manifest.fetch_manifest_entry(table.io, False)] + assert entries and all(entry.status == ManifestEntryStatus.DELETED for entry in entries) + table.maintenance.expire_snapshots().by_ids([first.snapshot_id, second.snapshot_id]).commit() + + table.maintenance.expire_snapshots().clean_expired_files().by_id(deleted.snapshot_id).commit() + + assert not any(_exists(entry.data_file.file_path) for entry in entries) + + +def test_expiration_on_append_only_table_reads_no_manifests(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + table = _expiration_table(tmp_path, appends_only=True) + *expired, current = table.snapshots() + reads = _record_manifest_reads(monkeypatch) + + table.maintenance.expire_snapshots().clean_expired_files().by_ids(_ids(expired)).commit() + + assert reads == [] + assert not any(_exists(snapshot.manifest_list) for snapshot in expired) + assert sorted(table.scan().to_arrow().column("x").to_pylist()) == [1, 2, 3] + + +class _DeleteManifestWriter(ManifestWriterV2): + def content(self) -> ManifestContent: + return ManifestContent.DELETES + + @property + def _meta(self) -> dict[str, str]: + return {**super()._meta, "content": "deletes"} + + +def _delete_file(table: Table, name: str) -> DataFile: + # The cleanup only looks at manifest entries, so the file body does not need to be a valid delete file. + path = f"{table.location()}/data/{name}" + with table.io.new_output(path).create() as stream: + stream.write(b"delete file") + return DataFile.from_args( + content=DataFileContent.POSITION_DELETES, + file_path=path, + file_format=FileFormat.PARQUET, + partition=Record(), + record_count=1, + file_size_in_bytes=11, + ) + + +def _commit_delete_manifest(table: Table, added: DataFile, replaced: ManifestFile | None = None) -> ManifestFile: + """Commit a snapshot with a delete manifest that adds `added` and marks the entries of `replaced` as deleted.""" + with table.transaction() as transaction, transaction.update_snapshot().fast_append() as append: + writer = _DeleteManifestWriter( + table.spec(), table.schema(), append.new_manifest_output(), append._snapshot_id, append._compression + ) + with writer: + writer.add(ManifestEntry.from_args(status=ManifestEntryStatus.ADDED, data_file=added)) + if replaced is not None: + for entry in replaced.fetch_manifest_entry(table.io): + writer.delete(entry) + manifest = writer.to_manifest_file() + kept = [existing for existing in append._existing_manifests() if existing != replaced] + [manifest] + append._existing_manifests = lambda: kept # type: ignore[method-assign] + return manifest + + +def test_expiration_cleans_delete_files(local_expiration_table: Table) -> None: + table = local_expiration_table + expired_deletes = _delete_file(table, "expired-deletes.parquet") + retained_deletes = _delete_file(table, "retained-deletes.parquet") + expired_manifest = _commit_delete_manifest(table, expired_deletes) + retained_manifest = _commit_delete_manifest(table, retained_deletes, replaced=expired_manifest) + *expired, current = table.snapshots() + + table.maintenance.expire_snapshots().clean_expired_files().by_ids(_ids(expired)).commit() + + assert not _exists(expired_deletes.file_path) + assert not _exists(expired_manifest.manifest_path) + assert _exists(retained_deletes.file_path) + assert _exists(retained_manifest.manifest_path) + + +def _add_statistics(table: Table, snapshot: Snapshot, path: str, partition: bool = False) -> None: + if not _exists(path): + with table.io.new_output(path).create() as stream: + stream.write(b"statistics") + if partition: + statistics = PartitionStatisticsFile(snapshot_id=snapshot.snapshot_id, statistics_path=path, file_size_in_bytes=10) + with table.transaction() as transaction: + transaction._apply((SetPartitionStatisticsUpdate(partition_statistics=statistics),)) + else: + statistics_file = StatisticsFile( + snapshot_id=snapshot.snapshot_id, + statistics_path=path, + file_size_in_bytes=10, + file_footer_size_in_bytes=4, + blob_metadata=[], + ) + table.update_statistics().set_statistics(statistics_file).commit() + + +def test_expiration_cleans_statistics_files(local_expiration_table: Table, monkeypatch: pytest.MonkeyPatch) -> None: + # A statistics file of an expired snapshot is kept while the table metadata after the commit still lists its path. + table = local_expiration_table + first, second, deleted, current = table.snapshots() + expired, expired_partition, shared, shared_with_partition = ( + f"{table.location()}/metadata/{name}.stats" for name in ("a", "b", "c", "d") + ) + _add_statistics(table, first, expired) + _add_statistics(table, first, expired_partition, partition=True) + _add_statistics(table, second, shared) + _add_statistics(table, current, shared) + _add_statistics(table, deleted, shared_with_partition) + _add_statistics(table, current, shared_with_partition, partition=True) + original_commit_table = table.catalog.commit_table + + def commit_dropping_partition_statistics(*args: Any, **kwargs: Any) -> CommitTableResponse: + # Like a REST catalog that also drops the partition statistics of removed snapshots. + response = original_commit_table(*args, **kwargs) + snapshot_ids = set(_ids(response.metadata.snapshots)) + partition_statistics = [s for s in response.metadata.partition_statistics if s.snapshot_id in snapshot_ids] + metadata = response.metadata.model_copy(update={"partition_statistics": partition_statistics}) + return CommitTableResponse(metadata=metadata, metadata_location=response.metadata_location) + + monkeypatch.setattr(table.catalog, "commit_table", commit_dropping_partition_statistics) + table.maintenance.expire_snapshots().clean_expired_files().by_ids(_ids([first, second, deleted])).commit() + + assert not _exists(expired) + assert not _exists(expired_partition) + assert _exists(shared) + assert _exists(shared_with_partition) + + +@pytest.mark.parametrize("error_type", [CommitFailedException, CommitStateUnknownException]) +def test_expiration_does_not_delete_when_commit_fails( + local_expiration_table: Table, monkeypatch: pytest.MonkeyPatch, error_type: type[Exception] +) -> None: + table = local_expiration_table + expired = table.snapshots()[:-1] + paths = _snapshot_files(table, expired) + monkeypatch.setattr("pyiceberg.table.Transaction.commit_transaction", Mock(side_effect=error_type("injected commit failure"))) + + with pytest.raises(error_type, match="injected commit failure"): + table.maintenance.expire_snapshots().clean_expired_files().by_ids(_ids(expired)).commit() + + assert all(_exists(path) for path in paths) + + +def test_expiration_stays_committed_when_retained_manifest_is_corrupt( + local_expiration_table: Table, caplog: pytest.LogCaptureFixture +) -> None: + table = local_expiration_table + *expired, current = table.snapshots() + paths = _snapshot_files(table, expired) + (retained_manifest,) = current.manifests(table.io) + with table.io.new_output(retained_manifest.manifest_path).create(overwrite=True) as stream: + stream.write(b"truncated") + + table.maintenance.expire_snapshots().clean_expired_files().by_ids(_ids(expired)).commit() + + assert _ids(table.catalog.load_table(table.name()).snapshots()) == [current.snapshot_id] + assert all(_exists(path) for path in paths) + assert "Failed to clean up expired snapshot files" in caplog.text + + +def test_expiration_cleanup_requires_autocommit(local_expiration_table: Table) -> None: + with pytest.raises(ValueError, match="autocommit"): + ExpireSnapshots(Transaction(local_expiration_table)).clean_expired_files().commit() + + +def test_expiration_cleanup_with_no_selected_snapshots(local_expiration_table: Table, monkeypatch: pytest.MonkeyPatch) -> None: + _forbid_manifest_list_reads(monkeypatch) + local_expiration_table.maintenance.expire_snapshots().clean_expired_files().commit() + assert len(local_expiration_table.snapshots()) == 4 + + +def test_expiration_respects_gc_enabled_in_latest_metadata(local_expiration_table: Table) -> None: + table = local_expiration_table + *expired, current = table.snapshots() + paths = _snapshot_files(table, table.snapshots()) + with table.catalog.load_table(table.name()).transaction() as transaction: + transaction.set_properties({"gc.enabled": "false"}) + + with pytest.raises(ValueError, match="gc.enabled is false"): + table.maintenance.expire_snapshots().clean_expired_files().by_ids(_ids(expired)).commit() + assert len(table.catalog.load_table(table.name()).snapshots()) == 4 + + table.maintenance.expire_snapshots().by_ids(_ids(expired)).commit() + assert _ids(table.snapshots()) == [current.snapshot_id] + assert all(_exists(path) for path in paths) + + +@pytest.mark.parametrize("change", ["tag", "rollback", "tag_then_refresh"]) +def test_expiration_cleanup_rejects_snapshot_referenced_in_latest_metadata(local_expiration_table: Table, change: str) -> None: + # Without the check, the catalog drops a ref that another writer put on the snapshot, and its files are deleted. + table = local_expiration_table + first = table.snapshots()[0] + paths = _snapshot_files(table, table.snapshots()) + expiration = table.maintenance.expire_snapshots().clean_expired_files().by_ids(_ids(table.snapshots()[:-1])) + other_writer = table.catalog.load_table(table.name()) + if change == "rollback": + other_writer.manage_snapshots().rollback_to_snapshot(first.snapshot_id).commit() + else: + other_writer.manage_snapshots().create_tag(first.snapshot_id, "keep").commit() + if change == "tag_then_refresh": + table.refresh() + + with pytest.raises(ValueError, match=rf"references in the latest table metadata: \[{first.snapshot_id}\]"): + expiration.commit() + + assert len(table.catalog.load_table(table.name()).snapshots()) == 4 + assert all(_exists(path) for path in paths) + + +def test_expiration_cleanup_fails_when_ref_moves_after_reload( + local_expiration_table: Table, monkeypatch: pytest.MonkeyPatch +) -> None: + table = local_expiration_table + first, retained, _, current = table.snapshots() + table.manage_snapshots().create_tag(current.snapshot_id, "other").create_tag(retained.snapshot_id, "keep").commit() + paths = _snapshot_files(table, table.snapshots()) + + def move_tag(other_writer: Table) -> None: + other_writer.manage_snapshots().remove_tag("keep").commit() + other_writer.manage_snapshots().create_tag(first.snapshot_id, "keep").commit() + + _run_after_next_load(table, monkeypatch, move_tag) + with pytest.raises(CommitFailedException, match="keep has changed"): + table.maintenance.expire_snapshots().clean_expired_files().by_id(first.snapshot_id).commit() + + assert len(table.catalog.load_table(table.name()).snapshots()) == 4 + assert all(_exists(path) for path in paths) + + +def test_expiration_cleanup_allows_concurrent_append_to_main( + local_expiration_table: Table, monkeypatch: pytest.MonkeyPatch +) -> None: + table = local_expiration_table + expired = table.snapshots()[:-1] + _run_after_next_load(table, monkeypatch, lambda other_writer: other_writer.append(pa.table({"x": [4]}))) + + table.maintenance.expire_snapshots().clean_expired_files().by_ids(_ids(expired)).commit() + + assert len(table.snapshots()) == 2 + assert sorted(table.scan().to_arrow().column("x").to_pylist()) == [3, 4] + + +def test_expiration_keeps_files_of_concurrently_committed_snapshots(local_expiration_table: Table) -> None: + # The files to keep come from the metadata the commit returned, including snapshots and refs other writers committed. + table = local_expiration_table + *expired, _ = table.snapshots() + (entry,) = [entry for manifest in expired[0].manifests(table.io) for entry in manifest.fetch_manifest_entry(table.io)] + with ( + table.catalog.load_table(table.name()).transaction() as transaction, + transaction.update_snapshot().fast_append() as append, + ): + append.append_data_file(entry.data_file) + other_writer = table.catalog.load_table(table.name()) + other_writer.manage_snapshots().create_tag(other_writer.snapshots()[-1].snapshot_id, "other").commit() + + table.maintenance.expire_snapshots().clean_expired_files().by_ids(_ids(expired)).commit() + + assert _exists(entry.data_file.file_path) + assert sorted(table.scan().to_arrow().column("x").to_pylist()) == [1, 3]