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 b07f7ffbb2 [python] Add Java-compatible CommitMessage serialization
and native commit (#10081)
b07f7ffbb2 is described below
commit b07f7ffbb21ffdc2853bec6757545e8d2b9ea092
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Sep 22 17:36:31 2026 +0800
[python] Add Java-compatible CommitMessage serialization and native commit
(#10081)
---
paimon-python/README.md | 37 +++
.../pypaimon/common/options/core_options.py | 11 +
.../tests/commit_message_serializer_test.py | 113 +++++++
.../pypaimon/tests/file_store_commit_test.py | 47 +++
paimon-python/pypaimon/tests/native_commit_test.py | 335 +++++++++++++++++++++
paimon-python/pypaimon/tests/table_commit_test.py | 3 +
paimon-python/pypaimon/write/commit_message.py | 13 +-
.../pypaimon/write/commit_message_serializer.py | 265 ++++++++++++++++
paimon-python/pypaimon/write/file_store_commit.py | 56 +++-
paimon-python/pypaimon/write/native_commit.py | 102 +++++++
paimon-python/pypaimon/write/table_commit.py | 42 ++-
11 files changed, 1014 insertions(+), 10 deletions(-)
diff --git a/paimon-python/README.md b/paimon-python/README.md
index a711f71345..8e45ba7153 100644
--- a/paimon-python/README.md
+++ b/paimon-python/README.md
@@ -119,6 +119,43 @@ table = table.copy({"parquet.filter.columnindex.enabled":
"false"})
Unsupported reads use the normal path. Reading fewer bytes may require more
object-store requests.
+# Native commit
+
+PyPaimon can submit append commits through the optional `pypaimon-rust`
+runtime. Enable it independently of native planning and reading:
+
+```python
+native_table = table.copy({"commit.native.enabled": "true"})
+builder = native_table.new_batch_write_builder()
+writer, commit = builder.new_write(), builder.new_commit()
+try:
+ writer.write_arrow(data)
+ commit.commit(writer.prepare_commit())
+finally:
+ writer.close()
+ commit.close()
+```
+
+The Python writer still produces files. Its commit messages cross the Java v14
+wire format into `CommitMessage.deserialize()` and are committed by Rust. Batch
+and stream append commits retain the Python builder's commit user, identifier,
+empty-commit option, and batch one-shot lifecycle. Explicit abort also supports
+native cleanup of uncommitted files.
+
+This requires a runtime containing the commit bindings merged in
+[paimon-rust #912](https://github.com/apache/paimon-rust/pull/912). Older or
missing
+runtimes automatically use Python. The current native route supports
main-branch
+tables using filesystem/JDBC catalogs or `FileStoreTable.from_path()` with
+standard FileIO. Overwrite, truncate, REST/catalog-managed publication, custom
+FileIO/environments, commit callbacks, and snapshot properties use Python.
+Data-evolution updates that need Python's row-id conflict rewriting also retain
+the Python path. Compact increments remain unsupported by both committers.
+
+Fallback is limited to capability checks, table construction and message
+conversion before a native mutation starts. A native commit error propagates;
+the adapter neither retries it through Python nor aborts files, since the
+snapshot may already have been published. The option is disabled by default.
+
# Native scan planning
PyPaimon can plan splits with the optional `pypaimon-rust` package. Planning
and
diff --git a/paimon-python/pypaimon/common/options/core_options.py
b/paimon-python/pypaimon/common/options/core_options.py
index 54bd771310..b329f62d19 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -677,6 +677,14 @@ class CoreOptions:
"routes fall back to pypaimon.")
)
+ COMMIT_NATIVE_ENABLED: ConfigOption[bool] = (
+ ConfigOptions.key("commit.native.enabled")
+ .boolean_type()
+ .default_value(False)
+ .with_description("Commit append messages via pypaimon_rust.
Unsupported "
+ "operations use Python before any native commit is
attempted.")
+ )
+
CHANGELOG_PRODUCER: ConfigOption[ChangelogProducer] = (
ConfigOptions.key("changelog-producer")
.enum_type(ChangelogProducer)
@@ -1592,6 +1600,9 @@ class CoreOptions:
def native_read_enabled(self, default=None):
return self.options.get(CoreOptions.READ_NATIVE_ENABLED, default)
+ def native_commit_enabled(self, default=None):
+ return self.options.get(CoreOptions.COMMIT_NATIVE_ENABLED, default)
+
def changelog_producer(self, default=None):
return self.options.get(CoreOptions.CHANGELOG_PRODUCER, default)
diff --git a/paimon-python/pypaimon/tests/commit_message_serializer_test.py
b/paimon-python/pypaimon/tests/commit_message_serializer_test.py
new file mode 100644
index 0000000000..45ef8e3bef
--- /dev/null
+++ b/paimon-python/pypaimon/tests/commit_message_serializer_test.py
@@ -0,0 +1,113 @@
+# 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.
+
+import base64
+import unittest
+
+from pypaimon.data.timestamp import Timestamp
+from pypaimon.globalindex.global_index_meta import GlobalIndexMeta
+from pypaimon.index.deletion_vector_meta import DeletionVectorMeta
+from pypaimon.index.index_file_meta import IndexFileMeta
+from pypaimon.manifest.index_manifest_entry import IndexManifestEntry
+from pypaimon.manifest.schema.data_file_meta import DataFileMeta
+from pypaimon.manifest.schema.simple_stats import SimpleStats
+from pypaimon.schema.data_types import AtomicType, DataField
+from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.write.commit_message import CommitMessage
+from pypaimon.write.commit_message_serializer import (
+ deserialize_commit_message, serialize_commit_message)
+
+
+# Captured from Java CommitMessageSerializer v14 with an empty partition,
+# bucket 3 and checkFromSnapshot 7. The second body has totalBuckets 5 and a
+# data-increment IndexFileMeta("I", "index", 9, 2).
+_JAVA_EMPTY = base64.b64decode(
+
'AAAADAAAAAAAAAAAAAAAAAAAAAMAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAEAAAAAAAAABw==')
+_JAVA_INDEXED = base64.b64decode(
+
'AAAADAAAAAAAAAAAAAAAAAAAAAMBAAAABQAAAAAAAAAAAAAAAAAAAAEAAABAAHAAAAAAAABJAAAAAAAAgWluZGV4AACFCQAA'
+
'AAAAAAACAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAEAAAAAAAAABw==')
+_JAVA_RICH_INDEXED = base64.b64decode(
+
'AAAADAAAAAAAAAAAAAAAAAAAAAMBAAAABQAAAAAAAAAAAAAAAAAAAAEAAADgAAAAAAAAAABEVgAAAAAAgmluZGV4AACFCQAA'
+
'AAAAAAACAAAAAAAAAEgAAABAAAAACwAAAIgAAABIAAAAmAAAAAEAAAAAAAAAOAAAABAAAAAAAAAAAAAAAAwAAAAoAAAAAwAA'
+
'AAAAAAAMAAAAAAAAAAIAAAAAAAAAZGF0YS5wYXJxdWV0AAAAAGZpbGU6L2luZGV4AAAAAAAAAAAAAAAAAAoAAAAAAAAAFAAA'
+
'AAAAAAADAAAAAAAAABAAAAA4AAAAAQIAAAAAAIIDAAAAAAAAgQIAAAAAAAAAAQAAAAQAAAAAAAAAAAAAAAAAAAAAAAAAAAAA'
+ 'AAAAAAABAAAAAAAAAAc=')
+
+
+class CommitMessageSerializerTest(unittest.TestCase):
+
+ def test_java_v14_golden(self):
+ for payload in (_JAVA_EMPTY, _JAVA_INDEXED, _JAVA_RICH_INDEXED):
+ message = deserialize_commit_message(payload, [])
+ self.assertEqual(serialize_commit_message(message, []), payload)
+ self.assertEqual(message.check_from_snapshot, 7)
+ indexed = deserialize_commit_message(_JAVA_INDEXED, [])
+ self.assertEqual(indexed.total_buckets, 5)
+ self.assertEqual(indexed.index_adds[0].index_file.file_name, 'index')
+ self.assertEqual(serialize_commit_message(
+ CommitMessage((), 3, [], check_from_snapshot=7), []), _JAVA_EMPTY)
+
+ def test_rich_index_java_golden(self):
+ meta = IndexFileMeta(
+ 'DV', 'index', 9, 2,
+ {'data.parquet': DeletionVectorMeta('data.parquet', 3, 12, 2)},
+ 'file:/index', GlobalIndexMeta(10, 20, 3, [1, 4], b'\x01\x02',
b'\x03'))
+ entry = IndexManifestEntry(0, GenericRow([], []), 3, meta)
+ message = CommitMessage(
+ (), 3, [], check_from_snapshot=7, index_adds=[entry],
total_buckets=5)
+ self.assertEqual(serialize_commit_message(message, []),
_JAVA_RICH_INDEXED)
+
+ def test_files_and_compaction_stay_separate(self):
+ stats = SimpleStats.empty_stats()
+ file = DataFileMeta(
+ 'data.parquet', 100, 4, GenericRow([], []), GenericRow([], []),
+ stats, stats, 1, 4, 2, 0, [], Timestamp(0), first_row_id=10,
+ write_cols_sequences=[4])
+ index = IndexManifestEntry(
+ 0, GenericRow([], []), 3, IndexFileMeta('I', 'index', 9, 2))
+ message = CommitMessage(
+ (), 3, [file], check_from_snapshot=7,
+ compact_before=[file], compact_after=[file],
+ compact_index_adds=[index])
+ decoded = deserialize_commit_message(serialize_commit_message(message,
[]), [])
+ self.assertEqual(decoded.new_files[0].file_name, 'data.parquet')
+ self.assertEqual(decoded.new_files[0].first_row_id, 10)
+ self.assertEqual(decoded.new_files[0].write_cols_sequences, [4])
+ self.assertEqual(len(decoded.compact_before), 1)
+ self.assertEqual(len(decoded.compact_after), 1)
+ self.assertEqual(decoded.compact_index_adds[0].index_file.file_name,
'index')
+
+ def test_bad_version_and_truncation(self):
+ with self.assertRaises(ValueError):
+ deserialize_commit_message(_JAVA_EMPTY, [], version=13)
+ with self.assertRaises(ValueError):
+ deserialize_commit_message(_JAVA_EMPTY[:-1], [])
+ with self.assertRaises(ValueError):
+ deserialize_commit_message(_JAVA_EMPTY + b'\x00', [])
+
+ def test_primary_key_files_require_key_fields_for_decode(self):
+ key_fields = [DataField(0, 'k', AtomicType('INT'))]
+ stats = SimpleStats.empty_stats()
+ file = DataFileMeta(
+ 'pk.parquet', 10, 1, GenericRow([1], key_fields),
+ GenericRow([1], key_fields), stats, stats, 0, 0, 1, 0, [])
+ payload = serialize_commit_message(CommitMessage((), 0, [file]), [])
+ with self.assertRaisesRegex(ValueError, 'key_fields are required'):
+ deserialize_commit_message(payload, [])
+ decoded = deserialize_commit_message(payload, [], key_fields)
+ self.assertEqual(decoded.new_files[0].min_key.values, [1])
+ self.assertEqual(serialize_commit_message(decoded, []), payload)
diff --git a/paimon-python/pypaimon/tests/file_store_commit_test.py
b/paimon-python/pypaimon/tests/file_store_commit_test.py
index 649812d7fd..4f8778a23a 100644
--- a/paimon-python/pypaimon/tests/file_store_commit_test.py
+++ b/paimon-python/pypaimon/tests/file_store_commit_test.py
@@ -38,11 +38,55 @@ from pypaimon.write.file_store_commit import (
RollbackRetryResult,
RewriteResult,
_abort_commit_messages,
+ _reject_compact_increment,
+ _row_id_check_from_messages,
)
+class TestRowIdCheckFromMessages(unittest.TestCase):
+
+ def test_minimum_baseline_and_invalid_messages(self):
+ tagged = CommitMessage((), 0, [], check_from_snapshot=7)
+ self.assertEqual(_row_id_check_from_messages([tagged, tagged]), 7)
+ self.assertIsNone(_row_id_check_from_messages([CommitMessage((), 0,
[])]))
+
+ newer = CommitMessage((), 0, [], check_from_snapshot=8)
+ for messages in ([tagged, newer], [newer, tagged]):
+ self.assertEqual(_row_id_check_from_messages(messages), 7)
+ with self.assertRaisesRegex(ValueError, 'Invalid row-id check
snapshot'):
+ _row_id_check_from_messages([
+ CommitMessage((), 0, [], check_from_snapshot=-1)])
+ with self.assertRaisesRegex(ValueError, 'missing its check-from
snapshot'):
+ _row_id_check_from_messages([
+ tagged, CommitMessage((), 0, [Mock(first_row_id=1)])])
+
+ def test_compaction_lists_are_not_flattened_into_append(self):
+ message = CommitMessage(
+ (), 0, [], compact_before=[Mock(file_name='before')],
+ compact_after=[Mock(file_name='after')],
+ compact_changelog_files=[Mock(file_name='changelog')])
+ with self.assertRaisesRegex(NotImplementedError, 'separate COMPACT
snapshot'):
+ _reject_compact_increment([message])
+ _reject_compact_increment([CommitMessage((), 0, [])])
+
+ def test_overwrite_validates_message_baseline(self):
+ commit = FileStoreCommit.__new__(FileStoreCommit)
+ with self.assertRaisesRegex(ValueError, 'Invalid row-id check
snapshot'):
+ commit.overwrite(
+ None, [CommitMessage((), 0, [], check_from_snapshot=-1)], 1)
+
+
class TestAbortCommitMessages(unittest.TestCase):
+ def test_reconstructs_local_path_after_wire_decode(self):
+ table = Mock()
+ table.path_factory.return_value.bucket_path.return_value =
'/table/p=1/bucket-0'
+ file = Mock(file_name='data.parquet', external_path=None,
file_path=None)
+ message = CommitMessage((1,), 0, [file])
+ _abort_commit_messages(table, [message])
+ table.file_io.delete_quietly.assert_called_once_with(
+ '/table/p=1/bucket-0/data.parquet')
+
def test_index_path_failure_does_not_escape_abort(self):
table = Mock()
table.path_factory.side_effect = RuntimeError("path lookup failed")
@@ -51,6 +95,9 @@ class TestAbortCommitMessages(unittest.TestCase):
new_files=[],
changelog_files=[],
index_adds=[Mock(index_file=index_file)],
+ compact_after=[],
+ compact_changelog_files=[],
+ compact_index_adds=[],
)
with self.assertLogs(
diff --git a/paimon-python/pypaimon/tests/native_commit_test.py
b/paimon-python/pypaimon/tests/native_commit_test.py
new file mode 100644
index 0000000000..fde7def37c
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_commit_test.py
@@ -0,0 +1,335 @@
+# 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.
+
+from unittest.mock import Mock, patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.catalog.catalog_environment import CatalogEnvironment
+from pypaimon.common.identifier import Identifier
+from pypaimon.common.options.core_options import CoreOptions
+from pypaimon.common.options.options import Options
+from pypaimon.snapshot.snapshot import BATCH_COMMIT_IDENTIFIER
+from pypaimon.table.file_store_table import FileStoreTable
+from pypaimon.write.commit_message import CommitMessage
+from pypaimon.write.native_commit import (
+ create_native_commit, native_commit_available, native_messages_supported)
+
+
+requires_native = pytest.mark.skipif(
+ not native_commit_available(), reason='pypaimon-rust CommitMessage/commit
API required')
+
+
+def _table(tmp_path, mode='append', backend='filesystem'):
+ options = {'warehouse': str(tmp_path / 'warehouse')}
+ if backend == 'jdbc':
+ options.update({'metastore': 'jdbc', 'uri': 'jdbc:sqlite:' +
str(tmp_path / 'catalog.db')})
+ catalog = CatalogFactory.create(options)
+ catalog.create_database('default', True)
+ options = {'file.format': 'parquet', 'commit.native.enabled': 'true'}
+ if mode == 'pk':
+ options['bucket'] = '1'
+ elif mode == 'de':
+ options.update({'data-evolution.enabled': 'true',
'row-tracking.enabled': 'true'})
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ pa.schema([('id', pa.int64()), ('pt', pa.string())]),
+ options=options, primary_keys=['id'] if mode == 'pk' else [],
+ partition_keys=[] if mode == 'pk' else ['pt']), False)
+ table = catalog.get_table('default.t')
+ if backend == 'path':
+ table = FileStoreTable.from_path(table.table_path)
+ if backend == 'jdbc':
+ catalog.close()
+ return table
+
+
+def _prepare(builder, rows, identifier=None):
+ writer = builder.new_write()
+ try:
+ writer.write_arrow(pa.Table.from_pylist(
+ rows, schema=pa.schema([('id', pa.int64()), ('pt', pa.string())])))
+ return writer.prepare_commit() if identifier is None else
writer.prepare_commit(identifier)
+ finally:
+ writer.close()
+
+
+def _rows(table):
+ builder = table.new_read_builder()
+ return sorted(builder.new_read().to_arrow(
+ builder.new_scan().plan().splits()).to_pylist(), key=lambda row:
row['id'])
+
+
+def _must_not_fallback(commit):
+ return patch.object(commit.file_store_commit, 'commit',
side_effect=AssertionError('Python fallback'))
+
+
+def test_native_commit_is_opt_in():
+ assert not CoreOptions(Options({})).native_commit_enabled()
+ assert CoreOptions(Options({'commit.native.enabled':
'true'})).native_commit_enabled()
+
+
+@requires_native
[email protected]('backend', ['filesystem', 'path', 'jdbc'])
[email protected]('mode', ['append', 'pk', 'de'])
+def test_native_batch_roundtrip_preserves_identity(tmp_path, backend, mode):
+ table = _table(tmp_path, mode, backend)
+ builder = table.new_batch_write_builder()
+ rows = [{'id': 1, 'pt': 'a'}, {'id': 2, 'pt': None}]
+ messages = _prepare(builder, rows)
+ commit = builder.new_commit()
+ try:
+ with _must_not_fallback(commit):
+ commit.commit(messages)
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ assert snapshot.commit_user == builder.commit_user
+ assert snapshot.commit_identifier == BATCH_COMMIT_IDENTIFIER
+ assert _rows(table) == rows
+ with pytest.raises(RuntimeError, match='one-time'):
+ commit.commit(messages)
+ finally:
+ commit.close()
+
+
+@requires_native
+def test_native_stream_reuses_commit_user_and_identifiers(tmp_path):
+ table = _table(tmp_path)
+ builder = table.new_stream_write_builder()
+ commit = builder.new_commit()
+ try:
+ with _must_not_fallback(commit):
+ for identifier in (7, 8):
+ commit.commit(_prepare(builder, [{'id': identifier, 'pt':
None}], identifier), identifier)
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ assert snapshot.commit_user == builder.commit_user
+ assert snapshot.commit_identifier == identifier
+ assert _rows(table) == [{'id': 7, 'pt': None}, {'id': 8, 'pt': None}]
+ finally:
+ commit.close()
+
+
+@requires_native
[email protected]('mode', ['batch', 'stream'])
[email protected]('ignore', [True, False])
+def test_native_empty_commit_preserves_python_option(tmp_path, mode, ignore):
+ table = _table(tmp_path).copy({'snapshot.ignore-empty-commit':
str(ignore).lower()})
+ commit = getattr(table, 'new_' + mode + '_write_builder')().new_commit()
+ try:
+ with _must_not_fallback(commit):
+ if mode == 'batch':
+ commit.commit([])
+ else:
+ commit.commit([], 7)
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ if ignore:
+ assert snapshot is None
+ else:
+ assert snapshot.commit_identifier == (BATCH_COMMIT_IDENTIFIER if
mode == 'batch' else 7)
+ assert snapshot.total_record_count == 0
+ finally:
+ commit.close()
+
+
+@requires_native
+def test_native_abort_removes_uncommitted_files(tmp_path):
+ table = _table(tmp_path)
+ builder = table.new_batch_write_builder()
+ messages = _prepare(builder, [{'id': 1, 'pt': None}])
+ files = list(tmp_path.rglob('data-*.parquet'))
+ assert files
+ commit = builder.new_commit()
+ try:
+ with patch.object(commit.file_store_commit, 'abort',
side_effect=AssertionError('Python fallback')):
+ commit.abort(messages)
+ assert not any(path.exists() for path in files)
+ assert table.snapshot_manager().get_latest_snapshot() is None
+ finally:
+ commit.close()
+
+
[email protected]('failure', ['missing', 'construction', 'conversion'])
+def test_preflight_failure_uses_python(tmp_path, failure):
+ table = _table(tmp_path)
+ builder = table.new_batch_write_builder()
+ messages = _prepare(builder, [{'id': 1, 'pt': 'a'}])
+ commit = builder.new_commit()
+ native = Mock()
+ with patch('pypaimon.write.native_commit.create_native_commit',
return_value=native) as create:
+ if failure == 'missing':
+ create.return_value = None
+ elif failure == 'construction':
+ create.side_effect = RuntimeError('unsupported FileIO')
+ with patch('pypaimon.write.native_commit.to_native_commit_messages',
+ side_effect=ValueError('unsupported payload')):
+ commit.commit(messages)
+ native.commit.assert_not_called()
+ assert _rows(table) == [{'id': 1, 'pt': 'a'}]
+ commit.close()
+
+
[email protected]('method', ['commit', 'abort'])
+def test_native_mutation_failure_never_falls_back_or_aborts(tmp_path, method):
+ table = _table(tmp_path)
+ builder = table.new_batch_write_builder()
+ messages = _prepare(builder, [{'id': 1, 'pt': 'a'}])
+ commit = builder.new_commit()
+ native = Mock()
+ getattr(native, method).side_effect = OSError('outcome unknown')
+ with patch('pypaimon.write.native_commit.create_native_commit',
return_value=native), \
+ patch('pypaimon.write.native_commit.to_native_commit_messages',
return_value=['wire']), \
+ patch.object(commit.file_store_commit, 'commit') as fallback, \
+ patch.object(commit.file_store_commit, 'abort') as abort:
+ with pytest.raises(OSError, match='outcome unknown'):
+ getattr(commit, method)(messages)
+ fallback.assert_not_called()
+ abort.assert_not_called()
+ if method == 'commit':
+ native.abort.assert_not_called()
+ with pytest.raises(RuntimeError, match='one-time'):
+ commit.commit(messages)
+ assert list(tmp_path.rglob('data-*.parquet'))
+ commit.close()
+
+
+@requires_native
+def
test_publication_response_loss_does_not_duplicate_or_delete_files(tmp_path):
+ table = _table(tmp_path)
+ builder = table.new_batch_write_builder()
+ messages = _prepare(builder, [{'id': 1, 'pt': None}])
+ native = create_native_commit(table, builder.commit_user)
+ proxy = Mock(wraps=native)
+
+ def publish_then_fail(identifier, messages):
+ native.commit(identifier, messages)
+ raise OSError('response lost')
+
+ proxy.commit.side_effect = publish_then_fail
+ commit = builder.new_commit()
+ try:
+ with patch('pypaimon.write.native_commit.create_native_commit',
return_value=proxy), \
+ _must_not_fallback(commit), \
+ patch.object(commit.file_store_commit, 'abort',
side_effect=AssertionError('abort')):
+ with pytest.raises(OSError, match='response lost'):
+ commit.commit(messages)
+ assert table.snapshot_manager().get_latest_snapshot().id == 1
+ assert _rows(table) == [{'id': 1, 'pt': None}]
+ proxy.abort.assert_not_called()
+ finally:
+ commit.close()
+
+
[email protected]('properties', [{}, {'source': 'python'}])
+def test_snapshot_properties_select_python_before_native(tmp_path, properties):
+ table = _table(tmp_path)
+ builder = table.new_batch_write_builder()
+ commit = builder.new_commit()
+ with patch('pypaimon.write.native_commit.create_native_commit',
+ side_effect=AssertionError('must not initialize native')) as
create:
+ commit.commit(_prepare(builder, [{'id': 1, 'pt': None}]),
snapshot_properties=properties)
+ create.assert_not_called()
+ assert table.snapshot_manager().get_latest_snapshot().properties ==
(properties or None)
+ commit.close()
+
+
[email protected]('warmup', [False, True])
+def test_callbacks_added_after_construction_select_python(tmp_path, warmup):
+ if warmup and not native_commit_available():
+ pytest.skip('native warmup requires the commit bindings')
+ table = _table(tmp_path)
+ builder = table.new_stream_write_builder()
+ commit = builder.new_commit()
+ if warmup:
+ with _must_not_fallback(commit):
+ commit.commit(_prepare(builder, [{'id': 6, 'pt': None}], 6), 6)
+ callback = Mock()
+ commit.add_commit_callback(callback)
+ with patch('pypaimon.write.native_commit.create_native_commit') as create:
+ commit.commit(_prepare(builder, [{'id': 1, 'pt': None}], 7), 7)
+ create.assert_not_called()
+ callback.call.assert_called_once()
+ commit.close()
+ callback.close.assert_called_once()
+
+
+def test_overwrite_and_truncate_use_python(tmp_path):
+ table = _table(tmp_path)
+ builder = table.new_batch_write_builder().overwrite({'pt': 'a'})
+ commit = builder.new_commit()
+ with patch('pypaimon.write.native_commit.create_native_commit') as create:
+ commit.commit(_prepare(builder, [{'id': 1, 'pt': 'a'}]))
+
table.new_batch_write_builder().new_commit().truncate_partitions([{'pt': 'a'}])
+ table.new_batch_write_builder().new_commit().truncate_table()
+ create.assert_not_called()
+ assert _rows(table) == []
+ commit.close()
+
+
+def test_disabled_option_never_initializes_native(tmp_path):
+ table = _table(tmp_path).copy({'commit.native.enabled': 'false'})
+ builder = table.new_batch_write_builder()
+ commit = builder.new_commit()
+ with patch('pypaimon.write.native_commit.create_native_commit') as create:
+ commit.commit(_prepare(builder, [{'id': 1, 'pt': None}]))
+ create.assert_not_called()
+ assert _rows(table) == [{'id': 1, 'pt': None}]
+ commit.close()
+
+
[email protected]('kind', ['version-managed', 'custom-env', 'branch',
'custom-io'])
+def test_incompatible_publication_environment_is_not_reconstructed(tmp_path,
kind):
+ table = _table(tmp_path)
+ if kind == 'version-managed':
+ table.catalog_environment.supports_version_management = True
+ elif kind == 'custom-env':
+ class CustomEnvironment(CatalogEnvironment):
+ pass
+ table.catalog_environment = CustomEnvironment()
+ elif kind == 'branch':
+ table.identifier = Identifier('default', 't', branch='dev')
+ else:
+ table.file_io = Mock()
+ with patch('pypaimon.write.native_commit.native_commit_available',
return_value=True), \
+
patch('pypaimon.write.native_commit._resolved_schema_file_io_options') as
resolve:
+ assert create_native_commit(table, 'job') is None
+ resolve.assert_not_called()
+
+
+def test_missing_capability_falls_back_without_reconstructing_table(tmp_path):
+ table = _table(tmp_path)
+ with patch('pypaimon.write.native_commit.native_method_available',
+ side_effect=lambda cls, method: (cls, method) !=
('CommitMessage', 'deserialize')), \
+
patch('pypaimon.write.native_commit._resolved_schema_file_io_options') as
resolve:
+ assert create_native_commit(table, 'job') is None
+ resolve.assert_not_called()
+
+
+def
test_partial_row_id_and_compact_messages_preserve_python_recovery(tmp_path):
+ table = _table(tmp_path, 'de')
+ assert not native_messages_supported(table, [CommitMessage((), 0, [],
check_from_snapshot=7)])
+ assert not native_messages_supported(table, [CommitMessage((), 0,
[Mock(first_row_id=1)])])
+ assert not native_messages_supported(table, [CommitMessage((), 0, [],
compact_after=[Mock()])])
+
+
+def test_close_releases_python_resources_even_if_native_close_fails(tmp_path):
+ commit = _table(tmp_path).new_batch_write_builder().new_commit()
+ commit._native_commit = Mock()
+ commit._native_commit.close.side_effect = OSError('close failure')
+ with patch.object(commit.file_store_commit, 'close') as close:
+ with pytest.raises(OSError, match='close failure'):
+ commit.close()
+ close.assert_called_once()
diff --git a/paimon-python/pypaimon/tests/table_commit_test.py
b/paimon-python/pypaimon/tests/table_commit_test.py
index 2c7079bcc7..5b71403b19 100644
--- a/paimon-python/pypaimon/tests/table_commit_test.py
+++ b/paimon-python/pypaimon/tests/table_commit_test.py
@@ -78,10 +78,13 @@ class TestTableCommit(unittest.TestCase):
commit = cls.__new__(cls)
commit.table = Mock()
commit.table.identifier = 'default.test_table'
+ commit.table.options.native_commit_enabled.return_value = False
commit.commit_user = 'test_user'
commit.overwrite_partition = overwrite_partition
commit.file_store_commit = Mock()
commit.batch_committed = False
+ commit._commit_callbacks = []
+ commit._native_commit = None
return commit, commit.file_store_commit
# -- Overwrite mode: should always call overwrite(), even with empty
messages --
diff --git a/paimon-python/pypaimon/write/commit_message.py
b/paimon-python/pypaimon/write/commit_message.py
index e3c6e15287..1325260cd2 100644
--- a/paimon-python/pypaimon/write/commit_message.py
+++ b/paimon-python/pypaimon/write/commit_message.py
@@ -29,12 +29,18 @@ class CommitMessage:
partition: Tuple
bucket: int
new_files: List[DataFileMeta]
- check_from_snapshot: Optional[int] = -1
+ check_from_snapshot: Optional[int] = None
deleted_files: List[DataFileMeta] = field(default_factory=list)
index_adds: List['IndexManifestEntry'] = field(default_factory=list)
index_deletes: List['IndexManifestEntry'] = field(default_factory=list)
changelog_files: List[DataFileMeta] = field(default_factory=list)
total_buckets: Optional[int] = None
+ # Java CommitMessageImpl keeps compaction changes separate from data
changes.
+ compact_before: List[DataFileMeta] = field(default_factory=list)
+ compact_after: List[DataFileMeta] = field(default_factory=list)
+ compact_changelog_files: List[DataFileMeta] = field(default_factory=list)
+ compact_index_adds: List['IndexManifestEntry'] =
field(default_factory=list)
+ compact_index_deletes: List['IndexManifestEntry'] =
field(default_factory=list)
def is_empty(self):
return (
@@ -43,4 +49,9 @@ class CommitMessage:
and not self.index_adds
and not self.index_deletes
and not self.changelog_files
+ and not self.compact_before
+ and not self.compact_after
+ and not self.compact_changelog_files
+ and not self.compact_index_adds
+ and not self.compact_index_deletes
)
diff --git a/paimon-python/pypaimon/write/commit_message_serializer.py
b/paimon-python/pypaimon/write/commit_message_serializer.py
new file mode 100644
index 0000000000..6b2b2ba2c7
--- /dev/null
+++ b/paimon-python/pypaimon/write/commit_message_serializer.py
@@ -0,0 +1,265 @@
+# 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.
+
+"""Java ``CommitMessageSerializer`` v14 body, without an outer version header.
+
+The caller must carry version 14 alongside these bytes, as Java's
+``ManifestCommittableSerializer`` does. The data and compaction increments are
+kept separate on the wire even when only the data increment is populated.
+"""
+
+from dataclasses import replace
+import struct
+from typing import List, Optional
+
+from pypaimon.globalindex.global_index_meta import GlobalIndexMeta
+from pypaimon.index.deletion_vector_meta import DeletionVectorMeta
+from pypaimon.index.index_file_meta import IndexFileMeta
+from pypaimon.manifest.index_manifest_entry import IndexManifestEntry
+from pypaimon.read.split_serializer import (
+ _DFM_FIELDS, _Reader, _Writer, _binary_array_header,
+ _datafilemeta_from_row, _round_to_word,
+ _serialize_data_file_meta,
+)
+from pypaimon.schema.data_types import AtomicType, DataField
+from pypaimon.table.row.binary_row import BinaryRow
+from pypaimon.table.row.generic_row import (
+ GenericRow, GenericRowDeserializer, GenericRowSerializer,
+)
+from pypaimon.write.commit_message import CommitMessage
+
+VERSION = 14
+
+
+def _fields(types):
+ return [DataField(i, str(i), AtomicType(t)) for i, t in enumerate(types)]
+
+
+_INDEX_FIELDS = _fields(['STRING', 'STRING', 'BIGINT', 'BIGINT', 'BYTES',
+ 'STRING', 'BYTES'])
+_DV_FIELDS = _fields(['STRING', 'INT', 'INT', 'BIGINT'])
+_GLOBAL_FIELDS = _fields(['BIGINT', 'BIGINT', 'INT', 'BYTES', 'BYTES',
'BYTES'])
+
+
+def _row(values, fields):
+ return GenericRowSerializer.to_bytes(GenericRow(values, fields))[4:]
+
+
+def _array_rows(rows: List[bytes]) -> bytes:
+ count = len(rows)
+ header = _binary_array_header(count)
+ data = bytearray(_round_to_word(header + count * 8))
+ struct.pack_into('<i', data, 0, count)
+ for i, row in enumerate(rows):
+ offset = len(data)
+ data.extend(row)
+ data.extend(b'\x00' * (_round_to_word(len(row)) - len(row)))
+ struct.pack_into('<Q', data, header + i * 8,
+ (offset << 32) | len(row))
+ return bytes(data)
+
+
+def _read_array_rows(data: bytes) -> List[bytes]:
+ if len(data) < 4:
+ raise ValueError('truncated BinaryArray<row>')
+ count = struct.unpack_from('<i', data)[0]
+ if count < 0:
+ raise ValueError('negative BinaryArray<row> count')
+ header = _binary_array_header(count)
+ fixed = _round_to_word(header + count * 8)
+ if fixed > len(data):
+ raise ValueError('truncated BinaryArray<row> fixed part')
+ result = []
+ for i in range(count):
+ if data[4 + i // 8] & (1 << (i % 8)):
+ raise ValueError('null BinaryArray<row> element')
+ slot = struct.unpack_from('<Q', data, header + i * 8)[0]
+ offset, length = slot >> 32, slot & 0xFFFFFFFF
+ if offset < fixed or offset + length > len(data):
+ raise ValueError('BinaryArray<row> element outside buffer')
+ result.append(data[offset:offset + length])
+ return result
+
+
+def _int_array(values: List[int]) -> bytes:
+ header = _binary_array_header(len(values))
+ data = bytearray(_round_to_word(header + len(values) * 4))
+ struct.pack_into('<i', data, 0, len(values))
+ for i, value in enumerate(values):
+ struct.pack_into('<i', data, header + i * 4, value)
+ return bytes(data)
+
+
+def _read_int_array(data: bytes) -> List[int]:
+ if len(data) < 4:
+ raise ValueError('truncated BinaryArray<int>')
+ count = struct.unpack_from('<i', data)[0]
+ header = _binary_array_header(count)
+ if count < 0 or header + count * 4 > len(data):
+ raise ValueError('invalid BinaryArray<int> size')
+ return [struct.unpack_from('<i', data, header + i * 4)[0]
+ for i in range(count)]
+
+
+def _index_row(meta: IndexFileMeta) -> bytes:
+ dv = None
+ if meta.dv_ranges is not None:
+ dv = _array_rows([
+ _row([value.data_file_name, value.offset, value.length,
+ value.cardinality], _DV_FIELDS)
+ for value in meta.dv_ranges.values()
+ ])
+ global_meta = None
+ if meta.global_index_meta is not None:
+ value = meta.global_index_meta
+ global_meta = _row([
+ value.row_range_start, value.row_range_end, value.index_field_id,
+ _int_array(value.extra_field_ids)
+ if value.extra_field_ids is not None else None,
+ value.index_meta, value.source_meta,
+ ], _GLOBAL_FIELDS)
+ return _row([meta.index_type, meta.file_name, meta.file_size,
+ meta.row_count, dv, meta.external_path, global_meta],
+ _INDEX_FIELDS)
+
+
+def _index_from_row(data: bytes) -> IndexFileMeta:
+ row = BinaryRow(struct.pack('>i', 7) + data, _INDEX_FIELDS)
+ dv = None
+ if row.get_field(4) is not None:
+ dv = {}
+ for raw in _read_array_rows(row.get_field(4)):
+ entry = BinaryRow(struct.pack('>i', 4) + raw, _DV_FIELDS)
+ value = DeletionVectorMeta(entry.get_field(0), entry.get_field(1),
+ entry.get_field(2), entry.get_field(3))
+ dv[value.data_file_name] = value
+ global_meta = None
+ if row.get_field(6) is not None:
+ value = BinaryRow(struct.pack('>i', 6) + row.get_field(6),
_GLOBAL_FIELDS)
+ extra = value.get_field(3)
+ global_meta = GlobalIndexMeta(
+ value.get_field(0), value.get_field(1), value.get_field(2),
+ _read_int_array(extra) if extra is not None else None,
+ value.get_field(4), value.get_field(5))
+ return IndexFileMeta(row.get_field(0), row.get_field(1), row.get_field(2),
+ row.get_field(3), dv, row.get_field(5), global_meta)
+
+
+def _write_list(writer: _Writer, values, encode):
+ writer.i32(len(values))
+ for value in values:
+ row = encode(value)
+ writer.i32(len(row))
+ writer.take(row)
+
+
+def _read_list(reader: _Reader, decode):
+ count = reader.i32()
+ if count < 0:
+ raise ValueError('negative CommitMessage list count')
+ return [decode(reader.take(reader.i32())) for _ in range(count)]
+
+
+def _file_row(meta):
+ # A local file_path is not Java's externalPath field.
+ return _serialize_data_file_meta(replace(meta, file_path=None), '')
+
+
+def _file_from_row(raw, key_fields):
+ if not key_fields:
+ row = BinaryRow(struct.pack('>i', len(_DFM_FIELDS)) + raw, _DFM_FIELDS)
+ for pos in (3, 4):
+ key = row.get_field(pos)
+ if key is not None and (len(key) < 4 or struct.unpack_from('>i',
key)[0] != 0):
+ raise ValueError('key_fields are required to decode
primary-key files')
+ meta = _datafilemeta_from_row(raw, '', len(_DFM_FIELDS), key_fields)
+ meta.file_path = None
+ return meta
+
+
+def serialize_commit_message(message: CommitMessage,
+ partition_fields: List[DataField]) -> bytes:
+ """Return the unframed Java v14 ``CommitMessageSerializer.serialize``
bytes."""
+ if len(message.partition) != len(partition_fields):
+ raise ValueError('partition arity does not match partition fields')
+ writer = _Writer()
+ partition = GenericRowSerializer.to_bytes(
+ GenericRow(list(message.partition), partition_fields))
+ writer.i32(len(partition))
+ writer.take(partition)
+ writer.i32(message.bucket)
+ writer.u8(1 if message.total_buckets is not None else 0)
+ if message.total_buckets is not None:
+ writer.i32(message.total_buckets)
+ for values in (message.new_files, message.deleted_files,
+ message.changelog_files):
+ _write_list(writer, values, _file_row)
+ for values in (message.index_adds, message.index_deletes):
+ _write_list(writer, values, lambda entry: _index_row(entry.index_file))
+ for values in (message.compact_before, message.compact_after,
+ message.compact_changelog_files):
+ _write_list(writer, values, _file_row)
+ for values in (message.compact_index_adds, message.compact_index_deletes):
+ _write_list(writer, values, lambda entry: _index_row(entry.index_file))
+ check = message.check_from_snapshot
+ writer.u8(0 if check is None else 1)
+ if check is not None:
+ writer.i64(check)
+ return writer.finish()
+
+
+def deserialize_commit_message(data: bytes, partition_fields: List[DataField],
+ key_fields: Optional[List[DataField]] = None,
+ version: int = VERSION) -> CommitMessage:
+ """Read a Java v14 body; ``version`` is supplied by the enclosing
format."""
+ if version != VERSION:
+ raise ValueError('unsupported CommitMessage version %d' % version)
+ reader = _Reader(data)
+ partition_bytes = reader.take(reader.i32())
+ if (len(partition_bytes) < 4 or
+ struct.unpack_from('>i', partition_bytes)[0] !=
len(partition_fields)):
+ raise ValueError('partition arity does not match partition fields')
+ partition = tuple(GenericRowDeserializer.from_bytes(
+ partition_bytes, partition_fields).values)
+ bucket = reader.i32()
+ total_buckets = reader.i32() if reader.u8() else None
+ file_reader = lambda raw: _file_from_row(raw, key_fields)
+ new_files = _read_list(reader, file_reader)
+ deleted_files = _read_list(reader, file_reader)
+ changelog_files = _read_list(reader, file_reader)
+
+ def index_list(kind):
+ return [IndexManifestEntry(kind, GenericRow(list(partition),
partition_fields),
+ bucket, meta)
+ for meta in _read_list(reader, _index_from_row)]
+
+ index_adds, index_deletes = index_list(0), index_list(1)
+ compact_before = _read_list(reader, file_reader)
+ compact_after = _read_list(reader, file_reader)
+ compact_changelog_files = _read_list(reader, file_reader)
+ compact_index_adds, compact_index_deletes = index_list(0), index_list(1)
+ check = reader.i64() if reader.u8() else None
+ reader.finish()
+ return CommitMessage(
+ partition=partition, bucket=bucket, new_files=new_files,
+ check_from_snapshot=check, deleted_files=deleted_files,
+ index_adds=index_adds, index_deletes=index_deletes,
+ changelog_files=changelog_files, total_buckets=total_buckets,
+ compact_before=compact_before, compact_after=compact_after,
+ compact_changelog_files=compact_changelog_files,
+ compact_index_adds=compact_index_adds,
+ compact_index_deletes=compact_index_deletes)
diff --git a/paimon-python/pypaimon/write/file_store_commit.py
b/paimon-python/pypaimon/write/file_store_commit.py
index 3fc49b1243..afe7476620 100644
--- a/paimon-python/pypaimon/write/file_store_commit.py
+++ b/paimon-python/pypaimon/write/file_store_commit.py
@@ -53,13 +53,51 @@ from pypaimon.write.commit_message import CommitMessage
logger = logging.getLogger(__name__)
+def _row_id_check_from_messages(messages: List[CommitMessage]) ->
Optional[int]:
+ """Check conflicts from the earliest baseline, matching the native
committer."""
+ check_from_snapshot = None
+ for message in messages:
+ snapshot = message.check_from_snapshot
+ if snapshot is None:
+ continue
+ if snapshot < 0:
+ raise ValueError('Invalid row-id check snapshot: %s' % snapshot)
+ check_from_snapshot = (snapshot if check_from_snapshot is None
+ else min(check_from_snapshot, snapshot))
+ if check_from_snapshot is not None:
+ for message in messages:
+ if message.check_from_snapshot is not None:
+ continue
+ if any(file.first_row_id is not None
+ for file in message.new_files + message.deleted_files):
+ raise ValueError(
+ 'A row-id commit message is missing its check-from
snapshot.')
+ return check_from_snapshot
+
+
+def _reject_compact_increment(messages: List[CommitMessage]):
+ # Java commits this increment as a separate COMPACT snapshot.
+ for message in messages:
+ if (message.compact_before or message.compact_after or
+ message.compact_changelog_files or
+ message.compact_index_adds or message.compact_index_deletes):
+ raise NotImplementedError(
+ 'Committing a compact increment requires a separate COMPACT
snapshot.')
+
+
def _abort_commit_messages(table, commit_messages: List[CommitMessage]):
"""Delete files created by messages known to be uncommitted."""
for message in commit_messages:
- for file in list(message.new_files) + list(message.changelog_files):
+ for file in (list(message.new_files) + list(message.changelog_files)
+ + list(message.compact_after)
+ + list(message.compact_changelog_files)):
path = None
try:
path = file.external_path or file.file_path
+ if not path:
+ bucket_path = table.path_factory().bucket_path(
+ tuple(message.partition), message.bucket)
+ path = '%s/%s' % (bucket_path.rstrip('/'), file.file_name)
if path:
table.file_io.delete_quietly(str(path))
except Exception as error:
@@ -68,7 +106,7 @@ def _abort_commit_messages(table, commit_messages:
List[CommitMessage]):
path,
error,
)
- for entry in message.index_adds:
+ for entry in message.index_adds + message.compact_index_adds:
file_name = None
try:
index_file = entry.index_file
@@ -197,11 +235,10 @@ class FileStoreCommit:
if not commit_messages and ignore_empty_commit:
return
- # Extract the minimum check_from_snapshot from commit messages
- valid_snapshots = [msg.check_from_snapshot for msg in commit_messages
- if msg.check_from_snapshot != -1]
- if valid_snapshots:
- self.conflict_detection._row_id_check_from_snapshot =
min(valid_snapshots)
+ _reject_compact_increment(commit_messages)
+ check_from_snapshot = _row_id_check_from_messages(commit_messages)
+ # A committer can be reused; an untagged commit clears the prior
baseline.
+ self.conflict_detection._row_id_check_from_snapshot =
check_from_snapshot
logger.info(
"Ready to commit to table %s, number of commit messages: %d",
@@ -226,7 +263,7 @@ class FileStoreCommit:
updated_cols = set()
written_partitions = set()
for msg in commit_messages:
- if msg.check_from_snapshot == -1:
+ if msg.check_from_snapshot is None:
continue
for f in msg.new_files:
write_cols =
self.table.table_schema.partial_file_write_cols(
@@ -282,6 +319,9 @@ class FileStoreCommit:
commit_identifier: int,
snapshot_properties: Optional[Dict[str, str]] = None):
"""Commit the given commit messages in overwrite mode."""
+ _reject_compact_increment(commit_messages)
+ self.conflict_detection._row_id_check_from_snapshot = (
+ _row_id_check_from_messages(commit_messages))
logger.info(
"Ready to overwrite to table %s, number of commit messages: %d",
self.table.identifier,
diff --git a/paimon-python/pypaimon/write/native_commit.py
b/paimon-python/pypaimon/write/native_commit.py
new file mode 100644
index 0000000000..decba9dcdc
--- /dev/null
+++ b/paimon-python/pypaimon/write/native_commit.py
@@ -0,0 +1,102 @@
+# 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 append commits using the Java CommitMessage v14 bridge."""
+
+from pypaimon.common.json_util import JSON
+from pypaimon.read.native_plan import (
+ _option_value_to_string, _resolved_schema_file_io_options,
+ native_method_available)
+from pypaimon.write.commit_message_serializer import serialize_commit_message
+
+
+def native_commit_available() -> bool:
+ """Probe capabilities instead of assuming every 0.4 development wheel has
them."""
+ return all(native_method_available(type_name, method) for type_name,
method in (
+ ('Table', 'from_resolved_schema'),
+ ('Table', 'new_stream_write_builder'),
+ ('StreamWriteBuilder', 'with_commit_user'),
+ ('StreamWriteBuilder', 'new_commit'),
+ ('StreamTableCommit', 'commit'),
+ ('StreamTableCommit', 'abort'),
+ ('StreamTableCommit', 'close'),
+ ('CommitMessage', 'deserialize'),
+ ))
+
+
+def native_messages_supported(table, messages) -> bool:
+ for message in messages:
+ if (message.compact_before or message.compact_after
+ or message.compact_changelog_files
+ or message.compact_index_adds or
message.compact_index_deletes):
+ return False
+ # Python can rewrite stale row-id files before retrying. The native
+ # committer does not yet implement that recovery path.
+ if table.options.data_evolution_enabled() and (
+ message.check_from_snapshot is not None
+ or any(file.first_row_id is not None
+ for file in message.new_files + message.deleted_files)):
+ return False
+ return True
+
+
+def create_native_commit(table, commit_user):
+ """Return a native committer only when its publication protocol matches
Python."""
+ from pypaimon.catalog.catalog_environment import CatalogEnvironment
+ from pypaimon.filesystem.local_file_io import LocalFileIO
+ from pypaimon.filesystem.pyarrow_file_io import PyArrowFileIO
+ from pypaimon.filesystem.resolving_file_io import ResolvingFileIO
+ from pypaimon.table.file_store_table import FileStoreTable
+
+ if not native_commit_available():
+ return None
+ # Native branch writes are not supported. Catalog-backed publication (REST
+ # or custom version management) must continue through Python's environment.
+ environment = table.catalog_environment
+ if (type(table) is not FileStoreTable
+ or type(environment) is not CatalogEnvironment
+ or environment.supports_version_management
+ or table.current_branch() != 'main'
+ or table.options.query_auth_enabled
+ or type(table.file_io) not in (LocalFileIO, PyArrowFileIO,
ResolvingFileIO)):
+ return None
+ file_io_options = _resolved_schema_file_io_options(table)
+ if file_io_options is None:
+ return None
+
+ from pypaimon_rust.datafusion import Table as NativeTable
+ # Preserve the resolved schema and all effective table options, including
+ # copy() overrides. Do not inject scan options or reload catalog schemas.
+ options = {str(key): _option_value_to_string(value)
+ for key, value in table.table_schema.options.items() if value
is not None}
+ native_table = NativeTable.from_resolved_schema(
+ table.table_path,
JSON.to_json(table.table_schema.copy(new_options=options)),
+ database=table.identifier.get_database_name(),
+ table=table.identifier.get_table_name(),
+ options=file_io_options)
+ # Both Python modes already enforce their public lifecycle/empty-commit
+ # contract. The stream builder preserves their existing writer identity;
+ # a native batch builder would mint a different commit user.
+ return
native_table.new_stream_write_builder().with_commit_user(commit_user).new_commit()
+
+
+def to_native_commit_messages(table, messages):
+ from pypaimon_rust.datafusion import CommitMessage as NativeCommitMessage
+ # Convert the whole batch before any native mutation can begin.
+ return [NativeCommitMessage.deserialize(
+ serialize_commit_message(message, table.partition_keys_fields),
version=14)
+ for message in messages]
diff --git a/paimon-python/pypaimon/write/table_commit.py
b/paimon-python/pypaimon/write/table_commit.py
index 35ee83f2fc..c6ba7bc3fa 100644
--- a/paimon-python/pypaimon/write/table_commit.py
+++ b/paimon-python/pypaimon/write/table_commit.py
@@ -52,6 +52,7 @@ class TableCommit:
raise RuntimeError("Table does not provide a SnapshotCommit
instance")
self._commit_callbacks: List[CommitCallback] = []
+ self._native_commit = None
self.file_store_commit = FileStoreCommit(
snapshot_commit, table, commit_user,
commit_callbacks=self._commit_callbacks)
@@ -95,13 +96,52 @@ class TableCommit:
"Committing table %s, %d non-empty messages",
self.table.identifier, len(non_empty_messages)
)
+ if snapshot_properties is None:
+ prepared = self._prepare_native_commit(non_empty_messages)
+ if prepared is not None:
+ native, messages = prepared
+ # Mutation is deliberately outside the fallback boundary:
+ # an exception can mean the snapshot was already published.
+ native.commit(commit_identifier, messages)
+ return
self.file_store_commit.commit(**commit_kwargs)
+ def _prepare_native_commit(self, messages):
+ if (not self.table.options.native_commit_enabled()
+ or self.overwrite_partition is not None
+ or self._commit_callbacks):
+ return None
+ try:
+ from pypaimon.write.native_commit import (
+ create_native_commit, native_messages_supported,
+ to_native_commit_messages)
+ if not native_messages_supported(self.table, messages):
+ return None
+ if self._native_commit is None:
+ self._native_commit = create_native_commit(self.table,
self.commit_user)
+ if self._native_commit is None:
+ return None
+ return self._native_commit, to_native_commit_messages(self.table,
messages)
+ except Exception as error:
+ # No native mutation has started. Preserve the normal Python path
+ # when the optional runtime, FileIO or wire bridge is unavailable.
+ logger.debug("Native commit preparation failed; using Python: %s",
error)
+ return None
+
def abort(self, commit_messages: List[CommitMessage]):
+ prepared = self._prepare_native_commit(commit_messages)
+ if prepared is not None:
+ native, messages = prepared
+ native.abort(messages)
+ return
self.file_store_commit.abort(commit_messages)
def close(self):
- self.file_store_commit.close()
+ try:
+ if self._native_commit is not None:
+ self._native_commit.close()
+ finally:
+ self.file_store_commit.close()
class BatchTableCommit(TableCommit):
