address comments

Author: geruh

pyiceberg/table/__init__.py

@@ -1856,21 +1856,6 @@ def _min_sequence_number(manifests: list[ManifestFile]) -> int:
         return INITIAL_SEQUENCE_NUMBER
 
 
-def _match_deletes_to_data_file(data_entry: ManifestEntry, delete_file_index: DeleteFileIndex) -> set[DataFile]:
-    """Check if delete files are relevant for the data file.
-
-    Args:
-        data_entry (ManifestEntry): The manifest entry of the data file.
-        delete_file_index (DeleteFileIndex): Index containing all delete files.
-
-    Returns:
-        A set of delete files that are relevant for the data file.
-    """
-    return delete_file_index.for_data_file(
-        data_entry.sequence_number or 0, data_entry.data_file, partition_key=data_entry.data_file.partition
-    )
-
-
 class DataScan(TableScan):
     def _build_partition_projection(self, spec_id: int) -> BooleanExpression:
         project = inclusive_projection(self.table_metadata.schema(), self.table_metadata.specs()[spec_id], self.case_sensitive)
@@ -2008,13 +1993,13 @@ def plan_files(self) -> Iterable[FileScanTask]:
                 raise ValueError("PyIceberg does not yet support equality deletes: https://github.com/apache/iceberg/issues/6568")
             else:
                 raise ValueError(f"Unknown DataFileContent ({data_file.content}): {manifest_entry}")
-
         return [
             FileScanTask(
                 data_entry.data_file,
-                delete_files=_match_deletes_to_data_file(
-                    data_entry,
-                    delete_index,
+                delete_files=delete_index.for_data_file(
+                    data_entry.sequence_number or INITIAL_SEQUENCE_NUMBER,
+                    data_entry.data_file,
+                    partition_key=data_entry.data_file.partition,
                 ),
                 residual=residual_evaluators[data_entry.data_file.spec_id](data_entry.data_file).residual_for(
                     data_entry.data_file.partition

tests/table/test_init.py

@@ -32,17 +32,14 @@
     In,
 )
 from pyiceberg.io import PY_IO_IMPL, load_file_io
-from pyiceberg.manifest import DataFile, DataFileContent, FileFormat, ManifestEntry, ManifestEntryStatus
 from pyiceberg.partitioning import PartitionField, PartitionSpec
 from pyiceberg.schema import Schema
 from pyiceberg.table import (
     CommitTableRequest,
     StaticTable,
     Table,
     TableIdentifier,
-    _match_deletes_to_data_file,
 )
-from pyiceberg.table.delete_file_index import DeleteFileIndex
 from pyiceberg.table.metadata import TableMetadataUtil, TableMetadataV2, _generate_snapshot_id
 from pyiceberg.table.refs import MAIN_BRANCH, SnapshotRef, SnapshotRefType
 from pyiceberg.table.snapshots import (
@@ -371,119 +368,6 @@ def test_static_table_io_does_not_exist(metadata_location: str) -> None:
         StaticTable.from_metadata(metadata_location, {PY_IO_IMPL: "pyiceberg.does.not.exist.FileIO"})
 
 
-def test_match_deletes_to_datafile() -> None:
-    data_file = DataFile.from_args(
-        content=DataFileContent.DATA,
-        file_path="s3://bucket/0000.parquet",
-        file_format=FileFormat.PARQUET,
-        partition={},
-        record_count=3,
-        file_size_in_bytes=3,
-    )
-    data_file._spec_id = 0
-    data_entry = ManifestEntry.from_args(
-        status=ManifestEntryStatus.ADDED,
-        sequence_number=1,
-        data_file=data_file,
-    )
-    delete_file_1 = DataFile.from_args(
-        content=DataFileContent.POSITION_DELETES,
-        file_path="s3://bucket/0001-delete.parquet",
-        file_format=FileFormat.PARQUET,
-        partition={},
-        record_count=3,
-        file_size_in_bytes=3,
-    )
-    delete_file_1._spec_id = 0
-    delete_entry_1 = ManifestEntry.from_args(
-        status=ManifestEntryStatus.ADDED,
-        sequence_number=0,  # Older than the data
-        data_file=delete_file_1,
-    )
-    delete_file_2 = DataFile.from_args(
-        content=DataFileContent.POSITION_DELETES,
-        file_path="s3://bucket/0002-delete.parquet",
-        file_format=FileFormat.PARQUET,
-        partition={},
-        record_count=3,
-        file_size_in_bytes=3,
-    )
-    delete_file_2._spec_id = 0
-    delete_entry_2 = ManifestEntry.from_args(
-        status=ManifestEntryStatus.ADDED,
-        sequence_number=3,
-        data_file=delete_file_2,
-    )
-
-    delete_file_index = DeleteFileIndex()
-    delete_file_index.add_delete_file(delete_entry_1)
-    delete_file_index.add_delete_file(delete_entry_2)
-
-    assert _match_deletes_to_data_file(
-        data_entry,
-        delete_file_index,
-    ) == {
-        delete_entry_2.data_file,
-    }
-
-
-def test_match_deletes_to_datafile_duplicate_number() -> None:
-    data_file = DataFile.from_args(
-        content=DataFileContent.DATA,
-        file_path="s3://bucket/0000.parquet",
-        file_format=FileFormat.PARQUET,
-        partition={},
-        record_count=3,
-        file_size_in_bytes=3,
-    )
-    data_file._spec_id = 0
-    data_entry = ManifestEntry.from_args(
-        status=ManifestEntryStatus.ADDED,
-        sequence_number=1,
-        data_file=data_file,
-    )
-    delete_file_1 = DataFile.from_args(
-        content=DataFileContent.POSITION_DELETES,
-        file_path="s3://bucket/0001-delete.parquet",
-        file_format=FileFormat.PARQUET,
-        partition={},
-        record_count=3,
-        file_size_in_bytes=3,
-    )
-    delete_file_1._spec_id = 0
-    delete_entry_1 = ManifestEntry.from_args(
-        status=ManifestEntryStatus.ADDED,
-        sequence_number=3,
-        data_file=delete_file_1,
-    )
-    delete_file_2 = DataFile.from_args(
-        content=DataFileContent.POSITION_DELETES,
-        file_path="s3://bucket/0002-delete.parquet",
-        file_format=FileFormat.PARQUET,
-        partition={},
-        record_count=3,
-        file_size_in_bytes=3,
-    )
-    delete_file_2._spec_id = 0
-    delete_entry_2 = ManifestEntry.from_args(
-        status=ManifestEntryStatus.ADDED,
-        sequence_number=3,
-        data_file=delete_file_2,
-    )
-
-    delete_file_index = DeleteFileIndex()
-    delete_file_index.add_delete_file(delete_entry_1)
-    delete_file_index.add_delete_file(delete_entry_2)
-
-    assert _match_deletes_to_data_file(
-        data_entry,
-        delete_file_index,
-    ) == {
-        delete_entry_1.data_file,
-        delete_entry_2.data_file,
-    }
-
-
 def test_serialize_set_properties_updates() -> None:
     assert (
         SetPropertiesUpdate(updates={"abc": "🤪"}).model_dump_json() == """{"action":"set-properties","updates":{"abc":"🤪"}}"""