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 547c720e4a [python] Add opt-in native writes to PyPaimon (#10154)
547c720e4a is described below
commit 547c720e4a04574ec60db369b4177e53dbb8e2be
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Sep 24 11:00:52 2026 +0800
[python] Add opt-in native writes to PyPaimon (#10154)
---
paimon-python/README.md | 40 ++++-
.../pypaimon/common/options/core_options.py | 12 ++
paimon-python/pypaimon/tests/native_write_test.py | 164 +++++++++++++++++++++
paimon-python/pypaimon/write/native_commit.py | 27 ++--
paimon-python/pypaimon/write/native_write.py | 162 ++++++++++++++++++++
paimon-python/pypaimon/write/write_builder.py | 21 ++-
6 files changed, 409 insertions(+), 17 deletions(-)
diff --git a/paimon-python/README.md b/paimon-python/README.md
index 3295d70df3..96f7ba0d9f 100644
--- a/paimon-python/README.md
+++ b/paimon-python/README.md
@@ -119,6 +119,36 @@ table = table.copy({"parquet.filter.columnindex.enabled":
"false"})
Unsupported reads use the normal path. Reading fewer bytes may require more
object-store requests.
+# Native write and commit
+
+PyPaimon can write Arrow batches through the optional `pypaimon-rust` runtime.
+Enable it on a table independently of native commit:
+
+```python
+native_table = table.copy({"write.native.enabled": "true",
+ "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 native writer returns ordinary PyPaimon commit messages, so the Python
+committer also works when `commit.native.enabled` is false. Batch overwrite and
+reusable stream writers retain the builder's commit user and identifier. Native
+write is currently limited to Parquet tables without BLOB fields or
+data-evolution mode, on the same filesystem/JDBC publication route as native
+commit. Writer methods requiring Python's specialized path select the Python
+writer before native data is written. If the runtime or table route is
+unavailable, write uses Python. Once Rust starts writing a batch, errors
+propagate without retrying that batch through Python.
+
+Both native options are disabled by default.
+
# Native commit
PyPaimon can submit append and batch overwrite commits through the optional
@@ -136,11 +166,11 @@ finally:
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.
+When only native commit is enabled, the Python writer 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.
For batch overwrite, configure the Python builder as usual:
diff --git a/paimon-python/pypaimon/common/options/core_options.py
b/paimon-python/pypaimon/common/options/core_options.py
index ee7cb43d7c..84fd675349 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -685,6 +685,15 @@ class CoreOptions:
"operations use Python before any native commit is
attempted.")
)
+ WRITE_NATIVE_ENABLED: ConfigOption[bool] = (
+ ConfigOptions.key("write.native.enabled")
+ .boolean_type()
+ .default_value(False)
+ .with_description("Write Arrow data via pypaimon_rust when the table
and "
+ "writer API are supported. Unsupported routes use
Python "
+ "before any native data is written.")
+ )
+
CHANGELOG_PRODUCER: ConfigOption[ChangelogProducer] = (
ConfigOptions.key("changelog-producer")
.enum_type(ChangelogProducer)
@@ -1603,6 +1612,9 @@ class CoreOptions:
def native_commit_enabled(self, default=None):
return self.options.get(CoreOptions.COMMIT_NATIVE_ENABLED, default)
+ def native_write_enabled(self, default=None):
+ return self.options.get(CoreOptions.WRITE_NATIVE_ENABLED, default)
+
def changelog_producer(self, default=None):
return self.options.get(CoreOptions.CHANGELOG_PRODUCER, default)
diff --git a/paimon-python/pypaimon/tests/native_write_test.py
b/paimon-python/pypaimon/tests/native_write_test.py
new file mode 100644
index 0000000000..7224dbe09a
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_write_test.py
@@ -0,0 +1,164 @@
+# 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 data writer bridge."""
+
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.common.options.core_options import CoreOptions
+from pypaimon.common.options.options import Options
+from pypaimon.write.native_write import NativeTableWrite,
native_write_available
+
+
+requires_native = pytest.mark.skipif(
+ not native_write_available(), reason='pypaimon-rust writer required')
+
+
+def _table(tmp_path, primary_key=False, commit_native=True):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ options = {'file.format': 'parquet', 'write.native.enabled': 'true',
+ 'commit.native.enabled': str(commit_native).lower()}
+ if primary_key:
+ options['bucket'] = '1'
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ pa.schema([('id', pa.int64()), ('pt', pa.string())]),
+ options=options, primary_keys=['id'] if primary_key else [],
+ partition_keys=[] if primary_key else ['pt']), False)
+ return catalog.get_table('default.t')
+
+
+def _batch(ids, partitions):
+ return pa.record_batch(
+ [pa.array(ids, pa.int64()), pa.array(partitions, pa.string())],
+ names=['id', 'pt'])
+
+
+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 test_native_write_is_opt_in():
+ assert not CoreOptions(Options({})).native_write_enabled()
+ assert CoreOptions(Options({'write.native.enabled':
'true'})).native_write_enabled()
+
+
+@requires_native
[email protected]('primary_key', [False, True])
[email protected]('commit_native', [False, True])
+def test_batch_native_write_commits_through_both_committers(
+ tmp_path, primary_key, commit_native):
+ table = _table(tmp_path, primary_key, commit_native)
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ assert isinstance(writer, NativeTableWrite)
+ try:
+ writer.write_arrow_batch(_batch([1], ['a']))
+ writer.write_arrow(pa.Table.from_batches([_batch([2], ['b'])]))
+ messages = writer.prepare_commit()
+ assert messages and sum(file.row_count for msg in messages
+ for file in msg.new_files) == 2
+ commit = builder.new_commit()
+ try:
+ commit.commit(messages)
+ finally:
+ commit.close()
+ finally:
+ writer.close()
+ assert _rows(table) == [{'id': 1, 'pt': 'a'}, {'id': 2, 'pt': 'b'}]
+ assert table.snapshot_manager().get_latest_snapshot().commit_user ==
builder.commit_user
+
+
+@requires_native
+def test_stream_native_write_reuses_writer_across_checkpoints(tmp_path):
+ table = _table(tmp_path)
+ builder = table.new_stream_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ assert isinstance(writer, NativeTableWrite)
+ try:
+ for identifier in (7, 8):
+ writer.write_arrow_batch(_batch([identifier], ['a']))
+ commit.commit(writer.prepare_commit(identifier), identifier)
+ assert _rows(table) == [{'id': 7, 'pt': 'a'}, {'id': 8, 'pt': 'a'}]
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ assert snapshot.commit_user == builder.commit_user
+ assert snapshot.commit_identifier == 8
+ finally:
+ writer.close()
+ commit.close()
+
+
+@requires_native
+def test_native_overwrite_and_empty_overwrite(tmp_path):
+ table = _table(tmp_path)
+ seed = table.new_batch_write_builder()
+ seed_writer = seed.new_write()
+ seed_writer.write_arrow_batch(_batch([1, 2], ['a', 'b']))
+ seed.new_commit().commit(seed_writer.prepare_commit())
+ seed_writer.close()
+
+ overwrite = table.new_batch_write_builder().overwrite({'pt': 'a'})
+ writer = overwrite.new_write()
+ assert isinstance(writer, NativeTableWrite)
+ writer.write_arrow_batch(_batch([3], ['a']))
+ overwrite.new_commit().commit(writer.prepare_commit())
+ writer.close()
+ assert _rows(table) == [{'id': 2, 'pt': 'b'}, {'id': 3, 'pt': 'a'}]
+
+ static_table = table.copy({'dynamic-partition-overwrite': 'false'})
+ empty = static_table.new_batch_write_builder().overwrite({'pt': 'a'})
+ empty_writer = empty.new_write()
+ assert isinstance(empty_writer, NativeTableWrite)
+ empty.new_commit().commit(empty_writer.prepare_commit())
+ empty_writer.close()
+ assert _rows(table) == [{'id': 2, 'pt': 'b'}]
+
+
+@requires_native
+def test_advanced_api_switches_before_write_and_rejects_late_switch(tmp_path):
+ table = _table(tmp_path)
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ assert isinstance(writer, NativeTableWrite)
+ assert writer.with_write_type(['id', 'pt']) is writer._python_writer
+ assert writer._native_writer is None
+ writer.write_arrow_batch(_batch([1], ['a']))
+ builder.new_commit().commit(writer.prepare_commit())
+ writer.close()
+ assert _rows(table) == [{'id': 1, 'pt': 'a'}]
+
+ native = table.new_batch_write_builder().new_write()
+ native.write_arrow_batch(_batch([2], ['a']))
+ with pytest.raises(RuntimeError, match='after native data'):
+ native.with_write_type(['id'])
+ native.abort()
+
+
+@requires_native
+def
test_unavailable_native_writer_falls_back_before_table_reconstruction(tmp_path):
+ table = _table(tmp_path)
+ with patch('pypaimon.write.native_write.native_write_available',
return_value=False), \
+ patch('pypaimon.write.native_write.create_native_write_table',
+ side_effect=AssertionError('must not reconstruct')):
+ writer = table.new_batch_write_builder().new_write()
+ assert not isinstance(writer, NativeTableWrite)
+ writer.close()
diff --git a/paimon-python/pypaimon/write/native_commit.py
b/paimon-python/pypaimon/write/native_commit.py
index 97d5316582..7c54ccb851 100644
--- a/paimon-python/pypaimon/write/native_commit.py
+++ b/paimon-python/pypaimon/write/native_commit.py
@@ -52,14 +52,28 @@ def native_messages_supported(table, messages) -> bool:
def create_native_commit(table, commit_user, overwrite_partition=None):
"""Return a native committer only when its publication protocol matches
Python."""
+ if not native_commit_available():
+ return None
+ native_table = create_native_write_table(table)
+ if native_table is None:
+ return None
+ 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()
+
+
+def create_native_write_table(table):
+ """Reconstruct a resolved table only for the filesystem publication
route."""
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
@@ -84,18 +98,11 @@ def create_native_commit(table, commit_user,
overwrite_partition=None):
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(
+ return 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)
- 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()
def to_native_commit_messages(table, messages):
diff --git a/paimon-python/pypaimon/write/native_write.py
b/paimon-python/pypaimon/write/native_write.py
new file mode 100644
index 0000000000..fe2952ed0c
--- /dev/null
+++ b/paimon-python/pypaimon/write/native_write.py
@@ -0,0 +1,162 @@
+# 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 Rust data writer behind PyPaimon's batch and stream builders."""
+
+import pyarrow as pa
+
+from pypaimon.read.native_plan import native_method_available
+from pypaimon.schema.arrow_schema import normalize_arrow_strings
+from pypaimon.schema.data_types import PyarrowFieldParser, is_blob_file_field
+from pypaimon.write.commit_message_serializer import deserialize_commit_message
+from pypaimon.write.native_commit import create_native_write_table
+from pypaimon.write.row_utils import row_to_named_values,
row_values_to_arrow_table
+
+
+def native_write_available() -> bool:
+ """Check every binding entry point used by the writer bridge."""
+ return all(native_method_available(type_name, method) for type_name,
method in (
+ ('Table', 'from_resolved_schema'),
+ ('BatchWriteBuilder', '_with_commit_user'),
+ ('BatchWriteBuilder', 'with_overwrite'),
+ ('BatchTableWrite', 'write_arrow'),
+ ('BatchTableWrite', 'prepare_commit'),
+ ('StreamWriteBuilder', 'with_commit_user'),
+ ('StreamTableWrite', 'write_arrow'),
+ ('StreamTableWrite', 'prepare_commit'),
+ ('CommitMessage', 'serialize'),
+ ))
+
+
+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."""
+ if (not native_write_available()
+ or table.options.data_evolution_enabled()
+ or table.options.file_format() != 'parquet'
+ or any(is_blob_file_field(field) for field in
table.table_schema.fields)):
+ return None
+ native_table = create_native_write_table(table)
+ if native_table is None:
+ return None
+ if stream:
+ builder =
native_table.new_stream_write_builder().with_commit_user(commit_user)
+ else:
+ builder =
native_table.new_batch_write_builder()._with_commit_user(commit_user)
+ if static_partition is not None:
+ builder = builder.with_overwrite(static_partition)
+ return NativeTableWrite(table, commit_user, static_partition, stream,
+ builder.new_write())
+
+
+class NativeTableWrite:
+ """Use Rust for Arrow batches while retaining PyPaimon's commit-message
API.
+
+ Advanced Python writer methods switch to the Python writer before the first
+ native data write. Once Rust has written data, switching would split one
+ logical write across two writers, so it is rejected.
+ """
+
+ def __init__(self, table, commit_user, static_partition, stream,
native_writer):
+ self.table = table
+ self.commit_user = commit_user
+ self.static_partition = static_partition
+ self.stream = stream
+ self._native_writer = native_writer
+ self._python_writer = None
+ self._written = False
+
+ def _switch_to_python(self):
+ if self._python_writer is not None:
+ return self._python_writer
+ if self._written:
+ raise RuntimeError(
+ 'Cannot switch to the Python writer after native data was
written')
+ self._native_writer.close()
+ self._native_writer = None
+ from pypaimon.write.table_write import BatchTableWrite,
StreamTableWrite
+ if self.stream:
+ self._python_writer = StreamTableWrite(self.table,
self.commit_user)
+ else:
+ self._python_writer = BatchTableWrite(
+ self.table, self.commit_user, self.static_partition)
+ return self._python_writer
+
+ def __getattr__(self, name):
+ if name.startswith('_'):
+ raise AttributeError(name)
+ return getattr(self._switch_to_python(), name)
+
+ def write_arrow(self, data):
+ if self._python_writer is not None:
+ return self._python_writer.write_arrow(data)
+ if isinstance(data, pa.RecordBatch):
+ return self.write_arrow_batch(data)
+ for batch in data.to_batches():
+ self.write_arrow_batch(batch)
+
+ def write_arrow_batch(self, data):
+ if self._python_writer is not None:
+ return self._python_writer.write_arrow_batch(data)
+ data = normalize_arrow_strings(data)
+ if data.num_rows:
+ # A failed native write may already have produced files. Never
+ # retry that batch through Python after this point.
+ self._written = True
+ self._native_writer.write_arrow(data)
+
+ def write_pandas(self, dataframe):
+ if self._python_writer is not None:
+ return self._python_writer.write_pandas(dataframe)
+ schema =
PyarrowFieldParser.from_paimon_schema(self.table.table_schema.fields)
+ self.write_arrow_batch(pa.RecordBatch.from_pandas(dataframe,
schema=schema))
+
+ def write_row(self, row):
+ if self._python_writer is not None:
+ return self._python_writer.write_row(row)
+ values = row_to_named_values(row, self.table.table_schema.fields)
+ names = list(self.table.field_names)
+ self.write_arrow_batch(row_values_to_arrow_table(
+ values, self.table.table_schema.fields, names).to_batches()[0])
+
+ def prepare_commit(self, commit_identifier=None):
+ if self._python_writer is not None:
+ if self.stream:
+ return self._python_writer.prepare_commit(commit_identifier)
+ return self._python_writer.prepare_commit()
+ if self.stream:
+ if commit_identifier is None:
+ raise TypeError('StreamTableWrite.prepare_commit requires an
identifier')
+ messages = self._native_writer.prepare_commit(True,
commit_identifier)
+ else:
+ if commit_identifier is not None:
+ raise TypeError('BatchTableWrite.prepare_commit accepts no
identifier')
+ messages = self._native_writer.prepare_commit()
+ return [deserialize_commit_message(
+ message.serialize(), self.table.partition_keys_fields,
+ self.table.trimmed_primary_keys_fields) for message in messages]
+
+ def close(self):
+ if self._python_writer is not None:
+ self._python_writer.close()
+ elif self._native_writer is not None:
+ self._native_writer.close()
+ self._native_writer = None
+
+ def abort(self):
+ if self._python_writer is not None:
+ self._python_writer.abort()
+ else:
+ self.close()
diff --git a/paimon-python/pypaimon/write/write_builder.py
b/paimon-python/pypaimon/write/write_builder.py
index 1c64d616e2..5180b8b08f 100644
--- a/paimon-python/pypaimon/write/write_builder.py
+++ b/paimon-python/pypaimon/write/write_builder.py
@@ -15,6 +15,7 @@
# specific language governing permissions and limitations
# under the License.
+import logging
import uuid
from abc import ABC
from typing import Optional
@@ -26,6 +27,8 @@ from pypaimon.write.table_update import (BatchTableUpdate,
StreamTableUpdate,
from pypaimon.write.table_write import (BatchTableWrite, StreamTableWrite,
TableWrite)
+logger = logging.getLogger(__name__)
+
class WriteBuilder(ABC):
def __init__(self, table):
@@ -50,6 +53,18 @@ class WriteBuilder(ABC):
else:
return str(uuid.uuid4())
+ def _native_write(self, static_partition=None, stream=False):
+ if not self.table.options.native_write_enabled():
+ return None
+ try:
+ from pypaimon.write.native_write import create_native_write
+ return create_native_write(self.table, self.commit_user,
+ static_partition, stream)
+ except Exception as error:
+ # Construction has not written any data; the normal writer is safe.
+ logger.debug('Native writer preparation failed; using Python: %s',
error)
+ return None
+
class BatchWriteBuilder(WriteBuilder):
@@ -62,7 +77,8 @@ class BatchWriteBuilder(WriteBuilder):
return self
def new_write(self) -> BatchTableWrite:
- return BatchTableWrite(self.table, self.commit_user,
self.static_partition)
+ return (self._native_write(self.static_partition)
+ or BatchTableWrite(self.table, self.commit_user,
self.static_partition))
def new_update(self) -> BatchTableUpdate:
return BatchTableUpdate(self.table, self.commit_user)
@@ -75,7 +91,8 @@ class BatchWriteBuilder(WriteBuilder):
class StreamWriteBuilder(WriteBuilder):
def new_write(self) -> StreamTableWrite:
- return StreamTableWrite(self.table, self.commit_user)
+ return (self._native_write(stream=True)
+ or StreamTableWrite(self.table, self.commit_user))
def new_update(self) -> StreamTableUpdate:
return StreamTableUpdate(self.table, self.commit_user)
