From b55d927f5ef8cce99a7b354a76417e6b8471ebc6 Mon Sep 17 00:00:00 2001
From: KK <37928252+dragonknightk@users.noreply.github.com>
Date: Fri, 30 Oct 2026 18:26:25 +0800
Subject: [PATCH] Use the full column name in the dynamic_partition_overwrite
 delete filter
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Closes issue 4196

# Rationale for this change

`dynamic_partition_overwrite` built its delete filter from `find_field(source_id).name`, which is only the leaf name. For a partition sourced from the nested column `s.x`, the filter referenced `x`, so:

- with a top-level `x`, it deleted rows by the wrong column and kept the old rows of the replaced partition
- without one, binding the filter raised `ValueError`

The filter now uses `find_column_name`, the full column name, the same way `_determine_partitions` does since #2204 (｡•̀ᴗ-)✧

## Are these changes tested?

Yes. `test_dynamic_partition_overwrite_by_nested_source_column` in `tests/table/test_init.py` runs the example from the issue against the memory and SQL catalogs:

- with this change, only partition `s.x = 5` is replaced: `[(5, 1), (9, 5)]`
- without it, the test fails with `[(1, 5), (9, 5)]`

## Are there any user-facing changes?

Yes, a bug fix: `dynamic_partition_overwrite` replaces the right partitions on tables partitioned by a nested column. Tables partitioned only by top-level columns behave as before (・ω・)b

## AI Disclosure

Human-led contribution. Grok Bot, working for dragonknightk, assisted with this change. All submitted content was reviewed by a human before submission.
---
 pyiceberg/table/__init__.py | 10 +++++++---
 tests/table/test_init.py    | 24 ++++++++++++++++++++++++
 2 files changed, 31 insertions(+), 3 deletions(-)

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