From bcdda7bb02ab0a138ce73d8a663a2b2a39791459 Mon Sep 17 00:00:00 2001 From: Gabriel Igliozzi Date: Wed, 12 Aug 2026 23:31:46 +0200 Subject: [PATCH 1/2] fix transform eval --- pyiceberg/table/__init__.py | 14 +++++----- pyiceberg/table/update/snapshot.py | 9 ++++--- tests/table/test_init.py | 5 ++-- tests/table/test_upsert.py | 41 +++++++++++++++++++++++++++++- 4 files changed, 55 insertions(+), 14 deletions(-) diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py index 3dffc2270c..bb879dfbce 100644 --- a/pyiceberg/table/__init__.py +++ b/pyiceberg/table/__init__.py @@ -390,19 +390,16 @@ def _set_ref_snapshot( return updates, requirements - def _build_partition_predicate( - self, partition_records: set[Record], spec: PartitionSpec, schema: Schema - ) -> BooleanExpression: + def _build_partition_predicate(self, partition_records: set[Record], partition_fields: list[str]) -> BooleanExpression: """Build a filter predicate matching any of the input partition records. Args: partition_records: A set of partition records to match - spec: An optional partition spec, if none then defaults to current - schema: An optional schema, if none then defaults to current + partition_fields: The field names to reference for each position in a partition record + Returns: 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() @@ -622,8 +619,11 @@ def dynamic_partition_overwrite( ) partitions_to_overwrite = {data_file.partition for data_file in data_files} + partitions_fields = [ + self.table_metadata.schema().find_field(field.source_id).name for field in self.table_metadata.spec().fields + ] delete_filter = self._build_partition_predicate( - partition_records=partitions_to_overwrite, spec=self.table_metadata.spec(), schema=self.table_metadata.schema() + partition_records=partitions_to_overwrite, partition_fields=partitions_fields ) self.delete( delete_filter=delete_filter, diff --git a/pyiceberg/table/update/snapshot.py b/pyiceberg/table/update/snapshot.py index 3c58f8ff44..bcbca4b240 100644 --- a/pyiceberg/table/update/snapshot.py +++ b/pyiceberg/table/update/snapshot.py @@ -528,11 +528,12 @@ def _build_delete_files_partition_predicate(self) -> None: group.add(data_file.partition) for spec_id, partition_records in partition_to_overwrite.items(): - self.delete_by_predicate( + self.partition_filters[spec_id] = Or( + self.partition_filters[spec_id], self._transaction._build_partition_predicate( - partition_records=partition_records, schema=self.schema(), spec=self.spec(spec_id) - ), - self._case_sensitive, + partition_records=partition_records, + partition_fields=[field.name for field in self.spec(spec_id).fields], + ) ) diff --git a/tests/table/test_init.py b/tests/table/test_init.py index 3f1e97768c..ec1c44fdb4 100644 --- a/tests/table/test_init.py +++ b/tests/table/test_init.py @@ -1982,10 +1982,11 @@ def test_check_uuid_passes_when_match(table_v2: Table, example_table_metadata_v2 def test_build_large_partition_predicate(table_v2: Table) -> None: with table_v2.transaction() as tx: + schema = table_v2.schema() + spec = table_v2.spec() expr = tx._build_partition_predicate( partition_records={Record(i) for i in range(5000)}, - spec=table_v2.metadata.spec(), - schema=table_v2.metadata.schema(), + partition_fields=[(schema.find_field(field.source_id).name) for field in spec.fields], ) bind(table_v2.metadata.schema(), expr, case_sensitive=True) diff --git a/tests/table/test_upsert.py b/tests/table/test_upsert.py index 08f90c6600..78ddbc7c5c 100644 --- a/tests/table/test_upsert.py +++ b/tests/table/test_upsert.py @@ -14,6 +14,7 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. +from datetime import datetime from pathlib import PosixPath import pyarrow as pa @@ -26,11 +27,13 @@ from pyiceberg.expressions import AlwaysTrue, And, EqualTo, Reference from pyiceberg.expressions.literals import LongLiteral from pyiceberg.io.pyarrow import schema_to_pyarrow +from pyiceberg.partitioning import PartitionField, PartitionSpec from pyiceberg.schema import Schema from pyiceberg.table import Table, UpsertResult from pyiceberg.table.snapshots import Operation from pyiceberg.table.upsert_util import create_match_filter -from pyiceberg.types import IntegerType, NestedField, StringType, StructType +from pyiceberg.transforms import DayTransform +from pyiceberg.types import IntegerType, NestedField, StringType, StructType, TimestampType from tests.catalog.test_base import InMemoryCatalog @@ -714,6 +717,42 @@ def test_upsert_with_nulls(catalog: Catalog) -> None: ) +def test_upsert_on_table_partitioned_by_transform(catalog: Catalog) -> None: + """Upsert has to rewrite the matched file on a table partitioned by a non-identity transform. + + The manifest pruning in the overwrite builds its predicate from the partition records of + the deleted files. Those records hold already-transformed values, so referencing the source + column would send them through the transform twice, prune away the only relevant manifest + and leave the replaced row behind as a duplicate. + """ + identifier = "default.test_upsert_on_table_partitioned_by_transform" + _drop_table(catalog, identifier) + + schema = Schema( + NestedField(1, "k", StringType(), required=False), + NestedField(2, "v", IntegerType(), required=False), + NestedField(3, "ts", TimestampType(), required=False), + ) + spec = PartitionSpec(PartitionField(source_id=3, field_id=1000, transform=DayTransform(), name="ts_day")) + table = catalog.create_table(identifier, schema, partition_spec=spec) + + arrow_schema = schema_to_pyarrow(schema) + # A timestamp whose day ordinal is far from the value it would be read as if the + # DayTransform were applied a second time. + ts = datetime(2026, 1, 6, 12) + + def rows(pairs: list[tuple[str, int]]) -> pa_table: + return pa.Table.from_pylist([{"k": k, "v": v, "ts": ts} for k, v in pairs], schema=arrow_schema) + + table.append(rows([("a", 1), ("b", 1)])) + + res = table.upsert(rows([("a", 2)]), join_cols=["k"]) + assert_upsert_result(res, expected_updated=1, expected_inserted=0) + + arrow = table.scan().to_arrow() + assert sorted(zip(arrow["k"].to_pylist(), arrow["v"].to_pylist(), strict=True)) == [("a", 2), ("b", 1)] + + def test_transaction(catalog: Catalog) -> None: """Test the upsert within a Transaction. Make sure that if something fails the entire Transaction is rolled back.""" From e14a8dc3b836ba9c28b4e7166a02f4794826f40b Mon Sep 17 00:00:00 2001 From: Gabriel Igliozzi Date: Wed, 12 Aug 2026 23:38:55 +0200 Subject: [PATCH 2/2] lint --- 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 bcbca4b240..57215dca04 100644 --- a/pyiceberg/table/update/snapshot.py +++ b/pyiceberg/table/update/snapshot.py @@ -533,7 +533,7 @@ def _build_delete_files_partition_predicate(self) -> None: self._transaction._build_partition_predicate( partition_records=partition_records, partition_fields=[field.name for field in self.spec(spec_id).fields], - ) + ), )