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 e42cd13023 Use Rust core for native Arrow upserts and table updates
(#10185)
e42cd13023 is described below
commit e42cd130237d9a3fd1fb4dee1fce93701b048eea
Author: Jingsong Lee <[email protected]>
AuthorDate: Sat Sep 26 21:59:39 2026 +0800
Use Rust core for native Arrow upserts and table updates (#10185)
---
.github/workflows/ci-python.yml | 7 +-
paimon-python/conftest.py | 43 +-
.../pypaimon/ray/data_evolution_merge_join.py | 10 +-
.../pypaimon/tests/deletion_vector_path_test.py | 5 +-
paimon-python/pypaimon/tests/native_update_test.py | 1024 ++++++++++++++++++++
paimon-python/pypaimon/tests/table_update_test.py | 3 +
paimon-python/pypaimon/write/native_commit.py | 16 +-
paimon-python/pypaimon/write/native_update.py | 304 ++++++
paimon-python/pypaimon/write/native_write.py | 27 +-
paimon-python/pypaimon/write/table_update.py | 73 +-
.../pypaimon/write/table_update_by_row_id.py | 3 +
.../pypaimon/write/table_upsert_by_key.py | 23 +
12 files changed, 1511 insertions(+), 27 deletions(-)
diff --git a/.github/workflows/ci-python.yml b/.github/workflows/ci-python.yml
index 6c756f578e..267e4fd0da 100644
--- a/.github/workflows/ci-python.yml
+++ b/.github/workflows/ci-python.yml
@@ -196,10 +196,10 @@ jobs:
curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s --
-y --default-toolchain stable --profile minimal
echo "$HOME/.cargo/bin" >> $GITHUB_PATH
- - name: Build paimon-rust with REST native commit support
+ - name: Build paimon-rust with native update support
run: |
python -m pip install --upgrade pip
- python -m pip install --upgrade
"git+https://github.com/apache/paimon-rust.git@aa60a6ac7cbe9299aca19effd9ba8301c1bc55da#subdirectory=bindings/python";
+ python -m pip install --upgrade
"git+https://github.com/apache/paimon-rust.git@780bc7957f24a78828b7910369e079bbff30e92c#subdirectory=bindings/python";
- name: Install Python test dependencies
run: |
@@ -227,12 +227,13 @@ jobs:
'commit', source['vcs_info']['commit_id'])
PY
- - name: Run Python tests with native planning, reading, writing and
committing
+ - name: Run Python tests with native planning, reading, writing,
updating and committing
shell: bash
env:
PYPAIMON_TEST_NATIVE_PLAN: '1'
PYPAIMON_TEST_NATIVE_READ: '1'
PYPAIMON_TEST_NATIVE_WRITE: '1'
+ PYPAIMON_TEST_NATIVE_UPDATE: '1'
PYPAIMON_TEST_NATIVE_COMMIT: '1'
run: |
bash paimon-python/dev/lint-python.sh -i pytest
diff --git a/paimon-python/conftest.py b/paimon-python/conftest.py
index 272428f9e9..ef2e3b5ec1 100644
--- a/paimon-python/conftest.py
+++ b/paimon-python/conftest.py
@@ -22,15 +22,18 @@ import pytest
_NATIVE_PLAN_ENV = "PYPAIMON_TEST_NATIVE_PLAN"
_NATIVE_READ_ENV = "PYPAIMON_TEST_NATIVE_READ"
_NATIVE_WRITE_ENV = "PYPAIMON_TEST_NATIVE_WRITE"
+_NATIVE_UPDATE_ENV = "PYPAIMON_TEST_NATIVE_UPDATE"
_NATIVE_COMMIT_ENV = "PYPAIMON_TEST_NATIVE_COMMIT"
_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)
_force_native_for_test = False
_force_native_read_for_test = False
_force_native_write_for_test = False
_force_native_commit_for_test = False
+_force_native_update_for_test = False
def pytest_addoption(parser):
@@ -81,6 +84,10 @@ def _native_commit_enabled():
return os.environ.get(_NATIVE_COMMIT_ENV) == "1"
+def _native_update_enabled():
+ return os.environ.get(_NATIVE_UPDATE_ENV) == "1"
+
+
def pytest_configure(config):
config.addinivalue_line(
"markers", "python_plan: keep Python planner assertions on the Python
lane")
@@ -149,6 +156,29 @@ def pytest_configure(config):
TableCommit._prepare_native_commit = tracked_prepare
+ if _native_update_enabled():
+ from pypaimon.write.native_update import (
+ NativeBatchTableUpdate, NativePredicateTableUpdate,
+ NativeTableUpdateByRowId, NativeTableUpsert,
+ )
+
+ def track_update(cls, method, kind):
+ original = getattr(cls, method)
+
+ def tracked(self, *args, **kwargs):
+ messages = original(self, *args, **kwargs)
+ if messages and _force_native_update_for_test:
+ _native_update_counts[kind] += 1
+ return messages
+
+ setattr(cls, method, tracked)
+
+ track_update(NativeBatchTableUpdate, 'update_by_arrow_with_row_id',
'row_id')
+ track_update(NativeBatchTableUpdate,
'update_by_arrow_batches_with_row_id', 'grouped')
+ track_update(NativePredicateTableUpdate, 'update', 'predicate')
+ track_update(NativeTableUpsert, 'upsert', 'upsert')
+ track_update(NativeTableUpdateByRowId, 'update_columns', 'incremental')
+
def pytest_collection_modifyitems(items):
if _native_plan_enabled():
@@ -163,6 +193,7 @@ def pytest_collection_modifyitems(items):
def enable_native_backends(request, monkeypatch):
global _force_native_for_test, _force_native_read_for_test
global _force_native_write_for_test, _force_native_commit_for_test
+ global _force_native_update_for_test
python_plan = request.node.get_closest_marker("python_plan") is not None
python_read = request.node.get_closest_marker("python_read") is not None
python_write = request.node.get_closest_marker("python_write") is not None
@@ -175,7 +206,8 @@ def enable_native_backends(request, monkeypatch):
and not native_plan_test)
force_write = _native_write_enabled() and not python_write
force_commit = _native_commit_enabled() and not python_commit
- if not (force_plan or force_read or force_write or force_commit):
+ force_update = _native_update_enabled() and not python_write
+ if not (force_plan or force_read or force_write or force_commit or
force_update):
yield
return
@@ -195,7 +227,7 @@ def enable_native_backends(request, monkeypatch):
return original_read(self, True if default is None else default)
monkeypatch.setattr(CoreOptions, "native_read_enabled", read_enabled)
- if force_write:
+ if force_write or force_update:
original_write = CoreOptions.native_write_enabled
def write_enabled(self, default=None):
@@ -213,6 +245,7 @@ def enable_native_backends(request, monkeypatch):
_force_native_read_for_test = force_read
_force_native_write_for_test = force_write
_force_native_commit_for_test = force_commit
+ _force_native_update_for_test = force_update
try:
yield
finally:
@@ -220,6 +253,7 @@ def enable_native_backends(request, monkeypatch):
_force_native_read_for_test = False
_force_native_write_for_test = False
_force_native_commit_for_test = False
+ _force_native_update_for_test = False
def pytest_sessionfinish(session, exitstatus):
@@ -227,6 +261,7 @@ def pytest_sessionfinish(session, exitstatus):
if ((_native_plan_enabled() and _native_plan_count == 0)
or (_native_read_enabled() and _native_read_count == 0)
or (_native_write_enabled() and _native_write_count == 0)
+ or (_native_update_enabled() and not
all(_native_update_counts.values()))
or (_native_commit_enabled() and _native_commit_count == 0)):
session.exitstatus = pytest.ExitCode.TESTS_FAILED
@@ -244,3 +279,7 @@ def pytest_terminal_summary(terminalreporter):
if _native_commit_enabled():
terminalreporter.write_line(
"native commits exercised: %d" % _native_commit_count)
+ if _native_update_enabled():
+ terminalreporter.write_line(
+ "native updates exercised: " + ', '.join(
+ '%s=%d' % item for item in _native_update_counts.items()))
diff --git a/paimon-python/pypaimon/ray/data_evolution_merge_join.py
b/paimon-python/pypaimon/ray/data_evolution_merge_join.py
index b86ac7b363..3cccc3dafb 100644
--- a/paimon-python/pypaimon/ray/data_evolution_merge_join.py
+++ b/paimon-python/pypaimon/ray/data_evolution_merge_join.py
@@ -715,7 +715,8 @@ def distributed_update_apply(
table.copy({CoreOptions.SCAN_SNAPSHOT_ID.key(): str(base_snapshot_id)})
if base_snapshot_id is not None else table
)
- planner =
scan_table.new_batch_write_builder().new_update().new_update_by_row_id()
+ from pypaimon.write.row_id_file_index import RowIdFileIndex
+ planner = RowIdFileIndex.from_table(scan_table)
sorted_first_row_ids = list(planner.first_row_ids)
if not sorted_first_row_ids:
return [], 0, []
@@ -742,7 +743,7 @@ def distributed_update_apply(
# detection covers the read→planner window.
from dataclasses import replace
files_info = replace(
- planner._snapshot_files_info(),
+ planner,
snapshot_id=check_from_snapshot,
)
precomputed_info_ref = ray.put(files_info)
@@ -916,7 +917,8 @@ def distributed_read_by_row_id(
empty_out = _read_output_schema(table, read_cols).empty_table()
# The caller pinned the resolved snapshot, including any retained tag
metadata.
- planner =
table.new_batch_write_builder().new_update().new_update_by_row_id()
+ from pypaimon.write.row_id_file_index import RowIdFileIndex
+ planner = RowIdFileIndex.from_table(table)
sorted_first_row_ids = list(planner.first_row_ids)
if not sorted_first_row_ids:
return None
@@ -929,7 +931,7 @@ def distributed_read_by_row_id(
data_context=data_context,
)
- precomputed_info_ref = ray.put(planner._snapshot_files_info())
+ precomputed_info_ref = ray.put(planner)
frid_col = "_FIRST_ROW_ID"
sorted_arr = np.asarray(sorted_first_row_ids, dtype=np.int64)
valid_ranges = planner.valid_row_id_ranges
diff --git a/paimon-python/pypaimon/tests/deletion_vector_path_test.py
b/paimon-python/pypaimon/tests/deletion_vector_path_test.py
index b1d9d4ce72..23db62775f 100644
--- a/paimon-python/pypaimon/tests/deletion_vector_path_test.py
+++ b/paimon-python/pypaimon/tests/deletion_vector_path_test.py
@@ -151,8 +151,9 @@ def
test_legacy_python_index_directory_remains_readable_and_new_deletes_use_buck
# Old Python writers ignored the option and placed the index under
table/index.
legacy_factory = table.path_factory()
legacy_factory.index_file_in_data_file_dir = False
- with patch.object(table, 'path_factory', return_value=legacy_factory):
- _delete(table, [0, 2])
+ legacy_table = table.copy({'write.native.enabled': 'false'})
+ with patch.object(legacy_table, 'path_factory',
return_value=legacy_factory):
+ _delete(legacy_table, [0, 2])
old_paths = [Path(table.table_path) / 'index' / entry.index_file.file_name
for entry in _entries(table, 2)]
assert all(path.is_file() for path in old_paths)
_read(table, planner, 2, [1, 3])
diff --git a/paimon-python/pypaimon/tests/native_update_test.py
b/paimon-python/pypaimon/tests/native_update_test.py
new file mode 100644
index 0000000000..1c01054aef
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_update_test.py
@@ -0,0 +1,1024 @@
+# 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.
+
+"""End-to-end coverage of the optional native batch row-ID update bridge."""
+
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.write.native_update import create_native_delete,
create_native_update
+from pypaimon.write.table_delete import TableDeleteByRowId
+from pypaimon.read.table_read import TableRead
+from pypaimon.write.table_update import BatchTableUpdate, StreamTableUpdate
+from pypaimon.write.table_update_by_row_id import TableUpdateByRowId
+from pypaimon.write.table_upsert_by_key import TableUpsertByKey
+
+
+pytestmark = pytest.mark.native_plan
+
+
[email protected]('stream', [False, True])
+def
test_incremental_row_id_updater_uses_core_and_accumulates_columns(tmp_path,
stream):
+ from pypaimon.snapshot.snapshot import BATCH_COMMIT_IDENTIFIER
+ from pypaimon.table.row.generic_row import GenericRow
+ from pypaimon.write.native_update import NativeTableUpdateByRowId
+
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ schema = pa.schema([('id', pa.int32()), ('name', pa.string()), ('age',
pa.int32())])
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(schema,
options={
+ 'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true',
+ 'write.native.enabled': 'true',
+ }), False)
+ table = catalog.get_table('default.t')
+ seed = table.new_batch_write_builder()
+ writer = seed.new_write()
+ writer.write_arrow(pa.table({'id': [1, 2], 'name': ['a', 'b'], 'age': [10,
20]}, schema=schema))
+ seed.new_commit().commit(writer.prepare_commit())
+ writer.close()
+ snapshot_id = table.snapshot_manager().get_latest_snapshot().id
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ commit_id = 27 if stream else BATCH_COMMIT_IDENTIFIER
+ # Factory configuration must not fix columns for this incremental API.
+ update = builder.new_update().with_update_type(['id'])
+ with patch.object(TableUpdateByRowId, '_load_existing_files_info',
+ side_effect=AssertionError('Python loaded the file
index')):
+ updater = update.new_update_by_row_id(commit_id) if stream else
update.new_update_by_row_id()
+ assert isinstance(updater, NativeTableUpdateByRowId)
+ assert updater.commit_user == builder.commit_user
+ assert updater.commit_identifier == commit_id
+ assert updater.commit_messages == []
+ first = updater.update_columns(pa.table({'_ROW_ID': [0], 'age':
[11]}), ['age'])
+ messages = updater.update_row_columns(
+ GenericRow(['B'], fields=[table.field_dict['name']]), [1],
['name'])
+ assert len(first) == 1
+ assert len(messages) == len(updater.commit_messages) == 2
+ assert all(message.check_from_snapshot == snapshot_id for message in
messages)
+ staged = set(tmp_path.rglob('*.parquet'))
+ with pytest.raises(ValueError, match='overlapping first_row_ids'):
+ updater.update_columns(pa.table({'_ROW_ID': [1], 'age': [99]}),
['age'])
+ assert set(tmp_path.rglob('*.parquet')) == staged
+ assert len(updater.commit_messages) == 2
+ commit = builder.new_commit()
+ if stream:
+ commit.commit(updater.commit_messages, commit_id)
+ else:
+ commit.commit(updater.commit_messages)
+ commit.close()
+ assert table.snapshot_manager().get_latest_snapshot().commit_identifier ==
commit_id
+ read = table.new_read_builder()
+ actual =
read.new_read().to_arrow(read.new_scan().plan().splits()).sort_by('id')
+ assert actual.to_pydict() == {'id': [1, 2], 'name': ['a', 'B'], 'age':
[11, 20]}
+
+
[email protected]('stream', [False, True])
+def test_partitioned_upsert_and_incremental_update_use_core(tmp_path, stream):
+ from pypaimon.write.native_update import NativeTableUpdateByRowId
+ from pypaimon.table.row.generic_row import GenericRow
+
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ schema = pa.schema([('p', pa.string()), ('id', pa.int32()), ('v',
pa.int32())])
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(schema,
partition_keys=['p'], options={
+ 'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true',
+ 'write.native.enabled': 'true',
+ }), False)
+ table = catalog.get_table('default.t')
+ seed = table.new_batch_write_builder()
+ writer = seed.new_write()
+ writer.write_arrow(pa.table({'p': ['a', 'b', None], 'id': [1, 1, 1], 'v':
[10, 20, 30]}, schema=schema))
+ seed.new_commit().commit(writer.prepare_commit())
+ writer.close()
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ update = builder.new_update()
+ input_rows = pa.table({'p': ['a', 'c', None], 'id': [1, 1, 1], 'v': [11,
40, 31]}, schema=schema)
+ with patch.object(TableUpsertByKey, '_upsert_partition',
+ side_effect=AssertionError('Python partition upsert
selected')):
+ messages = (update.upsert_by_arrow_with_key(input_rows, ['id'], 1) if
stream
+ else update.upsert_by_arrow_with_key(input_rows, ['id']))
+ commit = builder.new_commit()
+ commit.commit(messages, 1) if stream else commit.commit(messages)
+ commit.close()
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ update = builder.new_update()
+ with patch.object(TableUpsertByKey, '_upsert_row_partition',
+ side_effect=AssertionError('Python partition row upsert
selected')):
+ rows = [GenericRow(['a', 1, 12], table.fields)]
+ messages = (update.upsert_by_key(rows, ['id'], 2) if stream
+ else update.upsert_by_key(rows, ['id']))
+ commit = builder.new_commit()
+ commit.commit(messages, 2) if stream else commit.commit(messages)
+ commit.close()
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ update = builder.new_update()
+ read = table.new_read_builder().with_projection(['p', 'id', 'v',
'_ROW_ID'])
+ current = read.new_read().to_arrow(read.new_scan().plan().splits())
+ selected = current.filter(pa.compute.equal(current['p'], 'a'))
+ updater = update.new_update_by_row_id(3) if stream else
update.new_update_by_row_id()
+ assert isinstance(updater, NativeTableUpdateByRowId)
+ messages = updater.update_columns(selected.select(['p', '_ROW_ID']), ['p'])
+ commit = builder.new_commit()
+ commit.commit(messages, 3) if stream else commit.commit(messages)
+ commit.close()
+ read = table.new_read_builder()
+ actual =
read.new_read().to_arrow(read.new_scan().plan().splits()).sort_by('p')
+ assert actual.to_pydict() == {'p': ['a', 'b', 'c', None], 'id': [1, 1, 1,
1], 'v': [12, 20, 40, 31]}
+
+
+def test_batch_row_id_update_uses_rust_and_python_commit(tmp_path):
+
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ schema = pa.schema([
+ ('id', pa.int32()), ('name', pa.string()), ('age', pa.int32()),
+ ])
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(schema,
options={
+ 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true',
+ 'write.native.enabled': 'true',
+ }), False)
+ table = catalog.get_table('default.t')
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ writer.write_arrow(pa.Table.from_pydict({
+ 'id': [1, 2, 3], 'name': ['a', 'b', 'c'], 'age': [10, 20, 30],
+ }, schema=schema))
+ builder.new_commit().commit(writer.prepare_commit())
+ writer.close()
+
+ pinned = table.copy({
+ 'scan.snapshot-id':
str(table.snapshot_manager().get_latest_snapshot().id),
+ })
+ assert create_native_update(pinned, builder.commit_user, ['name']) is None
+
+ update_builder = table.new_batch_write_builder()
+ update = update_builder.new_update().with_update_type(['name', 'age'])
+ changed = pa.Table.from_batches([
+ pa.record_batch([
+ pa.array([2], type=pa.int64()), pa.array(['C']),
+ pa.array([31], type=pa.int32()),
+ ], names=['_ROW_ID', 'name', 'age']),
+ pa.record_batch([
+ pa.array([0], type=pa.int64()), pa.array(['A']),
+ pa.array([11], type=pa.int32()),
+ ], names=['_ROW_ID', 'name', 'age']),
+ ])
+ with patch.object(TableUpdateByRowId, 'update_columns',
+ side_effect=AssertionError('Python update was
selected')):
+ messages = update.update_by_arrow_with_row_id(changed)
+ assert messages
+ assert all(file.file_path and table.file_io.exists(file.file_path)
+ for message in messages for file in message.new_files)
+ update_builder.new_commit().commit(messages)
+
+ read_builder = table.new_read_builder()
+ actual = read_builder.new_read().to_arrow(
+ read_builder.new_scan().plan().splits()).sort_by('id')
+ assert actual.select(['id', 'name', 'age']).to_pydict() == {
+ 'id': [1, 2, 3], 'name': ['A', 'b', 'C'], 'age': [11, 20, 31],
+ }
+
+
+def test_batch_row_id_delete_uses_rust_deletion_vectors(tmp_path):
+
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ schema = pa.schema([('id', pa.int32()), ('name', pa.string())])
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(schema,
options={
+ 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true',
+ 'deletion-vectors.enabled': 'true',
+ 'write.native.enabled': 'true',
+ }), False)
+ table = catalog.get_table('default.t')
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ writer.write_arrow(pa.Table.from_pydict({
+ 'id': [1, 2, 3], 'name': ['a', 'b', 'c'],
+ }, schema=schema))
+ builder.new_commit().commit(writer.prepare_commit())
+ writer.close()
+
+ pinned = table.copy({
+ 'scan.snapshot-id':
str(table.snapshot_manager().get_latest_snapshot().id),
+ })
+ assert create_native_delete(pinned, builder.commit_user) is None
+
+ with patch.object(TableDeleteByRowId, 'delete',
+ side_effect=AssertionError('Python delete was
selected')):
+ delete_builder = table.new_batch_write_builder()
+ messages = delete_builder.new_update().delete_by_row_id([0, 2, 2])
+ assert messages and sum(len(message.index_adds) for message in
messages) == 1
+ delete_builder.new_commit().commit(messages)
+
+ predicate_builder = table.new_read_builder().new_predicate_builder()
+ delete_builder = table.new_batch_write_builder()
+ with patch.object(TableRead, 'to_arrow',
+ side_effect=AssertionError('Python predicate match
selected')):
+ messages = delete_builder.new_update().delete_by_predicate(
+ predicate_builder.equal('id', 2))
+ assert messages
+ delete_builder.new_commit().commit(messages)
+
+ read_builder = table.new_read_builder()
+ actual =
read_builder.new_read().to_arrow(read_builder.new_scan().plan().splits())
+ assert actual.num_rows == 0
+
+
+def test_stream_update_and_delete_use_native_writers(tmp_path):
+
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ schema = pa.schema([('id', pa.int32()), ('age', pa.int32())])
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(schema,
options={
+ 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true',
+ 'deletion-vectors.enabled': 'true',
+ 'write.native.enabled': 'true',
+ }), False)
+ table = catalog.get_table('default.t')
+ initial = table.new_batch_write_builder()
+ writer = initial.new_write()
+ writer.write_arrow(pa.Table.from_pydict({
+ 'id': [1, 2, 3, 4], 'age': [10, 20, 30, 40],
+ }, schema=schema))
+ initial.new_commit().commit(writer.prepare_commit())
+ writer.close()
+
+ stream = table.new_stream_write_builder()
+ committer = stream.new_commit()
+ update = stream.new_update()
+ with patch.object(TableUpdateByRowId, 'update_columns',
+ side_effect=AssertionError('Python row-ID update
selected')):
+ messages = update.update_by_arrow_with_row_id(
+ pa.Table.from_pydict({'_ROW_ID': [0], 'age': [11]}), 10)
+ committer.commit(messages, 10)
+
+ predicate = table.new_read_builder().new_predicate_builder()
+ with patch.object(StreamTableUpdate, '_build_predicate_update_table',
+ side_effect=AssertionError('Python assignments
selected')), \
+ patch.object(TableRead, 'to_arrow',
+ side_effect=AssertionError('Python predicate read
selected')):
+ messages = update.update_by_predicate(
+ predicate.equal('id', 2), {'age': 22}, 20)
+ committer.commit(messages, 20)
+
+ with patch.object(TableDeleteByRowId, 'delete',
+ side_effect=AssertionError('Python row-ID delete
selected')):
+ messages = update.delete_by_row_id([2], 30)
+ committer.commit(messages, 30)
+ with patch.object(TableRead, 'to_arrow',
+ side_effect=AssertionError('Python predicate read
selected')):
+ messages = update.delete_by_predicate(predicate.equal('id', 4), 40)
+ committer.commit(messages, 40)
+ committer.close()
+
+ read = table.new_read_builder()
+ actual =
read.new_read().to_arrow(read.new_scan().plan().splits()).sort_by('id')
+ assert actual.select(['id', 'age']).to_pydict() == {
+ 'id': [1, 2], 'age': [11, 22],
+ }
+ assert table.snapshot_manager().get_latest_snapshot().commit_identifier ==
40
+
+
+def test_native_batch_update_preserves_input_table_boundaries(tmp_path):
+
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ schema = pa.schema([('id', pa.int32()), ('age', pa.int32())])
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(schema,
options={
+ 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true',
+ 'write.native.enabled': 'true',
+ }), False)
+ table = catalog.get_table('default.t')
+ for ids in ([1, 2], [3, 4]):
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ writer.write_arrow(pa.Table.from_pydict({
+ 'id': ids, 'age': [value * 10 for value in ids],
+ }, schema=schema))
+ builder.new_commit().commit(writer.prepare_commit())
+ writer.close()
+
+ first = pa.Table.from_batches([
+ pa.record_batch([pa.array([0], type=pa.int64()), pa.array([11])],
+ names=['_ROW_ID', 'age']),
+ pa.record_batch([pa.array([1], type=pa.int64()), pa.array([22])],
+ names=['_ROW_ID', 'age']),
+ ])
+ with patch.object(BatchTableUpdate, '_update_by_arrow_batches_with_row_id',
+ side_effect=AssertionError('Python batch update
selected')):
+ with pytest.raises(ValueError, match='overlapping first_row_ids.*0'):
+ table.new_batch_write_builder().new_update().with_update_type(
+ ['age']).update_by_arrow_batches_with_row_id(iter([
+ first.slice(0, 1),
+ pa.Table.from_pydict({'_ROW_ID': [1], 'age': [23]}),
+ ]))
+
+ builder = table.new_batch_write_builder()
+ messages =
builder.new_update().update_by_arrow_batches_with_row_id(iter([
+ first,
+ pa.Table.from_pydict({'_ROW_ID': [2], 'age': [33]}),
+ pa.Table.from_pydict({'_ROW_ID': [0], 'id': [10]}),
+ ]))
+ builder.new_commit().commit(messages)
+ read = table.new_read_builder()
+ actual =
read.new_read().to_arrow(read.new_scan().plan().splits()).sort_by('id')
+ assert actual.select(['id', 'age']).to_pydict() == {
+ 'id': [2, 3, 4, 10], 'age': [22, 33, 40, 11],
+ }
+
+ before = set(tmp_path.rglob('*.parquet'))
+
+ def failed_input():
+ yield first
+ raise RuntimeError('input generator failed')
+
+ with pytest.raises(RuntimeError, match='input generator failed'):
+
table.new_batch_write_builder().new_update().update_by_arrow_batches_with_row_id(
+ failed_input())
+ assert set(tmp_path.rglob('*.parquet')) == before
+
+ read_snapshot_id = table.snapshot_manager().get_latest_snapshot().id
+
+ def interleaved_tables():
+ yield pa.Table.from_pydict({'_ROW_ID': [0], 'age': [12]})
+ concurrent = table.new_batch_write_builder()
+ changed = concurrent.new_update().with_update_type(['age'])
+ concurrent.new_commit().commit(changed.update_by_arrow_with_row_id(
+ pa.Table.from_pydict({'_ROW_ID': [1], 'age': [23]})))
+
+ with patch.object(BatchTableUpdate, '_update_by_arrow_batches_with_row_id',
+ side_effect=AssertionError('Python batch update
selected')):
+ staged = (table.new_batch_write_builder().new_update()
+ .with_update_type(['age'])
+ .update_by_arrow_batches_with_row_id(interleaved_tables()))
+ assert staged and all(message.check_from_snapshot == read_snapshot_id
+ for message in staged)
+ from pypaimon.write.file_store_commit import _abort_commit_messages
+ _abort_commit_messages(table, staged)
+
+
+def test_native_predicate_update_invokes_callable_by_file_group(tmp_path):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ schema = pa.schema([('id', pa.int32()), ('age', pa.int32())])
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(schema,
options={
+ 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true',
+ 'write.native.enabled': 'true',
+ }), False)
+ table = catalog.get_table('default.t')
+ for ids in ([1, 2], [3, 4]):
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ writer.write_arrow(pa.Table.from_pydict({
+ 'id': ids, 'age': [value * 10 for value in ids],
+ }, schema=schema))
+ builder.new_commit().commit(writer.prepare_commit())
+ writer.close()
+
+ seen = []
+ predicate =
table.new_read_builder().new_predicate_builder().greater_or_equal('id', 2)
+ builder = table.new_batch_write_builder()
+ with patch.object(BatchTableUpdate, '_build_predicate_update_table',
+ side_effect=AssertionError('Python assignments
selected')), \
+ patch.object(BatchTableUpdate, '_matched_update_scan_table',
+ side_effect=AssertionError('Python scan planning
selected')):
+ messages = builder.new_update().update_by_predicate(
+ predicate,
+ {'age': lambda matched: (
+ seen.append(matched.num_rows) or
+ pa.compute.add(matched['age'], 1)
+ )},
+ read_columns=['age'],
+ )
+ assert sorted(seen) == [1, 2]
+ builder.new_commit().commit(messages)
+ read_builder = table.new_read_builder()
+ actual = read_builder.new_read().to_arrow(
+ read_builder.new_scan().plan().splits()).sort_by('id')
+ assert actual.select(['id', 'age']).to_pydict() == {
+ 'id': [1, 2, 3, 4], 'age': [10, 21, 31, 41],
+ }
+
+
+def test_native_upsert_matches_duplicate_source_and_target_keys(tmp_path):
+ from pypaimon.table.row.generic_row import GenericRow
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ schema = pa.schema([('id', pa.int32()), ('age', pa.int32())])
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(schema,
options={
+ 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true',
+ 'write.native.enabled': 'true',
+ }), False)
+ table = catalog.get_table('default.t')
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ writer.write_arrow(pa.Table.from_pydict({
+ 'id': [1, 1, 2], 'age': [10, 11, 20],
+ }, schema=schema))
+ builder.new_commit().commit(writer.prepare_commit())
+ writer.close()
+
+ updates = pa.Table.from_pydict({
+ 'id': [1, 1, 3], 'age': [100, 101, 30],
+ }, schema=schema)
+ builder = table.new_batch_write_builder()
+ with patch.object(TableUpsertByKey, '_upsert_partition',
+ side_effect=AssertionError('Python upsert selected')):
+ messages = builder.new_update().upsert_by_arrow_with_key(updates,
['id'])
+ builder.new_commit().commit(messages)
+ builder = table.new_batch_write_builder()
+ with patch.object(TableUpsertByKey, '_upsert_row_partition',
+ side_effect=AssertionError('Python row upsert
selected')):
+ messages = builder.new_update().upsert_by_key([
+ GenericRow([1, 201], table.fields),
+ GenericRow([4, 40], table.fields),
+ ], ['id'])
+ builder.new_commit().commit(messages)
+ read_builder = table.new_read_builder()
+ actual = read_builder.new_read().to_arrow(
+ read_builder.new_scan().plan().splits()).sort_by([('id', 'ascending'),
+ ('age',
'ascending')])
+ assert actual.select(['id', 'age']).to_pydict() == {
+ 'id': [1, 1, 2, 3, 4], 'age': [201, 201, 20, 30, 40],
+ }
+
+ stream = table.new_stream_write_builder()
+ stream_updates = pa.Table.from_pydict({
+ 'id': [2, 5], 'age': [25, 50],
+ }, schema=schema)
+ with patch.object(TableUpsertByKey, '_upsert_partition',
+ side_effect=AssertionError('Python stream selected')):
+ messages = stream.new_update().upsert_by_arrow_with_key(
+ stream_updates, ['id'], 77)
+ stream.new_commit().commit(messages, 77)
+ read_builder = table.new_read_builder()
+ actual = read_builder.new_read().to_arrow(
+ read_builder.new_scan().plan().splits())
+ actual = actual.sort_by([('id', 'ascending'), ('age', 'ascending')])
+ assert actual.select(['id', 'age']).to_pydict() == {
+ 'id': [1, 1, 2, 3, 4, 5],
+ 'age': [201, 201, 25, 30, 40, 50],
+ }
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ assert snapshot.commit_identifier == 77
+
+ # If the core operation is unavailable, fall back for the whole upsert.
+ # No private matcher can leave Python coordinating native result indices.
+ builder = table.new_batch_write_builder()
+ fallback_updates = pa.Table.from_pydict({
+ 'id': [2, 6], 'age': [26, 60],
+ }, schema=schema)
+ with patch('pypaimon.write.native_update.create_native_upsert',
return_value=None), \
+ patch('pypaimon.write.native_update.create_native_update',
return_value=None):
+ messages =
builder.new_update().upsert_by_arrow_with_key(fallback_updates, ['id'])
+ builder.new_commit().commit(messages)
+ read_builder = table.new_read_builder()
+ actual = read_builder.new_read().to_arrow(
+ read_builder.new_scan().plan().splits())
+ actual = actual.sort_by([('id', 'ascending'), ('age', 'ascending')])
+ assert actual.select(['id', 'age']).to_pydict() == {
+ 'id': [1, 1, 2, 3, 4, 5, 6],
+ 'age': [201, 201, 26, 30, 40, 50, 60],
+ }
+
+ from pypaimon.write.native_update import NativeTableUpsert
+ before = set(tmp_path.rglob('*.parquet'))
+ snapshot_id = table.snapshot_manager().get_latest_snapshot().id
+ with patch.object(NativeTableUpsert, 'upsert',
side_effect=RuntimeError('core upsert failed')), \
+ patch.object(TableUpsertByKey, '_upsert_partition',
+ side_effect=AssertionError('retried as Python Arrow
upsert')), \
+ patch.object(TableUpsertByKey, '_upsert_row_partition',
+ side_effect=AssertionError('retried as Python row
upsert')):
+ with pytest.raises(RuntimeError, match='core upsert failed'):
+ builder.new_update().upsert_by_arrow_with_key(fallback_updates,
['id'])
+ with pytest.raises(RuntimeError, match='core upsert failed'):
+ builder.new_update().upsert_by_key([GenericRow([2, 27],
table.fields)], ['id'])
+ assert table.snapshot_manager().get_latest_snapshot().id == snapshot_id
+ assert set(tmp_path.rglob('*.parquet')) == before
+
+
[email protected]('native', [False, True])
[email protected]('values', [
+ pa.array(['bad']), pa.array([2147483648], type=pa.int64()),
+])
+def test_row_id_cast_failure_does_not_write_nulls(tmp_path, native, values):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ schema = pa.schema([('id', pa.int32()), ('value', pa.int32())])
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(schema,
options={
+ 'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true',
+ 'write.native.enabled': str(native).lower(),
+ }), False)
+ table = catalog.get_table('default.t')
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ writer.write_arrow(pa.Table.from_pydict({'id': [1], 'value': [10]},
schema=schema))
+ builder.new_commit().commit(writer.prepare_commit())
+ writer.close()
+ before = set(tmp_path.rglob('*.parquet'))
+ update = pa.Table.from_arrays([pa.array([0], type=pa.int64()), values],
+ names=['_ROW_ID', 'value'])
+ with pytest.raises((ValueError, pa.ArrowInvalid)):
+
table.new_batch_write_builder().new_update().update_by_arrow_with_row_id(update)
+ assert set(tmp_path.rglob('*.parquet')) == before
+ read = table.new_read_builder()
+ actual = read.new_read().to_arrow(read.new_scan().plan().splits())
+ assert actual.select(['id', 'value']).to_pydict() == {'id': [1], 'value':
[10]}
+
+
[email protected]('native', [False, True])
[email protected]('values,target', [
+ (pa.array([1.0]), pa.string()),
+ (pa.array([1e-6], type=pa.float32()), pa.string()),
+ (pa.array([-0.0]), pa.int32()),
+ (pa.array([1.234]), pa.decimal128(10, 2)),
+ (pa.array(['-1000e-5']), pa.decimal128(10, 2)),
+ (pa.array(['1.234']), pa.decimal128(10, 2)),
+ (pa.array([16777218], type=pa.int64()), pa.float32()),
+ (pa.array([1], type=pa.timestamp('us')), pa.string()),
+ (pa.array([1234], type=pa.timestamp('us')), pa.timestamp('ms')),
+ (pa.array([1234], type=pa.timestamp('us')), pa.date32()),
+ (pa.array([1.5]).dictionary_encode(), pa.int32()),
+ (pa.array(['1.234']).dictionary_encode(), pa.decimal128(10, 2)),
+ (pa.array([1.0]).dictionary_encode(), pa.string()),
+ (pa.array([[1.5]]), pa.list_(pa.int32())),
+ (pa.array([[1.0, None]]), pa.list_(pa.int32())),
+ (pa.array([1234567], type=pa.timestamp('ns')), pa.time32('ms')),
+ (pa.array([86400001000000], type=pa.timestamp('ns')), pa.time32('ms')),
+ (pa.array([-1000000], type=pa.timestamp('ns')), pa.time32('ms')),
+ (pa.array(['yes']), pa.bool_()),
+ (pa.array(['f']), pa.bool_()),
+ (pa.array(['TRUE']), pa.bool_()),
+ (pa.array(['1970-01-01 00:00:00.123456']), pa.timestamp('ms')),
+ (pa.array(['1970-01-01 00:00:00.0000']), pa.timestamp('ms')),
+ (pa.array(['1970-01-01 00:00:00.123']), pa.timestamp('ms')),
+ (pa.array(['1970-01-01T01']), pa.timestamp('ms')),
+ (pa.array(['1970-01-01 00:00:00Z']), pa.timestamp('ms')),
+ (pa.array([0], type=pa.date64()), pa.string()),
+ (pa.array([0], type=pa.timestamp('us', 'Asia/Shanghai')),
pa.timestamp('us')),
+ (pa.array([-1000], type=pa.timestamp('us', 'Asia/Shanghai')),
pa.timestamp('ms')),
+ (pa.array([-1001], type=pa.timestamp('us', 'Asia/Shanghai')),
pa.timestamp('ms')),
+ (pa.array([42], type=pa.int64()), pa.binary()),
+ (pa.array([None], type=pa.int64()), pa.binary()),
+ (pa.array([b'1.25']), pa.float64()),
+ (pa.array([b'1.25'], type=pa.large_binary()), pa.float64()),
+ (pa.array([b'1.25']), pa.decimal128(6, 2)),
+ (pa.array([b'1.234']), pa.decimal128(6, 2)),
+ (pa.array([b'0042']), pa.int32()),
+ (pa.array([b'0x2a']), pa.int32()),
+ (pa.array([b'0Xffffffff'], type=pa.large_binary()), pa.int32()),
+ (pa.array([b'+42']), pa.int32()),
+ (pa.array(['0x2a']), pa.int32()),
+ (pa.array(['+42']), pa.int32()),
+ (pa.array([b'true']), pa.bool_()),
+ (pa.array([b'yes']), pa.bool_()),
+ (pa.array(['00:00:01.234567']), pa.time32('ms')),
+ (pa.array(['2024-01-02 12:34:56']), pa.date32()),
+ (pa.array(['2024-01-02']), pa.date32()),
+ (pa.array([1.5]), pa.timestamp('ms')),
+ (pa.array([1000000], type=pa.timestamp('us')), pa.float64()),
+ (pa.array([1.5]), pa.date32()),
+ (pa.array([1.5]), pa.time32('ms')),
+ (pa.array([1], type=pa.int32()), pa.timestamp('ms')),
+ (pa.array([2.675]), pa.decimal128(10, 2)),
+ (pa.array([2.25], type=pa.float32()), pa.decimal128(10, 1)),
+ (pa.array([0.1]), pa.decimal128(38, 20)),
+])
+def test_native_assignment_cast_matches_pyarrow(tmp_path, native, values,
target):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ schema = pa.schema([('id', pa.int32()), ('value', target)])
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(schema,
options={
+ 'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true',
+ 'write.native.enabled': str(native).lower(),
+ }), False)
+ table = catalog.get_table('default.t')
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ writer.write_arrow(pa.Table.from_pydict({'id': [1], 'value': [None]},
schema=schema))
+ builder.new_commit().commit(writer.prepare_commit())
+ writer.close()
+ before = set(tmp_path.rglob('*.parquet'))
+ update = table.new_batch_write_builder()
+ try:
+ expected = values.cast(target).to_pylist()
+ except pa.ArrowException:
+ with pytest.raises((ValueError, pa.ArrowException)):
+ update.new_update().update_by_predicate(None, {'value': values})
+ assert set(tmp_path.rglob('*.parquet')) == before
+ expected = [None]
+ else:
+ messages = update.new_update().update_by_predicate(None, {'value':
values})
+ update.new_commit().commit(messages)
+ read = table.new_read_builder()
+ actual = read.new_read().to_arrow(read.new_scan().plan().splits())
+ assert actual['value'].to_pylist() == expected
+
+
[email protected]('native', [False, True])
[email protected]('callable_assignment', [False, True])
+def test_predicate_assignment_order_with_later_partial_file(
+ tmp_path, native, callable_assignment):
+ from contextlib import nullcontext
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ schema = pa.schema([('id', pa.int32()), ('age', pa.int32())])
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(schema,
options={
+ 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true',
+ 'write.native.enabled': str(native).lower(),
+ }), False)
+ table = catalog.get_table('default.t')
+ for ids in ([1, 2], [3, 4]):
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ writer.write_arrow(pa.Table.from_pydict({
+ 'id': ids, 'age': [value * 10 for value in ids],
+ }, schema=schema))
+ builder.new_commit().commit(writer.prepare_commit())
+ writer.close()
+ builder = table.new_batch_write_builder()
+ messages = builder.new_update().update_by_arrow_with_row_id(
+ pa.table({'_ROW_ID': [2], 'age': [33]}))
+ builder.new_commit().commit(messages)
+
+ seen = []
+
+ def assign(matched):
+ seen.append(matched['id'].to_pylist())
+ return pa.compute.add(matched['age'], 1)
+
+ assignment = assign if callable_assignment else pa.chunked_array(
+ [[101], [102, 103, 104]], type=pa.int32())
+ guard = (patch.object(BatchTableUpdate, '_matched_update_scan_table',
+ side_effect=AssertionError('Python planning
selected'))
+ if native else nullcontext())
+ with guard:
+ messages = builder.new_update().update_by_predicate(
+ None, {'age': assignment},
+ read_columns=['id', 'age'] if callable_assignment else None)
+ builder.new_commit().commit(messages)
+ if callable_assignment:
+ assert seen == [[1, 2], [3, 4]]
+ read = table.new_read_builder()
+ actual =
read.new_read().to_arrow(read.new_scan().plan().splits()).sort_by('id')
+ assert actual.select(['id', 'age']).to_pydict() == {
+ 'id': [1, 2, 3, 4],
+ 'age': [11, 21, 34, 41] if callable_assignment else [101, 102, 103,
104],
+ }
+
+
[email protected]('stream', [False, True])
+def
test_native_row_upsert_uses_public_operation_with_composite_null_keys(tmp_path,
stream):
+ from pypaimon.table.row.generic_row import GenericRow
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ schema = pa.schema([
+ ('id', pa.int32()), ('part', pa.string()), ('age', pa.int32()),
('keep', pa.string()),
+ ])
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(schema,
options={
+ 'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true',
+ 'write.native.enabled': 'true',
+ }), False)
+ table = catalog.get_table('default.t')
+ seed = table.new_batch_write_builder()
+ writer = seed.new_write()
+ writer.write_arrow(pa.Table.from_pydict({
+ 'id': [1, 1, None, 2], 'part': ['a', 'a', 'n', 'b'],
+ 'age': [10, 11, 20, 30], 'keep': ['old'] * 4,
+ }, schema=schema))
+ seed.new_commit().commit(writer.prepare_commit())
+ writer.close()
+ fields = list(reversed(table.fields))
+ rows = (GenericRow(values, fields) for values in [
+ ['new', 101, 'a', 1], ['new', 102, 'a', 1],
+ ['new', None, 'n', None], ['new', 40, 'c', 3],
+ ])
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ update = builder.new_update().with_update_type(['age'])
+ with patch.object(TableUpsertByKey, '_upsert_row_partition',
+ side_effect=AssertionError('Python upsert selected')):
+ if stream:
+ messages = update.upsert_by_key(rows, ['id', 'part'], 88)
+ builder.new_commit().commit(messages, 88)
+ else:
+ messages = update.upsert_by_key(rows, ['id', 'part'])
+ builder.new_commit().commit(messages)
+ read = table.new_read_builder()
+ actual =
read.new_read().to_arrow(read.new_scan().plan().splits()).sort_by('id')
+ assert actual.to_pydict() == {
+ 'id': [1, 1, 2, 3, None], 'part': ['a', 'a', 'b', 'c', 'n'],
+ 'age': [102, 102, 30, 40, None], 'keep': ['old', 'old', 'old', 'new',
'old'],
+ }
+
+
[email protected]('case', ['partial', 'float-key', 'empty-columns'])
+def test_row_upsert_unsupported_inputs_keep_python_semantics(tmp_path, case):
+ from pypaimon.table.row.generic_row import GenericRow
+ from pypaimon.write.native_update import NativeTableUpsert
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ schema = pa.schema([
+ ('id', pa.float64() if case == 'float-key' else pa.int32()),
+ ('age', pa.int32()), ('region', pa.string()),
+ ])
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ schema, options={
+ 'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true',
+ 'write.native.enabled': 'true',
+ }), False)
+ table = catalog.get_table('default.t')
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ writer.write_arrow(pa.Table.from_pydict({
+ 'id': [1, 2], 'age': [10, 20], 'region': ['east', 'west'],
+ }, schema=schema))
+ builder.new_commit().commit(writer.prepare_commit())
+ writer.close()
+ if case == 'partial':
+ rows = [GenericRow([1, 11], table.fields[:2])]
+ else:
+ rows = [GenericRow(values, table.fields) for values in [
+ [1, 11, 'east'], [3, 30, 'east'],
+ ]]
+ if case == 'empty-columns':
+ before = set(tmp_path.rglob('*.parquet'))
+ with patch.object(NativeTableUpsert, 'upsert',
+ side_effect=AssertionError('Empty row update became
all columns')), \
+ pytest.raises(ValueError, match='column_names cannot be
empty'):
+ builder.new_update().with_update_type([]).upsert_by_key(rows,
['id'])
+ assert set(tmp_path.rglob('*.parquet')) == before
+ read = table.new_read_builder()
+ actual =
read.new_read().to_arrow(read.new_scan().plan().splits()).sort_by('id')
+ assert actual.to_pydict() == {
+ 'id': [1, 2], 'age': [10, 20], 'region': ['east', 'west'],
+ }
+ return
+ with patch.object(NativeTableUpsert, 'upsert',
+ side_effect=AssertionError('Unsupported native upsert
selected')):
+ messages =
builder.new_update().with_update_type(['age']).upsert_by_key(rows, ['id'])
+ builder.new_commit().commit(messages)
+ read = table.new_read_builder()
+ actual =
read.new_read().to_arrow(read.new_scan().plan().splits()).sort_by('id')
+ assert actual.to_pydict() == {
+ 'id': [1, 2] if case == 'partial' else [1, 2, 3],
+ 'age': [11, 20] if case == 'partial' else [11, 20, 30],
+ 'region': ['east', 'west'] if case == 'partial' else ['east', 'west',
'east'],
+ }
+
+
[email protected]('native', [False, True])
[email protected]('grouped', [False, True])
[email protected]('empty_chunks', [False, True])
[email protected]('values,target,row_id_type', [
+ (pa.array([99], type=pa.int32()), pa.int32(), pa.int32()),
+ (pa.array([99], type=pa.int32()), pa.int32(), pa.uint64()),
+ (pa.array([99], type=pa.int32()), pa.int32(),
+ pa.dictionary(pa.int8(), pa.int32())),
+ (pa.array([1.0]), pa.string(), pa.int64()),
+ (pa.array([0], type=pa.timestamp('us')), pa.string(), pa.int64()),
+ (pa.array(['yes']), pa.bool_(), pa.int64()),
+ (pa.array([1.5]), pa.int32(), pa.int64()),
+ (pa.array([2 ** 63 - 1], type=pa.int64()), pa.float64(), pa.int64()),
+ (pa.array([-1234567], type=pa.timestamp('ns')), pa.timestamp('ms'),
pa.int64()),
+ (pa.array([1234567], type=pa.timestamp('ns')), pa.time32('ms'),
pa.int64()),
+ (pa.array([99], type=pa.int32()), pa.int32(), pa.uint32()),
+ (pa.array(['1970-01-01 00:00:00.123456']), pa.timestamp('ms'), pa.int32()),
+ (pa.array(['1970-01-01 00:00:00.123']), pa.timestamp('ms'), pa.int32()),
+ (pa.array([0], type=pa.date64()), pa.string(), pa.int32()),
+ (pa.array([0], type=pa.timestamp('us', 'Asia/Shanghai')),
pa.timestamp('us'), pa.int32()),
+ (pa.array([-1001], type=pa.timestamp('us', 'Asia/Shanghai')),
pa.timestamp('ms'), pa.int32()),
+ (pa.array([42], type=pa.int64()), pa.binary(), pa.int32()),
+ (pa.array([None], type=pa.int64()), pa.binary(), pa.int32()),
+ (pa.array([b'1.25']), pa.float64(), pa.int32()),
+ (pa.array([b'1.25'], type=pa.large_binary()), pa.float64(), pa.int32()),
+ (pa.array([b'1.234']), pa.decimal128(6, 2), pa.int32()),
+ (pa.array([b'0x2a']), pa.int32(), pa.int32()),
+ (pa.array([b'+42']), pa.int32(), pa.int32()),
+ (pa.array(['00:00:01.234567']), pa.time32('ms'), pa.int64()),
+ (pa.array(['2024-01-02 12:34:56']), pa.date32(), pa.int64()),
+ (pa.array(['2024-01-02']), pa.date32(), pa.int64()),
+ (pa.array([1.5]), pa.timestamp('ms'), pa.int64()),
+ (pa.array([1000000], type=pa.timestamp('us')), pa.float64(), pa.int64()),
+ (pa.array([1.5]), pa.date32(), pa.int64()),
+ (pa.array([1.5]), pa.time32('ms'), pa.int64()),
+ (pa.array([1], type=pa.int32()), pa.timestamp('ms'), pa.int64()),
+ (pa.array([2.675]), pa.decimal128(10, 2), pa.int64()),
+ (pa.array([2.25], type=pa.float32()), pa.decimal128(10, 1), pa.int64()),
+ (pa.array([0.1]), pa.decimal128(38, 20), pa.int64()),
+])
+def test_native_row_id_input_conversion_matches_python(
+ tmp_path, native, grouped, empty_chunks, values, target, row_id_type):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ schema = pa.schema([('id', pa.int32()), ('value', target)])
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(schema,
options={
+ 'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true',
+ 'write.native.enabled': str(native).lower(),
+ }), False)
+ table = catalog.get_table('default.t')
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ writer.write_arrow(pa.table({'id': [1], 'value': [None]}, schema=schema))
+ builder.new_commit().commit(writer.prepare_commit())
+ writer.close()
+ data = pa.table({'_ROW_ID': pa.array([0], type=row_id_type), 'value':
values})
+ if empty_chunks:
+ batch = data.to_batches()[0]
+ data = pa.Table.from_batches([batch.slice(0, 0), batch, batch.slice(1,
0)])
+ before = set(tmp_path.rglob('*.parquet'))
+ updater = builder.new_update()
+
+ def prepare():
+ if grouped:
+ return updater.update_by_arrow_batches_with_row_id(iter([data]))
+ return updater.update_by_arrow_with_row_id(data)
+
+ try:
+ expected = TableUpdateByRowId._coerce_column(values,
target).to_pylist()
+ except (ValueError, pa.ArrowException):
+ with pytest.raises((ValueError, pa.ArrowException)):
+ prepare()
+ expected = [None]
+ assert set(tmp_path.rglob('*.parquet')) == before
+ else:
+ # Ensure a native case does not silently exercise the Python fallback.
+ if native:
+ with patch.object(TableUpdateByRowId, 'update_columns',
+ side_effect=AssertionError('Python update
selected')):
+ messages = prepare()
+ else:
+ messages = prepare()
+ builder.new_commit().commit(messages)
+ read = table.new_read_builder()
+ result = read.new_read().to_arrow(read.new_scan().plan().splits())
+ assert result['value'].to_pylist() == expected
+
+
[email protected]('native', [False, True])
[email protected]('duplicate', ['_ROW_ID', 'value', 'unused'])
+def test_row_id_update_rejects_only_referenced_duplicate_columns(tmp_path,
native, duplicate):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ schema = pa.schema([('id', pa.int32()), ('value', pa.int32())])
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(schema,
options={
+ 'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true',
+ 'write.native.enabled': str(native).lower(),
+ }), False)
+ table = catalog.get_table('default.t')
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ writer.write_arrow(pa.table({'id': [1, 2], 'value': [10, 20]},
schema=schema))
+ builder.new_commit().commit(writer.prepare_commit())
+ writer.close()
+ data = pa.Table.from_arrays([
+ pa.array([0]), pa.array([100]), pa.array([1]), pa.array([2]),
+ ], names=['_ROW_ID', 'value', duplicate, duplicate])
+ before = set(tmp_path.rglob('*.parquet'))
+ # Repeating configuration names remains valid; ambiguity is in input
fields.
+ updater = builder.new_update().with_update_type(['value', 'value'])
+ if duplicate == 'unused':
+ messages = updater.update_by_arrow_with_row_id(data)
+ builder.new_commit().commit(messages)
+ else:
+ with pytest.raises((KeyError, ValueError, pa.ArrowException)):
+ updater.update_by_arrow_with_row_id(data)
+ assert set(tmp_path.rglob('*.parquet')) == before
+ read = table.new_read_builder()
+ actual =
read.new_read().to_arrow(read.new_scan().plan().splits()).sort_by('id')
+ assert actual['value'].to_pylist() == ([100, 20] if duplicate == 'unused'
else [10, 20])
+
+
[email protected]('native', [False, True])
[email protected]('values,target', [
+ (pa.array([{'a': 1, 'b': 1.5}], type=pa.struct([('a', pa.int32()), ('b',
pa.float64())])),
+ pa.struct([('a', pa.string()), ('b', pa.int32())])),
+ (pa.array([[(1, 1.5)]], type=pa.map_(pa.int32(), pa.float64())),
pa.map_(pa.string(), pa.int32())),
+ (pa.array([{'a': 1, 'b': 2.0}], type=pa.struct([('a', pa.int32()), ('b',
pa.float64())])),
+ pa.struct([('a', pa.string()), ('b', pa.int32())])),
+ (pa.StructArray.from_arrays([pa.array([2 ** 31, 7])], names=['a'],
mask=pa.array([True, False])),
+ pa.struct([('a', pa.int32())])),
+ (pa.ListArray.from_arrays([0, 1, 2], pa.array([2 ** 31, 7]),
mask=pa.array([True, False])),
+ pa.list_(pa.int32())),
+ (pa.MapArray.from_arrays([0, 1, 2], pa.array(['a', 'b']), pa.array([1.5,
7.5]),
+ mask=pa.array([True, False])),
pa.map_(pa.string(), pa.int32())),
+])
+def
test_core_nested_row_update_uses_whole_column_constructor_fallback(tmp_path,
native, values, target):
+ from pypaimon.write.native_commit import create_native_write_table
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ schema = pa.schema([('id', pa.int32()), ('value', target)])
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(schema,
options={
+ 'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true',
+ 'write.native.enabled': 'false',
+ }), False)
+ table = catalog.get_table('default.t')
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ writer.write_arrow(pa.table({'id': list(range(len(values))), 'value':
[None] * len(values)}, schema=schema))
+ builder.new_commit().commit(writer.prepare_commit())
+ writer.close()
+ # Exercise core nested conversion directly even when dispatch would fall
back.
+ builder = (create_native_write_table(table) if native else
table).new_batch_write_builder()
+ data = pa.table({'_ROW_ID': list(range(len(values))), 'value': values})
+ before = set(tmp_path.rglob('*.parquet'))
+ try:
+ expected = TableUpdateByRowId._coerce_column(values,
target).to_pylist()
+ except (ValueError, pa.ArrowException):
+ with pytest.raises((ValueError, pa.ArrowException)):
+ builder.new_update().update_by_arrow_with_row_id(data)
+ assert set(tmp_path.rglob('*.parquet')) == before
+ expected = [None] * len(values)
+ else:
+ messages = builder.new_update().update_by_arrow_with_row_id(data)
+ builder.new_commit().commit(messages)
+ read = table.new_read_builder()
+ actual =
read.new_read().to_arrow(read.new_scan().plan().splits()).sort_by('id')
+ assert actual['value'].to_pylist() == expected
+
+
[email protected]('layout', [
+ 'legacy', 'canonical', 'ordinary', 'directory', 'external', 'float',
'double', 'external-binary',
+])
[email protected]('operation', ['row', 'predicate'])
+def test_update_dispatch_preserves_existing_partition_paths(tmp_path, layout,
operation):
+ from pypaimon.write.native_commit import create_native_write_table
+ from pypaimon.write.native_update import NativeBatchTableUpdate,
NativePredicateTableUpdate
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path / 'warehouse')})
+ catalog.create_database('default', True)
+ part_type = {'float': pa.float32(), 'double': pa.float64(),
'external-binary': pa.binary()}.get(
+ layout, pa.string())
+ schema = pa.schema([('id', pa.int32()), ('part', part_type), ('value',
pa.int32())])
+ options = {
+ 'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true',
+ 'write.native.enabled': str(layout == 'canonical').lower(),
+ }
+ if layout == 'directory':
+ options['data-file.path-directory'] = 'data'
+ if layout.startswith('external'):
+ options['data-file.external-paths'] = (tmp_path / 'external').as_uri()
+ options['data-file.external-paths.strategy'] = 'round-robin'
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ schema, partition_keys=['part'], options=options), False)
+ table = catalog.get_table('default.t')
+ seed_table = create_native_write_table(table) if layout == 'canonical'
else table
+ builder = seed_table.new_batch_write_builder()
+ writer = builder.new_write()
+ parts = ['east', 'west'] if layout == 'ordinary' else ['a/b', 'a%b']
+ if layout in ('float', 'double'):
+ parts = [1.5, 2.25]
+ elif layout == 'external-binary':
+ parts = [b'x', b'a/b']
+ data = pa.table({'id': [1, 2], 'part': parts, 'value': [10, 20]},
schema=schema)
+ writer.write_arrow(data.to_batches()[0] if layout == 'canonical' else data)
+ builder.new_commit().commit(writer.prepare_commit())
+ if layout != 'canonical':
+ writer.close()
+ table = table.copy({'write.native.enabled': 'true'})
+ native_calls = []
+ wrapper = NativeBatchTableUpdate if operation == 'row' else
NativePredicateTableUpdate
+ method = 'update_by_arrow_with_row_id' if operation == 'row' else 'update'
+ original = getattr(wrapper, method)
+
+ def track(*args, **kwargs):
+ native_calls.append(operation)
+ return original(*args, **kwargs)
+
+ # Check a second update too: fallback outputs must remain readable on the
+ # next operation, and canonical files must continue using native updates.
+ with patch.object(wrapper, method, track):
+ for value in [11, 12]:
+ builder = table.new_batch_write_builder()
+ if operation == 'row':
+ messages = builder.new_update().update_by_arrow_with_row_id(
+ pa.table({'_ROW_ID': [0, 1], 'value': pa.array([value,
value], pa.int32())}))
+ else:
+ messages = builder.new_update().update_by_predicate(None,
{'value': value})
+ builder.new_commit().commit(messages)
+ read = table.new_read_builder()
+ actual =
read.new_read().to_arrow(read.new_scan().plan().splits()).sort_by('id')
+ assert actual.to_pydict() == {'id': [1, 2], 'part': parts,
'value': [value, value]}
+ expected_calls = 2 if layout in ('canonical', 'ordinary', 'external') else 0
+ assert len(native_calls) == expected_calls
diff --git a/paimon-python/pypaimon/tests/table_update_test.py
b/paimon-python/pypaimon/tests/table_update_test.py
index 25116b3c3b..262b9726c7 100644
--- a/paimon-python/pypaimon/tests/table_update_test.py
+++ b/paimon-python/pypaimon/tests/table_update_test.py
@@ -336,6 +336,7 @@ class _TableUpdateTestBase(DataEvolutionTestBase):
result['name'].to_pylist(),
)
+ @pytest.mark.python_write
def test_update_by_predicate_processes_one_file_group_at_a_time(self):
from pypaimon.write.table_update_by_row_id import TableUpdateByRowId
@@ -368,6 +369,7 @@ class _TableUpdateTestBase(DataEvolutionTestBase):
self._read_all(table)['city'].to_pylist(),
)
+ @pytest.mark.python_write
def test_predicate_update_aborts_groups_after_later_failure(self):
from pypaimon.write.table_update_by_row_id import TableUpdateByRowId
@@ -458,6 +460,7 @@ class _TableUpdateTestBase(DataEvolutionTestBase):
self._read_all(table)['age'].to_pylist(),
)
+ @pytest.mark.python_write
def test_literal_predicate_update_projects_only_row_id(self):
table = self._create_seeded_table()
pb = table.new_read_builder().new_predicate_builder()
diff --git a/paimon-python/pypaimon/write/native_commit.py
b/paimon-python/pypaimon/write/native_commit.py
index 1eb50e09bb..39b86cabe0 100644
--- a/paimon-python/pypaimon/write/native_commit.py
+++ b/paimon-python/pypaimon/write/native_commit.py
@@ -24,7 +24,9 @@ from pypaimon.read.native_plan import (
_catalog_context_options, _catalog_metastore, _option_value_to_string,
_resolved_schema_file_io_options)
from pypaimon.utils.file_store_path_factory import canonical_data_file_path
-from pypaimon.write.commit_message_serializer import serialize_commit_message
+from pypaimon.write.commit_message_serializer import (
+ deserialize_commit_message, serialize_commit_message,
+)
_DEFAULT_MANIFEST_TARGET_SIZE = 8 * 1024 * 1024
@@ -183,3 +185,15 @@ def to_native_commit_messages(table, messages):
return [NativeCommitMessage.deserialize(
serialize_commit_message(message, table.partition_keys_fields),
version=14)
for message in messages]
+
+
+def from_native_commit_messages(table, messages):
+ """Decode native v14 messages for PyPaimon's existing commit API."""
+ decoded = [deserialize_commit_message(
+ message.serialize(), table.partition_keys_fields,
+ table.trimmed_primary_keys_fields) for message in messages]
+ for message in decoded:
+ for file in message.new_files + message.changelog_files:
+ file.file_path = file.external_path or canonical_data_file_path(
+ table, message.partition, message.bucket, file.file_name)
+ return decoded
diff --git a/paimon-python/pypaimon/write/native_update.py
b/paimon-python/pypaimon/write/native_update.py
new file mode 100644
index 0000000000..d4c47367b0
--- /dev/null
+++ b/paimon-python/pypaimon/write/native_update.py
@@ -0,0 +1,304 @@
+# 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.
+
+"""Optional native row-ID updates for batch data-evolution tables."""
+
+import pyarrow as pa
+
+from pypaimon.schema.data_types import PyarrowFieldParser
+from pypaimon.snapshot.snapshot import BATCH_COMMIT_IDENTIFIER
+from pypaimon.snapshot.time_travel_util import SCAN_KEYS
+from pypaimon.table.file_store_table import FileStoreTable
+from pypaimon.write.native_commit import (
+ create_native_write_table, from_native_commit_messages,
+)
+from pypaimon.write.native_write import native_write_available,
_native_partition_types_supported
+from pypaimon.write.table_update_by_row_id import TableUpdateByRowId,
_RowIdUpdateFileWriter
+from pypaimon.write.row_utils import value_for_arrow
+
+
+def _native_row_id_table(table):
+ """Resolve an eligible table before creating a native row-ID writer."""
+ if (type(table) is not FileStoreTable
+ or not table.options.native_write_enabled()
+ or not table.options.data_evolution_enabled()
+ or not table.options.row_tracking_enabled()
+ or table.options.data_file_path_directory() is not None
+ or not _RowIdUpdateFileWriter.supports_table(table)
+ or any(table.options.options.contains_key(key) for key in
SCAN_KEYS)
+ or not native_write_available()):
+ return None
+ schema = PyarrowFieldParser.from_paimon_schema(table.table_schema.fields)
+ if not _native_partition_types_supported(schema, table.partition_keys):
+ return None
+ if not _native_update_paths_supported(table):
+ return None
+ return create_native_write_table(table)
+
+
+def _native_update_paths_supported(table):
+ """Keep legacy Python partition directories on the path-aware Python
updater."""
+ if not table.partition_keys:
+ return True
+ factory = table.path_factory()
+ # Core update scans for itself, unlike native reads which receive splits
+ # with repaired file paths. Decide before callbacks or staging can start.
+ splits = table.new_read_builder().new_scan().plan_for_write().splits()
+ bucket_files = {}
+ for split in splits:
+ partition = tuple(split.partition.values)
+ bucket_path = factory.bucket_path(partition, split.bucket)
+ if bucket_path == factory.bucket_path(partition, split.bucket,
canonical_partition=True):
+ continue
+ candidates = [file for file in split.files if not file.external_path]
+ if not candidates:
+ continue
+ if bucket_path not in bucket_files:
+ bucket_files[bucket_path] = {
+ status.base_name for status in
table.file_io.list_status(bucket_path)
+ }
+ # Python reads prefer an existing legacy path even if a canonical copy
+ # also exists. Canonical-only files and explicit external paths are
safe.
+ if any(file.file_name in bucket_files[bucket_path] for file in
candidates):
+ return False
+ return True
+
+
+def _native_update_table(table, columns=None):
+ schema = PyarrowFieldParser.from_paimon_schema(table.table_schema.fields)
+ selected = schema.names if columns is None else columns
+ if any(pa.types.is_nested(schema.field(name).type)
+ for name in selected if name in schema.names):
+ return None
+ return _native_row_id_table(table)
+
+
+def create_native_update(table, commit_user, columns):
+ """Use the public core updater for direct and grouped row-ID updates."""
+ native_table = _native_update_table(table, columns)
+ if native_table is None:
+ return None
+ writer = (native_table.new_batch_write_builder()
+ ._with_commit_user(commit_user)
+ .new_update())
+ if columns is not None:
+ writer.with_update_type(columns)
+ return NativeBatchTableUpdate(table, writer)
+
+
+def create_native_update_by_row_id(table, commit_user, commit_identifier):
+ """Create a core updater sharing one snapshot across incremental calls."""
+ native_table = _native_update_table(table)
+ if native_table is None:
+ return None
+ if commit_identifier == BATCH_COMMIT_IDENTIFIER:
+ writer =
(native_table.new_batch_write_builder()._with_commit_user(commit_user)
+ .new_update().new_update_by_row_id())
+ else:
+ writer =
(native_table.new_stream_write_builder().with_commit_user(commit_user)
+ .new_update().new_update_by_row_id(commit_identifier))
+ return NativeTableUpdateByRowId(
+ table, commit_user, commit_identifier, writer)
+
+
+def _supported_upsert_key_type(data_type):
+ return any(check(data_type) for check in (
+ pa.types.is_boolean, pa.types.is_integer, pa.types.is_string,
+ pa.types.is_large_string, pa.types.is_binary, pa.types.is_large_binary,
+ pa.types.is_fixed_size_binary, pa.types.is_date, pa.types.is_decimal,
+ ))
+
+
+def create_native_upsert(table, commit_user, data, keys, columns):
+ """Prepare one core upsert from full Arrow rows or named row values."""
+ native_table = _native_row_id_table(table)
+ if native_table is None:
+ return None
+ fields = table.table_schema.fields
+ schema = PyarrowFieldParser.from_paimon_schema(fields)
+ if (any(not _supported_upsert_key_type(schema.field(key).type) for key in
set(keys + table.partition_keys))
+ or any(pa.types.is_nested(schema.field(name).type) for name in
columns)):
+ return None
+ if not isinstance(data, pa.Table):
+ # Missing fields retain their row-object semantics on the fallback
+ # path; converting them to Arrow NULLs would change update behavior.
+ if not columns or any(set(values) != set(schema.names) for values in
data):
+ return None
+ data = pa.Table.from_pydict({
+ field.name: [value_for_arrow(values[field.name], field) for values
in data]
+ for field in fields
+ }, schema=schema)
+ if (len(data.column_names) != len(schema.names)
+ or set(data.column_names) != set(schema.names)
+ or any(data.schema.field(name).type != schema.field(name).type
+ for name in schema.names)):
+ return None
+ writer = (native_table.new_batch_write_builder()
+ ._with_commit_user(commit_user)
+ .new_update()
+ .with_update_type(columns))
+ return NativeTableUpsert(table, writer, keys, data)
+
+
+def create_native_predicate_update(table, commit_user, columns, predicate):
+ """Prepare a public core operation before any assignment can run."""
+ schema = PyarrowFieldParser.from_paimon_schema(table.table_schema.fields)
+ if any(pa.types.is_nested(schema.field(name).type)
+ for name in columns if name in schema.names):
+ return None
+ native_table = _native_row_id_table(table)
+ if native_table is None:
+ return None
+ from pypaimon.read.native_plan import _predicate_to_native
+ native_predicate = None if predicate is None else
_predicate_to_native(predicate)
+ if native_predicate is not None:
+ # Check predicate translation while fallback is still safe. Planning,
+ # reading and callback execution belong to the core operation below.
+ native_table.new_read_builder().with_filter(native_predicate)
+ writer = (native_table.new_batch_write_builder()
+ ._with_commit_user(commit_user).new_update())
+ return NativePredicateTableUpdate(table, writer, native_predicate)
+
+
+def native_predicate_row_ids(scan_table, predicate, splits):
+ """Match a batch delete predicate in Rust and return its row IDs."""
+ from pypaimon.read.native_plan import (
+ _prepare_native_read, native_split_bridge_available,
+ native_split_from_python,
+ )
+ if not native_split_bridge_available():
+ return None
+ reader = _prepare_native_read(
+ scan_table, predicate=predicate, projection=['_ROW_ID']
+ )
+ row_ids = []
+ for split in splits:
+ for batch in reader([native_split_from_python(split)]):
+ row_ids.extend(batch.column('_ROW_ID').to_pylist())
+ return row_ids
+
+
+def create_native_delete(table, commit_user):
+ """Select Rust's deletion-vector writer for supported batch deletes."""
+ if not table.options.deletion_vectors_enabled(False):
+ return None
+ native_table = _native_row_id_table(table)
+ if native_table is None:
+ return None
+ writer = (native_table.new_batch_write_builder()
+ ._with_commit_user(commit_user)
+ .new_update())
+ return NativeBatchTableUpdate(table, writer)
+
+
+def _raise_native_row_id_error(error):
+ detail = str(error)
+ if 'duplicate UPDATE operations' in detail:
+ raise ValueError('duplicate _ROW_ID: ' + detail) from error
+ if 'No file found for _ROW_ID' in detail:
+ raise ValueError(
+ detail + ' does not belong to any valid range') from error
+ raise error
+
+
+class NativeBatchTableUpdate:
+ """Wrap public core operations and decode their commit messages."""
+
+ def __init__(self, table, writer):
+ self.table = table
+ self.writer = writer
+
+ def update_by_arrow_with_row_id(self, data: pa.Table):
+ try:
+ messages = self.writer.update_by_arrow_with_row_id(data)
+ except ValueError as error:
+ _raise_native_row_id_error(error)
+ return from_native_commit_messages(self.table, messages)
+
+ def update_by_arrow_batches_with_row_id(self, tables):
+ try:
+ messages = self.writer.update_by_arrow_batches_with_row_id(tables)
+ except ValueError as error:
+ _raise_native_row_id_error(error)
+ return from_native_commit_messages(self.table, messages)
+
+ def delete_by_row_id(self, row_ids):
+ ids = []
+ for row_id in row_ids:
+ if row_id is None:
+ raise ValueError('_ROW_ID value must not be null.')
+ ids.append(int(row_id))
+ try:
+ messages = self.writer.delete_by_row_id(ids)
+ except ValueError as error:
+ _raise_native_row_id_error(error)
+ return from_native_commit_messages(self.table, messages)
+
+
+class NativeTableUpdateByRowId(TableUpdateByRowId):
+ """Adapt Python row inputs around the core incremental Arrow updater."""
+
+ def __init__(self, table, commit_user, commit_identifier, writer):
+ self.table = table
+ self.commit_user = commit_user
+ self.commit_identifier = commit_identifier
+ self.writer = writer
+
+ @property
+ def commit_messages(self):
+ return from_native_commit_messages(self.table,
self.writer.commit_messages)
+
+ def update_columns(self, data, column_names):
+ try:
+ messages = self.writer.update_columns(data, column_names)
+ except ValueError as error:
+ _raise_native_row_id_error(error)
+ return from_native_commit_messages(self.table, messages)
+
+ def _write_row_columns(self, data, column_names, blob_object_columns):
+ # This adapter is selected only for plain Parquet tables without blobs.
+ return self.update_columns(data, column_names)
+
+
+class NativeTableUpsert:
+ """Submit full Arrow rows to the core Rust upsert writer."""
+
+ def __init__(self, table, writer, keys, data):
+ self.table = table
+ self.writer = writer
+ self.keys = keys
+ self.data = data
+
+ def upsert(self):
+ return from_native_commit_messages(
+ self.table,
+ self.writer.upsert_by_arrow_with_key(self.data, self.keys))
+
+
+class NativePredicateTableUpdate:
+ """Convert operation inputs and commit messages around the core updater."""
+
+ def __init__(self, table, writer, predicate):
+ self.table = table
+ self.writer = writer
+ self.predicate = predicate
+
+ def update(self, assignments, read_columns):
+ return from_native_commit_messages(
+ self.table,
+ self.writer.update_by_predicate(
+ self.predicate, dict(assignments), list(read_columns or ())))
diff --git a/paimon-python/pypaimon/write/native_write.py
b/paimon-python/pypaimon/write/native_write.py
index 5185731614..2d631019c6 100644
--- a/paimon-python/pypaimon/write/native_write.py
+++ b/paimon-python/pypaimon/write/native_write.py
@@ -24,9 +24,9 @@ from pypaimon.common.options.core_options import CoreOptions,
MergeEngine
from pypaimon.schema.arrow_schema import arrow_schemas_compatible,
normalize_arrow_strings
from pypaimon.schema.data_types import PyarrowFieldParser, is_blob_file_field
from pypaimon.table.bucket_mode import BucketMode
-from pypaimon.utils.file_store_path_factory import canonical_data_file_path
-from pypaimon.write.commit_message_serializer import deserialize_commit_message
-from pypaimon.write.native_commit import create_native_write_table
+from pypaimon.write.native_commit import (
+ create_native_write_table, from_native_commit_messages,
+)
from pypaimon.write.row_utils import row_to_named_values,
row_values_to_arrow_table
@@ -39,6 +39,14 @@ def native_write_available() -> bool:
return True
+def _native_partition_types_supported(schema, partition_keys):
+ """Partition keys which Rust can encode and use to locate existing
files."""
+ return not any(
+ pa.types.is_binary(data_type) or pa.types.is_large_binary(data_type)
+ or pa.types.is_fixed_size_binary(data_type) or
pa.types.is_floating(data_type)
+ for data_type in (schema.field(name).type for name in partition_keys))
+
+
def create_native_write(table, commit_user, static_partition=None,
stream=False):
"""Return a native writer if the table can use the filesystem write
path."""
schema = PyarrowFieldParser.from_paimon_schema(table.table_schema.fields)
@@ -49,7 +57,6 @@ def create_native_write(table, commit_user,
static_partition=None, stream=False)
if (not table.options.sequence_field_sort_order_is_ascending()
or any(pa.types.is_floating(schema.field(name).type) for name
in sequence_fields)):
return None
- partition_types = [schema.field(name).type for name in
table.partition_keys]
if (not native_write_available()
or table.options.data_evolution_enabled()
or table.options.data_file_external_paths()
@@ -69,8 +76,7 @@ def create_native_write(table, commit_user,
static_partition=None, stream=False)
or any(pa.types.is_nested(field.type)
or pa.types.is_fixed_size_binary(field.type) for field in
schema)
# Rust cannot encode these partition keys yet.
- or any(pa.types.is_binary(type_) or pa.types.is_large_binary(type_)
- or pa.types.is_floating(type_) for type_ in partition_types)
+ or not _native_partition_types_supported(schema,
table.partition_keys)
or any(is_blob_file_field(field) for field in
table.table_schema.fields)):
return None
native_table = create_native_write_table(table)
@@ -179,14 +185,7 @@ class NativeTableWrite:
if commit_identifier is not None:
raise TypeError('BatchTableWrite.prepare_commit accepts no
identifier')
messages = self._native_writer.prepare_commit()
- decoded = [deserialize_commit_message(
- message.serialize(), self.table.partition_keys_fields,
- self.table.trimmed_primary_keys_fields) for message in messages]
- for message in decoded:
- for file in message.new_files + message.changelog_files:
- file.file_path = file.external_path or
canonical_data_file_path(
- self.table, message.partition, message.bucket,
file.file_name)
- return decoded
+ return from_native_commit_messages(self.table, messages)
def close(self):
if self._python_writer is not None:
diff --git a/paimon-python/pypaimon/write/table_update.py
b/paimon-python/pypaimon/write/table_update.py
index 494127d791..5c5958cb1e 100644
--- a/paimon-python/pypaimon/write/table_update.py
+++ b/paimon-python/pypaimon/write/table_update.py
@@ -16,6 +16,7 @@
# under the License.
from collections import defaultdict
+import logging
from typing import Any, Iterable, List, Mapping, Optional, Sequence, Tuple
import pyarrow
@@ -36,6 +37,7 @@ from pypaimon.manifest.index_manifest_file import
IndexManifestFile
from pypaimon.manifest.manifest_list_manager import ManifestListManager
from pypaimon.manifest.schema.data_file_meta import DataFileMeta
from pypaimon.read.scanner.file_scanner import FileScanner
+from pypaimon.read.query_auth_split import QueryAuthSplit
from pypaimon.read.scanner.data_evolution_split_generator import (
DataEvolutionSplitGenerator,
)
@@ -53,6 +55,8 @@ from pypaimon.write.table_upsert_by_key import
TableUpsertByKey
from pypaimon.write.writer.data_writer import DataWriter
from pypaimon.write.writer.append_only_data_writer import AppendOnlyDataWriter
+logger = logging.getLogger(__name__)
+
def _filter_by_whole_file_shard(splits: List[DataSplit], sub_task_id: int,
total_tasks: int) -> List[DataSplit]:
list_ranges = []
@@ -129,6 +133,16 @@ class TableUpdate:
def _new_row_id_updater(
self, commit_identifier: int, _precomputed_files_info=None
) -> TableUpdateByRowId:
+ if _precomputed_files_info is None:
+ from pypaimon.write.native_update import
create_native_update_by_row_id
+ try:
+ native = create_native_update_by_row_id(
+ self.table, self.commit_user, commit_identifier)
+ except Exception as error:
+ logger.debug('Native row-id updater preparation failed: %s',
error)
+ else:
+ if native is not None:
+ return native
return TableUpdateByRowId(
self.table,
self.commit_user,
@@ -174,6 +188,16 @@ class TableUpdate:
cols = self.update_cols if self.update_cols is not None else [
c for c in table.column_names if c != SpecialFields.ROW_ID.name
]
+ if (table.num_rows and cols
+ and SpecialFields.ROW_ID.name in table.column_names):
+ try:
+ from pypaimon.write.native_update import create_native_update
+ native = create_native_update(self.table, self.commit_user,
cols)
+ except Exception as error:
+ logger.debug('Native update preparation failed; using Python:
%s', error)
+ else:
+ if native is not None:
+ return native.update_by_arrow_with_row_id(table)
return self._new_row_id_updater(commit_identifier).update_columns(
table, cols)
@@ -278,6 +302,17 @@ class TableUpdate:
assignments, read_columns, has_callable, has_array
)
+ if self.table.options.native_write_enabled():
+ from pypaimon.write.native_update import
create_native_predicate_update
+ try:
+ native = create_native_predicate_update(
+ self.table, self.commit_user, list(assignments.keys()),
predicate)
+ except Exception as error:
+ logger.debug('Native predicate update preparation failed: %s',
error)
+ else:
+ if native is not None:
+ return native.update(assignments, read_columns)
+
scan_table = self._matched_update_scan_table()
read_builder = scan_table.new_read_builder()
if predicate is not None:
@@ -509,7 +544,7 @@ class TableUpdate:
return self._delete_by_partition_filter(partition_filter)
row_ids = self._matched_delete_row_ids(predicate)
- return TableDeleteByRowId(self.table).delete(row_ids)
+ return self._delete_by_row_id(row_ids, commit_identifier)
def _delete_by_partition_filter(
self, partition_filter: Predicate) -> List[CommitMessage]:
@@ -576,6 +611,15 @@ class TableUpdate:
row_ids: Sequence[int],
commit_identifier: int,
) -> List[CommitMessage]:
+ if len(row_ids):
+ try:
+ from pypaimon.write.native_update import create_native_delete
+ native = create_native_delete(self.table, self.commit_user)
+ except Exception as error:
+ logger.debug('Native delete preparation failed; using Python:
%s', error)
+ else:
+ if native is not None:
+ return native.delete_by_row_id(row_ids)
return TableDeleteByRowId(self.table).delete(list(row_ids))
def _partition_only_delete_filter(
@@ -637,6 +681,19 @@ class TableUpdate:
scan = read_builder.new_scan()
splits = scan.plan_for_write().splits()
+ if (splits
+ and self.table.options.native_write_enabled()
+ and self.table.options.data_file_path_directory() is None
+ and not any(isinstance(split, QueryAuthSplit)
+ for split in splits)):
+ try:
+ from pypaimon.write.native_update import
native_predicate_row_ids
+ row_ids = native_predicate_row_ids(scan_table, predicate,
splits)
+ except Exception as error:
+ logger.debug('Native predicate delete match failed: %s', error)
+ else:
+ if row_ids is not None:
+ return row_ids
matched = read_builder.new_read().to_arrow(splits)
if matched.num_rows == 0:
return []
@@ -666,6 +723,15 @@ class BatchTableUpdate(TableUpdate):
file groups. Conflicting overlap is rejected and all files staged by
earlier batches are aborted.
"""
+ try:
+ from pypaimon.write.native_update import create_native_update
+ native = create_native_update(
+ self.table, self.commit_user, self.update_cols)
+ except Exception as error:
+ logger.debug('Native batch update preparation failed: %s', error)
+ else:
+ if native is not None:
+ return native.update_by_arrow_batches_with_row_id(tables)
return self._update_by_arrow_batches_with_row_id(
tables, BATCH_COMMIT_IDENTIFIER)
@@ -710,6 +776,11 @@ class BatchTableUpdate(TableUpdate):
"""Delete rows by ``_ROW_ID`` using deletion vectors."""
return self._delete_by_row_id(row_ids, BATCH_COMMIT_IDENTIFIER)
+ def _delete_by_row_id(
+ self, row_ids: Sequence[int], commit_identifier: int
+ ) -> List[CommitMessage]:
+ return super()._delete_by_row_id(row_ids, commit_identifier)
+
def merge_into(
self,
source: Any,
diff --git a/paimon-python/pypaimon/write/table_update_by_row_id.py
b/paimon-python/pypaimon/write/table_update_by_row_id.py
index a8af846c9c..b1c65a1ed7 100644
--- a/paimon-python/pypaimon/write/table_update_by_row_id.py
+++ b/paimon-python/pypaimon/write/table_update_by_row_id.py
@@ -355,6 +355,9 @@ class TableUpdateByRowId:
fields.append(arrow_field)
update_data = pa.Table.from_arrays(arrays, schema=pa.schema(fields))
+ return self._write_row_columns(update_data, column_names,
blob_object_columns)
+
+ def _write_row_columns(self, update_data, column_names,
blob_object_columns):
data_with_first_row_id = self._calculate_first_row_id(update_data)
self._write_by_first_row_id(
data_with_first_row_id, column_names, blob_object_columns)
diff --git a/paimon-python/pypaimon/write/table_upsert_by_key.py
b/paimon-python/pypaimon/write/table_upsert_by_key.py
index aebbdf6176..3ffb6ad30e 100644
--- a/paimon-python/pypaimon/write/table_upsert_by_key.py
+++ b/paimon-python/pypaimon/write/table_upsert_by_key.py
@@ -94,6 +94,12 @@ class TableUpsertByKey:
else:
effective_update_cols = update_cols
+ columns = (list(effective_update_cols) if effective_update_cols
+ else list(self.table.field_names))
+ native = self._create_native_upsert(data, upsert_keys, columns)
+ if native is not None:
+ return native.upsert()
+
all_commit_messages: List[CommitMessage] = []
# Process each partition independently
@@ -123,6 +129,13 @@ class TableUpsertByKey:
else:
effective_update_cols = update_cols
+ columns = (list(effective_update_cols) if effective_update_cols is not
None
+ else list(self.table.field_names))
+ native = self._create_native_upsert(
+ [values for _, values in row_items], upsert_keys, columns)
+ if native is not None:
+ return native.upsert()
+
commit_messages: List[CommitMessage] = []
for partition_spec, partition_items in
self._group_rows_by_partition(row_items):
commit_messages.extend(
@@ -135,6 +148,16 @@ class TableUpsertByKey:
)
return commit_messages
+ def _create_native_upsert(self, data, upsert_keys, columns):
+ try:
+ from pypaimon.write.native_update import create_native_upsert
+ return create_native_upsert(
+ self.table, self.commit_user, data, upsert_keys, columns)
+ except Exception as error:
+ logger.debug(
+ 'Native upsert preparation failed; using Python: %s', error)
+ return None
+
@staticmethod
def _normalize_rows(rows) -> List:
if isinstance(rows, InternalRow):
