dynamic_partition_overwrite by identity(s.x) deletes rows by top-level x #4196

import tempfile
import pyarrow as pa
from pyiceberg.catalog.sql import SqlCatalog
from pyiceberg.partitioning import PartitionField, PartitionSpec
from pyiceberg.schema import Schema
from pyiceberg.transforms import IdentityTransform
from pyiceberg.types import LongType, NestedField, StructType

cat = SqlCatalog("c", uri="sqlite:///:memory:", warehouse=f"file://{tempfile.mkdtemp()}")
cat.create_namespace("ns")
schema = Schema(NestedField(1, "x", LongType()), NestedField(2, "s", StructType(NestedField(3, "x", LongType()))))
t = cat.create_table("ns.t", schema=schema, partition_spec=PartitionSpec(PartitionField(3, 1000, IdentityTransform(), "s_x")))
df = lambda *rs: pa.Table.from_pylist([{"x": x, "s": {"x": sx}} for x, sx in rs], schema=schema.as_arrow())
t.append(df((1, 5), (5, 1)))

t.dynamic_partition_overwrite(df((9, 5)))
print(sorted((r["x"], r["s"]["x"]) for r in t.scan().to_arrow().to_pylist()))
# [(1, 5), (9, 5)], should be [(5, 1), (9, 5)]
# rows are (x, s.x): partition s.x = 1 lost its row and s.x = 5 kept its old one