This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 27dc9b24f5 [python] Preserve fields from later Python MERGE source
rows (#10093)
27dc9b24f5 is described below
commit 27dc9b24f5f2e083f6d4477346abd2f8010f8e40
Author: chaoyang <[email protected]>
AuthorDate: Thu Sep 24 14:39:09 2026 +0800
[python] Preserve fields from later Python MERGE source rows (#10093)
---
paimon-python/pypaimon/multimodal/table.py | 6 ++-
.../pypaimon/tests/multimodal_table_test.py | 46 ++++++++++++++++++++++
2 files changed, 51 insertions(+), 1 deletion(-)
diff --git a/paimon-python/pypaimon/multimodal/table.py
b/paimon-python/pypaimon/multimodal/table.py
index 81f302c499..d988b91e07 100644
--- a/paimon-python/pypaimon/multimodal/table.py
+++ b/paimon-python/pypaimon/multimodal/table.py
@@ -599,7 +599,11 @@ def _to_arrow_table(data, target_schema=None):
elif isinstance(data, pa.RecordBatch):
table = pa.Table.from_batches([data])
elif isinstance(data, list):
- table = pa.Table.from_pylist(data)
+ # Inference from the first row alone drops fields in later source rows.
+ names = dict.fromkeys(name for row in data for name in row)
+ table = pa.Table.from_pydict({
+ name: [row.get(name) for row in data] for name in names
+ })
elif isinstance(data, dict):
table = pa.Table.from_pydict(data)
elif hasattr(data, "__dataframe__") or
data.__class__.__module__.startswith("pandas"):
diff --git a/paimon-python/pypaimon/tests/multimodal_table_test.py
b/paimon-python/pypaimon/tests/multimodal_table_test.py
index a61477f100..6551e743e0 100644
--- a/paimon-python/pypaimon/tests/multimodal_table_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_table_test.py
@@ -2395,6 +2395,52 @@ class MultimodalTableTest(unittest.TestCase):
rows,
)
+ def test_merge_preserves_fields_first_present_in_later_source_rows(self):
+ users = self.conn.create_table(
+ "sparse_merge",
+ data=[
+ {"id": 1, "name": "Alice", "age": 30},
+ {"id": 2, "name": "Bob", "age": 25},
+ ],
+ schema=_schema({
+ "id": pa.int32(), "name": pa.string(), "age": pa.int32(),
+ }),
+ options=_PARQUET_OPTIONS,
+ )
+ source = [{"id": 1}, {"id": 2, "name": "Bobby"},
+ {"id": 3, "name": "Carol"}]
+
+ users.merge("id").when_matched_update() \
+ .when_not_matched_insert().execute(source)
+
+ self.assertEqual([
+ {"id": 1, "name": None, "age": 30},
+ {"id": 2, "name": "Bobby", "age": 25},
+ {"id": 3, "name": "Carol", "age": None},
+ ], sorted(users.scan().to_list(), key=lambda row: row["id"]))
+ self.assertEqual({"id": 1}, source[0])
+
+ def test_merge_preserves_later_source_only_fields_with_mapped_keys(self):
+ users = self.conn.create_table(
+ "sparse_mapped_merge",
+ data=[{"id": 1, "name": "Alice"}],
+ schema=_schema({"id": pa.int32(), "name": pa.string()}),
+ options=_PARQUET_OPTIONS,
+ )
+ values = {"name": source_col("new_name")}
+
+ users.merge({"id": "source_id"}).when_matched_update(values) \
+ .when_not_matched_insert({
+ "id": source_col("source_id"), **values,
+ }).execute([
+ {"source_id": 1},
+ {"source_id": 2, "new_name": "Bob"},
+ ])
+
+ self.assertEqual([
+ {"id": 1, "name": None}, {"id": 2, "name": "Bob"},
+ ], sorted(users.scan().to_list(), key=lambda row: row["id"]))
+
def test_merge_supports_source_key_mapping(self):
users = self.conn.create_table(
"users",
