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 <lliangyu@amazon.com>
Co-authored-by: Fokko Driesprong <fokko@apache.org>

Author: 3 people

pyiceberg/table/update/snapshot.py

@@ -407,7 +407,8 @@ def _compute_deletes(self) -> Tuple[List[ManifestFile], List[ManifestEntry], boo
         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,

tests/integration/test_deletes.py

@@ -24,7 +24,7 @@
 
 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)