PR comments

Author: rambleraptor

pyiceberg/table/delete_file_index.py

@@ -153,16 +153,10 @@ def __init__(self, schema: Schema | None = None) -> None:
         self._schema = schema
         self._by_partition: dict[tuple[int, Record], PositionDeletes] = {}
         self._by_path: dict[str, PositionDeletes] = {}
-        self._eq_by_partition: dict[tuple[int, Record], EqualityDeletes] = {}
-        self._global_eq_deletes: EqualityDeletes = EqualityDeletes()
+        self._eq_deletes: dict[tuple[int, Record] | None, EqualityDeletes] = {}
 
     def is_empty(self) -> bool:
-        return (
-            not self._by_partition
-            and not self._by_path
-            and not self._eq_by_partition
-            and not self._global_eq_deletes.referenced_delete_files()
-        )
+        return not self._by_partition and not self._by_path and not self._eq_deletes
 
     def add_delete_file(self, manifest_entry: ManifestEntry, partition_key: Record | None = None) -> None:
         delete_file = manifest_entry.data_file
@@ -178,12 +172,9 @@ def add_delete_file(self, manifest_entry: ManifestEntry, partition_key: Record |
                 deletes = self._by_partition.setdefault(key, PositionDeletes())
                 deletes.add(delete_file, seq)
         elif delete_file.content == DataFileContent.EQUALITY_DELETES:
-            if partition_key is None or len(partition_key) == 0:
-                self._global_eq_deletes.add(delete_file, seq)
-            else:
-                key = _partition_key(delete_file.spec_id or 0, partition_key)
-                deletes = self._eq_by_partition.setdefault(key, EqualityDeletes())
-                deletes.add(delete_file, seq)
+            eq_key = _partition_key(delete_file.spec_id or 0, partition_key) if partition_key else None
+            deletes = self._eq_deletes.setdefault(eq_key, EqualityDeletes())
+            deletes.add(delete_file, seq)
 
     def for_data_file(self, seq_num: int, data_file: DataFile, partition_key: Record | None = None) -> set[DataFile]:
         if self.is_empty():
@@ -206,11 +197,13 @@ def for_data_file(self, seq_num: int, data_file: DataFile, partition_key: Record
 
         # Add equality deletes
         candidate_eq_deletes: list[DataFile] = []
-        partition_eq_deletes = self._eq_by_partition.get(key)
+        partition_eq_deletes = self._eq_deletes.get(key)
         if partition_eq_deletes:
             candidate_eq_deletes.extend(partition_eq_deletes.filter_by_seq(seq_num))
 
-        candidate_eq_deletes.extend(self._global_eq_deletes.filter_by_seq(seq_num))
+        global_eq_deletes = self._eq_deletes.get(None)
+        if global_eq_deletes:
+            candidate_eq_deletes.extend(global_eq_deletes.filter_by_seq(seq_num))
 
         for eq_delete_file in candidate_eq_deletes:
             if self._schema and not _eq_applies_to_data_file(eq_delete_file, data_file, self._schema):
@@ -228,9 +221,7 @@ def referenced_delete_files(self) -> list[DataFile]:
         for deletes in self._by_path.values():
             data_files.extend(deletes.referenced_delete_files())
 
-        for deletes in self._eq_by_partition.values():
+        for deletes in self._eq_deletes.values():
             data_files.extend(deletes.referenced_delete_files())
 
-        data_files.extend(self._global_eq_deletes.referenced_delete_files())
-
         return data_files

tests/table/test_delete_file_index.py

@@ -239,6 +239,38 @@ def test_equality_delete_sequence_number_filtering() -> None:
     assert len(index.for_data_file(3, data_file)) == 0
 
 
+def test_equality_delete_sequence_number_unpartitioned() -> None:
+    data_file = _create_data_file()
+
+    # Create both types of deletes at sequence number 10
+    pos_delete = _create_positional_delete(sequence_number=10)
+    eq_delete = _create_equality_delete(sequence_number=10)
+
+    index = DeleteFileIndex()
+    index.add_delete_file(pos_delete)
+    index.add_delete_file(eq_delete)
+
+    # Sequence 10
+    deletes = index.for_data_file(10, data_file)
+    assert len(deletes) == 1
+    # Position deletes will apply (applies to seq <= 10)
+    assert pos_delete.data_file in deletes
+    # Equality deletes will not (applies to seq < 10)
+    assert eq_delete.data_file not in deletes
+
+    # Sequence 9
+    # Both deletes will apply.
+    deletes = index.for_data_file(9, data_file)
+    assert len(deletes) == 2
+    assert pos_delete.data_file in deletes
+    assert eq_delete.data_file in deletes
+
+    # At sequence 111
+    # Neither should apply.
+    deletes = index.for_data_file(11, data_file)
+    assert len(deletes) == 0
+
+
 def test_global_equality_deletes() -> None:
     index = DeleteFileIndex()