diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py index 8d79472af5..6848f88bf0 100644 --- a/pyiceberg/table/__init__.py +++ b/pyiceberg/table/__init__.py @@ -634,9 +634,13 @@ 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 - ] + schema = self.table_metadata.schema() + partitions_fields = [] + for field in self.table_metadata.spec().fields: + column_name = schema.find_column_name(field.source_id) + if column_name is None: + raise ValueError(f"Could not find column name for field ID: {field.source_id}") + partitions_fields.append(column_name) delete_filter = self._build_partition_predicate( partition_records=partitions_to_overwrite, partition_fields=partitions_fields ) diff --git a/tests/table/test_init.py b/tests/table/test_init.py index f6b984f1a3..ee56e2a3f4 100644 --- a/tests/table/test_init.py +++ b/tests/table/test_init.py @@ -2190,6 +2190,30 @@ def test_overwrite_delete_data_file_on_bucket_partition(catalog: Catalog) -> Non assert table.scan().to_arrow().num_rows == 0 +def test_dynamic_partition_overwrite_by_nested_source_column(catalog: Catalog) -> None: + """Overwrite partitions by the nested source column s.x, not the top-level column x with the same name.""" + import pyarrow as pa + + catalog.create_namespace("default") + schema = Schema( + NestedField(1, "x", LongType(), required=False), + NestedField(2, "s", StructType(NestedField(3, "x", LongType(), required=False)), required=False), + ) + spec = PartitionSpec(PartitionField(source_id=3, field_id=1000, transform=IdentityTransform(), name="s_x")) + table = catalog.create_table("default.nested_partition_overwrite", schema=schema, partition_spec=spec) + + def to_arrow(*rows: tuple[int, int]) -> pa.Table: + return pa.Table.from_pylist([{"x": x, "s": {"x": s_x}} for x, s_x in rows], schema=schema.as_arrow()) + + # Rows are (x, s.x): one row in partition s.x = 5 and one in s.x = 1 + table.append(to_arrow((1, 5), (5, 1))) + table.dynamic_partition_overwrite(to_arrow((9, 5))) + + # Only partition s.x = 5 is replaced, so (1, 5) becomes (9, 5) and (5, 1) is kept + result = sorted((row["x"], row["s"]["x"]) for row in table.scan().to_arrow().to_pylist()) + assert result == [(5, 1), (9, 5)] + + def test_static_table_forwards_location_to_table_file_io(metadata_location: str, monkeypatch: pytest.MonkeyPatch) -> None: seen_locations: list[str | None] = [] real_load_file_io = load_file_io