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 3dffc2270c..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, 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 diff --git a/pyiceberg/table/update/snapshot.py b/pyiceberg/table/update/snapshot.py index 3c58f8ff44..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, 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"]): 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 diff --git a/tests/table/test_snapshot_manifest_pruning.py b/tests/table/test_snapshot_manifest_pruning.py new file mode 100644 index 0000000000..522fdd31e2 --- /dev/null +++ b/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}