(iceberg-python) branch main updated: Fix: use new snapshot id in deleted manifest entry unless is existing entry (#2266)

[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 cf987c68 Fix: use new snapshot id in deleted manifest entry unless is existing entry (#2266)
cf987c68 is described below

commit cf987c68cd760b51c477c69e761a45e66aec2fb7
Author: Leon Lin <li...@gmail.com>
AuthorDate: Mon Aug 18 21:07:12 2025 -0700

    Fix: use new snapshot id in deleted manifest entry unless is existing entry (#2266)
    
    <!--
    Thanks for opening a pull request!
    -->
    <!-- In the case this PR will resolve an issue, please replace
    ${GITHUB_ISSUE_ID} below with the actual Github issue id. -->
    <!-- Closes #${GITHUB_ISSUE_ID} -->
    
    # Rationale for this change
    
    Based on iceberg spec, when a manifest entry is marked as deleted, the
    snapshot id for when the entry was deleted should be used.
    https://iceberg.apache.org/spec/?h=deletes#manifest-entry-fields
    ```
    When a file is replaced or deleted from the dataset, its manifest entry fields store the snapshot ID in which the file was deleted and status 2 (deleted). The file may be deleted from the file system when the snapshot in which it was deleted is garbage collected, assuming that older snapshots have also been garbage collected [1].
    ```
    
    
    https://github.com/apache/iceberg/blob/main/core/src/main/java/org/apache/iceberg/ManifestWriter.java#L178-L179
    
    Incorrect snapshot id could lead to data being deleted during garbage
    collection when not supposed to.
    
    # Are these changes tested?
    
    Added in `test_deletes.py`
    
    # Are there any user-facing changes?
    
    <!-- In the case of user-facing changes, please add the changelog label.
    -->
    
    No
    
    ---------
    
    Co-authored-by: Leon Lin <ll...@amazon.com>
    Co-authored-by: Fokko Driesprong <fo...@apache.org>
---
 pyiceberg/table/update/snapshot.py |  3 ++-
 tests/integration/test_deletes.py  | 54 +++++++++++++++++++++++++++++++++++++-
 2 files changed, 55 insertions(+), 2 deletions(-)

diff --git a/pyiceberg/table/update/snapshot.py b/pyiceberg/table/update/snapshot.py
index 018f6614..2ed03ced 100644
--- a/pyiceberg/table/update/snapshot.py
+++ b/pyiceberg/table/update/snapshot.py
@@ -407,7 +407,8 @@ class _DeleteFiles(_SnapshotProducer["_DeleteFiles"]):
         def _copy_with_new_status(entry: ManifestEntry, status: ManifestEntryStatus) -> ManifestEntry:
             return ManifestEntry.from_args(
                 status=status,
-                snapshot_id=entry.snapshot_id,
+                # When a file is replaced or deleted from the dataset, its manifest entry fields store the snapshot ID in which the file was deleted and status 2 (deleted).
+                snapshot_id=self.snapshot_id if status == ManifestEntryStatus.DELETED else entry.snapshot_id,
                 sequence_number=entry.sequence_number,
                 file_sequence_number=entry.file_sequence_number,
                 data_file=entry.data_file,
diff --git a/tests/integration/test_deletes.py b/tests/integration/test_deletes.py
index abf8502a..21c3d129 100644
--- a/tests/integration/test_deletes.py
+++ b/tests/integration/test_deletes.py
@@ -24,7 +24,7 @@ from pyspark.sql import SparkSession
 
 from pyiceberg.catalog.rest import RestCatalog
 from pyiceberg.exceptions import NoSuchTableError
-from pyiceberg.expressions import AlwaysTrue, EqualTo
+from pyiceberg.expressions import AlwaysTrue, EqualTo, LessThanOrEqual
 from pyiceberg.manifest import ManifestEntryStatus
 from pyiceberg.partitioning import PartitionField, PartitionSpec
 from pyiceberg.schema import Schema
@@ -923,3 +923,55 @@ def test_delete_on_empty_table(spark: SparkSession, session_catalog: RestCatalog
 
     # Assert that no new snapshot was created because no rows were deleted
     assert len(tbl.snapshots()) == 0
+
+
+@pytest.mark.integration
+def test_manifest_entry_after_deletes(session_catalog: RestCatalog) -> None:
+    identifier = "default.test_manifest_entry_after_deletes"
+    try:
+        session_catalog.drop_table(identifier)
+    except NoSuchTableError:
+        pass
+
+    schema = pa.schema(
+        [
+            ("id", pa.int32()),
+            ("name", pa.string()),
+        ]
+    )
+
+    table = session_catalog.create_table(identifier, schema)
+    data = pa.Table.from_pylist(
+        [
+            {"id": 1, "name": "foo"},
+            {"id": 2, "name": "bar"},
+            {"id": 3, "name": "bar"},
+            {"id": 4, "name": "bar"},
+        ],
+        schema=schema,
+    )
+    table.append(data)
+
+    def assert_manifest_entry(expected_status: ManifestEntryStatus, expected_snapshot_id: int) -> None:
+        current_snapshot = table.refresh().current_snapshot()
+        assert current_snapshot is not None
+
+        manifest_files = current_snapshot.manifests(table.io)
+        assert len(manifest_files) == 1
+
+        entries = manifest_files[0].fetch_manifest_entry(table.io, discard_deleted=False)
+        assert len(entries) == 1
+        entry = entries[0]
+        assert entry.status == expected_status
+        assert entry.snapshot_id == expected_snapshot_id
+
+    before_delete_snapshot = table.current_snapshot()
+    assert before_delete_snapshot is not None
+
+    assert_manifest_entry(ManifestEntryStatus.ADDED, before_delete_snapshot.snapshot_id)
+
+    table.delete(LessThanOrEqual("id", 4))
+    after_delete_snapshot = table.refresh().current_snapshot()
+    assert after_delete_snapshot is not None
+
+    assert_manifest_entry(ManifestEntryStatus.DELETED, after_delete_snapshot.snapshot_id)