From bbea3166a4610f38e716e4d95aee8622f56b465e Mon Sep 17 00:00:00 2001 From: Alex Stephen Date: Fri, 21 Aug 2026 19:02:53 +0000 Subject: [PATCH 1/2] Fail when explicitly deleted data file is missing --- pyiceberg/table/update/snapshot.py | 23 ++++++++- tests/table/test_snapshots.py | 83 ++++++++++++++++++++++++++++++ 2 files changed, 104 insertions(+), 2 deletions(-) diff --git a/pyiceberg/table/update/snapshot.py b/pyiceberg/table/update/snapshot.py index 7931edacdd..ef983ca861 100644 --- a/pyiceberg/table/update/snapshot.py +++ b/pyiceberg/table/update/snapshot.py @@ -26,6 +26,7 @@ from typing import TYPE_CHECKING, Generic from pyiceberg.avro.codecs import AvroCompressionCodec +from pyiceberg.exceptions import ValidationException from pyiceberg.expressions import AlwaysFalse, BooleanExpression, Or from pyiceberg.expressions.visitors import ( ROWS_MIGHT_NOT_MATCH, @@ -667,9 +668,27 @@ def _get_entries(manifest: ManifestFile) -> list[ManifestEntry]: ] list_of_entries = executor.map(_get_entries, previous_snapshot.manifests(self._io)) - return list(itertools.chain(*list_of_entries)) + deleted_entries = list(itertools.chain(*list_of_entries)) else: - return [] + deleted_entries = [] + + self._validate_required_deletes(deleted_entries) + + return deleted_entries + + def _validate_required_deletes(self, deleted_entries: list[ManifestEntry]) -> None: + """Validate that every explicitly deleted data file is present in the current manifests. + + A data file that was passed to `delete_data_file` can already be absent from the base + snapshot, for example when it was removed by an earlier commit. Committing anyway would + silently reintroduce the data of its replacement files, and skew the snapshot summary. + + Raises: + ValidationException: If a data file to delete is missing from the current manifests. + """ + found_data_files = {entry.data_file for entry in deleted_entries} + if missing := [data_file.file_path for data_file in self._deleted_data_files if data_file not in found_data_files]: + raise ValidationException(f"Missing required files to delete: {', '.join(sorted(missing))}") class UpdateSnapshot: diff --git a/tests/table/test_snapshots.py b/tests/table/test_snapshots.py index 5f1680ed59..423f5f2cc2 100644 --- a/tests/table/test_snapshots.py +++ b/tests/table/test_snapshots.py @@ -15,10 +15,16 @@ # specific language governing permissions and limitations # under the License. # pylint:disable=redefined-outer-name,eval-used +import re +import uuid from typing import cast +import pyarrow as pa import pytest +from pyiceberg.catalog import Catalog +from pyiceberg.exceptions import ValidationException +from pyiceberg.io.pyarrow import _dataframe_to_data_files from pyiceberg.manifest import DataFile, DataFileContent, ManifestContent, ManifestFile from pyiceberg.partitioning import PartitionField, PartitionSpec from pyiceberg.schema import Schema @@ -649,3 +655,80 @@ def summary_calls(n_files: int) -> int: f"_MergeAppendFiles.__init__ made {merge_init - fast_init} extra update_table_metadata " "calls over its superclass; expected 1 (hoisted)" ) + + +def _rewrite(table: Table, df: pa.Table) -> DataFile: + return next( + iter( + _dataframe_to_data_files( + table_metadata=table.metadata, + df=df, + io=table.io, + write_uuid=uuid.uuid4(), + ) + ) + ) + + +def _total_data_files(table: Table) -> str: + snapshot = table.current_snapshot() + assert snapshot is not None and snapshot.summary is not None + return snapshot.summary.additional_properties["total-data-files"] + + +def test_overwrite_replaces_a_file_that_is_present(catalog: Catalog, arrow_table_simple: pa.Table) -> None: + catalog.create_namespace("default") + table = catalog.create_table("default.overwrite", arrow_table_simple.schema) + table.append(arrow_table_simple) + + data_file = list(table.scan().plan_files())[0].file + replacement = _rewrite(table, arrow_table_simple.slice(0, 1)) + + with table.transaction() as tx: + with tx.update_snapshot().overwrite() as overwrite: + overwrite.delete_data_file(data_file) + overwrite.append_data_file(replacement) + + assert table.scan().to_arrow()["foo"].to_pylist() == ["a"] + assert _total_data_files(table) == "1" + + +def test_overwrite_rejects_file_missing_from_base(catalog: Catalog, arrow_table_simple: pa.Table) -> None: + catalog.create_namespace("default") + table = catalog.create_table("default.overwrite", arrow_table_simple.schema) + table.append(arrow_table_simple) + + stale_file = list(table.scan().plan_files())[0].file + stale_rows = table.scan().to_arrow() + + # Delete the file before the replacement transaction begins + with catalog.load_table("default.overwrite").transaction() as tx: + with tx.update_snapshot().overwrite() as overwrite: + overwrite.delete_data_file(stale_file) + + current = catalog.load_table("default.overwrite") + replacement = _rewrite(current, stale_rows) + + with pytest.raises(ValidationException, match=re.escape(f"Missing required files to delete: {stale_file.file_path}")): + with current.transaction() as tx: + with tx.update_snapshot().overwrite() as overwrite: + overwrite.delete_data_file(stale_file) + overwrite.append_data_file(replacement) + + committed = catalog.load_table("default.overwrite") + assert committed.scan().to_arrow()["foo"].to_pylist() == [] + assert _total_data_files(committed) == "0" + + +def test_overwrite_rejects_deletes_without_a_parent_snapshot(catalog: Catalog, arrow_table_simple: pa.Table) -> None: + catalog.create_namespace("default") + table = catalog.create_table("default.overwrite", arrow_table_simple.schema) + table.append(arrow_table_simple) + + stale_file = list(table.scan().plan_files())[0].file + empty = catalog.create_table("default.empty", arrow_table_simple.schema) + + with pytest.raises(ValidationException, match=re.escape(f"Missing required files to delete: {stale_file.file_path}")): + with empty.transaction() as tx: + with tx.update_snapshot().overwrite() as overwrite: + overwrite.delete_data_file(stale_file) From 0a6c5fc666ee82c1e7e6196fd3dcf4dbc808a9f5 Mon Sep 17 00:00:00 2001 From: Kevin Liu Date: Sat, 22 Aug 2026 20:36:34 -0700 Subject: [PATCH 2/2] Avoid writing manifests before overwrite validation --- pyiceberg/table/update/snapshot.py | 17 ++++--- tests/table/test_snapshots.py | 75 ++++++++++++++++-------------- 2 files changed, 50 insertions(+), 42 deletions(-) diff --git a/pyiceberg/table/update/snapshot.py b/pyiceberg/table/update/snapshot.py index ef983ca861..afbeef7e99 100644 --- a/pyiceberg/table/update/snapshot.py +++ b/pyiceberg/table/update/snapshot.py @@ -197,8 +197,6 @@ def _write_added_manifest() -> list[ManifestFile]: return [] def _write_delete_manifest() -> list[ManifestFile]: - # Check if we need to mark the files as deleted - deleted_entries = self._deleted_entries() if len(deleted_entries) > 0: deleted_manifests = [] partition_groups: dict[int, list[ManifestEntry]] = defaultdict(list) @@ -215,6 +213,8 @@ def _write_delete_manifest() -> list[ManifestFile]: # Updates self._predicate with computed partition predicate for manifest pruning self._build_delete_files_partition_predicate() + # Plan deletes before starting manifest writers so validation failures do not leave orphaned manifests + deleted_entries = self._deleted_entries() executor = ExecutorFactory.get_or_create() @@ -677,14 +677,17 @@ def _get_entries(manifest: ManifestFile) -> list[ManifestEntry]: return deleted_entries def _validate_required_deletes(self, deleted_entries: list[ManifestEntry]) -> None: - """Validate that every explicitly deleted data file is present in the current manifests. + """Validate that explicitly deleted data files exist in the parent snapshot. - A data file that was passed to `delete_data_file` can already be absent from the base - snapshot, for example when it was removed by an earlier commit. Committing anyway would - silently reintroduce the data of its replacement files, and skew the snapshot summary. + Files passed to `delete_data_file` are required deletes. If one is absent, an overwrite + could commit replacement files without the corresponding deletion and produce incorrect + snapshot summary totals. + + Args: + deleted_entries: Live parent-snapshot entries selected for deletion. Raises: - ValidationException: If a data file to delete is missing from the current manifests. + ValidationException: If a required data file is missing. """ found_data_files = {entry.data_file for entry in deleted_entries} if missing := [data_file.file_path for data_file in self._deleted_data_files if data_file not in found_data_files]: diff --git a/tests/table/test_snapshots.py b/tests/table/test_snapshots.py index 423f5f2cc2..0f72b08087 100644 --- a/tests/table/test_snapshots.py +++ b/tests/table/test_snapshots.py @@ -17,7 +17,9 @@ # pylint:disable=redefined-outer-name,eval-used import re import uuid +from pathlib import Path from typing import cast +from urllib.parse import urlparse import pyarrow as pa import pytest @@ -657,12 +659,20 @@ def summary_calls(n_files: int) -> int: ) -def _rewrite(table: Table, df: pa.Table) -> DataFile: +@pytest.fixture +def overwrite_table(catalog: Catalog, arrow_table_simple: pa.Table) -> Table: + catalog.create_namespace("default") + table = catalog.create_table("default.overwrite", arrow_table_simple.schema) + table.append(arrow_table_simple) + return table + + +def _write_data_file(table: Table, rows: pa.Table) -> DataFile: return next( iter( _dataframe_to_data_files( table_metadata=table.metadata, - df=df, + df=rows, io=table.io, write_uuid=uuid.uuid4(), ) @@ -670,65 +680,60 @@ def _rewrite(table: Table, df: pa.Table) -> DataFile: ) -def _total_data_files(table: Table) -> str: +def _total_data_file_count(table: Table) -> int: snapshot = table.current_snapshot() assert snapshot is not None and snapshot.summary is not None - return snapshot.summary.additional_properties["total-data-files"] - + return int(snapshot.summary.additional_properties["total-data-files"]) -def test_overwrite_replaces_a_file_that_is_present(catalog: Catalog, arrow_table_simple: pa.Table) -> None: - catalog.create_namespace("default") - table = catalog.create_table("default.overwrite", arrow_table_simple.schema) - table.append(arrow_table_simple) - data_file = list(table.scan().plan_files())[0].file - replacement = _rewrite(table, arrow_table_simple.slice(0, 1)) +def test_overwrite_replaces_existing_file(overwrite_table: Table, arrow_table_simple: pa.Table) -> None: + original_file = next(iter(overwrite_table.scan().plan_files())).file + replacement = _write_data_file(overwrite_table, arrow_table_simple.slice(0, 1)) - with table.transaction() as tx: + with overwrite_table.transaction() as tx: with tx.update_snapshot().overwrite() as overwrite: - overwrite.delete_data_file(data_file) + overwrite.delete_data_file(original_file) overwrite.append_data_file(replacement) - assert table.scan().to_arrow()["foo"].to_pylist() == ["a"] - assert _total_data_files(table) == "1" + assert overwrite_table.scan().to_arrow()["foo"].to_pylist() == ["a"] + assert _total_data_file_count(overwrite_table) == 1 -def test_overwrite_rejects_file_missing_from_base(catalog: Catalog, arrow_table_simple: pa.Table) -> None: - catalog.create_namespace("default") - table = catalog.create_table("default.overwrite", arrow_table_simple.schema) - table.append(arrow_table_simple) - - stale_file = list(table.scan().plan_files())[0].file - stale_rows = table.scan().to_arrow() +def test_overwrite_rejects_explicit_delete_missing_from_base_snapshot(catalog: Catalog, overwrite_table: Table) -> None: + stale_file = next(iter(overwrite_table.scan().plan_files())).file + stale_rows = overwrite_table.scan().to_arrow() # Delete the file before the replacement transaction begins - with catalog.load_table("default.overwrite").transaction() as tx: + with catalog.load_table(overwrite_table.name()).transaction() as tx: with tx.update_snapshot().overwrite() as overwrite: overwrite.delete_data_file(stale_file) - current = catalog.load_table("default.overwrite") - replacement = _rewrite(current, stale_rows) + current = catalog.load_table(overwrite_table.name()) + replacement = _write_data_file(current, stale_rows) + expected_error = re.escape(f"Missing required files to delete: {stale_file.file_path}") - with pytest.raises(ValidationException, match=re.escape(f"Missing required files to delete: {stale_file.file_path}")): + with pytest.raises(ValidationException, match=expected_error): with current.transaction() as tx: with tx.update_snapshot().overwrite() as overwrite: overwrite.delete_data_file(stale_file) overwrite.append_data_file(replacement) - committed = catalog.load_table("default.overwrite") - assert committed.scan().to_arrow()["foo"].to_pylist() == [] - assert _total_data_files(committed) == "0" + metadata_path = Path(urlparse(current.location()).path) / "metadata" + assert not any(metadata_path.glob(f"{overwrite.commit_uuid}-m*.avro")) + committed = catalog.load_table(overwrite_table.name()) + assert committed.scan().to_arrow()["foo"].to_pylist() == [] + assert _total_data_file_count(committed) == 0 -def test_overwrite_rejects_deletes_without_a_parent_snapshot(catalog: Catalog, arrow_table_simple: pa.Table) -> None: - catalog.create_namespace("default") - table = catalog.create_table("default.overwrite", arrow_table_simple.schema) - table.append(arrow_table_simple) - stale_file = list(table.scan().plan_files())[0].file +def test_overwrite_rejects_explicit_delete_without_parent_snapshot( + catalog: Catalog, overwrite_table: Table, arrow_table_simple: pa.Table +) -> None: + stale_file = next(iter(overwrite_table.scan().plan_files())).file empty = catalog.create_table("default.empty", arrow_table_simple.schema) + expected_error = re.escape(f"Missing required files to delete: {stale_file.file_path}") - with pytest.raises(ValidationException, match=re.escape(f"Missing required files to delete: {stale_file.file_path}")): + with pytest.raises(ValidationException, match=expected_error): with empty.transaction() as tx: with tx.update_snapshot().overwrite() as overwrite: overwrite.delete_data_file(stale_file)