Make delete file sets read-only and clarify DV validation

Author: KaiqiJinWow

pyiceberg/table/__init__.py

@@ -2267,13 +2267,13 @@ def from_rest_response(
 
         data_file = _rest_file_to_data_file(rest_task.data_file)
 
-        resolved_deletes = DeleteFileSet()
+        resolved_deletes: list[DataFile] = []
         if rest_task.delete_file_references:
             for idx in rest_task.delete_file_references:
                 delete_file = delete_files[idx]
                 if isinstance(delete_file, RESTEqualityDeleteFile):
                     raise NotImplementedError(f"PyIceberg does not yet support equality deletes: {delete_file.file_path}")
-                resolved_deletes.add(_rest_file_to_data_file(delete_file))
+                resolved_deletes.append(_rest_file_to_data_file(delete_file))
 
         return FileScanTask(
             data_file=data_file,

pyiceberg/table/delete_file.py

@@ -16,9 +16,8 @@
 # under the License.
 from __future__ import annotations
 
-from collections.abc import Iterable, Iterator, MutableSet
+from collections.abc import Iterable, Iterator, Set
 from dataclasses import dataclass
-from typing import Any
 
 from pyiceberg.manifest import DataFile
 
@@ -41,15 +40,15 @@ def from_file(cls, delete_file: DataFile) -> DeleteFileKey:
         )
 
 
-class DeleteFileSet(MutableSet[DataFile]):
-    """Set-like delete-file collection keyed by location and content range."""
+class DeleteFileSet(Set[DataFile]):
+    """Read-only delete-file collection keyed by location and content range."""
 
     _files: dict[DeleteFileKey, DataFile]
 
     def __init__(self, delete_files: Iterable[DataFile] = ()) -> None:
         self._files = {}
         for delete_file in delete_files:
-            self.add(delete_file)
+            self._files.setdefault(DeleteFileKey.from_file(delete_file), delete_file)
 
     def __contains__(self, delete_file: object) -> bool:
         """Return whether the delete file is present."""
@@ -63,34 +62,25 @@ def __len__(self) -> int:
         """Return the number of delete files."""
         return len(self._files)
 
-    def add(self, delete_file: DataFile) -> None:
-        self._files.setdefault(DeleteFileKey.from_file(delete_file), delete_file)
-
-    def discard(self, delete_file: DataFile) -> None:
-        self._files.pop(DeleteFileKey.from_file(delete_file), None)
-
-    def update(self, delete_files: Iterable[DataFile]) -> None:
-        for delete_file in delete_files:
-            self.add(delete_file)
-
     def __repr__(self) -> str:
         """Return a string representation of the delete file set."""
         return f"{type(self).__name__}({list(self)!r})"
 
-    def __eq__(self, other: Any) -> bool:
+    def __eq__(self, other: object) -> bool:
         """Compare delete file sets by delete file identity."""
         if isinstance(other, DeleteFileSet):
             return self._files.keys() == other._files.keys()
 
-        if not isinstance(other, Iterable):
+        if not isinstance(other, Set):
+            return NotImplemented
+
+        if len(self) != len(other):
             return False
 
         other_keys: set[DeleteFileKey] = set()
-        other_count = 0
         for delete_file in other:
             if not isinstance(delete_file, DataFile):
                 return False
             other_keys.add(DeleteFileKey.from_file(delete_file))
-            other_count += 1
 
-        return len(other_keys) == other_count and set(self._files) == other_keys
+        return self._files.keys() == other_keys

pyiceberg/table/delete_file_index.py

@@ -130,21 +130,21 @@ def for_data_file(self, seq_num: int, data_file: DataFile, partition_key: Record
         if self.is_empty():
             return DeleteFileSet()
 
-        deletes = DeleteFileSet()
+        deletes: list[DataFile] = []
         spec_id = data_file.spec_id or 0
 
         key = _partition_key(spec_id, partition_key)
         partition_deletes = self._by_partition.get(key)
         if partition_deletes:
             for delete_file in partition_deletes.filter_by_seq(seq_num):
                 if _applies_to_data_file(delete_file, data_file):
-                    deletes.add(delete_file)
+                    deletes.append(delete_file)
 
         path_deletes = self._by_path.get(data_file.file_path)
         if path_deletes:
-            deletes.update(path_deletes.filter_by_seq(seq_num))
+            deletes.extend(path_deletes.filter_by_seq(seq_num))
 
-        return deletes
+        return DeleteFileSet(deletes)
 
     def referenced_delete_files(self) -> list[DataFile]:
         data_files: list[DataFile] = []

pyiceberg/table/deletion_vector.py

@@ -143,7 +143,11 @@ def _validate_deletion_vector_content(dv: "DataFile") -> None:
 
 
 def has_deletion_vector_content_reference(dv: "DataFile") -> bool:
-    """Return whether a deletion vector is described by manifest content-range metadata."""
+    """Return whether any manifest content-reference field is present.
+
+    Partial references must be validated as range reads. Whole-Puffin reads
+    are only used when all content-reference fields are absent.
+    """
     return dv.content_offset is not None or dv.content_size_in_bytes is not None or dv.referenced_data_file is not None
 
 

tests/io/test_pyarrow.py

@@ -1909,6 +1909,38 @@ def test_read_deletes_from_whole_puffin_file(tmp_path: Path) -> None:
     assert deletes == {referenced_data_file: pa.chunked_array([[1, 3, 5]])}
 
 
+@pytest.mark.parametrize(
+    ("missing_field", "error"),
+    [
+        ("content_offset", "content offset is missing"),
+        ("content_size_in_bytes", "content size is missing"),
+        ("referenced_data_file", "referenced data file is missing"),
+    ],
+)
+def test_read_deletes_rejects_partial_content_reference(tmp_path: Path, missing_field: str, error: str) -> None:
+    referenced_data_file = f"{tmp_path}/data.parquet"
+    delete_file_path = f"{tmp_path}/deletes.puffin"
+    puffin_payload = _deletion_vector_puffin_payload(referenced_data_file)
+    blob = PuffinFile(puffin_payload).footer.blobs[0]
+
+    with open(delete_file_path, "wb") as f:
+        f.write(puffin_payload)
+
+    dv = DataFile.from_args(
+        _table_format_version=3,
+        content=DataFileContent.POSITION_DELETES,
+        file_path=delete_file_path,
+        file_format=FileFormat.PUFFIN,
+        record_count=3,
+        content_offset=blob.offset if missing_field != "content_offset" else None,
+        content_size_in_bytes=blob.length if missing_field != "content_size_in_bytes" else None,
+        referenced_data_file=referenced_data_file if missing_field != "referenced_data_file" else None,
+    )
+
+    with pytest.raises(ValueError, match=error):
+        _read_deletes(PyArrowFileIO(), dv)
+
+
 def test_read_deletion_vector_blob_from_content_range(tmp_path: Path) -> None:
     referenced_data_file = f"{tmp_path}/data.parquet"
     dv_blob = _deletion_vector_blob(_deletion_vector_bitmap_payload())

tests/table/test_delete_file_index.py

@@ -14,6 +14,8 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
+from collections.abc import Callable, Iterable, Set
+
 import pytest
 
 from pyiceberg.manifest import DataFile, DataFileContent, FileFormat, ManifestEntry, ManifestEntryStatus
@@ -190,6 +192,33 @@ def test_delete_file_set_uses_content_range_identity() -> None:
     assert len(DeleteFileSet([first_dv, second_dv])) == 2
 
 
+@pytest.mark.parametrize("collection_type", [DeleteFileSet, set, frozenset])
+@pytest.mark.parametrize(("content_offset", "content_size_in_bytes", "equal"), [(4, 10, True), (40, 10, False), (4, 12, False)])
+def test_delete_file_set_equality(
+    collection_type: Callable[[Iterable[DataFile]], Set[DataFile]], content_offset: int, content_size_in_bytes: int, equal: bool
+) -> None:
+    dv = _create_deletion_vector(content_offset=4, content_size_in_bytes=10).data_file
+    other_dv = _create_deletion_vector(content_offset=content_offset, content_size_in_bytes=content_size_in_bytes).data_file
+    deletes = DeleteFileSet([dv])
+    other = collection_type([other_dv])
+
+    assert (deletes == other) is equal
+    assert (other == deletes) is equal
+    assert (deletes != other) is not equal
+
+
+def test_delete_file_set_does_not_compare_with_non_sets() -> None:
+    dv = _create_deletion_vector(content_offset=4, content_size_in_bytes=10).data_file
+    deletes = DeleteFileSet([dv])
+    iterator = iter([dv])
+
+    assert deletes != [dv]
+    assert deletes != (dv,)
+    assert deletes != iterator
+    assert next(iterator) is dv
+    assert DeleteFileSet() != []
+
+
 def test_dvs_with_same_file_path_and_different_content_ranges_are_not_deduped() -> None:
     index = DeleteFileIndex()
     shared_file_path = "s3://bucket/deletion-vectors.bin"