Add support for Equality Deletes on DeleteFileIndex

pyiceberg/io/pyarrow.py

@@ -1693,7 +1693,12 @@ def _task_to_record_batches(
 
 def _read_all_delete_files(io: FileIO, tasks: Iterable[FileScanTask]) -> dict[str, list[ChunkedArray]]:
     deletes_per_file: dict[str, list[ChunkedArray]] = {}
-    unique_deletes = set(itertools.chain.from_iterable([task.delete_files for task in tasks]))
+    unique_deletes = {
+        delete_file
+        for task in tasks
+        for delete_file in task.delete_files
+        if delete_file.content == DataFileContent.POSITION_DELETES
+    }
     if len(unique_deletes) > 0:
         executor = ExecutorFactory.get_or_create()
         deletes_per_files: Iterator[dict[str, ChunkedArray]] = executor.map(

pyiceberg/table/__init__.py

@@ -2059,7 +2059,7 @@ def _plan_files_server_side(self) -> Iterable[FileScanTask]:
     def _plan_files_local(self) -> Iterable[FileScanTask]:
         """Plan files locally by reading manifests."""
         data_entries: list[ManifestEntry] = []
-        delete_index = DeleteFileIndex()
+        delete_index = DeleteFileIndex(self.table_metadata.schema())
 
         residual_evaluators: dict[int, Callable[[DataFile], ResidualEvaluator]] = KeyDefaultDict(self._build_residual_evaluator)
 

pyiceberg/table/delete_file_index.py

@@ -17,12 +17,17 @@
 from __future__ import annotations
 
 from bisect import bisect_left
+from typing import TYPE_CHECKING
 
+from pyiceberg.conversions import from_bytes
 from pyiceberg.expressions import EqualTo
 from pyiceberg.expressions.visitors import _InclusiveMetricsEvaluator
-from pyiceberg.manifest import INITIAL_SEQUENCE_NUMBER, POSITIONAL_DELETE_SCHEMA, DataFile, ManifestEntry
+from pyiceberg.manifest import INITIAL_SEQUENCE_NUMBER, POSITIONAL_DELETE_SCHEMA, DataFile, DataFileContent, ManifestEntry
 from pyiceberg.typedef import Record
 
+if TYPE_CHECKING:
+    from pyiceberg.schema import Schema
+
 PATH_FIELD_ID = 2147483546
 
 
@@ -59,6 +64,15 @@ def referenced_delete_files(self) -> list[DataFile]:
         return [data_file for data_file, _ in self._files]
 
 
+class EqualityDeletes(PositionDeletes):
+    """Collects equality delete files and indexes them by sequence number."""
+
+    def add(self, delete_file: DataFile, seq_num: int) -> None:
+        # Equality deletes are indexed by sequence number - 1 to ensure they only
+        # apply to data files added in strictly earlier snapshots.
+        super().add(delete_file, seq_num - 1)
+
+
 def _has_path_bounds(delete_file: DataFile) -> bool:
     lower = delete_file.lower_bounds
     upper = delete_file.upper_bounds
@@ -76,6 +90,60 @@ def _applies_to_data_file(delete_file: DataFile, data_file: DataFile) -> bool:
     return evaluator.eval(delete_file)
 
 
+def _is_all_null(data_file: DataFile, field_id: int) -> bool:
+    null_counts = data_file.null_value_counts
+    value_counts = data_file.value_counts
+    if not null_counts or not value_counts:
+        return False
+    null_count = null_counts.get(field_id)
+    value_count = value_counts.get(field_id)
+    return null_count is not None and value_count is not None and null_count == value_count
+
+
+def _has_no_nulls(data_file: DataFile, field_id: int) -> bool:
+    null_counts = data_file.null_value_counts
+    if not null_counts:
+        return False
+    return null_counts.get(field_id) == 0
+
+
+def _eq_applies_to_data_file(eq_delete_file: DataFile, data_file: DataFile, schema: Schema) -> bool:
+    if not eq_delete_file.equality_ids:
+        return True
+
+    for field_id in eq_delete_file.equality_ids:
+        # Data is only nulls for this field, but the delete has no null rows.
+        if _is_all_null(data_file, field_id) and _has_no_nulls(eq_delete_file, field_id):
+            return False
+        # Delete only removes null rows, but the data has none.
+        if _is_all_null(eq_delete_file, field_id) and _has_no_nulls(data_file, field_id):
+            return False
+
+        if (
+            eq_delete_file.lower_bounds
+            and eq_delete_file.upper_bounds
+            and data_file.lower_bounds
+            and data_file.upper_bounds
+            and field_id in eq_delete_file.lower_bounds
+            and field_id in eq_delete_file.upper_bounds
+            and field_id in data_file.lower_bounds
+            and field_id in data_file.upper_bounds
+        ):
+            field_type = schema.find_type(field_id)
+            if not field_type.is_primitive:
+                continue
+
+            eq_lower = from_bytes(field_type, eq_delete_file.lower_bounds[field_id])
+            eq_upper = from_bytes(field_type, eq_delete_file.upper_bounds[field_id])
+            data_lower = from_bytes(field_type, data_file.lower_bounds[field_id])
+            data_upper = from_bytes(field_type, data_file.upper_bounds[field_id])
+
+            if eq_upper < data_lower or eq_lower > data_upper:
+                return False
+
+    return True
+
+
 def _referenced_data_file_path(delete_file: DataFile) -> str | None:
     """Return the path, if the path bounds evaluate to the same location."""
     lower_bounds = delete_file.lower_bounds
@@ -103,26 +171,33 @@ def _partition_key(spec_id: int, partition: Record | None) -> tuple[int, Record]
 
 
 class DeleteFileIndex:
-    """Indexes position delete files by partition and by exact data file path."""
+    """Indexes position and equality delete files by partition and by exact data file path."""
 
-    def __init__(self) -> None:
+    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_deletes: dict[tuple[int, Record] | None, EqualityDeletes] = {}
 
     def is_empty(self) -> bool:
-        return not self._by_partition and not self._by_path
+        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
         seq = manifest_entry.sequence_number or INITIAL_SEQUENCE_NUMBER
-        target_path = _referenced_data_file_path(delete_file)
 
-        if target_path:
-            deletes = self._by_path.setdefault(target_path, PositionDeletes())
-            deletes.add(delete_file, seq)
-        else:
-            key = _partition_key(delete_file.spec_id or 0, partition_key)
-            deletes = self._by_partition.setdefault(key, PositionDeletes())
+        if delete_file.content == DataFileContent.POSITION_DELETES:
+            target_path = _referenced_data_file_path(delete_file)
+            if target_path:
+                deletes = self._by_path.setdefault(target_path, PositionDeletes())
+                deletes.add(delete_file, seq)
+            else:
+                key = _partition_key(delete_file.spec_id or 0, partition_key)
+                deletes = self._by_partition.setdefault(key, PositionDeletes())
+                deletes.add(delete_file, seq)
+        elif delete_file.content == DataFileContent.EQUALITY_DELETES:
+            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]:
@@ -131,17 +206,33 @@ def for_data_file(self, seq_num: int, data_file: DataFile, partition_key: Record
 
         deletes: set[DataFile] = set()
         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):
+
+        # Add position deletes
+        partition_pos_deletes = self._by_partition.get(key)
+        if partition_pos_deletes:
+            for delete_file in partition_pos_deletes.filter_by_seq(seq_num):
                 if _applies_to_data_file(delete_file, data_file):
                     deletes.add(delete_file)
 
-        path_deletes = self._by_path.get(data_file.file_path)
-        if path_deletes:
-            deletes.update(path_deletes.filter_by_seq(seq_num))
+        path_pos_deletes = self._by_path.get(data_file.file_path)
+        if path_pos_deletes:
+            deletes.update(path_pos_deletes.filter_by_seq(seq_num))
+
+        # Add equality deletes
+        candidate_eq_deletes: list[DataFile] = []
+        partition_eq_deletes = self._eq_deletes.get(key)
+        if partition_eq_deletes:
+            candidate_eq_deletes.extend(partition_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):
+                continue
+            deletes.add(eq_delete_file)
 
         return deletes
 
@@ -154,4 +245,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_deletes.values():
+            data_files.extend(deletes.referenced_delete_files())
+
         return data_files

pyiceberg/table/update/validate.py

@@ -245,13 +245,13 @@ def _added_delete_files(
         DeleteFileIndex
     """
     if parent_snapshot is None or table.format_version < 2:
-        return DeleteFileIndex()
+        return DeleteFileIndex(table.schema())
 
     manifests, snapshot_ids = _validation_history(
         table, parent_snapshot, starting_snapshot, VALIDATE_ADDED_DELETE_FILES_OPERATIONS, ManifestContent.DELETES
     )
 
-    dfi = DeleteFileIndex()
+    dfi = DeleteFileIndex(table.schema())
 
     for manifest in manifests:
         for entry in manifest.fetch_manifest_entry(table.io, discard_deleted=True):