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 2420ac6b9d [python] Enable native write and REST commit in Native CI
(#10155)
2420ac6b9d is described below
commit 2420ac6b9db7a9d42b89b500df88ea573c8e0753
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Sep 24 18:06:54 2026 +0800
[python] Enable native write and REST commit in Native CI (#10155)
---
.github/workflows/ci-python.yml | 21 ++-
paimon-python/conftest.py | 109 +++++++++++++-
.../pypaimon/read/reader/format_pyarrow_reader.py | 14 +-
.../pypaimon/read/scanner/split_generator.py | 14 ++
paimon-python/pypaimon/tests/binary_row_test.py | 22 ++-
.../tests/data_evolution_row_rolling_test.py | 2 +
paimon-python/pypaimon/tests/native_commit_test.py | 161 ++++++++++++--------
.../pypaimon/tests/native_plan_integration_test.py | 1 +
paimon-python/pypaimon/tests/native_write_test.py | 165 ++++++++++++++++++++-
.../pypaimon/tests/parquet_metadata_cache_test.py | 19 +++
.../pypaimon/tests/ray_range_join_test.py | 2 +
.../rest/rest_catalog_commit_snapshot_test.py | 2 +
.../pypaimon/tests/rest/rest_simple_test.py | 5 +
.../pypaimon/tests/schema_evolution_read_test.py | 3 +
.../tests/write/changelog_producer_test.py | 5 +-
.../pypaimon/tests/write/table_write_test.py | 5 +
.../pypaimon/utils/file_store_path_factory.py | 16 ++
paimon-python/pypaimon/write/native_commit.py | 126 ++++++++++++----
paimon-python/pypaimon/write/native_write.py | 63 ++++++--
19 files changed, 624 insertions(+), 131 deletions(-)
diff --git a/.github/workflows/ci-python.yml b/.github/workflows/ci-python.yml
index ea2a988695..892a155b5c 100644
--- a/.github/workflows/ci-python.yml
+++ b/.github/workflows/ci-python.yml
@@ -35,7 +35,7 @@ jobs:
test:
name: Tests / Python ${{ matrix.python-version }}
env:
- # Real native planning is covered by the Rust main job below.
+ # Native planning, reading, writing and REST commits run in Native CI.
PYTEST_ADDOPTS: "-m 'not native_plan'"
timeout-minutes: 90
runs-on: ubuntu-latest
@@ -177,7 +177,7 @@ jobs:
./paimon-python/dev/lint-python.sh -e pytest_torch
rust-plan:
- name: Rust Plan
+ name: Native CI
timeout-minutes: 90
runs-on: ubuntu-latest
container: "python:3.11-slim"
@@ -196,10 +196,10 @@ jobs:
curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s --
-y --default-toolchain stable --profile minimal
echo "$HOME/.cargo/bin" >> $GITHUB_PATH
- - name: Build paimon-rust main
+ - name: Build paimon-rust with REST native commit support
run: |
python -m pip install --upgrade pip
- python -m pip install --upgrade
"git+https://github.com/apache/paimon-rust.git@main#subdirectory=bindings/python";
+ python -m pip install --upgrade
"git+https://github.com/apache/paimon-rust.git@aa60a6ac7cbe9299aca19effd9ba8301c1bc55da#subdirectory=bindings/python";
- name: Install Python test dependencies
run: |
@@ -213,30 +213,27 @@ jobs:
vortex-data==0.70.0 'paimon-mosaic>=0.1.0'
python -m pip check
- - name: Verify upstream native planning APIs
+ - name: Verify native wheel provenance
run: |
python - <<'PY'
import importlib.metadata as metadata
import json
- from pypaimon_rust.datafusion import PaimonCatalog, ReadBuilder,
Split
+ import pypaimon_rust.datafusion
distribution = metadata.distribution('pypaimon-rust')
source = json.loads(distribution.read_text('direct_url.json'))
print('pypaimon-rust', distribution.version,
'commit', source['vcs_info']['commit_id'])
- assert hasattr(PaimonCatalog, 'get_table'), 'Missing
PaimonCatalog.get_table'
- assert hasattr(Split, 'serialize'), 'Missing Split.serialize'
- assert hasattr(Split, 'is_streaming'), 'Missing stream-aware
Split.is_streaming'
- assert hasattr(ReadBuilder, 'new_incremental_scan'), 'Missing
ReadBuilder.new_incremental_scan'
- assert hasattr(ReadBuilder, 'with_row_ranges'), 'Missing
ReadBuilder.with_row_ranges'
PY
- - name: Run Python tests with Rust planning and reading
+ - name: Run Python tests with native planning, reading, writing and
committing
shell: bash
env:
PYPAIMON_TEST_NATIVE_PLAN: '1'
PYPAIMON_TEST_NATIVE_READ: '1'
+ PYPAIMON_TEST_NATIVE_WRITE: '1'
+ PYPAIMON_TEST_NATIVE_COMMIT: '1'
run: |
bash paimon-python/dev/lint-python.sh -i pytest
diff --git a/paimon-python/conftest.py b/paimon-python/conftest.py
index b5090ed771..272428f9e9 100644
--- a/paimon-python/conftest.py
+++ b/paimon-python/conftest.py
@@ -21,10 +21,16 @@ import pytest
_NATIVE_PLAN_ENV = "PYPAIMON_TEST_NATIVE_PLAN"
_NATIVE_READ_ENV = "PYPAIMON_TEST_NATIVE_READ"
+_NATIVE_WRITE_ENV = "PYPAIMON_TEST_NATIVE_WRITE"
+_NATIVE_COMMIT_ENV = "PYPAIMON_TEST_NATIVE_COMMIT"
_native_plan_count = 0
_native_read_count = 0
+_native_write_count = 0
+_native_commit_count = 0
_force_native_for_test = False
_force_native_read_for_test = False
+_force_native_write_for_test = False
+_force_native_commit_for_test = False
def pytest_addoption(parser):
@@ -34,6 +40,31 @@ def pytest_addoption(parser):
)
[email protected]
+def native_rest_catalog(tmp_path):
+ """A local REST catalog for native writer and committer integration
tests."""
+ import uuid
+
+ from pypaimon import CatalogFactory
+ from pypaimon.api.api_response import ConfigResponse
+ from pypaimon.api.auth import BearTokenAuthProvider
+ from pypaimon.tests.rest.rest_server import RESTCatalogServer
+
+ token = str(uuid.uuid4())
+ server = RESTCatalogServer(
+ data_path=str(tmp_path), auth_provider=BearTokenAuthProvider(token),
+ config=ConfigResponse(defaults={'prefix': 'native-test'}),
warehouse='warehouse')
+ server.start()
+ try:
+ catalog = CatalogFactory.create({
+ 'metastore': 'rest', 'uri': server.get_url(), 'warehouse':
'warehouse',
+ 'token.provider': 'bear', 'token': token, 'data-token.enabled':
'false'})
+ catalog.create_database('default', True)
+ yield catalog
+ finally:
+ server.shutdown()
+
+
def _native_plan_enabled():
return os.environ.get(_NATIVE_PLAN_ENV) == "1"
@@ -42,6 +73,14 @@ def _native_read_enabled():
return os.environ.get(_NATIVE_READ_ENV) == "1"
+def _native_write_enabled():
+ return os.environ.get(_NATIVE_WRITE_ENV) == "1"
+
+
+def _native_commit_enabled():
+ return os.environ.get(_NATIVE_COMMIT_ENV) == "1"
+
+
def pytest_configure(config):
config.addinivalue_line(
"markers", "python_plan: keep Python planner assertions on the Python
lane")
@@ -49,6 +88,10 @@ def pytest_configure(config):
"markers", "python_read: keep Python reader assertions on the Python
lane")
config.addinivalue_line(
"markers", "native_plan: exercise the real Rust planner in the Rust
main CI job")
+ config.addinivalue_line(
+ "markers", "python_write: keep Python writer assertions on the Python
lane")
+ config.addinivalue_line(
+ "markers", "python_commit: keep Python committer assertions on the
Python lane")
if _native_plan_enabled():
from pypaimon.read.table_scan import TableScan
@@ -77,6 +120,35 @@ def pytest_configure(config):
TableRead._try_native_batches = tracked_read
+ if _native_write_enabled():
+ from pypaimon.write.native_write import NativeTableWrite
+
+ original_write = NativeTableWrite.write_arrow_batch
+
+ def tracked_write(self, data):
+ global _native_write_count
+ native = self._native_writer is not None
+ result = original_write(self, data)
+ if native and data.num_rows and _force_native_write_for_test:
+ _native_write_count += 1
+ return result
+
+ NativeTableWrite.write_arrow_batch = tracked_write
+
+ if _native_commit_enabled():
+ from pypaimon.write.table_commit import TableCommit
+
+ original_prepare = TableCommit._prepare_native_commit
+
+ def tracked_prepare(self, messages):
+ global _native_commit_count
+ prepared = original_prepare(self, messages)
+ if prepared is not None and _force_native_commit_for_test:
+ _native_commit_count += 1
+ return prepared
+
+ TableCommit._prepare_native_commit = tracked_prepare
+
def pytest_collection_modifyitems(items):
if _native_plan_enabled():
@@ -88,17 +160,22 @@ def pytest_collection_modifyitems(items):
@pytest.fixture(autouse=True)
-def enable_native_plan_and_read(request, monkeypatch):
+def enable_native_backends(request, monkeypatch):
global _force_native_for_test, _force_native_read_for_test
+ global _force_native_write_for_test, _force_native_commit_for_test
python_plan = request.node.get_closest_marker("python_plan") is not None
python_read = request.node.get_closest_marker("python_read") is not None
+ python_write = request.node.get_closest_marker("python_write") is not None
+ python_commit = request.node.get_closest_marker("python_commit") is not
None
native_plan_test = request.path.name in (
"native_plan_test.py", "native_plan_integration_test.py",
"native_plan_capabilities_test.py")
force_plan = _native_plan_enabled() and not python_plan and not
native_plan_test
force_read = (_native_read_enabled() and not python_plan and not
python_read
and not native_plan_test)
- if not (force_plan or force_read):
+ force_write = _native_write_enabled() and not python_write
+ force_commit = _native_commit_enabled() and not python_commit
+ if not (force_plan or force_read or force_write or force_commit):
yield
return
@@ -118,19 +195,39 @@ def enable_native_plan_and_read(request, monkeypatch):
return original_read(self, True if default is None else default)
monkeypatch.setattr(CoreOptions, "native_read_enabled", read_enabled)
+ if force_write:
+ original_write = CoreOptions.native_write_enabled
+
+ def write_enabled(self, default=None):
+ return original_write(self, True if default is None else default)
+
+ monkeypatch.setattr(CoreOptions, "native_write_enabled", write_enabled)
+ if force_commit:
+ original_commit = CoreOptions.native_commit_enabled
+
+ def commit_enabled(self, default=None):
+ return original_commit(self, True if default is None else default)
+
+ monkeypatch.setattr(CoreOptions, "native_commit_enabled",
commit_enabled)
_force_native_for_test = force_plan
_force_native_read_for_test = force_read
+ _force_native_write_for_test = force_write
+ _force_native_commit_for_test = force_commit
try:
yield
finally:
_force_native_for_test = False
_force_native_read_for_test = False
+ _force_native_write_for_test = False
+ _force_native_commit_for_test = False
def pytest_sessionfinish(session, exitstatus):
if exitstatus == 0:
if ((_native_plan_enabled() and _native_plan_count == 0)
- or (_native_read_enabled() and _native_read_count == 0)):
+ or (_native_read_enabled() and _native_read_count == 0)
+ or (_native_write_enabled() and _native_write_count == 0)
+ or (_native_commit_enabled() and _native_commit_count == 0)):
session.exitstatus = pytest.ExitCode.TESTS_FAILED
@@ -141,3 +238,9 @@ def pytest_terminal_summary(terminalreporter):
if _native_read_enabled():
terminalreporter.write_line(
"native reads exercised: %d" % _native_read_count)
+ if _native_write_enabled():
+ terminalreporter.write_line(
+ "native writes exercised: %d" % _native_write_count)
+ if _native_commit_enabled():
+ terminalreporter.write_line(
+ "native commits exercised: %d" % _native_commit_count)
diff --git a/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
b/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
index 572f90a54b..ecb7db7e74 100644
--- a/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
+++ b/paimon-python/pypaimon/read/reader/format_pyarrow_reader.py
@@ -294,9 +294,17 @@ def _file_format_dataset(file_io: FileIO, file_format:
str, file_path: str,
fragment_options = {}
if known_size is not None and not _pyarrow_lt_7():
fragment_options["file_size"] = known_size
- fragment = parquet_format.make_fragment(
- file_path_for_pyarrow, filesystem=filesystem,
- **fragment_options)
+ try:
+ fragment = parquet_format.make_fragment(
+ file_path_for_pyarrow, filesystem=filesystem,
+ **fragment_options)
+ except TypeError as error:
+ # PyArrow 7-12 expose make_fragment but do not accept the
+ # file_size hint. The handler already received the size.
+ if not fragment_options or 'file_size' not in str(error):
+ raise
+ fragment = parquet_format.make_fragment(
+ file_path_for_pyarrow, filesystem=filesystem)
# Reuse this fragment's footer for schema discovery and scanning.
return ds.FileSystemDataset(
[fragment], fragment.physical_schema, parquet_format,
filesystem)
diff --git a/paimon-python/pypaimon/read/scanner/split_generator.py
b/paimon-python/pypaimon/read/scanner/split_generator.py
index 100fbec2e9..ac0e15625f 100644
--- a/paimon-python/pypaimon/read/scanner/split_generator.py
+++ b/paimon-python/pypaimon/read/scanner/split_generator.py
@@ -26,6 +26,7 @@ from pypaimon.read.split import Split
from pypaimon.read.split import DataSplit
from pypaimon.table.row.generic_row import GenericRow
from pypaimon.table.source.deletion_file import DeletionFile
+from pypaimon.utils.file_store_path_factory import canonical_data_file_path
class AbstractSplitGenerator(ABC):
@@ -94,6 +95,13 @@ class AbstractSplitGenerator(ABC):
Build splits from packed files.
"""
splits = []
+ if not packed_files or not file_entries:
+ return splits
+ partition = tuple(file_entries[0].partition.values)
+ path_factory = self.table.path_factory()
+ escaped_partition = (path_factory.bucket_path(
+ partition, file_entries[0].bucket, canonical_partition=True)
+ != path_factory.bucket_path(partition, file_entries[0].bucket))
for file_group in packed_files:
if use_optimized_path:
raw_convertible = True
@@ -109,6 +117,12 @@ class AbstractSplitGenerator(ABC):
file_entries[0].bucket,
self.default_part_value
)
+ if escaped_partition and not data_file.external_path:
+ canonical_path = canonical_data_file_path(
+ self.table, partition, file_entries[0].bucket,
+ data_file.file_name)
+ if self.table.file_io.exists(canonical_path):
+ data_file.file_path = canonical_path
if file_group:
# Get deletion files for this split
diff --git a/paimon-python/pypaimon/tests/binary_row_test.py
b/paimon-python/pypaimon/tests/binary_row_test.py
index 4fbe5940e2..5e8420e9eb 100644
--- a/paimon-python/pypaimon/tests/binary_row_test.py
+++ b/paimon-python/pypaimon/tests/binary_row_test.py
@@ -269,11 +269,16 @@ class BinaryRowTest(unittest.TestCase):
manifest_files =
file_scanner.manifest_list_manager.read_all(latest_snapshot)
manifest_entries =
file_scanner.manifest_file_manager.read(manifest_files[0].file_name)
self._transform_manifest_entries(manifest_entries, [])
- for i, entry in enumerate(manifest_entries):
+ for entry in manifest_entries:
+ # Manifest entry order is not part of the file format. Derive the
+ # synthetic statistics from this file's partition value.
+ partition_id = entry.partition.values[0]
entry.file.value_stats_cols = ['f2', 'f6', 'f8']
entry.file.value_stats = SimpleStats(
- GenericRow([10 * (i + 1), 100 * (i + 1), 5 - i],
[table.fields[2], table.fields[6], table.fields[8]]),
- GenericRow([10 * (i + 1), 100 * (i + 1), 5 - i],
[table.fields[2], table.fields[6], table.fields[8]]),
+ GenericRow([10 * partition_id, 100 * partition_id, 6 -
partition_id],
+ [table.fields[2], table.fields[6],
table.fields[8]]),
+ GenericRow([10 * partition_id, 100 * partition_id, 6 -
partition_id],
+ [table.fields[2], table.fields[6],
table.fields[8]]),
[0, 0, 0],
)
file_scanner.manifest_file_manager.write(manifest_files[0].file_name,
manifest_entries)
@@ -301,18 +306,21 @@ class BinaryRowTest(unittest.TestCase):
'f8': [0, -3, -4],
'f9': ['w5', 'w8', 'w9']
}
- self.assertEqual(expected_data, actual.to_pydict())
+ self.assertEqual(expected_data, actual.sort_by('f0').to_pydict())
file_scanner = FileScanner(table, lambda: ([], None))
latest_snapshot = file_scanner.snapshot_manager.get_latest_snapshot()
manifest_files =
file_scanner.manifest_list_manager.read_all(latest_snapshot)
manifest_entries =
file_scanner.manifest_file_manager.read(manifest_files[0].file_name)
self._transform_manifest_entries(manifest_entries, [])
- for i, entry in enumerate(manifest_entries):
+ for entry in manifest_entries:
+ partition_id = entry.partition.values[0]
entry.file.value_stats_cols = ['f2', 'f6', 'f8']
entry.file.value_stats = SimpleStats(
- GenericRow([0, 100 * (i + 1), 5 - i], [table.fields[2],
table.fields[6], table.fields[8]]),
- GenericRow([0, 100 * (i + 1), 5 - i], [table.fields[2],
table.fields[6], table.fields[8]]),
+ GenericRow([0, 100 * partition_id, 6 - partition_id],
+ [table.fields[2], table.fields[6],
table.fields[8]]),
+ GenericRow([0, 100 * partition_id, 6 - partition_id],
+ [table.fields[2], table.fields[6],
table.fields[8]]),
[0, 0, 0],
)
file_scanner.manifest_file_manager.write(manifest_files[0].file_name,
manifest_entries)
diff --git a/paimon-python/pypaimon/tests/data_evolution_row_rolling_test.py
b/paimon-python/pypaimon/tests/data_evolution_row_rolling_test.py
index bc797398e3..f2a5405621 100644
--- a/paimon-python/pypaimon/tests/data_evolution_row_rolling_test.py
+++ b/paimon-python/pypaimon/tests/data_evolution_row_rolling_test.py
@@ -23,6 +23,7 @@ import uuid
from unittest.mock import Mock
import pyarrow as pa
+import pytest
from pypaimon import CatalogFactory, Schema
from pypaimon.common.uri_reader import FileUriReader
@@ -193,6 +194,7 @@ class DataEvolutionRowRollingTest(unittest.TestCase):
self.assertEqual([1, 1, 1, 1], [f.row_count for f in files])
self.assertEqual(list(range(4)), self._read_ids(table))
+ @pytest.mark.python_write
def test_non_de_table_still_fails_fast(self):
table = self._create({'target-file-row-num': '3'})
wb = table.new_batch_write_builder()
diff --git a/paimon-python/pypaimon/tests/native_commit_test.py
b/paimon-python/pypaimon/tests/native_commit_test.py
index 8dcf130fd5..c996973b14 100644
--- a/paimon-python/pypaimon/tests/native_commit_test.py
+++ b/paimon-python/pypaimon/tests/native_commit_test.py
@@ -33,6 +33,9 @@ from pypaimon.write.native_commit import (
from pypaimon.write.table_write import StreamTableWrite
+pytestmark = pytest.mark.python_write
+
+
def requires_native(test):
# Native commit needs the from-source paimon-rust runtime built in the Rust
# Plan CI job; the PyPI pypaimon-rust wheel used by the standard test job
@@ -42,12 +45,13 @@ def requires_native(test):
not native_commit_available(), reason='pypaimon-rust runtime
required')(test)
-def _table(tmp_path, mode='append', backend='filesystem'):
+def _table(tmp_path, mode='append', backend='filesystem', catalog=None):
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)
+ if catalog is None:
+ catalog = CatalogFactory.create(options)
+ catalog.create_database('default', True)
options = {'file.format': 'parquet', 'commit.native.enabled': 'true'}
if mode == 'pk':
options['bucket'] = '1'
@@ -95,11 +99,25 @@ def _seed(table):
commit.close()
[email protected]_commit
def test_native_commit_is_opt_in():
assert not CoreOptions(Options({})).native_commit_enabled()
assert CoreOptions(Options({'commit.native.enabled':
'true'})).native_commit_enabled()
[email protected]('backend', ['filesystem', 'path', 'jdbc'])
+def test_non_rest_catalog_keeps_python_commit(tmp_path, backend):
+ table = _table(tmp_path, backend=backend)
+ assert create_native_commit(table, 'job') is None
+ builder = table.new_batch_write_builder()
+ commit = builder.new_commit()
+ try:
+ commit.commit(_prepare(builder, [{'id': 1, 'pt': 'a'}]))
+ assert _rows(table) == [{'id': 1, 'pt': 'a'}]
+ finally:
+ commit.close()
+
+
def test_overwrite_builder_api_is_batch_only(tmp_path):
table = _table(tmp_path)
batch = table.new_batch_write_builder()
@@ -113,10 +131,9 @@ def test_overwrite_builder_api_is_batch_only(tmp_path):
@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)
[email protected]('mode', ['append', 'pk'])
+def test_native_batch_roundtrip_preserves_identity(tmp_path,
native_rest_catalog, mode):
+ table = _table(tmp_path, mode, catalog=native_rest_catalog)
builder = table.new_batch_write_builder()
rows = [{'id': 1, 'pt': 'a'}, {'id': 2, 'pt': None}]
messages = _prepare(builder, rows)
@@ -135,8 +152,15 @@ def
test_native_batch_roundtrip_preserves_identity(tmp_path, backend, mode):
@requires_native
-def test_native_stream_reuses_commit_user_and_identifiers(tmp_path):
- table = _table(tmp_path)
+def test_replaced_rest_table_rejects_stale_commit(tmp_path,
native_rest_catalog):
+ table = _table(tmp_path, catalog=native_rest_catalog)
+ table.catalog_environment.uuid = 'stale-table-id'
+ assert create_native_commit(table, 'job') is None
+
+
+@requires_native
+def test_native_stream_reuses_commit_user_and_identifiers(tmp_path,
native_rest_catalog):
+ table = _table(tmp_path, catalog=native_rest_catalog)
builder = table.new_stream_write_builder()
commit = builder.new_commit()
try:
@@ -155,12 +179,12 @@ def
test_native_stream_reuses_commit_user_and_identifiers(tmp_path):
@pytest.mark.parametrize('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({
+def test_native_overwrite_replaces_only_target_rows(
+ tmp_path, native_rest_catalog, mode, dynamic, spec):
+ table = _table(tmp_path, mode, catalog=native_rest_catalog).copy({
'dynamic-partition-overwrite': str(dynamic).lower(),
'commit.user-prefix': 'python'})
_seed(table)
builder = table.new_batch_write_builder().overwrite(spec)
@@ -193,8 +217,10 @@ def
test_native_overwrite_replaces_only_target_rows(tmp_path, mode, dynamic, spe
('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()})
+def test_native_empty_overwrite_semantics(
+ tmp_path, native_rest_catalog, mode, dynamic, spec, remaining):
+ table = _table(tmp_path, mode, catalog=native_rest_catalog).copy({
+ 'dynamic-partition-overwrite': str(dynamic).lower()})
_seed(table)
builder = table.new_batch_write_builder().overwrite(spec)
commit = builder.new_commit()
@@ -215,8 +241,8 @@ def test_native_empty_overwrite_semantics(tmp_path, mode,
dynamic, spec, remaini
@requires_native
@pytest.mark.parametrize('value', ['off', '0', ' false '])
-def test_native_overwrite_normalizes_python_boolean_option(tmp_path, value):
- table = _table(tmp_path).copy({
+def test_native_overwrite_normalizes_python_boolean_option(tmp_path,
native_rest_catalog, value):
+ table = _table(tmp_path, catalog=native_rest_catalog).copy({
'dynamic-partition-overwrite': value, 'snapshot.ignore-empty-commit':
value})
_seed(table)
commit = table.new_batch_write_builder().overwrite({'pt':
'a'}).new_commit()
@@ -230,13 +256,14 @@ def
test_native_overwrite_normalizes_python_boolean_option(tmp_path, value):
@pytest.mark.parametrize('native', [False, pytest.param(True,
marks=pytest.mark.native_plan)])
@pytest.mark.parametrize('case', ['unpartitioned', 'static-empty',
'static-missing', 'dynamic-empty'])
-def test_empty_overwrite_records_java_snapshot(tmp_path, native, case):
+def test_empty_overwrite_records_java_snapshot(tmp_path, native_rest_catalog,
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(),
- })
+ table = _table(tmp_path, 'unpartitioned' if case == 'unpartitioned' else
'append',
+ catalog=native_rest_catalog).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(
@@ -264,8 +291,10 @@ def test_empty_overwrite_records_java_snapshot(tmp_path,
native, case):
@requires_native
@pytest.mark.parametrize('mode', ['batch', 'stream'])
@pytest.mark.parametrize('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()})
+def test_native_empty_commit_preserves_python_option(
+ tmp_path, native_rest_catalog, mode, ignore):
+ table = _table(tmp_path, catalog=native_rest_catalog).copy({
+ 'snapshot.ignore-empty-commit': str(ignore).lower()})
commit = getattr(table, 'new_' + mode + '_write_builder')().new_commit()
try:
with _must_not_fallback(commit):
@@ -285,8 +314,8 @@ def
test_native_empty_commit_preserves_python_option(tmp_path, mode, ignore):
@requires_native
@pytest.mark.parametrize('overwrite', [False, True])
-def test_native_abort_removes_uncommitted_files(tmp_path, overwrite):
- table = _table(tmp_path)
+def test_native_abort_removes_uncommitted_files(tmp_path, native_rest_catalog,
overwrite):
+ table = _table(tmp_path, catalog=native_rest_catalog)
builder = table.new_batch_write_builder()
if overwrite:
builder.overwrite()
@@ -303,6 +332,27 @@ def test_native_abort_removes_uncommitted_files(tmp_path,
overwrite):
commit.close()
+@requires_native
+def test_python_file_in_legacy_partition_uses_python_abort(
+ tmp_path, native_rest_catalog):
+ table = _table(tmp_path, catalog=native_rest_catalog)
+ builder = table.new_batch_write_builder()
+ messages = _prepare(builder, [{'id': 1, 'pt': 'a/b'}])
+ file = messages[0].new_files[0]
+ assert table.file_io.exists(file.file_path)
+ assert not native_messages_supported(table, messages)
+
+ commit = builder.new_commit()
+ try:
+ with patch.object(commit.file_store_commit, 'abort',
+ wraps=commit.file_store_commit.abort) as
python_abort:
+ commit.abort(messages)
+ python_abort.assert_called_once_with(messages)
+ assert not table.file_io.exists(file.file_path)
+ finally:
+ commit.close()
+
+
@pytest.mark.parametrize('failure', ['missing', 'construction', 'conversion'])
def test_preflight_failure_uses_python(tmp_path, failure):
table = _table(tmp_path)
@@ -354,8 +404,9 @@ def
test_native_mutation_failure_never_falls_back_or_aborts(tmp_path, method, ov
@requires_native
@pytest.mark.parametrize('overwrite', [False, True])
-def
test_publication_response_loss_does_not_duplicate_or_delete_files(tmp_path,
overwrite):
- table = _table(tmp_path)
+def test_publication_response_loss_does_not_duplicate_or_delete_files(
+ tmp_path, native_rest_catalog, overwrite):
+ table = _table(tmp_path, catalog=native_rest_catalog)
builder = table.new_batch_write_builder()
if overwrite:
builder.overwrite()
@@ -399,10 +450,11 @@ def
test_snapshot_properties_select_python_before_native(tmp_path, properties, o
@pytest.mark.parametrize('warmup', [False, pytest.param(True,
marks=pytest.mark.native_plan)])
-def test_callbacks_added_after_construction_select_python(tmp_path, warmup):
+def test_callbacks_added_after_construction_select_python(
+ tmp_path, native_rest_catalog, warmup):
if warmup and not native_commit_available():
pytest.skip('native warmup requires the commit bindings')
- table = _table(tmp_path)
+ table = _table(tmp_path, catalog=native_rest_catalog)
builder = table.new_stream_write_builder()
commit = builder.new_commit()
if warmup:
@@ -478,12 +530,11 @@ def
test_disabled_option_never_initializes_native(tmp_path):
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':
[email protected]('kind', ['custom-env', 'branch', 'custom-io'])
+def test_incompatible_publication_environment_is_not_reconstructed(
+ tmp_path, native_rest_catalog, kind):
+ table = _table(tmp_path, catalog=native_rest_catalog)
+ if kind == 'custom-env':
class CustomEnvironment(CatalogEnvironment):
pass
table.catalog_environment = CustomEnvironment()
@@ -492,43 +543,37 @@ def
test_incompatible_publication_environment_is_not_reconstructed(tmp_path, kin
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:
+ patch('pypaimon.write.native_commit._native_rest_table') as
resolve:
assert create_native_commit(table, 'job') is None
resolve.assert_not_called()
[email protected]('missing_type,missing_method', [
- ('Table', 'from_resolved_schema'),
- ('CommitMessage', 'deserialize'),
- ('StreamWriteBuilder', 'with_commit_user'),
- ('BatchWriteBuilder', '_with_commit_user'),
- ('BatchWriteBuilder', 'with_overwrite'),
-])
-def test_incomplete_runtime_falls_back_without_reconstructing_table(
- tmp_path, missing_type, missing_method):
- table = _table(tmp_path)
- with patch('pypaimon.write.native_commit.native_method_available',
- side_effect=lambda cls, method: (cls, method) != (missing_type,
missing_method)), \
-
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_runtime_falls_back_without_reconstructing_table(tmp_path):
- table = _table(tmp_path)
+def test_missing_runtime_falls_back_without_reconstructing_table(
+ tmp_path, native_rest_catalog):
+ table = _table(tmp_path, catalog=native_rest_catalog)
with patch('pypaimon.write.native_commit.native_commit_available',
return_value=False), \
-
patch('pypaimon.write.native_commit._resolved_schema_file_io_options') as
resolve:
+ patch('pypaimon.write.native_commit._native_rest_table') 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')
+def test_partial_row_id_and_compact_messages_preserve_python_recovery(
+ tmp_path, native_rest_catalog):
+ table = _table(tmp_path, 'de', catalog=native_rest_catalog)
+ assert create_native_commit(table, 'job') is None
+ assert not native_messages_supported(table, [CommitMessage((), 0, [])])
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_custom_manifest_target_uses_python_rolling(tmp_path,
native_rest_catalog):
+ table = _table(tmp_path, catalog=native_rest_catalog).copy({
+ 'manifest.target-file-size': '16 kb'})
+ assert create_native_commit(table, 'job') is None
+ assert not native_messages_supported(table, [CommitMessage((), 0, [])])
+
+
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()
diff --git a/paimon-python/pypaimon/tests/native_plan_integration_test.py
b/paimon-python/pypaimon/tests/native_plan_integration_test.py
index aa22c9a7ba..3f6cf52fe1 100644
--- a/paimon-python/pypaimon/tests/native_plan_integration_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_integration_test.py
@@ -1499,6 +1499,7 @@ class NativePlanIntegrationTest(unittest.TestCase):
self.assertEqual(native.split_count, len(normal.splits()))
self.assertEqual(native.split_count, 1)
+ @pytest.mark.python_write
def test_partitioned_table_matches_normal_plan(self):
# Native decoding restores PyPaimon's legacy unescaped partition path.
schema = pa.schema([('k', pa.int64()), ('p', pa.string())])
diff --git a/paimon-python/pypaimon/tests/native_write_test.py
b/paimon-python/pypaimon/tests/native_write_test.py
index 7224dbe09a..1b0f4972f3 100644
--- a/paimon-python/pypaimon/tests/native_write_test.py
+++ b/paimon-python/pypaimon/tests/native_write_test.py
@@ -24,20 +24,20 @@ 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
+from pypaimon.write.native_write import NativeTableWrite
-requires_native = pytest.mark.skipif(
- not native_write_available(), reason='pypaimon-rust writer required')
+requires_native = pytest.mark.native_plan
-def _table(tmp_path, primary_key=False, commit_native=True):
+def _table(tmp_path, primary_key=False, commit_native=True,
table_options=None):
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'
+ options.update(table_options or {})
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 [],
@@ -57,6 +57,7 @@ def _rows(table):
builder.new_scan().plan().splits()).to_pylist(), key=lambda row:
row['id'])
[email protected]_write
def test_native_write_is_opt_in():
assert not CoreOptions(Options({})).native_write_enabled()
assert CoreOptions(Options({'write.native.enabled':
'true'})).native_write_enabled()
@@ -77,6 +78,8 @@ def test_batch_native_write_commits_through_both_committers(
messages = writer.prepare_commit()
assert messages and sum(file.row_count for msg in messages
for file in msg.new_files) == 2
+ assert all(file.file_path and table.file_io.exists(file.file_path)
+ for msg in messages for file in msg.new_files)
commit = builder.new_commit()
try:
commit.commit(messages)
@@ -88,6 +91,82 @@ def test_batch_native_write_commits_through_both_committers(
assert table.snapshot_manager().get_latest_snapshot().commit_user ==
builder.commit_user
+@requires_native
+def test_escaped_partition_file_path_and_abort(tmp_path, native_rest_catalog):
+ catalog = native_rest_catalog
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ pa.schema([('id', pa.int64()), ('pt', pa.string())]),
+ options={'file.format': 'parquet', 'write.native.enabled': 'true',
+ 'commit.native.enabled': 'true'}, partition_keys=['pt']),
False)
+ table = catalog.get_table('default.t')
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ assert isinstance(writer, NativeTableWrite)
+ try:
+ writer.write_arrow_batch(_batch([1], ['a/b']))
+ messages = writer.prepare_commit()
+ file = messages[0].new_files[0]
+ assert 'pt=a%2Fb/bucket-0' in file.file_path
+ assert table.file_io.exists(file.file_path)
+
+ commit = builder.new_commit()
+ try:
+ with patch.object(commit.file_store_commit, 'abort',
+ side_effect=AssertionError('Python fallback')):
+ commit.abort(messages)
+ assert not table.file_io.exists(file.file_path)
+ finally:
+ commit.close()
+ finally:
+ writer.close()
+
+
+@requires_native
[email protected]_plan
[email protected]_read
+def test_escaped_partition_native_write_is_readable_by_python(tmp_path):
+ table = _table(tmp_path)
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ try:
+ writer.write_arrow_batch(_batch([1], ['a/b']))
+ messages = writer.prepare_commit()
+ builder.new_commit().commit(messages)
+ finally:
+ writer.close()
+ assert _rows(table) == [{'id': 1, 'pt': 'a/b'}]
+
+
+@requires_native
+def test_rest_native_write_and_commit(tmp_path, native_rest_catalog):
+ catalog = native_rest_catalog
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ pa.schema([('id', pa.int64()), ('pt', pa.string())]),
+ options={'file.format': 'parquet', 'write.native.enabled': 'true',
+ 'commit.native.enabled': 'true'}, partition_keys=['pt']),
False)
+ table = catalog.get_table('default.t')
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ assert isinstance(writer, NativeTableWrite)
+ commit = builder.new_commit()
+ try:
+ writer.write_arrow_batch(_batch([1, 2], ['a', 'b']))
+ with patch.object(commit.file_store_commit, 'commit',
+ side_effect=AssertionError('Python commit
fallback')):
+ commit.commit(writer.prepare_commit())
+ assert _rows(table) == [{'id': 1, 'pt': 'a'}, {'id': 2, 'pt': 'b'}]
+ shard_rows = []
+ for shard in range(3):
+ read_builder = table.new_read_builder()
+ splits = read_builder.new_scan().with_shard(shard,
3).plan().splits()
+
shard_rows.extend(read_builder.new_read().to_arrow(splits).to_pylist())
+ assert sorted(shard_rows, key=lambda row: row['id']) == _rows(table)
+ assert table.snapshot_manager().get_latest_snapshot().commit_user ==
builder.commit_user
+ finally:
+ writer.close()
+ commit.close()
+
+
@requires_native
def test_stream_native_write_reuses_writer_across_checkpoints(tmp_path):
table = _table(tmp_path)
@@ -162,3 +241,81 @@ def
test_unavailable_native_writer_falls_back_before_table_reconstruction(tmp_pa
writer = table.new_batch_write_builder().new_write()
assert not isinstance(writer, NativeTableWrite)
writer.close()
+
+
+@requires_native
[email protected]('primary_key', [False, True])
+def test_custom_prefix_uses_native_writer(tmp_path, primary_key):
+ table = _table(tmp_path, primary_key=primary_key).copy({
+ 'data-file.prefix': 'custom-'})
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ assert isinstance(writer, NativeTableWrite)
+ try:
+ writer.write_arrow_batch(_batch([1], ['a']))
+ messages = writer.prepare_commit()
+ assert messages
+ assert all(file.file_name.startswith('custom-')
+ for message in messages for file in message.new_files)
+ builder.new_commit().commit(messages)
+ finally:
+ writer.close()
+ assert _rows(table) == [{'id': 1, 'pt': 'a'}]
+
+
+def test_external_data_paths_fall_back_before_native_write(tmp_path):
+ table = _table(tmp_path).copy({
+ 'data-file.external-paths': 'file://' + str(tmp_path / 'external')})
+ with patch('pypaimon.write.native_write.native_write_available',
return_value=True), \
+ 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()
+
+
[email protected]('options', [
+ {'bucket': '-1'},
+ {'merge-engine': 'first-row'},
+ {'merge-engine': 'partial-update'},
+ {'merge-engine': 'aggregation'},
+ {'target-file-row-num': '5'},
+ {'changelog-file.format': 'orc'},
+ {'metadata.stats-mode': 'full'},
+])
+def test_unsupported_primary_key_write_falls_back_before_native_reconstruction(
+ tmp_path, options):
+ table = (_table(tmp_path, primary_key=True, table_options=options)
+ if 'bucket' in options else
+ _table(tmp_path, primary_key=True).copy(options))
+ with patch('pypaimon.write.native_write.native_write_available',
return_value=True), \
+ 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()
+
+
+@requires_native
+def test_native_write_validates_input_schema_before_writing(tmp_path):
+ table = _table(tmp_path)
+ writer = table.new_batch_write_builder().new_write()
+ assert isinstance(writer, NativeTableWrite)
+ wrong = pa.record_batch([pa.array([1], pa.int32()), pa.array(['a'])],
+ names=['id', 'pt'])
+ with pytest.raises(ValueError, match="Input schema isn't consistent"):
+ writer.write_arrow_batch(wrong)
+ assert not writer._written
+ writer.abort()
+
+
[email protected]('engine', ['partial-update', 'aggregation'])
+def test_deletion_vectors_with_merge_engine_fall_back(tmp_path, engine):
+ table = _table(tmp_path).copy({
+ 'deletion-vectors.enabled': 'true', 'merge-engine': engine})
+ with patch('pypaimon.write.native_write.native_write_available',
return_value=True), \
+ 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/tests/parquet_metadata_cache_test.py
b/paimon-python/pypaimon/tests/parquet_metadata_cache_test.py
index 711e73191e..3b5db10c24 100644
--- a/paimon-python/pypaimon/tests/parquet_metadata_cache_test.py
+++ b/paimon-python/pypaimon/tests/parquet_metadata_cache_test.py
@@ -266,6 +266,25 @@ class FileFormatMetadataCacheTest(unittest.TestCase):
self.assertIn(
("register_file_size", self.paths[0], file_size), handler.calls)
+ def test_old_pyarrow_retries_fragment_without_file_size(self):
+ parquet_format = unittest.mock.Mock()
+ fragment = unittest.mock.Mock(physical_schema=pa.schema([]))
+ parquet_format.make_fragment.side_effect = [
+ TypeError("make_fragment() got an unexpected keyword argument
'file_size'"),
+ fragment,
+ ]
+ with patch.object(reader_module, "_pyarrow_lt_7", return_value=False),
\
+ patch.object(reader_module.ds, "ParquetFileFormat",
return_value=parquet_format), \
+ patch.object(reader_module.ds, "FileSystemDataset",
+ return_value=unittest.mock.sentinel.dataset):
+ dataset = reader_module._file_format_dataset(
+ self.file_io, "parquet", self.paths[0], 0, 123)
+ self.assertIs(unittest.mock.sentinel.dataset, dataset)
+ self.assertEqual([
+ unittest.mock.call(self.paths[0],
filesystem=self.file_io.filesystem, file_size=123),
+ unittest.mock.call(self.paths[0],
filesystem=self.file_io.filesystem),
+ ], parquet_format.make_fragment.call_args_list)
+
def test_fragment_metadata_is_reused_without_io(self):
handler = _CountingFileSystemHandler()
self.file_io.filesystem = pafs.PyFileSystem(handler)
diff --git a/paimon-python/pypaimon/tests/ray_range_join_test.py
b/paimon-python/pypaimon/tests/ray_range_join_test.py
index 3696bf851f..9a7acfa212 100644
--- a/paimon-python/pypaimon/tests/ray_range_join_test.py
+++ b/paimon-python/pypaimon/tests/ray_range_join_test.py
@@ -418,6 +418,8 @@ class RayRangeJoinTest(unittest.TestCase):
name, self.catalog_options, None, "renamed")
self.assertEqual([(lo, hi) for _, lo, hi in ranged], [(3, 9)])
+ @pytest.mark.python_plan
+ @pytest.mark.python_write
def test_footer_failure_degrades_to_unknown(self):
schema = pa.schema([("k", pa.int64())])
self._table("default.rj_footer_failure", schema, [
diff --git
a/paimon-python/pypaimon/tests/rest/rest_catalog_commit_snapshot_test.py
b/paimon-python/pypaimon/tests/rest/rest_catalog_commit_snapshot_test.py
index 4b7ac5d8b0..be3b57339a 100644
--- a/paimon-python/pypaimon/tests/rest/rest_catalog_commit_snapshot_test.py
+++ b/paimon-python/pypaimon/tests/rest/rest_catalog_commit_snapshot_test.py
@@ -24,6 +24,7 @@ from dataclasses import replace
from unittest.mock import Mock, patch
import pyarrow as pa
+import pytest
from pypaimon import Schema
from pypaimon.api.api_response import CommitTableResponse
@@ -398,6 +399,7 @@ class TestRESTCommit(RESTBaseTest):
self.assertEqual(
sorted(actual.column('id').to_pylist()), [1, 2, 3, 4, 5, 6])
+ @pytest.mark.python_commit
def test_commit_succeeded_on_server_but_client_fails(self):
pa_schema = pa.schema([('id', pa.int32()), ('name', pa.string())])
opts = {
diff --git a/paimon-python/pypaimon/tests/rest/rest_simple_test.py
b/paimon-python/pypaimon/tests/rest/rest_simple_test.py
index 77183eef6d..1217661c29 100644
--- a/paimon-python/pypaimon/tests/rest/rest_simple_test.py
+++ b/paimon-python/pypaimon/tests/rest/rest_simple_test.py
@@ -18,6 +18,7 @@
import sys
import pyarrow as pa
+import pytest
from pypaimon import Schema
from pypaimon.catalog.catalog_exception import DatabaseAlreadyExistException,
TableAlreadyExistException, \
@@ -58,6 +59,8 @@ class RESTSimpleTest(RESTBaseTest):
}
self.expected = pa.Table.from_pydict(self.data, schema=self.pa_schema)
+ @pytest.mark.python_write
+ @pytest.mark.python_commit
def test_with_shard_ao_unaware_bucket(self):
schema = Schema.from_pyarrow_schema(self.pa_schema,
partition_keys=['dt'])
self.rest_catalog.drop_table('default.test_with_shard_ao_unaware_bucket', True)
@@ -169,6 +172,8 @@ class RESTSimpleTest(RESTBaseTest):
}, schema=self.pa_schema)
self.assertEqual(actual, expected)
+ @pytest.mark.python_write
+ @pytest.mark.python_commit
def test_with_shard_ao_fixed_bucket(self):
schema = Schema.from_pyarrow_schema(self.pa_schema,
partition_keys=['dt'],
options={'bucket': '5',
'bucket-key': 'item_id'})
diff --git a/paimon-python/pypaimon/tests/schema_evolution_read_test.py
b/paimon-python/pypaimon/tests/schema_evolution_read_test.py
index 60705d5cad..ea02304657 100644
--- a/paimon-python/pypaimon/tests/schema_evolution_read_test.py
+++ b/paimon-python/pypaimon/tests/schema_evolution_read_test.py
@@ -61,6 +61,7 @@ class SchemaEvolutionReadTest(unittest.TestCase):
shutil.rmtree(cls.tempdir, ignore_errors=True)
@pytest.mark.python_plan
+ @pytest.mark.python_write
def test_schema_evolution(self):
# schema 0
pa_schema = pa.schema([
@@ -132,6 +133,7 @@ class SchemaEvolutionReadTest(unittest.TestCase):
self.assertEqual(expected, actual)
@pytest.mark.python_plan
+ @pytest.mark.python_write
def test_schema_evolution_type(self):
# schema 0
pa_schema = pa.schema([
@@ -428,6 +430,7 @@ class SchemaEvolutionReadTest(unittest.TestCase):
self.assertEqual(1, len(entries)) # verify scan filter success for
schema evolution
@pytest.mark.python_plan
+ @pytest.mark.python_write
def test_schema_evolution_with_read_filter(self):
# schema 0
pa_schema = pa.schema([
diff --git a/paimon-python/pypaimon/tests/write/changelog_producer_test.py
b/paimon-python/pypaimon/tests/write/changelog_producer_test.py
index 88faa792ac..0ac4610ee8 100644
--- a/paimon-python/pypaimon/tests/write/changelog_producer_test.py
+++ b/paimon-python/pypaimon/tests/write/changelog_producer_test.py
@@ -17,14 +17,14 @@
################################################################################
import glob
+import json
import os
import shutil
import tempfile
import unittest
import pyarrow as pa
-
-import json
+import pytest
from pypaimon import CatalogFactory, Schema
from pypaimon.manifest.manifest_list_manager import ManifestListManager
@@ -337,6 +337,7 @@ class ChangelogProducerTest(unittest.TestCase):
table_write.close()
table_commit.close()
+ @pytest.mark.python_write
def test_failed_changelog_write_leaves_nothing_to_commit(self):
"""A data file and its changelog are committed together or not at all.
diff --git a/paimon-python/pypaimon/tests/write/table_write_test.py
b/paimon-python/pypaimon/tests/write/table_write_test.py
index 8e1bbffaf2..ee0ee42fd9 100644
--- a/paimon-python/pypaimon/tests/write/table_write_test.py
+++ b/paimon-python/pypaimon/tests/write/table_write_test.py
@@ -26,6 +26,7 @@ from unittest.mock import Mock, patch
from pypaimon import CatalogFactory, Schema
import pyarrow as pa
+import pytest
from parameterized import parameterized
from pypaimon.build_info import full_version as build_full_version
@@ -471,6 +472,7 @@ class TableWriteTest(unittest.TestCase):
with patch.object(pa.TableGroupBy, 'aggregate', raise_missing_kernel):
self.assertFalse(rk._probe_arrow_group_by())
+ @pytest.mark.python_commit
def test_write_snapshot(self):
schema = Schema.from_pyarrow_schema(self.pa_schema,
partition_keys=['dt'])
self.catalog.create_table('default.test_write_snapshot', schema, False)
@@ -591,6 +593,7 @@ class TableWriteTest(unittest.TestCase):
self.assertEqual(
expected.sort_by(sort_keys), self._read_sorted(table, sort_keys))
+ @pytest.mark.python_write
def test_multi_prepare_commit_ao(self):
schema = Schema.from_pyarrow_schema(self.pa_schema,
partition_keys=['dt'])
self.catalog.create_table('default.test_append_only_parquet', schema,
False)
@@ -715,6 +718,7 @@ class TableWriteTest(unittest.TestCase):
actual = table_read.to_arrow(splits).sort_by('user_id')
self.assertEqual(expected, actual)
+ @pytest.mark.python_write
def test_multi_prepare_commit_pk(self):
schema = Schema.from_pyarrow_schema(self.pa_schema,
partition_keys=['dt'], primary_keys=['user_id', 'dt'],
options={'bucket': '2'})
@@ -1970,6 +1974,7 @@ class TableWriteTest(unittest.TestCase):
actual = self._read_sorted(table, 'id')
self.assertEqual(expected, actual)
+ @pytest.mark.python_write
def test_validate_schema_allows_binary_family_for_write_cols(self):
pa_schema = pa.schema([
('id', pa.int32()),
diff --git a/paimon-python/pypaimon/utils/file_store_path_factory.py
b/paimon-python/pypaimon/utils/file_store_path_factory.py
index d712f7e815..35594c2963 100644
--- a/paimon-python/pypaimon/utils/file_store_path_factory.py
+++ b/paimon-python/pypaimon/utils/file_store_path_factory.py
@@ -42,6 +42,22 @@ def _escape_partition_component(value: str) -> str:
for char in value)
+def canonical_data_file_path(table, partition, bucket, file_name):
+ """Locate a file in Java's escaped partition directory."""
+ bucket_path = table.path_factory().bucket_path(
+ tuple(partition), bucket, canonical_partition=True)
+ path = f"{bucket_path.rstrip('/')}/{file_name}"
+ root = table.table_path.rstrip('/')
+ if (root.startswith('file:') and path.startswith(root + '/')
+ and '%' in path[len(root) + 1:]):
+ from pypaimon.filesystem.local_file_io import LocalFileIO
+
+ # File I/O wrappers eventually decode %2F in file URIs. Rust stores
+ # the literal escaped component, so use its physical local path.
+ return str(LocalFileIO()._to_file(root) / path[len(root) + 1:])
+ return path
+
+
def _floating_partition_string(value, single_precision: bool) -> str:
# Use a shortest round-tripping form for the initial lookup. Older JVMs
# can use different digits; the read fallback matches their stored values.
diff --git a/paimon-python/pypaimon/write/native_commit.py
b/paimon-python/pypaimon/write/native_commit.py
index 7c54ccb851..1eb50e09bb 100644
--- a/paimon-python/pypaimon/write/native_commit.py
+++ b/paimon-python/pypaimon/write/native_commit.py
@@ -17,42 +17,67 @@
"""Optional native commits using the Java CommitMessage v14 bridge."""
+from importlib import import_module
+
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)
+ _catalog_context_options, _catalog_metastore, _option_value_to_string,
+ _resolved_schema_file_io_options)
+from pypaimon.utils.file_store_path_factory import canonical_data_file_path
from pypaimon.write.commit_message_serializer import serialize_commit_message
+_DEFAULT_MANIFEST_TARGET_SIZE = 8 * 1024 * 1024
+
+
def native_commit_available() -> bool:
- """Whether the Rust runtime provides the required commit APIs."""
- return all(native_method_available(type_name, method) for type_name,
method in (
- ('Table', 'from_resolved_schema'),
- ('CommitMessage', 'deserialize'),
- ('StreamWriteBuilder', 'with_commit_user'),
- ('BatchWriteBuilder', '_with_commit_user'),
- ('BatchWriteBuilder', 'with_overwrite'),
- ))
+ """Whether the optional Rust bindings are installed."""
+ try:
+ import_module('pypaimon_rust.datafusion')
+ except ImportError:
+ return False
+ return True
+
+
+def _native_publication_supported(table) -> bool:
+ # Data evolution needs sidecar ranges and row-id recovery. Custom manifest
+ # targets need Java's forced size checks between manifest entry groups.
+ return (not table.options.data_evolution_enabled()
+ and table.options.manifest_target_size() ==
_DEFAULT_MANIFEST_TARGET_SIZE)
def native_messages_supported(table, messages) -> bool:
+ if not _native_publication_supported(table):
+ return False
+ path_factory = table.path_factory()
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
+ partition = tuple(message.partition)
+ bucket_path = path_factory.bucket_path(
+ partition, message.bucket, canonical_partition=True)
+ for file in message.new_files + message.changelog_files:
+ if file.external_path:
+ continue
+ expected = canonical_data_file_path(
+ table, partition, message.bucket, file.file_name)
+ if file.file_path:
+ if str(file.file_path) != expected:
+ return False
+ elif path_factory.bucket_path(partition, message.bucket) !=
bucket_path:
+ # The message does not say which of the two partition layouts
+ # contains the file. Use Python's path-aware commit and abort.
+ return False
return True
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():
+ if (not _native_publication_supported(table)
+ or not _rest_catalog_supported(table)
+ or not native_commit_available()):
return None
native_table = create_native_write_table(table)
if native_table is None:
@@ -66,26 +91,63 @@ def create_native_commit(table, commit_user,
overwrite_partition=None):
return
native_table.new_stream_write_builder().with_commit_user(commit_user).new_commit()
+def _rest_catalog_supported(table):
+ from pypaimon.catalog.catalog_environment import CatalogEnvironment
+ from pypaimon.catalog.rest.rest_token_file_io import RESTTokenFileIO
+ from pypaimon.filesystem.caching_file_io import CachingFileIO
+ from pypaimon.filesystem.local_file_io import LocalFileIO
+ from pypaimon.filesystem.oss_file_io import OssFileIO
+ from pypaimon.filesystem.pyarrow_file_io import PyArrowFileIO
+ from pypaimon.filesystem.resolving_file_io import ResolvingFileIO
+ from pypaimon.table.file_store_table import FileStoreTable
+
+ environment = table.catalog_environment
+ loader = getattr(environment, 'catalog_loader', None)
+ context = loader.context() if _catalog_metastore(loader) == 'rest' else
None
+ file_io = table.file_io
+ if type(file_io) is CachingFileIO:
+ file_io = file_io._delegate
+ return (type(table) is FileStoreTable
+ and type(environment) is CatalogEnvironment
+ and type(file_io) in (LocalFileIO, PyArrowFileIO, OssFileIO,
+ ResolvingFileIO, RESTTokenFileIO)
+ and environment.supports_version_management
+ and environment.uuid is not None
+ and context is not None
+ and context.options is not None
+ and all(getattr(context, attr, None) is None for attr in (
+ 'hadoop_conf', 'prefer_io_loader', 'fallback_io_loader')))
+
+
+def _native_rest_table(table, schema_json):
+ from pypaimon_rust.datafusion import PaimonCatalog as NativeCatalog
+
+ catalog_options = _catalog_context_options(table)
+ catalog_options['metastore'] = 'rest'
+ native_table = NativeCatalog(catalog_options).get_table((
+ table.identifier.get_database_name(),
table.identifier.get_table_name()))
+ if (native_table.location() != table.table_path
+ or native_table.rest_table_uuid() !=
table.catalog_environment.uuid):
+ return None
+ return native_table.copy_with_resolved_schema(schema_json)
+
+
def create_native_write_table(table):
- """Reconstruct a resolved table only for the filesystem publication
route."""
+ """Preserve the resolved schema and the catalog's 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.bucket_mode import BucketMode
from pypaimon.table.file_store_table import FileStoreTable
- # Native branch writes are not supported. Catalog-backed publication (REST
- # or custom version management) must continue through Python's environment.
+ # Native branch and postpone writes are not supported.
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:
+ or table.bucket_mode() == BucketMode.POSTPONE_MODE
+ or table.options.query_auth_enabled):
return None
from pypaimon_rust.datafusion import Table as NativeTable
@@ -98,8 +160,18 @@ def create_native_write_table(table):
table.options.dynamic_partition_overwrite())
options['snapshot.ignore-empty-commit'] = _option_value_to_string(
table.options.snapshot_ignore_empty_commit())
+ schema_json = JSON.to_json(table.table_schema.copy(new_options=options))
+ if environment.supports_version_management:
+ if not _rest_catalog_supported(table):
+ return None
+ return _native_rest_table(table, schema_json)
+ if 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
return NativeTable.from_resolved_schema(
- table.table_path,
JSON.to_json(table.table_schema.copy(new_options=options)),
+ table.table_path, schema_json,
database=table.identifier.get_database_name(),
table=table.identifier.get_table_name(),
options=file_io_options)
diff --git a/paimon-python/pypaimon/write/native_write.py
b/paimon-python/pypaimon/write/native_write.py
index fe2952ed0c..f96905f3e7 100644
--- a/paimon-python/pypaimon/write/native_write.py
+++ b/paimon-python/pypaimon/write/native_write.py
@@ -16,36 +16,54 @@
"""Optional Rust data writer behind PyPaimon's batch and stream builders."""
+from importlib import import_module
+
import pyarrow as pa
-from pypaimon.read.native_plan import native_method_available
-from pypaimon.schema.arrow_schema import normalize_arrow_strings
+from pypaimon.common.options.core_options import CoreOptions, MergeEngine
+from pypaimon.schema.arrow_schema import arrow_schemas_compatible,
normalize_arrow_strings
from pypaimon.schema.data_types import PyarrowFieldParser, is_blob_file_field
+from pypaimon.table.bucket_mode import BucketMode
+from pypaimon.utils.file_store_path_factory import canonical_data_file_path
from pypaimon.write.commit_message_serializer import deserialize_commit_message
from pypaimon.write.native_commit import create_native_write_table
from pypaimon.write.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'),
- ))
+ """Whether the optional Rust bindings are installed."""
+ try:
+ import_module('pypaimon_rust.datafusion')
+ except ImportError:
+ return False
+ return True
def create_native_write(table, commit_user, static_partition=None,
stream=False):
"""Return a native writer if the table can use the filesystem write
path."""
+ schema = PyarrowFieldParser.from_paimon_schema(table.table_schema.fields)
+ partition_types = [schema.field(name).type for name in
table.partition_keys]
if (not native_write_available()
or table.options.data_evolution_enabled()
+ or table.options.data_file_external_paths()
+ or table.bucket_mode() not in (BucketMode.HASH_FIXED,
+ BucketMode.BUCKET_UNAWARE)
+ or table.options.merge_engine() in (MergeEngine.FIRST_ROW,
+ MergeEngine.PARTIAL_UPDATE,
+ MergeEngine.AGGREGATE)
+ # Rust currently omits value stats for primary-key files.
+ or (table.is_primary_key_table and
table.options.metadata_stats_enabled())
+ or table.options.target_file_row_num()
+ != CoreOptions.TARGET_FILE_ROW_NUM.default_value()
+ or table.options.changelog_file_format() not in (None, 'parquet')
or table.options.file_format() != 'parquet'
+ # Rust validates nested Arrow child names strictly; PyPaimon
accepts
+ # equivalent layouts such as list<item> and list<element>.
+ or any(pa.types.is_nested(field.type)
+ or pa.types.is_fixed_size_binary(field.type) for field in
schema)
+ # Rust cannot encode these partition keys yet.
+ or any(pa.types.is_binary(type_) or pa.types.is_large_binary(type_)
+ or pa.types.is_floating(type_) for type_ in partition_types)
or any(is_blob_file_field(field) for field in
table.table_schema.fields)):
return None
native_table = create_native_write_table(table)
@@ -77,6 +95,7 @@ class NativeTableWrite:
self._native_writer = native_writer
self._python_writer = None
self._written = False
+ self._schema =
PyarrowFieldParser.from_paimon_schema(table.table_schema.fields)
def _switch_to_python(self):
if self._python_writer is not None:
@@ -110,6 +129,15 @@ class NativeTableWrite:
def write_arrow_batch(self, data):
if self._python_writer is not None:
return self._python_writer.write_arrow_batch(data)
+ if not arrow_schemas_compatible(
+ data.schema, self._schema, check_top_level_nullability=False,
+ allow_binary_compatibility=True):
+ raise ValueError(
+ "Input schema isn't consistent with table schema and write
cols. "
+ f"Input schema is: {data.schema} Table schema is:
{self._schema} "
+ "Write cols is: None")
+ if any(pa.types.is_fixed_size_binary(field.type) for field in
data.schema):
+ return self._switch_to_python().write_arrow_batch(data)
data = normalize_arrow_strings(data)
if data.num_rows:
# A failed native write may already have produced files. Never
@@ -144,9 +172,14 @@ class NativeTableWrite:
if commit_identifier is not None:
raise TypeError('BatchTableWrite.prepare_commit accepts no
identifier')
messages = self._native_writer.prepare_commit()
- return [deserialize_commit_message(
+ decoded = [deserialize_commit_message(
message.serialize(), self.table.partition_keys_fields,
self.table.trimmed_primary_keys_fields) for message in messages]
+ for message in decoded:
+ for file in message.new_files + message.changelog_files:
+ file.file_path = file.external_path or
canonical_data_file_path(
+ self.table, message.partition, message.bucket,
file.file_name)
+ return decoded
def close(self):
if self._python_writer is not None:
