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 d1675a7a64 [python] Support nested payloads in local MERGE INTO
(#10094)
d1675a7a64 is described below
commit d1675a7a64089fa24504ed53ca67a0e579198337
Author: chaoyang <[email protected]>
AuthorDate: Thu Sep 24 14:38:52 2026 +0800
[python] Support nested payloads in local MERGE INTO (#10094)
---
.../pypaimon/table/data_evolution_merge_into.py | 30 +++++-
.../pypaimon/tests/table_merge_into_test.py | 102 +++++++++++++++++++++
2 files changed, 129 insertions(+), 3 deletions(-)
diff --git a/paimon-python/pypaimon/table/data_evolution_merge_into.py
b/paimon-python/pypaimon/table/data_evolution_merge_into.py
index ac556b6675..7439823d08 100644
--- a/paimon-python/pypaimon/table/data_evolution_merge_into.py
+++ b/paimon-python/pypaimon/table/data_evolution_merge_into.py
@@ -486,7 +486,8 @@ def _build_matched_update_table(
target_renamed = _rename_with_prefix(target, "t.")
source_renamed = _rename_with_prefix(source_table, "s.")
- joined = target_renamed.join(
+ joined = _join_with_row_indices(
+ target_renamed,
source_renamed,
keys=["t.{}".format(c) for c in ctx.target_on_cols],
right_keys=["s.{}".format(c) for c in ctx.source_on_cols],
@@ -525,7 +526,8 @@ def _build_matched_delete_table(
target_renamed = _rename_with_prefix(target, "t.")
source_renamed = _rename_with_prefix(source_table, "s.")
- joined = target_renamed.join(
+ joined = _join_with_row_indices(
+ target_renamed,
source_renamed,
keys=["t.{}".format(c) for c in ctx.target_on_cols],
right_keys=["s.{}".format(c) for c in ctx.source_on_cols],
@@ -564,7 +566,8 @@ def _build_not_matched_insert_table(
unmatched = source_renamed
else:
target_renamed = _rename_with_prefix(target, "t.")
- unmatched = source_renamed.join(
+ unmatched = _join_with_row_indices(
+ source_renamed,
target_renamed,
keys=["s.{}".format(c) for c in ctx.source_on_cols],
right_keys=["t.{}".format(c) for c in ctx.target_on_cols],
@@ -577,6 +580,27 @@ def _build_not_matched_insert_table(
return transform(unmatched)
+def _join_with_row_indices(left, right, keys, right_keys, join_type):
+ """Keep nested payloads out of Arrow's join, then gather matched rows."""
+ key_names = ["key_%d" % index for index in range(len(keys))]
+ left_keys = left.select(keys).rename_columns(key_names).append_column(
+ "left_index", pa.array(range(left.num_rows), type=pa.int64()))
+ right_keys_table = right.select(right_keys).rename_columns(key_names)
+ if join_type == "inner":
+ right_keys_table = right_keys_table.append_column(
+ "right_index", pa.array(range(right.num_rows), type=pa.int64()))
+ indices = left_keys.join(right_keys_table, keys=key_names,
join_type=join_type)
+ result = left.take(indices["left_index"])
+ if join_type == "inner":
+ # Match Table.join's coalesced-key schema: keep only the left keys.
+ payload = right.select([
+ name for name in right.column_names if name not in right_keys
+ ]).take(indices["right_index"])
+ for field, column in zip(payload.schema, payload.columns):
+ result = result.append_column(field, column)
+ return result
+
+
def _prepare_commit_messages(
table,
update_table: Optional[pa.Table],
diff --git a/paimon-python/pypaimon/tests/table_merge_into_test.py
b/paimon-python/pypaimon/tests/table_merge_into_test.py
index 811de74d90..3f117c4893 100644
--- a/paimon-python/pypaimon/tests/table_merge_into_test.py
+++ b/paimon-python/pypaimon/tests/table_merge_into_test.py
@@ -623,6 +623,108 @@ class TableMergeIntoTest(BatchModeMixin,
DataEvolutionTestBase, unittest.TestCas
self._read_projected_sorted(target, ["id", "name", "payload"]),
)
+ def test_table_merge_into_nested_payloads(self):
+ cases = [
+ (pa.list_(pa.float32(), 2), [1., 2.], [3., 4.], [5., 6.]),
+ (pa.list_(pa.int64()), [1, None], [], None),
+ (pa.map_(pa.string(), pa.string()), [("old", "value")], [],
[("new", None)]),
+ (pa.struct([("label", pa.string()), ("count", pa.int64())]),
+ {"label": "old", "count": None}, None, {"label": None, "count":
2}),
+ ]
+ for payload_type, before, updated, inserted in cases:
+ with self.subTest(payload_type=payload_type):
+ schema = pa.schema([("id", pa.int32()), ("payload",
payload_type)])
+ target = self._create_table(pa_schema=schema, options=dict(
+ self.table_options, **{
+ "file.format": "parquet", "vector.file.format":
"parquet",
+ "deletion-vectors.enabled": "true",
+ }))
+ self._write_arrow(target, pa.Table.from_pylist([
+ {"id": 1, "payload": before}, {"id": 2, "payload": before},
+ ], schema=schema))
+ # Keep the source chunked, with a nonzero slice offset.
+ source = pa.concat_tables([
+ pa.Table.from_pylist([
+ {"id": 0, "payload": before}, {"id": 2, "payload":
updated},
+ ], schema=schema).slice(1),
+ pa.Table.from_pylist([{"id": 3, "payload": inserted}],
schema=schema),
+ ])
+
+ self._merge_and_commit(
+ target, source, on=["id"],
+ when_matched=[WhenMatched.update("*")],
+ when_not_matched=[WhenNotMatched(insert="*")])
+
+ self.assertEqual({"id": [1, 2, 3], "payload": [before,
updated, inserted]},
+ self._read_sorted(target))
+ # A delete must also accept unused nested source columns.
+ self._merge_and_commit(
+ target, source.slice(0, 1), on=["id"],
+ when_matched=[WhenMatched.delete()])
+ self.assertEqual({"id": [1, 3], "payload": [before, inserted]},
+ self._read_sorted(target))
+
+ @unittest.skipIf(_SKIP_CONDITION, _SKIP_REASON)
+ def test_nested_merge_keeps_mapped_composite_keys_and_null_semantics(self):
+ schema = pa.schema([
+ ("id", pa.int32()), ("group", pa.string()),
+ ("payload", pa.list_(pa.int32())), ("previous",
pa.list_(pa.int32())),
+ ])
+ target = self._create_table(pa_schema=schema, options=dict(
+ self.table_options, **{"file.format": "parquet"}))
+ self._write_arrow(target, pa.Table.from_pylist([
+ {"id": 1, "group": "a", "payload": [10], "previous": []},
+ {"id": 1, "group": "b", "payload": [20], "previous": []},
+ {"id": None, "group": "a", "payload": [30], "previous": []},
+ ], schema=schema))
+ source_schema = pa.schema([
+ ("source_id", pa.int32()), ("source_group", pa.string()),
+ ("payload", pa.list_(pa.int32())),
+ ])
+ source = pa.Table.from_pylist([
+ {"source_id": 1, "source_group": "a", "payload": [11]},
+ {"source_id": 1, "source_group": "b", "payload": [21]},
+ {"source_id": None, "source_group": "a", "payload": [31]},
+ {"source_id": 2, "source_group": "a", "payload": [40]},
+ ], schema=source_schema)
+
+ self._merge_and_commit(
+ target, source, on={"id": "source_id", "group": "source_group"},
+ when_matched=[WhenMatched.update({
+ "payload": source_col("payload"), "previous":
target_col("payload"),
+ }, condition="s.source_id = 1 AND t.group = 'a'")],
+ when_not_matched=[WhenNotMatched(insert={"payload":
source_col("payload")})])
+
+ self.assertCountEqual([
+ {"id": 1, "group": "a", "payload": [11], "previous": [10]},
+ {"id": 1, "group": "b", "payload": [20], "previous": []},
+ {"id": None, "group": "a", "payload": [30], "previous": []},
+ {"id": None, "group": "a", "payload": [31], "previous": None},
+ {"id": 2, "group": "a", "payload": [40], "previous": None},
+ ], self._read_all(target).to_pylist())
+
+ def test_nested_merge_rejects_duplicate_matches_before_commit(self):
+ schema = pa.schema([("id", pa.int32()), ("payload",
pa.list_(pa.int32()))])
+ target = self._create_table(pa_schema=schema, options=dict(
+ self.table_options, **{"file.format": "parquet"}))
+ self._write_arrow(target, pa.Table.from_pylist([
+ {"id": 1, "payload": [10]},
+ ], schema=schema))
+ snapshot_id = target.snapshot_manager().get_latest_snapshot().id
+ source = pa.Table.from_pylist([
+ {"id": 1, "payload": [11]}, {"id": 1, "payload": [12]},
+ {"id": 2, "payload": [20]},
+ ], schema=schema)
+
+ with self.assertRaisesRegex(ValueError, "multiple source rows"):
+ self._merge_and_commit(
+ target, source, on=["id"],
+ when_matched=[WhenMatched.update("*")],
+ when_not_matched=[WhenNotMatched(insert="*")])
+
+ self.assertEqual(snapshot_id,
target.snapshot_manager().get_latest_snapshot().id)
+ self.assertEqual({"id": [1], "payload": [[10]]},
self._read_sorted(target))
+
def test_table_merge_into_inserts_null_for_unspecified_blob_column(self):
blob_schema = pa.schema([
("id", pa.int32()),
