Fix: non-identity manifest pruning

pyiceberg/partitioning.py

@@ -33,6 +33,7 @@
 )
 
 from pyiceberg.exceptions import ValidationError
+from pyiceberg.expressions import AlwaysFalse, And, BooleanExpression, EqualTo, IsNull, Or, Reference
 from pyiceberg.schema import Schema
 from pyiceberg.transforms import (
     BucketTransform,
@@ -550,3 +551,32 @@ def _(type: IcebergType, value: uuid.UUID | int | bytes | None) -> bytes | int |
 @_to_partition_representation.register(PrimitiveType)
 def _(type: IcebergType, value: Any | None) -> Any | None:
     return value
+
+
+def build_field_value_predicate(field_names: list[str], field_values: Record) -> BooleanExpression:
+    """Build a predicate matching a single record via per-field EqualTo/IsNull, ANDed together.
+
+    Args:
+        field_names: The name to reference for each position in field_values.
+        field_values: The values to match, one per field name, by position.
+
+    Raises:
+        IndexError: If field_names is empty.
+    """
+    predicates: list[BooleanExpression] = [
+        EqualTo(Reference(name), field_values[pos]) if field_values[pos] is not None else IsNull(Reference(name))
+        for pos, name in enumerate(field_names)
+    ]
+    return And(*predicates) if len(predicates) > 1 else predicates[0]
+
+
+def build_records_predicate(field_names: list[str], records: set[Record]) -> BooleanExpression:
+    """Build a predicate matching any of the given records, ORing together per-record predicates.
+
+    Returns AlwaysFalse() if there are no fields or no records to match.
+    """
+    if not records or not field_names:
+        return AlwaysFalse()
+
+    per_record_exprs = [build_field_value_predicate(field_names, record) for record in records]
+    return Or(*per_record_exprs) if len(per_record_exprs) > 1 else per_record_exprs[0]

pyiceberg/table/__init__.py

@@ -35,7 +35,7 @@
 
 import pyiceberg.expressions.parser as parser
 from pyiceberg.exceptions import CommitFailedException, ValidationException
-from pyiceberg.expressions import AlwaysFalse, AlwaysTrue, And, BooleanExpression, EqualTo, IsNull, Or, Reference
+from pyiceberg.expressions import AlwaysFalse, AlwaysTrue, And, BooleanExpression, Or
 from pyiceberg.expressions.visitors import (
     ResidualEvaluator,
     _InclusiveMetricsEvaluator,
@@ -46,7 +46,13 @@
 )
 from pyiceberg.io import FileIO, load_file_io
 from pyiceberg.manifest import DataFile, DataFileContent, ManifestContent, ManifestEntry, ManifestEntryStatus, ManifestFile
-from pyiceberg.partitioning import PARTITION_FIELD_ID_START, UNPARTITIONED_PARTITION_SPEC, PartitionKey, PartitionSpec
+from pyiceberg.partitioning import (
+    PARTITION_FIELD_ID_START,
+    UNPARTITIONED_PARTITION_SPEC,
+    PartitionKey,
+    PartitionSpec,
+    build_records_predicate,
+)
 from pyiceberg.schema import Schema
 from pyiceberg.table.delete_file_index import DeleteFileIndex
 from pyiceberg.table.inspect import InspectTable
@@ -403,20 +409,7 @@ def _build_partition_predicate(
             A predicate matching any of the input partition records.
         """
         partition_fields = [schema.find_field(field.source_id).name for field in spec.fields]
-        if not partition_records or not partition_fields:
-            return AlwaysFalse()
-
-        per_record_exprs: list[BooleanExpression] = []
-        for partition_record in partition_records:
-            predicates: list[BooleanExpression] = [
-                EqualTo(Reference(partition_field), partition_record[pos])
-                if partition_record[pos] is not None
-                else IsNull(Reference(partition_field))
-                for pos, partition_field in enumerate(partition_fields)
-            ]
-            per_record_exprs.append(And(*predicates) if len(predicates) > 1 else predicates[0])
-
-        return Or(*per_record_exprs) if len(per_record_exprs) > 1 else per_record_exprs[0]
+        return build_records_predicate(partition_fields, partition_records)
 
     def _append_snapshot_producer(
         self, snapshot_properties: dict[str, str], branch: str | None = MAIN_BRANCH

pyiceberg/table/update/snapshot.py

@@ -29,7 +29,7 @@
 
 from pyiceberg.avro.codecs import AvroCompressionCodec
 from pyiceberg.exceptions import ValidationException
-from pyiceberg.expressions import AlwaysFalse, BooleanExpression, Or
+from pyiceberg.expressions import AlwaysFalse, AlwaysTrue, BooleanExpression, Or
 from pyiceberg.expressions.visitors import (
     ROWS_MIGHT_NOT_MATCH,
     ROWS_MUST_MATCH,
@@ -50,7 +50,7 @@
     write_manifest,
     write_manifest_list,
 )
-from pyiceberg.partitioning import PartitionSpec
+from pyiceberg.partitioning import PartitionSpec, build_records_predicate
 from pyiceberg.schema import Schema
 from pyiceberg.table.refs import MAIN_BRANCH, SnapshotRefType
 from pyiceberg.table.snapshots import (
@@ -140,6 +140,7 @@ class _SnapshotProducer(UpdateTableMetadata[U], Generic[U]):
     _compression: AvroCompressionCodec
     _target_branch: str | None
     _predicate: BooleanExpression
+    _delete_files_partition_filters: dict[int, BooleanExpression]
     _case_sensitive: bool
     _commit_window: CommitWindow | None
     _written_manifests: list[str]
@@ -177,6 +178,7 @@ def __init__(
         self._parent_snapshot_id = self._current_branch_head_id()
         self._starting_snapshot_id = self._parent_snapshot_id
         self._predicate = AlwaysFalse()
+        self._delete_files_partition_filters = {}
         self._case_sensitive = True
         self._commit_window = None
         self._isolation_operation: Operation = Operation.DELETE
@@ -263,7 +265,7 @@ def _write_delete_manifest() -> list[ManifestFile]:
             else:
                 return []
 
-        # Updates self._predicate with computed partition predicate for manifest pruning
+        # Populates self._delete_files_partition_filters for manifest pruning; does not touch self._predicate
         self._build_delete_files_partition_predicate()
 
         executor = ExecutorFactory.get_or_create()
@@ -514,26 +516,31 @@ def partition_filters(self) -> KeyDefaultDict[int, BooleanExpression]:
         return KeyDefaultDict(self._build_partition_projection)
 
     def _build_manifest_evaluator(self, spec_id: int) -> Callable[[ManifestFile], bool]:
-        return manifest_evaluator(self.spec(spec_id), self.schema(), self.partition_filters[spec_id], self._case_sensitive)
+        partition_filter = self.partition_filters[spec_id]
+        if delete_files_partition_filter := self._delete_files_partition_filters.get(spec_id):
+            partition_filter = Or(partition_filter, delete_files_partition_filter)
+        return manifest_evaluator(self.spec(spec_id), self.schema(), partition_filter, self._case_sensitive)
 
     def delete_by_predicate(self, predicate: BooleanExpression, case_sensitive: bool = True) -> None:
         self._predicate = Or(self._predicate, predicate)
         self._case_sensitive = case_sensitive
 
     def _build_delete_files_partition_predicate(self) -> None:
-        """Build BooleanExpression based on deleted data files partitions."""
+        """Build a partition-domain predicate per spec for deleted data files, used to prune manifests."""
+        self._delete_files_partition_filters = {}
         partition_to_overwrite: dict[int, set[Record]] = {}
         for data_file in self._deleted_data_files:
             group = partition_to_overwrite.setdefault(data_file.spec_id, set())
             group.add(data_file.partition)
 
         for spec_id, partition_records in partition_to_overwrite.items():
-            self.delete_by_predicate(
-                self._transaction._build_partition_predicate(
-                    partition_records=partition_records, schema=self.schema(), spec=self.spec(spec_id)
-                ),
-                self._case_sensitive,
-            )
+            # Bound against the partition struct (field.name), not the row schema, so this works for any transform.
+            partition_field_names = [field.name for field in self.spec(spec_id).fields]
+            if not partition_field_names:
+                # Unpartitioned spec: nothing to filter on, so the (single, empty) partition always matches.
+                self._delete_files_partition_filters[spec_id] = AlwaysTrue()
+            else:
+                self._delete_files_partition_filters[spec_id] = build_records_predicate(partition_field_names, partition_records)
 
 
 class _DeleteFiles(_SnapshotProducer["_DeleteFiles"]):

tests/table/test_commit_retry.py

@@ -457,7 +457,8 @@ def test_concurrent_deletes_on_different_partitions_succeed(catalog: Catalog) ->
 def test_concurrent_partial_deletes_on_different_partitions_succeed(catalog: Catalog) -> None:
     """Concurrent partial deletes (CoW rewrite) on different partitions should succeed.
 
-    This tests the auto-computed partition predicate from _build_delete_files_partition_predicate.
+    Conflict detection for this path uses the user's delete filter directly (matching Java),
+    not an auto-computed partition predicate.
     """
     from pyiceberg.partitioning import PartitionField, PartitionSpec
     from pyiceberg.transforms import IdentityTransform