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 5fa4a337ed [python] Support native batch overwrite commits (#10097)
5fa4a337ed is described below
commit 5fa4a337edbe8e20cdfd0c6c2bce690041d48b46
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Sep 22 23:06:43 2026 +0800
[python] Support native batch overwrite commits (#10097)
---
paimon-python/README.md | 32 ++-
.../pypaimon/common/options/core_options.py | 2 +-
paimon-python/pypaimon/tests/native_commit_test.py | 220 +++++++++++++++++++--
paimon-python/pypaimon/tests/ray_sink_test.py | 11 +-
paimon-python/pypaimon/tests/table_commit_test.py | 15 +-
paimon-python/pypaimon/write/file_store_commit.py | 5 +-
paimon-python/pypaimon/write/native_commit.py | 35 ++--
paimon-python/pypaimon/write/table_commit.py | 14 +-
paimon-python/pypaimon/write/table_write.py | 3 +
paimon-python/pypaimon/write/write_builder.py | 17 +-
10 files changed, 279 insertions(+), 75 deletions(-)
diff --git a/paimon-python/README.md b/paimon-python/README.md
index 8e45ba7153..7da6734823 100644
--- a/paimon-python/README.md
+++ b/paimon-python/README.md
@@ -121,8 +121,8 @@ object-store requests.
# Native commit
-PyPaimon can submit append commits through the optional `pypaimon-rust`
-runtime. Enable it independently of native planning and reading:
+PyPaimon can submit append and batch overwrite commits through the optional
+`pypaimon-rust` runtime. Enable it independently of native planning and
reading:
```python
native_table = table.copy({"commit.native.enabled": "true"})
@@ -142,16 +142,34 @@ 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
+For batch overwrite, configure the Python builder as usual:
+
+```python
+builder = native_table.new_batch_write_builder().overwrite({"pt":
"2026-09-22"})
+```
+
+Overwrite preserves the Python commit user and follows the table's
+`dynamic-partition-overwrite` option. Dynamic overwrite replaces only
partitions
+present in the messages and does nothing for empty input. Static overwrite
+replaces partitions matching the spec, including for empty input; an empty spec
+matches the whole table. Empty overwrite of an unpartitioned table truncates
it.
+Static and unpartitioned overwrite record an OVERWRITE snapshot even when no
+files match, following Java; this also applies when the operation uses Python.
+
+Overwrite is configured only through `BatchWriteBuilder`, following Java's
+batch/stream API split. `StreamWriteBuilder` does not expose overwrite.
+
+Native commits require a runtime containing the batch identity bridge in
+[paimon-rust #916](https://github.com/apache/paimon-rust/pull/916), built on
+[#915](https://github.com/apache/paimon-rust/pull/915). If the optional
runtime is
+not installed, commits 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
+standard FileIO. 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
+Fallback is limited to runtime availability, 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.
diff --git a/paimon-python/pypaimon/common/options/core_options.py
b/paimon-python/pypaimon/common/options/core_options.py
index b329f62d19..ee7cb43d7c 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -681,7 +681,7 @@ class CoreOptions:
ConfigOptions.key("commit.native.enabled")
.boolean_type()
.default_value(False)
- .with_description("Commit append messages via pypaimon_rust.
Unsupported "
+ .with_description("Commit append and batch overwrite messages via
pypaimon_rust. Unsupported "
"operations use Python before any native commit is
attempted.")
)
diff --git a/paimon-python/pypaimon/tests/native_commit_test.py
b/paimon-python/pypaimon/tests/native_commit_test.py
index fde7def37c..67c9656c39 100644
--- a/paimon-python/pypaimon/tests/native_commit_test.py
+++ b/paimon-python/pypaimon/tests/native_commit_test.py
@@ -30,10 +30,11 @@ 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)
+from pypaimon.write.table_write import StreamTableWrite
requires_native = pytest.mark.skipif(
- not native_commit_available(), reason='pypaimon-rust CommitMessage/commit
API required')
+ not native_commit_available(), reason='pypaimon-rust runtime required')
def _table(tmp_path, mode='append', backend='filesystem'):
@@ -50,7 +51,7 @@ def _table(tmp_path, mode='append', backend='filesystem'):
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)
+ partition_keys=[] if mode in ('pk', 'unpartitioned') else ['pt']),
False)
table = catalog.get_table('default.t')
if backend == 'path':
table = FileStoreTable.from_path(table.table_path)
@@ -75,8 +76,18 @@ def _rows(table):
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 _must_not_fallback(commit, method='commit'):
+ return patch.object(commit.file_store_commit, method,
side_effect=AssertionError('Python fallback'))
+
+
+def _seed(table):
+ builder = table.copy({'commit.native.enabled':
'false'}).new_batch_write_builder()
+ commit = builder.new_commit()
+ try:
+ commit.commit(_prepare(builder, [
+ {'id': 1, 'pt': 'a'}, {'id': 2, 'pt': 'b'}, {'id': 3, 'pt':
None}]))
+ finally:
+ commit.close()
def test_native_commit_is_opt_in():
@@ -84,6 +95,18 @@ def test_native_commit_is_opt_in():
assert CoreOptions(Options({'commit.native.enabled':
'true'})).native_commit_enabled()
+def test_overwrite_builder_api_is_batch_only(tmp_path):
+ table = _table(tmp_path)
+ batch = table.new_batch_write_builder()
+ assert batch.overwrite({'pt': 'a'}) is batch
+ assert batch.static_partition == {'pt': 'a'}
+ stream = table.new_stream_write_builder()
+ with pytest.raises(AttributeError):
+ stream.overwrite({'pt': 'a'})
+ with pytest.raises(TypeError):
+ StreamTableWrite(table, stream.commit_user, {'pt': 'a'})
+
+
@requires_native
@pytest.mark.parametrize('backend', ['filesystem', 'path', 'jdbc'])
@pytest.mark.parametrize('mode', ['append', 'pk', 'de'])
@@ -123,6 +146,116 @@ def
test_native_stream_reuses_commit_user_and_identifiers(tmp_path):
commit.close()
+@requires_native
[email protected]('mode,dynamic,spec', [
+ ('append', True, {'pt': 'ignored'}),
+ ('append', False, {'pt': 'a'}),
+ ('de', True, {}),
+ ('pk', True, {}),
+ ('unpartitioned', True, {}),
+])
+def test_native_overwrite_replaces_only_target_rows(tmp_path, mode, dynamic,
spec):
+ table = _table(tmp_path, mode).copy({
+ 'dynamic-partition-overwrite': str(dynamic).lower(),
'commit.user-prefix': 'python'})
+ _seed(table)
+ builder = table.new_batch_write_builder().overwrite(spec)
+ commit = builder.new_commit()
+ try:
+ messages = _prepare(builder, [{'id': 4, 'pt': 'a'}])
+ with _must_not_fallback(commit, 'overwrite'):
+ commit.commit(messages)
+ expected = [{'id': 4, 'pt': 'a'}]
+ if table.partition_keys:
+ expected = [{'id': 2, 'pt': 'b'}, {'id': 3, 'pt': None}] + expected
+ assert _rows(table) == expected
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ assert snapshot.commit_user == builder.commit_user
+ assert snapshot.commit_identifier == BATCH_COMMIT_IDENTIFIER
+ assert snapshot.commit_kind == 'OVERWRITE'
+ with pytest.raises(RuntimeError, match='one-time'):
+ commit.commit(messages)
+ finally:
+ commit.close()
+
+
+@requires_native
[email protected]('mode,dynamic,spec,remaining', [
+ ('append', True, {}, [1, 2, 3]),
+ ('append', False, {'pt': 'a'}, [2, 3]),
+ ('append', False, {'pt': None}, [1, 2]),
+ ('append', False, {'pt': '__DEFAULT_PARTITION__'}, [1, 2]),
+ ('append', False, {}, []),
+ ('unpartitioned', True, {}, []),
+ ('pk', True, {}, []),
+])
+def test_native_empty_overwrite_semantics(tmp_path, mode, dynamic, spec,
remaining):
+ table = _table(tmp_path, mode).copy({'dynamic-partition-overwrite':
str(dynamic).lower()})
+ _seed(table)
+ builder = table.new_batch_write_builder().overwrite(spec)
+ commit = builder.new_commit()
+ try:
+ with _must_not_fallback(commit, 'overwrite'):
+ commit.commit([])
+ assert [row['id'] for row in _rows(table)] == remaining
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ if remaining == [1, 2, 3]:
+ assert snapshot.id == 1
+ else:
+ assert snapshot.id == 2
+ assert snapshot.commit_user == builder.commit_user
+ assert snapshot.commit_kind == 'OVERWRITE'
+ finally:
+ commit.close()
+
+
+@requires_native
[email protected]('value', ['off', '0', ' false '])
+def test_native_overwrite_normalizes_python_boolean_option(tmp_path, value):
+ table = _table(tmp_path).copy({
+ 'dynamic-partition-overwrite': value, 'snapshot.ignore-empty-commit':
value})
+ _seed(table)
+ commit = table.new_batch_write_builder().overwrite({'pt':
'a'}).new_commit()
+ try:
+ with _must_not_fallback(commit, 'overwrite'):
+ commit.commit([])
+ assert [row['id'] for row in _rows(table)] == [2, 3]
+ finally:
+ commit.close()
+
+
[email protected]('native', [False, True])
[email protected]('case', ['unpartitioned', 'static-empty',
'static-missing', 'dynamic-empty'])
+def test_empty_overwrite_records_java_snapshot(tmp_path, native, case):
+ if native and not native_commit_available():
+ pytest.skip('pypaimon-rust runtime required')
+ table = _table(tmp_path, 'unpartitioned' if case == 'unpartitioned' else
'append').copy({
+ 'commit.native.enabled': str(native).lower(),
+ 'dynamic-partition-overwrite': str(case == 'dynamic-empty').lower(),
+ })
+ if case == 'static-missing':
+ _seed(table)
+ builder = table.new_batch_write_builder().overwrite(
+ {'pt': 'missing'} if case.startswith('static') else {})
+ commit = builder.new_commit()
+ try:
+ if native:
+ with _must_not_fallback(commit, 'overwrite'):
+ commit.commit([])
+ else:
+ commit.commit([])
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ if case == 'dynamic-empty':
+ assert snapshot is None
+ else:
+ assert snapshot.id == (2 if case == 'static-missing' else 1)
+ assert snapshot.commit_user == builder.commit_user
+ assert snapshot.commit_kind == 'OVERWRITE'
+ assert snapshot.total_record_count == (3 if case ==
'static-missing' else 0)
+ assert [row['id'] for row in _rows(table)] == ([1, 2, 3] if case ==
'static-missing' else [])
+ finally:
+ commit.close()
+
+
@requires_native
@pytest.mark.parametrize('mode', ['batch', 'stream'])
@pytest.mark.parametrize('ignore', [True, False])
@@ -146,9 +279,12 @@ def
test_native_empty_commit_preserves_python_option(tmp_path, mode, ignore):
@requires_native
-def test_native_abort_removes_uncommitted_files(tmp_path):
[email protected]('overwrite', [False, True])
+def test_native_abort_removes_uncommitted_files(tmp_path, overwrite):
table = _table(tmp_path)
builder = table.new_batch_write_builder()
+ if overwrite:
+ builder.overwrite()
messages = _prepare(builder, [{'id': 1, 'pt': None}])
files = list(tmp_path.rglob('data-*.parquet'))
assert files
@@ -183,9 +319,12 @@ def test_preflight_failure_uses_python(tmp_path, failure):
@pytest.mark.parametrize('method', ['commit', 'abort'])
-def test_native_mutation_failure_never_falls_back_or_aborts(tmp_path, method):
[email protected]('overwrite', [False, True])
+def test_native_mutation_failure_never_falls_back_or_aborts(tmp_path, method,
overwrite):
table = _table(tmp_path)
builder = table.new_batch_write_builder()
+ if overwrite:
+ builder.overwrite()
messages = _prepare(builder, [{'id': 1, 'pt': 'a'}])
commit = builder.new_commit()
native = Mock()
@@ -193,10 +332,12 @@ def
test_native_mutation_failure_never_falls_back_or_aborts(tmp_path, method):
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, 'overwrite') as
fallback_overwrite, \
patch.object(commit.file_store_commit, 'abort') as abort:
with pytest.raises(OSError, match='outcome unknown'):
getattr(commit, method)(messages)
fallback.assert_not_called()
+ fallback_overwrite.assert_not_called()
abort.assert_not_called()
if method == 'commit':
native.abort.assert_not_called()
@@ -207,22 +348,25 @@ def
test_native_mutation_failure_never_falls_back_or_aborts(tmp_path, method):
@requires_native
-def
test_publication_response_loss_does_not_duplicate_or_delete_files(tmp_path):
[email protected]('overwrite', [False, True])
+def
test_publication_response_loss_does_not_duplicate_or_delete_files(tmp_path,
overwrite):
table = _table(tmp_path)
builder = table.new_batch_write_builder()
+ if overwrite:
+ builder.overwrite()
messages = _prepare(builder, [{'id': 1, 'pt': None}])
- native = create_native_commit(table, builder.commit_user)
+ native = create_native_commit(table, builder.commit_user,
builder.static_partition)
proxy = Mock(wraps=native)
- def publish_then_fail(identifier, messages):
- native.commit(identifier, messages)
+ def publish_then_fail(*args):
+ native.commit(*args)
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), \
+ _must_not_fallback(commit, 'overwrite' if overwrite else
'commit'), \
patch.object(commit.file_store_commit, 'abort',
side_effect=AssertionError('abort')):
with pytest.raises(OSError, match='response lost'):
commit.commit(messages)
@@ -234,9 +378,12 @@ def
test_publication_response_loss_does_not_duplicate_or_delete_files(tmp_path):
@pytest.mark.parametrize('properties', [{}, {'source': 'python'}])
-def test_snapshot_properties_select_python_before_native(tmp_path, properties):
[email protected]('overwrite', [False, True])
+def test_snapshot_properties_select_python_before_native(tmp_path, properties,
overwrite):
table = _table(tmp_path)
builder = table.new_batch_write_builder()
+ if overwrite:
+ builder.overwrite()
commit = builder.new_commit()
with patch('pypaimon.write.native_commit.create_native_commit',
side_effect=AssertionError('must not initialize native')) as
create:
@@ -266,17 +413,53 @@ def
test_callbacks_added_after_construction_select_python(tmp_path, warmup):
callback.close.assert_called_once()
-def test_overwrite_and_truncate_use_python(tmp_path):
+def test_truncate_uses_python(tmp_path):
table = _table(tmp_path)
+ _seed(table)
+ with patch('pypaimon.write.native_commit.create_native_commit') as create:
+ for method, args in [('truncate_partitions', ([{'pt': 'a'}],)),
('truncate_table', ())]:
+ commit = table.new_batch_write_builder().new_commit()
+ try:
+ getattr(commit, method)(*args)
+ finally:
+ commit.close()
+ create.assert_not_called()
+ assert _rows(table) == []
+
+
+def test_overwrite_conversion_failure_preserves_partition_scope(tmp_path):
+ table = _table(tmp_path).copy({'dynamic-partition-overwrite': 'false'})
+ _seed(table)
builder = table.new_batch_write_builder().overwrite({'pt': 'a'})
commit = builder.new_commit()
+ native = Mock()
+ try:
+ messages = _prepare(builder, [{'id': 4, 'pt': 'a'}])
+ with patch('pypaimon.write.native_commit.create_native_commit',
return_value=native), \
+ patch('pypaimon.write.native_commit.to_native_commit_messages',
+ side_effect=ValueError('unsupported payload')):
+ commit.commit(messages)
+ native.commit.assert_not_called()
+ native.abort.assert_not_called()
+ assert _rows(table) == [{'id': 2, 'pt': 'b'}, {'id': 3, 'pt': None},
{'id': 4, 'pt': 'a'}]
+ assert table.snapshot_manager().get_latest_snapshot().commit_user ==
builder.commit_user
+ finally:
+ commit.close()
+
+
+def test_overwrite_callbacks_select_python(tmp_path):
+ table = _table(tmp_path)
+ builder = table.new_batch_write_builder().overwrite()
+ commit = builder.new_commit()
+ 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': '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) == []
+ assert _rows(table) == [{'id': 1, 'pt': 'a'}]
+ callback.call.assert_called_once()
commit.close()
+ callback.close.assert_called_once()
def test_disabled_option_never_initializes_native(tmp_path):
@@ -309,10 +492,9 @@ def
test_incompatible_publication_environment_is_not_reconstructed(tmp_path, kin
resolve.assert_not_called()
-def test_missing_capability_falls_back_without_reconstructing_table(tmp_path):
+def test_missing_runtime_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')), \
+ with patch('pypaimon.write.native_commit.native_commit_available',
return_value=False), \
patch('pypaimon.write.native_commit._resolved_schema_file_io_options') as
resolve:
assert create_native_commit(table, 'job') is None
resolve.assert_not_called()
diff --git a/paimon-python/pypaimon/tests/ray_sink_test.py
b/paimon-python/pypaimon/tests/ray_sink_test.py
index d5db6f3074..72c565828e 100644
--- a/paimon-python/pypaimon/tests/ray_sink_test.py
+++ b/paimon-python/pypaimon/tests/ray_sink_test.py
@@ -172,14 +172,11 @@ class RaySinkTest(unittest.TestCase):
finally:
batch_write.close()
- stream_write = (
- self.table
- .new_stream_write_builder()
- .overwrite({'dt': '2024-01-01'})
- .new_write()
- )
+ stream_builder = self.table.new_stream_write_builder()
+ self.assertFalse(hasattr(stream_builder, 'overwrite'))
+ stream_write = stream_builder.new_write()
try:
- self.assertEqual(stream_write.static_partition, {'dt':
'2024-01-01'})
+ self.assertIsNone(stream_write.static_partition)
finally:
stream_write.close()
diff --git a/paimon-python/pypaimon/tests/table_commit_test.py
b/paimon-python/pypaimon/tests/table_commit_test.py
index 5b71403b19..82cd49291c 100644
--- a/paimon-python/pypaimon/tests/table_commit_test.py
+++ b/paimon-python/pypaimon/tests/table_commit_test.py
@@ -166,18 +166,9 @@ class TestTableCommit(unittest.TestCase):
snapshot_properties={"source": "capture"},
)
- # -- StreamTableCommit overwrite should also reach overwrite() with empty
messages --
-
- def test_stream_commit_overwrite_empty_messages(self):
- commit, mock_fsc = self._create_commit(StreamTableCommit,
overwrite_partition={'dt': '2024-01-15'})
-
- commit.commit([], commit_identifier=42)
-
- mock_fsc.overwrite.assert_called_once_with(
- overwrite_partition={'dt': '2024-01-15'},
- commit_messages=[],
- commit_identifier=42,
- )
+ def test_stream_commit_does_not_accept_overwrite_configuration(self):
+ with self.assertRaises(TypeError):
+ StreamTableCommit(Mock(), 'job', {'dt': '2024-01-15'})
def test_stream_commit_forwards_snapshot_properties(self):
commit, mock_fsc = self._create_commit(
diff --git a/paimon-python/pypaimon/write/file_store_commit.py
b/paimon-python/pypaimon/write/file_store_commit.py
index afe7476620..5c0b6f7427 100644
--- a/paimon-python/pypaimon/write/file_store_commit.py
+++ b/paimon-python/pypaimon/write/file_store_commit.py
@@ -361,6 +361,8 @@ class FileStoreCommit:
index_deletes=index_deletes,
index_adds=index_adds,
snapshot_properties=snapshot_properties,
+ # Java records static/table overwrite even when no files match.
+ allow_empty_commit=True,
)
def _overwrite_hash_index_deletes(self, partition_filter, deletes):
@@ -476,8 +478,7 @@ class FileStoreCommit:
else commit_entries_plan(latest_snapshot)
)
- # Append can explicitly publish an empty snapshot for tagging.
- # No-op overwrite/drop operations retain their existing behavior.
+ # Callers opt in when the operation records an empty snapshot.
if (not allow_empty_commit and not commit_entries
and not index_deletes and not index_adds):
break
diff --git a/paimon-python/pypaimon/write/native_commit.py
b/paimon-python/pypaimon/write/native_commit.py
index decba9dcdc..86fc008942 100644
--- a/paimon-python/pypaimon/write/native_commit.py
+++ b/paimon-python/pypaimon/write/native_commit.py
@@ -15,27 +15,18 @@
# specific language governing permissions and limitations
# under the License.
-"""Optional native append commits using the Java CommitMessage v14 bridge."""
+"""Optional native 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)
+ _option_value_to_string, _resolved_schema_file_io_options)
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'),
- ))
+ """Whether the optional Rust runtime is installed."""
+ from importlib.util import find_spec
+ return find_spec('pypaimon_rust') is not None
def native_messages_supported(table, messages) -> bool:
@@ -54,7 +45,7 @@ def native_messages_supported(table, messages) -> bool:
return True
-def create_native_commit(table, commit_user):
+def create_native_commit(table, commit_user, overwrite_partition=None):
"""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
@@ -83,14 +74,22 @@ def create_native_commit(table, commit_user):
# 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}
+ # Python accepts boolean spellings such as "off"; pass the parsed value.
+ options['dynamic-partition-overwrite'] = _option_value_to_string(
+ table.options.dynamic_partition_overwrite())
+ options['snapshot.ignore-empty-commit'] = _option_value_to_string(
+ table.options.snapshot_ignore_empty_commit())
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.
+ if overwrite_partition is not None:
+ return (native_table.new_batch_write_builder()
+ ._with_commit_user(commit_user)
+ .with_overwrite(overwrite_partition).new_commit())
+ # Append commits use the stream committer with the Python writer's
identity;
+ # Python enforces each mode's lifecycle and empty-commit rules.
return
native_table.new_stream_write_builder().with_commit_user(commit_user).new_commit()
diff --git a/paimon-python/pypaimon/write/table_commit.py
b/paimon-python/pypaimon/write/table_commit.py
index c6ba7bc3fa..ef2595d93c 100644
--- a/paimon-python/pypaimon/write/table_commit.py
+++ b/paimon-python/pypaimon/write/table_commit.py
@@ -85,6 +85,13 @@ class TableCommit:
"Committing overwrite to 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
+ # Keep publication failures outside the preparation
fallback.
+ native.commit(messages)
+ return
self.file_store_commit.overwrite(
overwrite_partition=self.overwrite_partition,
**commit_kwargs)
@@ -108,7 +115,6 @@ class TableCommit:
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:
@@ -118,7 +124,8 @@ class TableCommit:
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)
+ self._native_commit = create_native_commit(
+ self.table, self.commit_user, self.overwrite_partition)
if self._native_commit is None:
return None
return self._native_commit, to_native_commit_messages(self.table,
messages)
@@ -185,6 +192,9 @@ class StreamTableCommit(TableCommit):
:meth:`StreamTableWrite.prepare_commit`.
"""
+ def __init__(self, table, commit_user: str):
+ super().__init__(table, commit_user, None)
+
def commit(
self,
commit_messages: List[CommitMessage],
diff --git a/paimon-python/pypaimon/write/table_write.py
b/paimon-python/pypaimon/write/table_write.py
index 37ee3fe4c8..87867b54b5 100644
--- a/paimon-python/pypaimon/write/table_write.py
+++ b/paimon-python/pypaimon/write/table_write.py
@@ -390,6 +390,9 @@ class BatchTableWrite(TableWrite):
class StreamTableWrite(TableWrite):
+ def __init__(self, table, commit_user):
+ super().__init__(table, commit_user, None)
+
def prepare_commit(self, commit_identifier) -> List[CommitMessage]:
messages = self._prepare_commit(commit_identifier)
self._release_prepared_indexes()
diff --git a/paimon-python/pypaimon/write/write_builder.py
b/paimon-python/pypaimon/write/write_builder.py
index f7a0459305..1c64d616e2 100644
--- a/paimon-python/pypaimon/write/write_builder.py
+++ b/paimon-python/pypaimon/write/write_builder.py
@@ -33,11 +33,6 @@ class WriteBuilder(ABC):
self.table: FileStoreTable = table
self.commit_user = self._create_commit_user()
- self.static_partition = None
-
- def overwrite(self, static_partition: Optional[dict] = None):
- self.static_partition = static_partition if static_partition is not
None else {}
- return self
def new_write(self) -> TableWrite:
"""Returns a table write."""
@@ -58,6 +53,14 @@ class WriteBuilder(ABC):
class BatchWriteBuilder(WriteBuilder):
+ def __init__(self, table):
+ super().__init__(table)
+ self.static_partition = None
+
+ def overwrite(self, static_partition: Optional[dict] = None):
+ self.static_partition = static_partition if static_partition is not
None else {}
+ return self
+
def new_write(self) -> BatchTableWrite:
return BatchTableWrite(self.table, self.commit_user,
self.static_partition)
@@ -72,11 +75,11 @@ class BatchWriteBuilder(WriteBuilder):
class StreamWriteBuilder(WriteBuilder):
def new_write(self) -> StreamTableWrite:
- return StreamTableWrite(self.table, self.commit_user,
self.static_partition)
+ return StreamTableWrite(self.table, self.commit_user)
def new_update(self) -> StreamTableUpdate:
return StreamTableUpdate(self.table, self.commit_user)
def new_commit(self) -> StreamTableCommit:
- commit = StreamTableCommit(self.table, self.commit_user,
self.static_partition)
+ commit = StreamTableCommit(self.table, self.commit_user)
return commit
