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 23e6f84c9b [python] Run data evolution MERGE through native core
(#10418)
23e6f84c9b is described below
commit 23e6f84c9be3d76e750fca545bfc845729cc7299
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Oct 8 09:32:49 2026 +0800
[python] Run data evolution MERGE through native core (#10418)
---
paimon-python/README.md | 16 ++
paimon-python/conftest.py | 4 +-
.../pypaimon/table/data_evolution_merge_into.py | 30 ++-
.../pypaimon/tests/native_merge_into_test.py | 230 +++++++++++++++++++++
paimon-python/pypaimon/write/native_merge_into.py | 97 +++++++++
5 files changed, 375 insertions(+), 2 deletions(-)
diff --git a/paimon-python/README.md b/paimon-python/README.md
index dedf06db7d..a6cbfcbd75 100644
--- a/paimon-python/README.md
+++ b/paimon-python/README.md
@@ -168,6 +168,22 @@ path select the Python writer before native data is
written. If the runtime or t
unavailable, write uses Python. Once Rust starts writing a batch, errors
propagate without retrying that batch through Python.
+Batch and stream `merge_into` also use Rust core for eligible data-evolution
+Parquet tables when `write.native.enabled=true`. Existing `WhenMatched` and
+`WhenNotMatched` clauses accept Arrow/pandas input or self-merge on `_ROW_ID`.
+Core pins the target snapshot, matches keys, selects the first satisfied clause
+and prepares updates, deletion vectors and inserts. SQL conditions, including
+subqueries, execute in the Rust DataFusion adapter; Python transports the
+normalized clauses and returned commit messages. Conditional MERGE requires
+Python 3.10+ and the existing `pypaimon[sql]` extra, which installs both Python
+DataFusion and `pypaimon-rust`. Shared clause validation checks that dependency
+before native execution. NULL keys do not match.
+Multiple source rows matching a target are rejected before action conditions,
+except for a sole unconditional DELETE, following Paimon Spark MERGE.
+Non-self table sources and packed-video inserts use the Python path selected
+before native execution starts. Native failures propagate without Python retry
+or deleting files from earlier prepared actions.
+
MAP columns configured with `fields.<name>.map.storage-layout=shared-shredding`
also use native Parquet writes and data-evolution updates, including predicate
updates and upserts. Rust applies Java's `plain`, `sequential` and `lru` column
diff --git a/paimon-python/conftest.py b/paimon-python/conftest.py
index 4288d9b997..0ca0ac11a0 100644
--- a/paimon-python/conftest.py
+++ b/paimon-python/conftest.py
@@ -28,7 +28,7 @@ _native_plan_count = 0
_native_read_count = 0
_native_write_count = 0
_native_commit_count = 0
-_native_update_counts = dict.fromkeys(('row_id', 'grouped', 'predicate',
'upsert', 'incremental'), 0)
+_native_update_counts = dict.fromkeys(('row_id', 'grouped', 'predicate',
'upsert', 'incremental', 'merge'), 0)
_force_native_for_test = False
_force_native_read_for_test = False
_force_native_write_for_test = False
@@ -160,6 +160,7 @@ def pytest_configure(config):
TableCommit._prepare_native_commit = tracked_prepare
if _native_update_enabled():
+ from pypaimon.write.native_merge_into import NativeTableMergeInto
from pypaimon.write.native_update import (
NativeBatchTableUpdate, NativePredicateTableUpdate,
NativeTableUpdateByRowId, NativeTableUpsert,
@@ -181,6 +182,7 @@ def pytest_configure(config):
track_update(NativePredicateTableUpdate, 'update', 'predicate')
track_update(NativeTableUpsert, 'upsert', 'upsert')
track_update(NativeTableUpdateByRowId, 'update_columns', 'incremental')
+ track_update(NativeTableMergeInto, 'prepare_commit', 'merge')
def pytest_collection_modifyitems(items):
diff --git a/paimon-python/pypaimon/table/data_evolution_merge_into.py
b/paimon-python/pypaimon/table/data_evolution_merge_into.py
index 58c02d4517..d5baff21bc 100644
--- a/paimon-python/pypaimon/table/data_evolution_merge_into.py
+++ b/paimon-python/pypaimon/table/data_evolution_merge_into.py
@@ -92,6 +92,14 @@ def merge_into(
The returned messages must be committed by the caller's matching
``TableCommit``.
"""
+ if target_table.options.native_write_enabled():
+ from pypaimon.write.native_merge_into import create_native_merge_into
+
+ native = create_native_merge_into(target_table, source, on,
when_matched, when_not_matched,
+ commit_user, commit_identifier)
+ if native is not None:
+ return native.prepare_commit()
+
base_snapshot = target_table.snapshot_manager().get_latest_snapshot()
source_table, matched_specs, not_matched_specs, ctx = _prepare(
target_table,
@@ -326,6 +334,14 @@ def _build_tables(
if matched_specs and base_snapshot is not None:
update_cols_union = _union_update_cols(matched_specs)
+ if not update_cols_union and not any(c.delete for c in matched_specs):
+ names = list(dict.fromkeys(ctx.target_on_cols +
[SpecialFields.ROW_ID.name]))
+ target = _read_table(target_table, projection=names,
snapshot_id=base_snapshot_id)
+ target = _rename_with_prefix(target, "t.")
+ _join_with_row_indices(target, _rename_with_prefix(source_table,
"s."),
+ ["t." + name for name in
ctx.target_on_cols],
+ ["s." + name for name in
ctx.source_on_cols],
+ "inner", check_cardinality=True)
if update_cols_union:
update_table = _build_matched_update_table(
target_table,
@@ -491,6 +507,7 @@ def _build_matched_update_table(
keys=["t.{}".format(c) for c in ctx.target_on_cols],
right_keys=["s.{}".format(c) for c in ctx.source_on_cols],
join_type="inner",
+ check_cardinality=True,
)
transform = _build_matched_transform(
clauses,
@@ -531,6 +548,8 @@ def _build_matched_delete_table(
keys=["t.{}".format(c) for c in ctx.target_on_cols],
right_keys=["s.{}".format(c) for c in ctx.source_on_cols],
join_type="inner",
+ check_cardinality=not (len(clauses) == 1 and clauses[0].delete and
clauses[0].condition is None),
+ deduplicate_left=len(clauses) == 1 and clauses[0].delete and
clauses[0].condition is None,
)
transform = _build_matched_delete_transform(
clauses,
@@ -579,7 +598,8 @@ def _build_not_matched_insert_table(
return transform(unmatched)
-def _join_with_row_indices(left, right, keys, right_keys, join_type):
+def _join_with_row_indices(left, right, keys, right_keys, join_type,
+ check_cardinality=False, deduplicate_left=False):
"""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(
@@ -589,6 +609,14 @@ def _join_with_row_indices(left, right, keys, right_keys,
join_type):
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)
+ # Spark checks source-match cardinality before action predicates. A sole
+ # unconditional DELETE is unambiguous and may collapse duplicate matches.
+ if check_cardinality and pc.count_distinct(indices["left_index"]).as_py()
!= indices.num_rows:
+ raise ValueError("MERGE matched multiple source rows to the same
target _ROW_ID. "
+ "Deduplicate the source before merging.")
+ if deduplicate_left:
+ indices = indices.group_by("left_index", use_threads=False).aggregate([
+ ("right_index", "first")]).rename_columns(["left_index",
"right_index"])
result = left.take(indices["left_index"])
if join_type == "inner":
# Match Table.join's coalesced-key schema: keep only the left keys.
diff --git a/paimon-python/pypaimon/tests/native_merge_into_test.py
b/paimon-python/pypaimon/tests/native_merge_into_test.py
new file mode 100644
index 0000000000..d7713a70a5
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_merge_into_test.py
@@ -0,0 +1,230 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""MERGE must match, select actions and stage files in native core."""
+
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.table.data_evolution_merge_into import WhenMatched,
WhenNotMatched, source_col, lit
+
+pytestmark = pytest.mark.native_plan
+_SCHEMA = pa.schema([('id', pa.int32()), ('value', pa.int32()), ('name',
pa.string())])
+
+
+def _table(tmp_path, rows=None):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('db', True)
+ catalog.create_table('db.t', Schema.from_pyarrow_schema(_SCHEMA, options={
+ 'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true',
+ 'write.native.enabled': 'true', 'deletion-vectors.enabled': 'true',
+ }), False)
+ table = catalog.get_table('db.t')
+ if rows:
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(pa.Table.from_pylist(rows, schema=_SCHEMA))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+ return table
+
+
+def _rows(table):
+ reader = table.new_read_builder()
+ return
sorted(reader.new_read().to_arrow(reader.new_scan().plan().splits()).to_pylist(),
+ key=lambda row: row['id'] if row['id'] is not None else -1)
+
+
+def test_native_merge_updates_and_inserts_without_python_join(tmp_path):
+ table = _table(tmp_path, [dict(id=1, value=10, name='old'), dict(id=2,
value=20, name='keep')])
+ source = pa.Table.from_pylist([dict(id=1, value=11, name='new'),
dict(id=3, value=30, name='insert')],
+ schema=_SCHEMA)
+ builder = table.new_batch_write_builder()
+ with patch('pypaimon.table.data_evolution_merge_into._build_tables',
+ side_effect=AssertionError('Python merge orchestration')):
+ messages = builder.new_update().merge_into(source, on=['id'],
+
when_matched=[WhenMatched.update({'value': source_col('value')})],
+
when_not_matched=[WhenNotMatched('*')])
+ builder.new_commit().commit(messages)
+ assert _rows(table) == [dict(id=1, value=11, name='old'), dict(id=2,
value=20, name='keep'),
+ dict(id=3, value=30, name='insert')]
+
+
[email protected]('native', [False, True])
+def test_duplicate_source_is_rejected_before_action_conditions(tmp_path,
native):
+ table = _table(tmp_path, [dict(id=1, value=10,
name='old')]).copy({'write.native.enabled': str(native).lower()})
+ source = pa.Table.from_pylist([dict(id=1, value=11, name='a'), dict(id=1,
value=12, name='b')], schema=_SCHEMA)
+ files = set(tmp_path.rglob('*.parquet'))
+ with pytest.raises(ValueError, match='multiple source rows'):
+ table.new_batch_write_builder().new_update().merge_into(
+ source, on=['id'], when_matched=[
+ WhenMatched.update({'value': source_col('value')},
condition='s.value = 11')])
+ assert set(tmp_path.rglob('*.parquet')) == files
+ assert _rows(table) == [dict(id=1, value=10, name='old')]
+
+
[email protected]('native', [False, True])
+def test_unconditional_delete_allows_multiple_source_matches(tmp_path, native):
+ table = _table(tmp_path, [dict(id=1, value=10, name='remove'), dict(id=2,
value=20, name='keep')])
+ table = table.copy({'write.native.enabled': str(native).lower()})
+ source = pa.Table.from_pylist([dict(id=1, value=11, name='a'), dict(id=1,
value=12, name='b')], schema=_SCHEMA)
+ builder = table.new_batch_write_builder()
+ messages = builder.new_update().merge_into(source, on=['id'],
when_matched=[WhenMatched.delete()])
+ builder.new_commit().commit(messages)
+ assert _rows(table) == [dict(id=2, value=20, name='keep')]
+
+
[email protected]('stream', [False, True])
+def test_ordered_conditions_null_keys_and_three_action_types(tmp_path, stream):
+ table = _table(tmp_path, [
+ dict(id=1, value=10, name='remove'), dict(id=2, value=20, name='keep'),
+ dict(id=None, value=90, name='null-old')])
+ source = pa.Table.from_pylist([dict(id=1, value=-1, name='remove'),
dict(id=2, value=None, name='new'),
+ dict(id=3, value=30, name='insert'),
dict(id=None, value=91, name='null-new')],
+ schema=_SCHEMA)
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ with patch('pypaimon.table.data_evolution_merge_into._build_tables',
+ side_effect=AssertionError('Python merge orchestration')):
+ update = builder.new_update()
+ kwargs = {'commit_identifier': 27} if stream else {}
+ messages = update.merge_into(source, on=['id'], when_matched=[
+ WhenMatched.delete('s.value < 0'),
+ WhenMatched.update({'value': source_col('value')},
condition='s.value > t.value'),
+ WhenMatched.update({'value': lit(73)}),
+ ], when_not_matched=[WhenNotMatched('*')], **kwargs)
+ commit = builder.new_commit()
+ commit.commit(messages, 27) if stream else commit.commit(messages)
+ assert sorted(_rows(table), key=lambda row: row['name']) == [
+ dict(id=3, value=30, name='insert'), dict(id=2, value=73, name='keep'),
+ dict(id=None, value=91, name='null-new'), dict(id=None, value=90,
name='null-old')]
+
+
+def test_native_self_merge_and_empty_target_insert(tmp_path):
+ table = _table(tmp_path, [dict(id=1, value=10, name='old'), dict(id=2,
value=20, name='keep')])
+ builder = table.new_batch_write_builder()
+ with patch('pypaimon.table.data_evolution_merge_into._build_tables',
+ side_effect=AssertionError('Python self merge orchestration')):
+ messages = builder.new_update().merge_into(table, on=['_ROW_ID'],
when_matched=[
+ WhenMatched.update({'value': lit(99)}, condition="s.id = 1 AND
t.name = 'old'")])
+ builder.new_commit().commit(messages)
+ assert _rows(table) == [dict(id=1, value=99, name='old'), dict(id=2,
value=20, name='keep')]
+ empty = _table(tmp_path / 'empty')
+ source = pa.Table.from_pylist([dict(id=3, value=30, name='new')],
schema=_SCHEMA)
+ builder = empty.new_batch_write_builder()
+ with patch('pypaimon.table.data_evolution_merge_into._build_tables',
+ side_effect=AssertionError('Python merge orchestration')):
+ messages = builder.new_update().merge_into(source, on=['id'],
when_not_matched=[
+ WhenNotMatched({'value': source_col('value'), 'name': lit('literal
s.id')}, condition='s.value > 10')])
+ builder.new_commit().commit(messages)
+ assert _rows(empty) == [dict(id=3, value=30, name='literal s.id')]
+
+
[email protected]('condition', [
+ 's.value > (SELECT 10)',
+ 's.value IN (SELECT v FROM (VALUES (11), (30)) AS m(v))',
+ 's.value > (SELECT MAX(v) FROM (VALUES (0), (10)) AS m(v))',
+])
+def test_native_merge_preserves_sql_subquery_conditions(tmp_path, condition):
+ table = _table(tmp_path, [dict(id=1, value=10, name='a'), dict(id=2,
value=20, name='b')])
+ source = pa.Table.from_pylist([dict(id=2, value=None, name='b'),
dict(id=1, value=11, name='a'),
+ dict(id=3, value=30, name='c')],
schema=_SCHEMA)
+ builder = table.new_batch_write_builder()
+ with patch('pypaimon.table.data_evolution_merge_into._build_tables',
+ side_effect=AssertionError('Python merge orchestration')):
+ messages = builder.new_update().merge_into(source, on=['id'],
when_matched=[
+ WhenMatched.update({'value': source_col('value')},
condition=condition)],
+ when_not_matched=[WhenNotMatched('*', condition=condition)])
+ builder.new_commit().commit(messages)
+ assert _rows(table) == [dict(id=1, value=11, name='a'), dict(id=2,
value=20, name='b'),
+ dict(id=3, value=30, name='c')]
+
+
+def test_native_merge_renamed_keys_pandas_and_duplicate_targets(tmp_path):
+ import pandas as pd
+
+ table = _table(tmp_path, [dict(id=1, value=10, name='a'), dict(id=1,
value=20, name='b')])
+ source = pd.DataFrame({'key': [1, 3], 'value': [99, 30], 'name':
['updated', 'new']})
+ # Preserve the declared ON type when converting a pandas frame to Arrow.
+ source['key'] = source['key'].astype('int32')
+ source['value'] = source['value'].astype('int32')
+ builder = table.new_batch_write_builder()
+ with patch('pypaimon.table.data_evolution_merge_into._build_tables',
+ side_effect=AssertionError('Python merge orchestration')):
+ messages = builder.new_update().merge_into(source, on={'id': 'key'},
when_matched=[
+ WhenMatched.update({'value': source_col('value')})],
when_not_matched=[WhenNotMatched('*')])
+ builder.new_commit().commit(messages)
+ assert _rows(table) == [dict(id=1, value=99, name='a'), dict(id=1,
value=99, name='b'),
+ dict(id=3, value=30, name='new')]
+
+
+def test_native_merge_defers_literals_until_a_clause_is_selected(tmp_path):
+ table = _table(tmp_path, [dict(id=1, value=10, name='a')])
+ source = pa.Table.from_pylist([dict(id=1, value=11, name='a')],
schema=_SCHEMA)
+ update = table.new_batch_write_builder().new_update()
+ files = set(tmp_path.rglob('*.parquet'))
+ messages = update.merge_into(source, on=['id'], when_matched=[
+ WhenMatched.update({'value': lit('invalid int')}, condition='s.value <
0')])
+ assert messages == []
+ with pytest.raises((ValueError, pa.ArrowInvalid)):
+ update.merge_into(source, on=['id'],
when_matched=[WhenMatched.update({'value': lit('invalid int')})])
+ assert set(tmp_path.rglob('*.parquet')) == files
+ assert _rows(table) == [dict(id=1, value=10, name='a')]
+
+
+def test_native_merge_keeps_video_inserts_on_python_path(tmp_path):
+ from pypaimon.write.native_merge_into import create_native_merge_into
+ from pypaimon.snapshot.snapshot import BATCH_COMMIT_IDENTIFIER
+
+ table = _table(tmp_path).copy({'video-frame-field': 'name'})
+ source = pa.Table.from_pylist([dict(id=1, value=10, name='frame')],
schema=_SCHEMA)
+ assert create_native_merge_into(table, source, ['id'], [],
[WhenNotMatched('*')],
+ 'merge', BATCH_COMMIT_IDENTIFIER) is None
+
+
[email protected]('condition', [None, 's.id = 1', 'EXISTS (SELECT 1
WHERE s.id = 1)'])
+def test_native_self_merge_does_not_read_unreferenced_blob(tmp_path,
condition):
+ from pypaimon.table.row.blob import BlobDescriptor
+
+ schema = pa.schema([('id', pa.int32()), ('name', pa.string()), ('payload',
pa.large_binary())])
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('db', True)
+ catalog.create_table('db.t', Schema.from_pyarrow_schema(schema, options={
+ 'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true',
+ 'blob-descriptor-field': 'payload', 'blob-as-descriptor': 'false',
'write.native.enabled': 'true',
+ }), False)
+ table = catalog.get_table('db.t')
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ descriptor = BlobDescriptor(str(tmp_path / 'unreadable.bin'), 0,
1).serialize()
+ try:
+ writer.write_arrow(pa.Table.from_pylist([dict(id=1, name='before',
payload=descriptor)], schema=schema))
+ builder.new_commit().commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ with patch('pypaimon.table.data_evolution_merge_into._build_tables',
+ side_effect=AssertionError('Python self merge orchestration')):
+ messages = builder.new_update().merge_into(table, on=['_ROW_ID'],
when_matched=[
+ WhenMatched.update({'name': lit('after')}, condition=condition)])
+ builder.new_commit().commit(messages)
+ reader = table.new_read_builder().with_projection(['id', 'name'])
+ assert
reader.new_read().to_arrow(reader.new_scan().plan().splits()).to_pylist() ==
[dict(id=1, name='after')]
diff --git a/paimon-python/pypaimon/write/native_merge_into.py
b/paimon-python/pypaimon/write/native_merge_into.py
new file mode 100644
index 0000000000..1e12185f9e
--- /dev/null
+++ b/paimon-python/pypaimon/write/native_merge_into.py
@@ -0,0 +1,97 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Transport existing MERGE inputs into the core operation."""
+
+import pyarrow as pa
+
+from pypaimon.snapshot.snapshot import BATCH_COMMIT_IDENTIFIER
+from pypaimon.write.native_commit import from_native_commit_messages
+from pypaimon.write.native_update import (
+ _native_row_id_table, _native_update_columns_supported,
_supported_upsert_key_type,
+)
+
+
+def create_native_merge_into(table, source, on, matched, not_matched,
commit_user, commit_identifier):
+ from pypaimon.table.data_evolution_merge_into import _is_self_merge,
_is_table_like, _prepare
+ from pypaimon.ray.data_evolution_merge_into import _normalize_on,
_union_update_cols
+ from pypaimon.ray.data_evolution_merge_transform import SourceColumnRef,
TargetColumnRef, LiteralValue
+ from pypaimon.ray.merge_condition import rewrite_condition
+
+ target_keys, source_keys = _normalize_on(on)
+ self_merge = _is_self_merge(table, source, target_keys, source_keys)
+ if not_matched and table.options.video_frame_fields():
+ return None
+ if _is_table_like(source) and not self_merge:
+ return None
+ native_table = _native_row_id_table(table)
+ if native_table is None:
+ return None
+ source, matched, not_matched, context = _prepare(table, source,
list(matched), list(not_matched), on)
+ if any(callable(value) for clause in matched + not_matched for value in
clause.spec.values()):
+ return None
+ if not _native_update_columns_supported(table,
_union_update_cols(matched)):
+ return None
+ for target, origin in zip(target_keys, source_keys):
+ target_type = pa.int64() if target == '_ROW_ID' else
context.full_pa_schema.field(target).type
+ source_type = target_type if self_merge else
source.schema.field(origin).type
+ if source_type != target_type or not
_supported_upsert_key_type(target_type):
+ return None
+
+ def encode(clause):
+ values = []
+ for name, value in clause.spec.items():
+ if isinstance(value, SourceColumnRef):
+ values.append((name, 'source', value.column))
+ elif isinstance(value, TargetColumnRef):
+ values.append((name, 'target', value.column))
+ elif isinstance(value, LiteralValue):
+ values.append((name, 'literal', value.value))
+ else:
+ return None
+ condition = None if clause.condition is None else
dict(sql=rewrite_condition(clause.condition))
+ return dict(assignments=values, condition=condition,
delete=clause.delete)
+
+ encoded_matched = [encode(clause) for clause in matched]
+ encoded_not_matched = [encode(clause) for clause in not_matched]
+ if any(clause is None for clause in encoded_matched + encoded_not_matched):
+ return None
+ builder =
(native_table.new_batch_write_builder()._with_commit_user(commit_user)
+ if commit_identifier == BATCH_COMMIT_IDENTIFIER else
+
native_table.new_stream_write_builder().with_commit_user(commit_user))
+ writer = builder.new_update()
+ return NativeTableMergeInto(table, writer, source, list(zip(target_keys,
source_keys)),
+ encoded_matched, encoded_not_matched,
commit_identifier)
+
+
+class NativeTableMergeInto:
+ def __init__(self, table, writer, source, on, matched, not_matched,
commit_identifier):
+ self.table = table
+ self.writer = writer
+ self.source = source
+ self.on = on
+ self.matched = matched
+ self.not_matched = not_matched
+ self.commit_identifier = commit_identifier
+
+ def prepare_commit(self):
+ # The core operation owns the snapshot, matching, action selection and
staging.
+ # Never retry via Python after it starts or delete prepared messages
on failure.
+ kwargs = ({} if self.commit_identifier == BATCH_COMMIT_IDENTIFIER else
+ dict(commit_identifier=self.commit_identifier))
+ return from_native_commit_messages(self.table, self.writer.merge_into(
+ self.source, on=self.on, when_matched=self.matched,
when_not_matched=self.not_matched, **kwargs))
