comments, thanks geruh

Author: gabeiglio

pyiceberg/table/delete_file_index.py

@@ -145,7 +145,7 @@ def for_data_file(self, seq_num: int, data_file: DataFile, partition_key: Record
 
         return deletes
 
-    def referenced_data_files(self) -> list[DataFile]:
+    def referenced_delete_files(self) -> list[DataFile]:
         data_files: list[DataFile] = []
 
         for deletes in self._by_partition.values():

pyiceberg/table/update/validate.py

@@ -254,7 +254,7 @@ def _added_delete_files(
     dfi = DeleteFileIndex()
 
     for manifest in manifests:
-        for entry in manifest.fetch_manifest_entry(table.io, discard_deleted=False):
+        for entry in manifest.fetch_manifest_entry(table.io, discard_deleted=True):
             if _filter_manifest_entries(
                 entry, snapshot_ids, data_filter, partition_set, ManifestEntryStatus.ADDED, table.schema()
             ):
@@ -274,9 +274,8 @@ def _starting_sequence_number(table: Table, starting_snapshot: Snapshot | None)
         Sequence number as int
     """
     if starting_snapshot is not None:
-        if snapshot := table.snapshot_by_id(starting_snapshot.snapshot_id):
-            if seq := snapshot.sequence_number:
-                return seq
+        if seq := starting_snapshot.sequence_number:
+            return seq
     return INITIAL_SEQUENCE_NUMBER
 
 
@@ -322,13 +321,13 @@ def _validate_no_new_delete_files(
     if deletes.is_empty():
         return
 
-    conflicting_delete_files = deletes.referenced_data_files()
+    conflicting_delete_paths = [file.file_path for file in deletes.referenced_delete_files()]
     raise ValidationException(
-        f"Found new conflicting delete files that can apply to records matching {data_filter}: {conflicting_delete_files}"
+        f"Found new conflicting delete files that can apply to records matching {data_filter}: {conflicting_delete_paths}"
     )
 
 
-def _validate_no_new_delete_files_for_data_files(
+def _validate_no_new_deletes_for_data_files(
     table: Table,
     starting_snapshot: Snapshot,
     data_filter: BooleanExpression | None,
@@ -353,6 +352,6 @@ def _validate_no_new_delete_files_for_data_files(
 
     # Fail to any delete file found that applies to files written in or before the starting snapshot
     for data_file in data_files:
-        delete_files = deletes.for_data_file(seq_num, data_file)
+        delete_files = deletes.for_data_file(seq_num, data_file, data_file.partition)
         if len(delete_files) > 0:
             raise ValidationException(f"Cannot commit, found new delete for replace data file {data_file}")

tests/table/test_validate.py

@@ -32,7 +32,7 @@
     _validate_added_data_files,
     _validate_deleted_data_files,
     _validate_no_new_delete_files,
-    _validate_no_new_delete_files_for_data_files,
+    _validate_no_new_deletes_for_data_files,
     _validation_history,
 )
 
@@ -356,7 +356,7 @@ class DummyEntry:
 
 
 @pytest.mark.parametrize("operation", [Operation.APPEND, Operation.REPLACE])
-def test_validate_added_delete_files_non_conflicting_count(
+def test_added_delete_files_non_conflicting_count(
     table_v2_with_extensive_snapshots_and_manifests: tuple[Table, dict[int, list[ManifestFile]]],
     operation: Operation,
 ) -> None:
@@ -403,11 +403,11 @@ def mock_fetch_manifest_entry(self: ManifestFile, io: FileIO, discard_deleted: b
         )
 
         assert dfi.is_empty()
-        assert len(dfi.referenced_data_files()) == 0
+        assert len(dfi.referenced_delete_files()) == 0
 
 
 @pytest.mark.parametrize("operation", [Operation.DELETE, Operation.OVERWRITE])
-def test_validate_added_delete_files_conflicting_count(
+def test_added_delete_files_conflicting_count(
     table_v2_with_extensive_snapshots_and_manifests: tuple[Table, dict[int, list[ManifestFile]]],
     operation: Operation,
 ) -> None:
@@ -442,21 +442,15 @@ def mock_read_manifest_side_effect(self: Snapshot, io: FileIO) -> list[ManifestF
         return []
 
     def mock_fetch_manifest_entry(self: ManifestFile, io: FileIO, discard_deleted: bool = True) -> list[ManifestEntry]:
-        result = [
+        return [
             ManifestEntry.from_args(
-                status=ManifestEntryStatus.ADDED, snapshot_id=self.added_snapshot_id, sequence_number=self.min_sequence_number
+                status=ManifestEntryStatus.ADDED,
+                snapshot_id=self.added_snapshot_id,
+                sequence_number=self.min_sequence_number,
+                data_file=mock_delete_file,
             )
         ]
 
-        result[-1] = ManifestEntry.from_args(
-            status=ManifestEntryStatus.ADDED,
-            snapshot_id=self.added_snapshot_id,
-            sequence_number=10000,
-            data_file=mock_delete_file,
-        )
-
-        return result
-
     with (
         patch("pyiceberg.table.snapshots.Snapshot.manifests", new=mock_read_manifest_side_effect),
         patch("pyiceberg.manifest.ManifestFile.fetch_manifest_entry", new=mock_fetch_manifest_entry),
@@ -470,7 +464,7 @@ def mock_fetch_manifest_entry(self: ManifestFile, io: FileIO, discard_deleted: b
         )
 
         assert not dfi.is_empty()
-        assert dfi.referenced_data_files()[0] == mock_delete_file
+        assert dfi.referenced_delete_files()[0] == mock_delete_file
 
 
 def test_validate_no_new_delete_files_raises_on_conflict(
@@ -502,7 +496,7 @@ def test_validate_no_new_delete_files_for_data_files_raises_on_conflict(
 
     with patch("pyiceberg.table.update.validate.DeleteFileIndex.for_data_file", return_value=[mocked_data_file]):
         with pytest.raises(ValidationException):
-            _validate_no_new_delete_files_for_data_files(
+            _validate_no_new_deletes_for_data_files(
                 table=table,
                 starting_snapshot=newest_snapshot,
                 data_filter=None,