From bfedc0c3bd4cb51b4a0d496341d3517d6c861ab8 Mon Sep 17 00:00:00 2001 From: QlikFrederic Date: Tue, 11 Aug 2026 14:52:54 +0200 Subject: [PATCH 1/2] fix delete_data_file overwrite pruning for non-identity partition specs --- pyiceberg/table/update/snapshot.py | 8 +- ...t_delete_data_file_manifest_pruning_bug.py | 83 +++++++++++++++++++ 2 files changed, 90 insertions(+), 1 deletion(-) create mode 100644 tests/table/test_delete_data_file_manifest_pruning_bug.py diff --git a/pyiceberg/table/update/snapshot.py b/pyiceberg/table/update/snapshot.py index 7931edacdd..4c468a5657 100644 --- a/pyiceberg/table/update/snapshot.py +++ b/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( diff --git a/tests/table/test_delete_data_file_manifest_pruning_bug.py b/tests/table/test_delete_data_file_manifest_pruning_bug.py new file mode 100644 index 0000000000..70d1fe027b --- /dev/null +++ b/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 From c1b0a70a011618f5eb9c7af3dfbbcc2601808bb9 Mon Sep 17 00:00:00 2001 From: QlikFrederic Date: Tue, 11 Aug 2026 22:54:38 +0200 Subject: [PATCH 2/2] Address review: clarify AlwaysTrue fallback and strengthen test - Comment explains that the AlwaysTrue fallback only disables the manifest-pruning optimization; deletion still happens by exact DataFile identity in _OverwriteFiles, so no rows are unexpectedly dropped. - Test now asserts on the file-path set before/after deletion (exact path removed, count drops by exactly one) instead of relying on row-count alone. --- pyiceberg/table/update/snapshot.py | 2 ++ tests/table/test_delete_data_file_manifest_pruning_bug.py | 7 +++++-- 2 files changed, 7 insertions(+), 2 deletions(-) diff --git a/pyiceberg/table/update/snapshot.py b/pyiceberg/table/update/snapshot.py index 4c468a5657..3ef6a91651 100644 --- a/pyiceberg/table/update/snapshot.py +++ b/pyiceberg/table/update/snapshot.py @@ -383,6 +383,8 @@ def _build_delete_files_partition_predicate(self) -> None: for spec_id in partition_to_overwrite: if any(not isinstance(field.transform, IdentityTransform) for field in self.spec(spec_id).fields): + # Disables the manifest-pruning optimization (not correctness): deletion of the + # specific data files still happens by exact DataFile identity in _OverwriteFiles. self.delete_by_predicate(AlwaysTrue()) return diff --git a/tests/table/test_delete_data_file_manifest_pruning_bug.py b/tests/table/test_delete_data_file_manifest_pruning_bug.py index 70d1fe027b..0cbc8c787e 100644 --- a/tests/table/test_delete_data_file_manifest_pruning_bug.py +++ b/tests/table/test_delete_data_file_manifest_pruning_bug.py @@ -70,6 +70,7 @@ def test_delete_data_file_manifest_pruning_bucket_transform_succeeds(catalog: Ca ) 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: @@ -77,7 +78,9 @@ def test_delete_data_file_manifest_pruning_bucket_transform_succeeds(catalog: Ca overwrite.delete_data_file(existing_file) after = table.scan().to_arrow() - remaining_paths = {task.file.file_path for task in table.scan().plan_files()} + after_paths = {task.file.file_path for task in table.scan().plan_files()} - assert existing_file.file_path not in remaining_paths + 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