fix: delete_data_file overwrite pruning for non-identity partition specs

pyiceberg/table/update/snapshot.py

@@ -26,7 +26,7 @@
 from typing import TYPE_CHECKING, Generic
 
 from pyiceberg.avro.codecs import AvroCompressionCodec
-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,
@@ -71,6 +71,7 @@
     UpdatesAndRequirements,
     UpdateTableMetadata,
 )
+from pyiceberg.transforms import IdentityTransform
 from pyiceberg.typedef import EMPTY_DICT, KeyDefaultDict, Record
 from pyiceberg.utils.bin_packing import ListPacker
 from pyiceberg.utils.concurrent import ExecutorFactory
@@ -380,6 +381,11 @@ def _build_delete_files_partition_predicate(self) -> None:
             group = partition_to_overwrite.setdefault(data_file.spec_id, set())
             group.add(data_file.partition)
 
+        for spec_id in partition_to_overwrite:
+            if any(not isinstance(field.transform, IdentityTransform) for field in self.spec(spec_id).fields):
+                self.delete_by_predicate(AlwaysTrue())
+                return
+
         for spec_id, partition_records in partition_to_overwrite.items():
             self.delete_by_predicate(
                 self._transaction._build_partition_predicate(

tests/table/test_delete_data_file_manifest_pruning_bug.py

@@ -0,0 +1,83 @@
+# 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.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 via non-pruning fallback.
+
+    For bucket-partitioned tables, the stored partition value is a bucket id and cannot
+    be safely mapped back to a source-column predicate. The fallback should therefore
+    disable pruning and still apply delete by exact DataFile identity.
+    """
+    catalog.create_namespace_if_not_exists("default")
+    identifier = f"default.bucket_delete_bug_{catalog.name}"
+
+    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()
+    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()
+    remaining_paths = {task.file.file_path for task in table.scan().plan_files()}
+
+    assert existing_file.file_path not in remaining_paths
+    assert after.num_rows < before.num_rows