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

tests/table/test_snapshot_manifest_pruning.py

@@ -0,0 +1,187 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import pyarrow as pa
+
+from pyiceberg.catalog import Catalog
+from pyiceberg.expressions import EqualTo, Reference
+from pyiceberg.partitioning import PartitionField, PartitionSpec
+from pyiceberg.schema import Schema
+from pyiceberg.transforms import BucketTransform
+from pyiceberg.types import IntegerType, NestedField, StringType
+
+
+def test_delete_data_file_manifest_pruning_bucket_transform_succeeds(catalog: Catalog) -> None:
+    """delete_data_file should work for non-identity specs.
+
+    Manifest-pruning predicates are built against the partition struct (using the
+    partition field name, e.g. the bucket id) rather than the source column, so this
+    works regardless of the partition transform.
+    """
+    catalog.create_namespace_if_not_exists("default")
+    identifier = "default.bucket_delete"
+
+    schema = Schema(
+        NestedField(1, "tenant_id", StringType(), required=True),
+        NestedField(2, "value", IntegerType(), required=True),
+    )
+    spec = PartitionSpec(
+        PartitionField(
+            source_id=1,
+            field_id=1000,
+            transform=BucketTransform(8),
+            name="tenant_id_bucket",
+        ),
+        spec_id=0,
+    )
+    table = catalog.create_table(
+        identifier=identifier,
+        schema=schema,
+        partition_spec=spec,
+        properties={"format-version": "2"},
+    )
+
+    table.append(
+        pa.Table.from_pylist(
+            [
+                {"tenant_id": "tenant-a", "value": 1},
+                {"tenant_id": "tenant-b", "value": 2},
+            ],
+            schema=pa.schema(
+                [
+                    pa.field("tenant_id", pa.string(), nullable=False),
+                    pa.field("value", pa.int32(), nullable=False),
+                ]
+            ),
+        )
+    )
+
+    before = table.scan().to_arrow()
+    before_paths = {task.file.file_path for task in table.scan().plan_files()}
+    existing_file = next(iter(table.scan().plan_files())).file
+
+    with table.transaction() as txn:
+        with txn.update_snapshot().overwrite() as overwrite:
+            overwrite.delete_data_file(existing_file)
+
+    after = table.scan().to_arrow()
+    after_paths = {task.file.file_path for task in table.scan().plan_files()}
+
+    assert existing_file.file_path not in after_paths
+    assert before_paths - after_paths == {existing_file.file_path}
+    assert len(after_paths) == len(before_paths) - 1
+    assert after.num_rows < before.num_rows
+
+
+def test_delete_data_file_manifest_pruning_predicate_uses_partition_field(catalog: Catalog) -> None:
+    """The manifest-pruning predicate must reference the partition field, not the source column.
+
+    `_OverwriteFiles` deletes by exact `DataFile` identity regardless of this predicate, so an
+    end-to-end delete would still succeed even if pruning silently degraded back to a
+    non-discriminating fallback. This test guards the pruning predicate itself.
+    """
+    catalog.create_namespace_if_not_exists("default")
+    identifier = "default.bucket_delete_pruning_predicate"
+
+    schema = Schema(
+        NestedField(1, "tenant_id", StringType(), required=True),
+        NestedField(2, "value", IntegerType(), required=True),
+    )
+    spec = PartitionSpec(
+        PartitionField(
+            source_id=1,
+            field_id=1000,
+            transform=BucketTransform(8),
+            name="tenant_id_bucket",
+        ),
+        spec_id=0,
+    )
+    table = catalog.create_table(
+        identifier=identifier,
+        schema=schema,
+        partition_spec=spec,
+        properties={"format-version": "2"},
+    )
+    table.append(
+        pa.Table.from_pylist(
+            [{"tenant_id": "tenant-a", "value": 1}],
+            schema=pa.schema(
+                [
+                    pa.field("tenant_id", pa.string(), nullable=False),
+                    pa.field("value", pa.int32(), nullable=False),
+                ]
+            ),
+        )
+    )
+    existing_file = next(iter(table.scan().plan_files())).file
+    expected_bucket_id = BucketTransform(8).transform(StringType())("tenant-a")
+
+    with table.transaction() as txn:
+        with txn.update_snapshot().overwrite() as overwrite:
+            overwrite.delete_data_file(existing_file)
+            overwrite._build_delete_files_partition_predicate()
+            predicate = overwrite._delete_files_partition_filters[existing_file.spec_id]
+
+    assert predicate == EqualTo(Reference("tenant_id_bucket"), expected_bucket_id)
+
+
+def test_delete_data_file_manifest_pruning_bucket_on_same_result_type_succeeds(catalog: Catalog) -> None:
+    """delete_data_file must not silently skip a manifest when the bucket id happens to share the source column's type.
+
+    Pre-fix, the buggy predicate compared the source column (an int) against the bucket id
+    (also an int), so binding succeeded instead of raising. The manifest's min/max stats for
+    that column then incorrectly ruled out the manifest containing the target file, so the
+    whole manifest was skipped and the file was silently never deleted - no exception, no
+    error, just a delete that quietly did nothing. A string source column can't hit this path
+    since it would fail to bind (see the other tests here), so this needs a same-result-type
+    source to catch a regression back to the source-column domain.
+    """
+    catalog.create_namespace_if_not_exists("default")
+    identifier = "default.bucket_delete_same_result_type"
+
+    schema = Schema(NestedField(1, "value", IntegerType(), required=True))
+    spec = PartitionSpec(
+        PartitionField(
+            source_id=1,
+            field_id=1000,
+            transform=BucketTransform(8),
+            name="value_bucket",
+        ),
+        spec_id=0,
+    )
+    table = catalog.create_table(
+        identifier=identifier,
+        schema=schema,
+        partition_spec=spec,
+    )
+    table.append(
+        pa.Table.from_pylist(
+            [{"value": 42}],
+            schema=pa.schema([pa.field("value", pa.int32(), nullable=False)]),
+        )
+    )
+
+    before_paths = {task.file.file_path for task in table.scan().plan_files()}
+    existing_file = next(iter(table.scan().plan_files())).file
+
+    with table.transaction() as txn:
+        with txn.update_snapshot().overwrite() as overwrite:
+            overwrite.delete_data_file(existing_file)
+
+    after_paths = {task.file.file_path for task in table.scan().plan_files()}
+
+    assert before_paths - after_paths == {existing_file.file_path}