From bfedc0c3bd4cb51b4a0d496341d3517d6c861ab8 Mon Sep 17 00:00:00 2001 From: QlikFrederic Date: Tue, 11 Aug 2026 14:52:54 +0200 Subject: [PATCH 1/8] 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 8a06d25174175c742b4629a10087f8f89313bef3 Mon Sep 17 00:00:00 2001 From: QlikFrederic Date: Tue, 11 Aug 2026 22:42:25 +0200 Subject: [PATCH 2/8] Prune manifests for delete_data_file using partition-domain predicates Build the manifest-pruning predicate against the partition struct (field.name) instead of the row schema, so it works for any partition transform, not just identity. Falls back to a plain AlwaysTrue for unpartitioned specs, and combines with existing predicate-based pruning via delete_by_predicate. --- pyiceberg/table/update/snapshot.py | 40 +++++++++++++------ ...t_delete_data_file_manifest_pruning_bug.py | 8 ++-- 2 files changed, 31 insertions(+), 17 deletions(-) diff --git a/pyiceberg/table/update/snapshot.py b/pyiceberg/table/update/snapshot.py index 4c468a5657..88811227eb 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, AlwaysTrue, BooleanExpression, Or +from pyiceberg.expressions import AlwaysFalse, AlwaysTrue, And, BooleanExpression, EqualTo, IsNull, Or, Reference from pyiceberg.expressions.visitors import ( ROWS_MIGHT_NOT_MATCH, ROWS_MUST_MATCH, @@ -71,7 +71,6 @@ 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 @@ -104,6 +103,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 def __init__( @@ -134,6 +134,7 @@ def __init__( snapshot.snapshot_id if (snapshot := self._transaction.table_metadata.snapshot_by_name(self._target_branch)) else None ) self._predicate = AlwaysFalse() + self._delete_files_partition_filters = {} self._case_sensitive = True def _validate_target_branch(self, branch: str | None) -> str | None: @@ -368,31 +369,44 @@ 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.""" 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 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( - partition_records=partition_records, schema=self.schema(), spec=self.spec(spec_id) - ) + # 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] + per_record_exprs = [ + self._build_partition_record_predicate(partition_field_names, partition_record) + for partition_record in partition_records + ] + self._delete_files_partition_filters[spec_id] = ( + Or(*per_record_exprs) if len(per_record_exprs) > 1 else per_record_exprs[0] ) + @staticmethod + def _build_partition_record_predicate(partition_field_names: list[str], partition_record: Record) -> BooleanExpression: + predicates: list[BooleanExpression] = [ + EqualTo(Reference(name), partition_record[pos]) if partition_record[pos] is not None else IsNull(Reference(name)) + for pos, name in enumerate(partition_field_names) + ] + if not predicates: + # Unpartitioned spec: nothing to filter on, so the (single, empty) partition always matches. + return AlwaysTrue() + return And(*predicates) if len(predicates) > 1 else predicates[0] + class _DeleteFiles(_SnapshotProducer["_DeleteFiles"]): """Will delete manifest entries from the current snapshot based on the 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 index 70d1fe027b..e4b66bd42d 100644 --- a/tests/table/test_delete_data_file_manifest_pruning_bug.py +++ b/tests/table/test_delete_data_file_manifest_pruning_bug.py @@ -25,11 +25,11 @@ 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. + """delete_data_file should work for non-identity specs. - 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. + 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 = f"default.bucket_delete_bug_{catalog.name}" From 349f5cd86da8831b60cbc419cf5b6be279c39444 Mon Sep 17 00:00:00 2001 From: QlikFrederic Date: Tue, 11 Aug 2026 22:47:44 +0200 Subject: [PATCH 3/8] Add test asserting delete_data_file pruning predicate uses partition field _OverwriteFiles deletes by exact DataFile identity, so an end-to-end delete would still pass even if pruning silently degraded to a non-discriminating fallback. This test inspects the built predicate directly to guard the partition-domain fix itself. --- ...t_delete_data_file_manifest_pruning_bug.py | 53 +++++++++++++++++++ 1 file changed, 53 insertions(+) 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 e4b66bd42d..497a0dbefa 100644 --- a/tests/table/test_delete_data_file_manifest_pruning_bug.py +++ b/tests/table/test_delete_data_file_manifest_pruning_bug.py @@ -18,6 +18,7 @@ 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 @@ -81,3 +82,55 @@ def test_delete_data_file_manifest_pruning_bucket_transform_succeeds(catalog: Ca assert existing_file.file_path not in remaining_paths 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 = f"default.bucket_delete_pruning_predicate_{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}], + 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) From c1b0a70a011618f5eb9c7af3dfbbcc2601808bb9 Mon Sep 17 00:00:00 2001 From: QlikFrederic Date: Tue, 11 Aug 2026 22:54:38 +0200 Subject: [PATCH 4/8] 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 From 6435109577832fff44e32b19699f1cb697bd720b Mon Sep 17 00:00:00 2001 From: QlikFrederic Date: Thu, 13 Aug 2026 09:42:07 +0200 Subject: [PATCH 5/8] re-order --- pyiceberg/table/update/snapshot.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyiceberg/table/update/snapshot.py b/pyiceberg/table/update/snapshot.py index 8d9b680dcc..369c1e475b 100644 --- a/pyiceberg/table/update/snapshot.py +++ b/pyiceberg/table/update/snapshot.py @@ -28,8 +28,8 @@ from typing import TYPE_CHECKING, Generic from pyiceberg.avro.codecs import AvroCompressionCodec -from pyiceberg.expressions import AlwaysFalse, AlwaysTrue, BooleanExpression, Or from pyiceberg.exceptions import ValidationException +from pyiceberg.expressions import AlwaysFalse, AlwaysTrue, BooleanExpression, Or from pyiceberg.expressions.visitors import ( ROWS_MIGHT_NOT_MATCH, ROWS_MUST_MATCH, From 6e88f9d575878699aaf70e867bcf9508f2b34adf Mon Sep 17 00:00:00 2001 From: QlikFrederic Date: Thu, 13 Aug 2026 13:14:48 +0200 Subject: [PATCH 6/8] Address review: fix stale comment, dedupe predicate builder, reset state - Fix stale comment in _manifests(): the call no longer touches self._predicate, only self._delete_files_partition_filters. - Extract build_field_value_predicate/build_records_predicate into pyiceberg.expressions, shared by Transaction._build_partition_predicate and _build_delete_files_partition_predicate, removing near-duplicate EqualTo/IsNull/And/Or construction. - Reset self._delete_files_partition_filters at the top of _build_delete_files_partition_predicate so it starts clean on every _manifests() pass, including retries. - Update stale test_commit_retry.py docstring: conflict detection for the CoW-rewrite path now uses the user's delete filter directly (self._predicate is no longer widened by the deleted-files partition predicate), matching Java's approach and fixing a latent inconsistency where _predicate accumulated across retries while partition_filters stayed frozen from the first attempt. This is a behavior change beyond the original bug fix and should be called out in the PR description. --- pyiceberg/expressions/__init__.py | 31 +++++++++++++++++++++++++++++- pyiceberg/table/__init__.py | 17 ++-------------- pyiceberg/table/update/snapshot.py | 28 ++++++++------------------- tests/table/test_commit_retry.py | 3 ++- 4 files changed, 42 insertions(+), 37 deletions(-) diff --git a/pyiceberg/expressions/__init__.py b/pyiceberg/expressions/__init__.py index ef4cb2506e..26089765ff 100644 --- a/pyiceberg/expressions/__init__.py +++ b/pyiceberg/expressions/__init__.py @@ -29,7 +29,7 @@ from pyiceberg.expressions.literals import AboveMax, BelowMin, Literal, literal from pyiceberg.schema import Accessor, Schema -from pyiceberg.typedef import IcebergBaseModel, IcebergRootModel, L, LiteralValue, StructProtocol +from pyiceberg.typedef import IcebergBaseModel, IcebergRootModel, L, LiteralValue, Record, StructProtocol from pyiceberg.types import DoubleType, FloatType, NestedField from pyiceberg.utils.singleton import Singleton @@ -1119,3 +1119,32 @@ def __invert__(self) -> StartsWith: @property def as_bound(self) -> type[BoundNotStartsWith]: # type: ignore return BoundNotStartsWith + + +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] diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py index 3dffc2270c..2f30c1fe91 100644 --- a/pyiceberg/table/__init__.py +++ b/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, build_records_predicate from pyiceberg.expressions.visitors import ( ResidualEvaluator, _InclusiveMetricsEvaluator, @@ -403,20 +403,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 diff --git a/pyiceberg/table/update/snapshot.py b/pyiceberg/table/update/snapshot.py index c2b44420f1..ee08b4a60a 100644 --- a/pyiceberg/table/update/snapshot.py +++ b/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, AlwaysTrue, And, BooleanExpression, EqualTo, IsNull, Or, Reference +from pyiceberg.expressions import AlwaysFalse, AlwaysTrue, BooleanExpression, Or, build_records_predicate from pyiceberg.expressions.visitors import ( ROWS_MIGHT_NOT_MATCH, ROWS_MUST_MATCH, @@ -265,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() @@ -527,6 +527,7 @@ def delete_by_predicate(self, predicate: BooleanExpression, case_sensitive: bool def _build_delete_files_partition_predicate(self) -> None: """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()) @@ -535,24 +536,11 @@ def _build_delete_files_partition_predicate(self) -> None: for spec_id, partition_records in partition_to_overwrite.items(): # 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] - per_record_exprs = [ - self._build_partition_record_predicate(partition_field_names, partition_record) - for partition_record in partition_records - ] - self._delete_files_partition_filters[spec_id] = ( - Or(*per_record_exprs) if len(per_record_exprs) > 1 else per_record_exprs[0] - ) - - @staticmethod - def _build_partition_record_predicate(partition_field_names: list[str], partition_record: Record) -> BooleanExpression: - predicates: list[BooleanExpression] = [ - EqualTo(Reference(name), partition_record[pos]) if partition_record[pos] is not None else IsNull(Reference(name)) - for pos, name in enumerate(partition_field_names) - ] - if not predicates: - # Unpartitioned spec: nothing to filter on, so the (single, empty) partition always matches. - return AlwaysTrue() - return And(*predicates) if len(predicates) > 1 else predicates[0] + 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"]): diff --git a/tests/table/test_commit_retry.py b/tests/table/test_commit_retry.py index ce5dca96aa..d2c155267b 100644 --- a/tests/table/test_commit_retry.py +++ b/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 From 96aef974003c0961726dc98905f21d52bc0ecd45 Mon Sep 17 00:00:00 2001 From: QlikFrederic Date: Thu, 13 Aug 2026 15:28:21 +0200 Subject: [PATCH 7/8] Address review: test naming, catalog-name no-op, silent-skip coverage - Rename tests/table/test_delete_data_file_manifest_pruning_bug.py to test_snapshot_manifest_pruning.py, matching the repo's component-named test file convention (no _bug suffix). - Drop the f"..._{catalog.name}" identifier suffix: all three catalog fixture params share name="test_catalog", so it was a no-op: isolation already comes from the per-test tmp_path. - Add test_delete_data_file_manifest_pruning_bucket_on_same_result_type_succeeds: a BucketTransform over an IntegerType source column, where the pre-fix predicate's type happened to match the source column's type. Unlike the string-source case (a loud TypeError), this variant let the buggy row-domain predicate bind successfully and silently prune away the manifest containing the target file, so delete_data_file reported success without deleting anything. Verified this fails (silently, no exception) against the pre-fix code and passes with the current fix. --- ...g.py => test_snapshot_manifest_pruning.py} | 52 ++++++++++++++++++- 1 file changed, 50 insertions(+), 2 deletions(-) rename tests/table/{test_delete_data_file_manifest_pruning_bug.py => test_snapshot_manifest_pruning.py} (70%) diff --git a/tests/table/test_delete_data_file_manifest_pruning_bug.py b/tests/table/test_snapshot_manifest_pruning.py similarity index 70% rename from tests/table/test_delete_data_file_manifest_pruning_bug.py rename to tests/table/test_snapshot_manifest_pruning.py index 487700660e..522fdd31e2 100644 --- a/tests/table/test_delete_data_file_manifest_pruning_bug.py +++ b/tests/table/test_snapshot_manifest_pruning.py @@ -33,7 +33,7 @@ def test_delete_data_file_manifest_pruning_bucket_transform_succeeds(catalog: Ca works regardless of the partition transform. """ catalog.create_namespace_if_not_exists("default") - identifier = f"default.bucket_delete_bug_{catalog.name}" + identifier = "default.bucket_delete" schema = Schema( NestedField(1, "tenant_id", StringType(), required=True), @@ -95,7 +95,7 @@ def test_delete_data_file_manifest_pruning_predicate_uses_partition_field(catalo non-discriminating fallback. This test guards the pruning predicate itself. """ catalog.create_namespace_if_not_exists("default") - identifier = f"default.bucket_delete_pruning_predicate_{catalog.name}" + identifier = "default.bucket_delete_pruning_predicate" schema = Schema( NestedField(1, "tenant_id", StringType(), required=True), @@ -137,3 +137,51 @@ def test_delete_data_file_manifest_pruning_predicate_uses_partition_field(catalo 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} From e3c462bbf397c03b6edb34d0674ffbacb985c120 Mon Sep 17 00:00:00 2001 From: QlikFrederic Date: Fri, 14 Aug 2026 10:29:40 +0200 Subject: [PATCH 8/8] Move build_field_value_predicate/build_records_predicate to partitioning.py They build predicates over Record values, not expression AST nodes, so pyiceberg/expressions/__init__.py (which only defines the expression classes themselves) was the wrong home. pyiceberg/partitioning.py already owns Record/PartitionSpec semantics and both call sites already import from it, so this is a natural fit with no new import needed and no circular-import risk (partitioning.py -> expressions is a new but one-directional edge; expressions/__init__.py does not import partitioning). --- pyiceberg/expressions/__init__.py | 31 +----------------------------- pyiceberg/partitioning.py | 30 +++++++++++++++++++++++++++++ pyiceberg/table/__init__.py | 10 ++++++++-- pyiceberg/table/update/snapshot.py | 4 ++-- 4 files changed, 41 insertions(+), 34 deletions(-) diff --git a/pyiceberg/expressions/__init__.py b/pyiceberg/expressions/__init__.py index 26089765ff..ef4cb2506e 100644 --- a/pyiceberg/expressions/__init__.py +++ b/pyiceberg/expressions/__init__.py @@ -29,7 +29,7 @@ from pyiceberg.expressions.literals import AboveMax, BelowMin, Literal, literal from pyiceberg.schema import Accessor, Schema -from pyiceberg.typedef import IcebergBaseModel, IcebergRootModel, L, LiteralValue, Record, StructProtocol +from pyiceberg.typedef import IcebergBaseModel, IcebergRootModel, L, LiteralValue, StructProtocol from pyiceberg.types import DoubleType, FloatType, NestedField from pyiceberg.utils.singleton import Singleton @@ -1119,32 +1119,3 @@ def __invert__(self) -> StartsWith: @property def as_bound(self) -> type[BoundNotStartsWith]: # type: ignore return BoundNotStartsWith - - -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] diff --git a/pyiceberg/partitioning.py b/pyiceberg/partitioning.py index 3074c30ea1..618037ed83 100644 --- a/pyiceberg/partitioning.py +++ b/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] diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py index 2f30c1fe91..da5471f833 100644 --- a/pyiceberg/table/__init__.py +++ b/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, Or, build_records_predicate +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 diff --git a/pyiceberg/table/update/snapshot.py b/pyiceberg/table/update/snapshot.py index ee08b4a60a..348ff5634e 100644 --- a/pyiceberg/table/update/snapshot.py +++ b/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, AlwaysTrue, BooleanExpression, Or, build_records_predicate +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 (