(iceberg-python) branch main updated: Fail when explicitly deleted data file is missing (#3818)

[ke...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

kevinjqliu pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-python.git


The following commit(s) were added to refs/heads/main by this push:
     new 75396614a Fail when explicitly deleted data file is missing (#3818)
75396614a is described below

commit 75396614a5f093cf24e93c7d916ee2cda0bdf44e
Author: Alex Stephen <13...@users.noreply.github.com>
AuthorDate: Sat Aug 22 21:24:19 2026 -0700

    Fail when explicitly deleted data file is missing (#3818)
    
    * Fail when explicitly deleted data file is missing
    
    * Avoid writing manifests before overwrite validation
    
    ---------
    
    Co-authored-by: Kevin Liu <ke...@gmail.com>
---
 pyiceberg/table/update/snapshot.py | 29 +++++++++++--
 tests/table/test_snapshots.py      | 88 ++++++++++++++++++++++++++++++++++++++
 2 files changed, 113 insertions(+), 4 deletions(-)

diff --git a/pyiceberg/table/update/snapshot.py b/pyiceberg/table/update/snapshot.py
index e500e09fa..6bddd2790 100644
--- a/pyiceberg/table/update/snapshot.py
+++ b/pyiceberg/table/update/snapshot.py
@@ -247,8 +247,6 @@ class _SnapshotProducer(UpdateTableMetadata[U], Generic[U]):
                 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)
@@ -265,6 +263,8 @@ class _SnapshotProducer(UpdateTableMetadata[U], Generic[U]):
 
         # 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()
 
@@ -850,9 +850,30 @@ class _OverwriteFiles(_SnapshotProducer["_OverwriteFiles"]):
                 ]
 
             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 explicitly deleted data files exist in the parent snapshot.
+
+        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 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]:
+            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 5f1680ed5..0f72b0808 100644
--- a/tests/table/test_snapshots.py
+++ b/tests/table/test_snapshots.py
@@ -15,10 +15,18 @@
 # specific language governing permissions and limitations
 # under the License.
 # 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
 
+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 +657,83 @@ def test_snapshot_producer_bounded_metadata_access(table_v2: Table) -> None:
             f"_MergeAppendFiles.__init__ made {merge_init - fast_init} extra update_table_metadata "
             "calls over its superclass; expected 1 (hoisted)"
         )
+
+
+@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=rows,
+                io=table.io,
+                write_uuid=uuid.uuid4(),
+            )
+        )
+    )
+
+
+def _total_data_file_count(table: Table) -> int:
+    snapshot = table.current_snapshot()
+    assert snapshot is not None and snapshot.summary is not None
+    return int(snapshot.summary.additional_properties["total-data-files"])
+
+
+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 overwrite_table.transaction() as tx:
+        with tx.update_snapshot().overwrite() as overwrite:
+            overwrite.delete_data_file(original_file)
+            overwrite.append_data_file(replacement)
+
+    assert overwrite_table.scan().to_arrow()["foo"].to_pylist() == ["a"]
+    assert _total_data_file_count(overwrite_table) == 1
+
+
+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(overwrite_table.name()).transaction() as tx:
+        with tx.update_snapshot().overwrite() as overwrite:
+            overwrite.delete_data_file(stale_file)
+
+    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=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)
+
+    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_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=expected_error):
+        with empty.transaction() as tx:
+            with tx.update_snapshot().overwrite() as overwrite:
+                overwrite.delete_data_file(stale_file)