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 702810b32b [Python] Complete native writer routing and guarded REST
publication (#10645)
702810b32b is described below
commit 702810b32b1b0c7189e31191e9234fca3d107c32
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Oct 19 14:30:16 2026 +0800
[Python] Complete native writer routing and guarded REST publication
(#10645)
---
docs/docs/pypaimon/writing.md | 26 ++
paimon-python/conftest.py | 4 +-
paimon-python/pypaimon/index/dynamic_bucket.py | 10 +-
.../pypaimon/read/scanner/file_scanner.py | 4 +-
.../tests/native_precomputed_bucket_test.py | 17 +-
.../tests/native_read_path_completion_test.py | 3 +-
.../tests/native_sorted_index_build_test.py | 13 +-
.../tests/native_test_instrumentation_test.py | 71 +++++
paimon-python/pypaimon/tests/native_update_test.py | 2 +
.../tests/native_write_path_completion_test.py | 82 ++++--
.../pypaimon/tests/native_write_routing_test.py | 326 +++++++++++++++++++++
paimon-python/pypaimon/tests/native_write_test.py | 5 +-
.../pypaimon/tests/query_auth_validation_test.py | 2 +
.../tests/ray_vector_filter_exactness_test.py | 1 +
.../scanner/file_scanner_schema_fields_test.py | 40 +++
paimon-python/pypaimon/tests/table_update_test.py | 1 +
.../pypaimon/tests/write/table_write_test.py | 4 +-
paimon-python/pypaimon/write/file_store_write.py | 17 +-
paimon-python/pypaimon/write/native_commit.py | 3 +-
paimon-python/pypaimon/write/native_update.py | 3 +
paimon-python/pypaimon/write/native_write.py | 35 ++-
.../pypaimon/write/postpone_batch_table_write.py | 4 +-
paimon-python/pypaimon/write/table_write.py | 33 +--
23 files changed, 621 insertions(+), 85 deletions(-)
diff --git a/docs/docs/pypaimon/writing.md b/docs/docs/pypaimon/writing.md
index 68d2c8aa0f..c74deca7c8 100644
--- a/docs/docs/pypaimon/writing.md
+++ b/docs/docs/pypaimon/writing.md
@@ -72,6 +72,32 @@ Use `new_postpone_fixed_bucket_write_builder()` to write
real buckets. New
partitions are buffered until `prepare_commit()`; existing ones are
incremental.
This builder currently supports `bucket-function.type=default` only.
+### Native Write and Commit
+
+With the `pypaimon-rust` extension installed, set `write.native.enabled=true`
+to use Rust for Parquet/BLOB writes and `commit.native.enabled=true` to use
+Rust for REST snapshot publication. Batch and stream writers retain the same
+prepare-and-commit lifecycle. Native commits also support overwrite,
+`truncate_table()`, `truncate_partitions()` and commit callbacks.
+
+Standard REST tables with `query-auth.enabled=true` can write and commit
+without requesting SELECT permission. The REST server still authorizes snapshot
+publication. Row filters and column masks continue to apply to explicit reads;
+internal sequence-number and HASH-index restoration reads manifest metadata.
+Predicate updates and deletes retain their existing read authorization checks.
+Query-authorized cross-partition index bootstrap remains unsupported.
+
+`write_ray()` uses the shared Python coordinator and lets workers select Native
+writers. For postpone fixed-bucket writes, overwrite plans use the new input
+unless an explicit bucket plan was supplied. A complete primary-key
+`with_write_type()` leaves the writer unchanged; primary-key column subsets
+remain unsupported.
+
+Native commit selection requires a standard REST catalog and a reconstructible
+built-in FileIO. Other catalogs and custom Python FileIO implementations retain
+the Python path. Binary partition keys require `partition.legacy-name=false`
for
+portable names across implementations.
+
To replace data, configure overwrite on the write builder before creating the
writer and commit object:
diff --git a/paimon-python/conftest.py b/paimon-python/conftest.py
index ee07dc8313..eeb1602b72 100644
--- a/paimon-python/conftest.py
+++ b/paimon-python/conftest.py
@@ -163,9 +163,9 @@ def pytest_configure(config):
original_prepare = TableCommit._prepare_native_commit
- def tracked_prepare(self, messages):
+ def tracked_prepare(self, messages, *args, **kwargs):
global _native_commit_count
- prepared = original_prepare(self, messages)
+ prepared = original_prepare(self, messages, *args, **kwargs)
if prepared is not None and _force_native_commit_for_test:
_native_commit_count += 1
return prepared
diff --git a/paimon-python/pypaimon/index/dynamic_bucket.py
b/paimon-python/pypaimon/index/dynamic_bucket.py
index d233cec1e0..dcface6d69 100644
--- a/paimon-python/pypaimon/index/dynamic_bucket.py
+++ b/paimon-python/pypaimon/index/dynamic_bucket.py
@@ -392,11 +392,11 @@ class HashBucketAssigner:
read_builder.with_filter(
predicate_builder.and_predicates(predicates)
)
- return {
- split.bucket
- for split in read_builder.new_scan().plan_all_files().splits()
- if split.files
- }
+ # HASH validation needs all live data-file buckets, not the subset a
+ # SELECT policy exposes. This internal manifest lookup returns no rows.
+ scanner = read_builder.new_scan().file_scanner
+ scanner.skip_level0 = False
+ return {entry.bucket for entry in scanner.plan_files()}
@dataclass
diff --git a/paimon-python/pypaimon/read/scanner/file_scanner.py
b/paimon-python/pypaimon/read/scanner/file_scanner.py
index b24624e69f..6fc822fe7f 100755
--- a/paimon-python/pypaimon/read/scanner/file_scanner.py
+++ b/paimon-python/pypaimon/read/scanner/file_scanner.py
@@ -613,7 +613,9 @@ class FileScanner:
return self.manifest_file_manager.read_entries_parallel(
manifest_files,
self._filter_manifest_entry,
- drop_stats=not keep_stats,
+ # Transported guarded splits must match the snapshot metadata when
+ # Rust restores their catalog grant, even if Python planned them.
+ drop_stats=not (keep_stats or
self.table.options.query_auth_enabled),
max_workers=max_workers,
early_entry_filter=self._build_early_bucket_filter(),
early_record_filter=early_row_filter,
diff --git a/paimon-python/pypaimon/tests/native_precomputed_bucket_test.py
b/paimon-python/pypaimon/tests/native_precomputed_bucket_test.py
index 5d070bccec..d096a609ce 100644
--- a/paimon-python/pypaimon/tests/native_precomputed_bucket_test.py
+++ b/paimon-python/pypaimon/tests/native_precomputed_bucket_test.py
@@ -17,7 +17,7 @@
"""Native writes of the partition/bucket groups produced by Daft."""
-from unittest.mock import patch
+from unittest.mock import Mock, patch
import pyarrow as pa
import pytest
@@ -377,7 +377,7 @@ def
test_empty_restore_ignores_existing_sequences_and_hashes(tmp_path, python_fa
@pytest.mark.parametrize('overwrite', [False, True])
-def test_empty_restore_preserves_python_write_authorization(tmp_path,
overwrite):
+def test_empty_restore_does_not_apply_select_policy_to_python_writes(tmp_path,
overwrite):
from pypaimon.catalog.catalog_exception import TableNoPermissionException
from pypaimon.catalog.table_query_auth import TableQueryAuthResult
@@ -389,14 +389,21 @@ def
test_empty_restore_preserves_python_write_authorization(tmp_path, overwrite)
# Exercise the ordinary writer selected before any Native data is staged.
writer._switch_to_python()
auth = TableQueryAuthResult(None, {'value': '{"name":"NULL"}'})
+ authorize = Mock(return_value=auth)
try:
with patch.object(type(table.catalog_environment), 'table_query_auth',
- return_value=lambda _: auth):
+ return_value=authorize):
+ writer.write_arrow(_batch([1], ['written']), bucket=0)
+ messages = writer.prepare_commit()
+ assert messages[0].new_files[0].min_sequence_number == 0
+ authorize.assert_not_called()
+ # Only the explicit whole-table read requires an unrestricted
grant.
with pytest.raises(TableNoPermissionException):
- writer.write_arrow(_batch([1], ['forbidden']), bucket=0)
+ table.new_read_builder().new_scan().plan_all_files()
+ builder.new_commit().commit(messages)
finally:
writer.close()
- assert table.snapshot_manager().get_latest_snapshot() is None
+ assert _rows(table) == [{'id': 1, 'value': 'written', 'pt': 'a'}]
def test_python_pk_file_sequence_range_matches_deduplicated_rows(tmp_path):
diff --git a/paimon-python/pypaimon/tests/native_read_path_completion_test.py
b/paimon-python/pypaimon/tests/native_read_path_completion_test.py
index 851d1b7d0b..d86e3040db 100644
--- a/paimon-python/pypaimon/tests/native_read_path_completion_test.py
+++ b/paimon-python/pypaimon/tests/native_read_path_completion_test.py
@@ -258,7 +258,8 @@ def
test_authorized_python_and_serialized_splits_use_native_reader(guarded_table
source = table.copy({'scan.native-plan.enabled': str(transport ==
'wire').lower(), 'read.native.enabled': 'false'})
splits = source.new_read_builder().new_scan().plan().splits()
if transport == 'wire':
- splits = [deserialize_split_v1(serialize_split_v1(split), []) for
split in splits]
+ splits = [deserialize_split_v1(serialize_split_v1(split), [],
table.trimmed_primary_keys_fields)
+ for split in splits]
assert all(getattr(split, '_native_split', None) is None for split in
splits)
builder = table.copy({'read.native.enabled':
'true'}).new_read_builder().with_projection(['id', 'value'])
with patch('pypaimon.read.table_read.TableRead._create_split_read',
diff --git a/paimon-python/pypaimon/tests/native_sorted_index_build_test.py
b/paimon-python/pypaimon/tests/native_sorted_index_build_test.py
index 1005659916..1242773d16 100644
--- a/paimon-python/pypaimon/tests/native_sorted_index_build_test.py
+++ b/paimon-python/pypaimon/tests/native_sorted_index_build_test.py
@@ -471,19 +471,18 @@ def
test_partition_set_predicates_follow_java_null_semantics(rest_catalog, kind,
@pytest.mark.parametrize('restricted', [False, True])
def test_query_authorization_remains_on_python_build_path(rest_catalog,
restricted):
- from pypaimon.catalog.rest.rest_catalog import RESTCatalog
- from pypaimon.catalog.table_query_auth import TableQueryAuthResult
+ from pypaimon.api.api_response import AuthTableQueryResponse
from pypaimon.catalog.catalog_exception import TableNoPermissionException
- catalog, _ = rest_catalog
+ catalog, server = rest_catalog
table = _create(catalog)
_append(table, ROWS)
table = table.copy({'query-auth.enabled': 'true'})
- auth = TableQueryAuthResult(None, {'name': '{"name":"NULL"}'} if
restricted else None)
+ server.table_query_auth_store[table.identifier.get_full_name()] =
AuthTableQueryResponse(
+ column_masking={'name': '{"name":"NULL"}'} if restricted else None)
original = GlobalIndexBuilder._build_sorted_index
- with patch.object(RESTCatalog, 'auth_table_query', return_value=auth), \
- patch('pypaimon.globalindex.native_index_build._native_table',
- side_effect=AssertionError('native auth build')), \
+ with patch('pypaimon.globalindex.native_index_build._native_table',
+ side_effect=AssertionError('native auth build')), \
patch.object(GlobalIndexBuilder, '_build_sorted_index',
autospec=True, side_effect=original) as build:
if restricted:
with pytest.raises(TableNoPermissionException):
diff --git a/paimon-python/pypaimon/tests/native_test_instrumentation_test.py
b/paimon-python/pypaimon/tests/native_test_instrumentation_test.py
new file mode 100644
index 0000000000..6faf44dc20
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_test_instrumentation_test.py
@@ -0,0 +1,71 @@
+# 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.
+
+"""Native CI instrumentation preserves the public mutation API."""
+
+import importlib.util
+from pathlib import Path
+from unittest.mock import Mock
+
+import pytest
+
+from pypaimon.write.table_commit import BatchTableCommit, TableCommit
+
+pytestmark = [pytest.mark.python_plan, pytest.mark.python_read,
+ pytest.mark.python_write, pytest.mark.python_commit]
+
+
[email protected]('operation,args', [
+ ('truncate_table', ()), ('truncate_partitions', ([{'pt': 'a'}],)),
+])
[email protected]('fail', [False, True])
+def test_native_commit_tracking_forwards_batch_truncate(monkeypatch,
operation, args, fail):
+ # Load the actual CI hook in isolation. Its patches are scoped to this test
+ # and must not replace the surrounding pytest session's counters.
+ path = Path(__file__).resolve().parents[2] / 'conftest.py'
+ spec = importlib.util.spec_from_file_location('native_instrumentation',
str(path))
+ instrumentation = importlib.util.module_from_spec(spec)
+ spec.loader.exec_module(instrumentation)
+ for key in ('PLAN', 'READ', 'WRITE', 'UPDATE'):
+ monkeypatch.setenv('PYPAIMON_TEST_NATIVE_' + key, '0')
+ monkeypatch.setenv('PYPAIMON_TEST_NATIVE_COMMIT', '1')
+ backend = Mock()
+ calls = []
+
+ def prepare(self, messages, *, batch=False):
+ calls.append((messages, batch))
+ return backend, []
+
+ monkeypatch.setattr(TableCommit, '_prepare_native_commit', prepare)
+ instrumentation.pytest_configure(Mock())
+ instrumentation._force_native_commit_for_test = True
+ commit = object.__new__(BatchTableCommit)
+ commit.batch_committed = False
+ commit.file_store_commit = Mock()
+ error = RuntimeError('snapshot may already have published')
+ if fail:
+ getattr(backend, operation).side_effect = error
+ with pytest.raises(RuntimeError) as raised:
+ getattr(commit, operation)(*args)
+ assert raised.value is error
+ else:
+ getattr(commit, operation)(*args)
+ assert calls == [([], True)]
+ assert instrumentation._native_commit_count == 1
+ getattr(backend, operation).assert_called_once_with(*args)
+ commit.file_store_commit.truncate_table.assert_not_called()
+ commit.file_store_commit.drop_partitions.assert_not_called()
diff --git a/paimon-python/pypaimon/tests/native_update_test.py
b/paimon-python/pypaimon/tests/native_update_test.py
index 8ed01dbf62..709c79d540 100644
--- a/paimon-python/pypaimon/tests/native_update_test.py
+++ b/paimon-python/pypaimon/tests/native_update_test.py
@@ -655,6 +655,8 @@ def test_predicate_assignment_order_with_later_partial_file(
'row-tracking.enabled': 'true',
'data-evolution.enabled': 'true',
'write.native.enabled': str(native).lower(),
+ # Compare against Python's scan order, independently of the CI lane.
+ 'scan.native-plan.enabled': 'false', 'read.native.enabled': 'false',
}), False)
table = catalog.get_table('default.t')
for ids in ([1, 2], [3, 4]):
diff --git a/paimon-python/pypaimon/tests/native_write_path_completion_test.py
b/paimon-python/pypaimon/tests/native_write_path_completion_test.py
index 939a7b3183..760d735978 100644
--- a/paimon-python/pypaimon/tests/native_write_path_completion_test.py
+++ b/paimon-python/pypaimon/tests/native_write_path_completion_test.py
@@ -453,41 +453,59 @@ def
test_native_truncate_partition_specs_use_java_typed_strings(tmp_path, native
commit.close()
-@native
-def
test_automatic_postpone_blob_rows_keep_unknown_sources_until_prepare(tmp_path):
[email protected]_plan
[email protected]_read
[email protected]('fail', [False, True])
+def test_postpone_blob_row_bridge_keeps_sources_until_prepare(fail):
+ from types import SimpleNamespace
+ from pypaimon.common.options.core_options import CoreOptions
+ from pypaimon.common.options.options import Options
from pypaimon.schema.data_types import AtomicType, DataField
from pypaimon.table.row.blob import Blob, BlobDescriptor
from pypaimon.table.row.generic_row import GenericRow
from pypaimon.write.native_write import NativePostponeFixedBucketTableWrite
- catalog = CatalogFactory.create({'warehouse': str(tmp_path /
'blob-postpone')})
- catalog.create_database('default', True)
+
+ # Exercise the bridge's deferred input lifetime without creating a table:
+ # Python catalog validation does not allow a primary-key BLOB schema.
fields = [DataField(0, 'id', AtomicType('INT')), DataField(1, 'pt',
AtomicType('STRING')),
DataField(2, 'payload', AtomicType('BLOB'))]
- catalog.create_table('default.t', Schema(
- fields, partition_keys=['pt'], primary_keys=['id', 'pt'], options={
- 'bucket': '-2', 'write.native.enabled': 'true',
- 'postpone.target-row-num-per-bucket': '1'}), False)
- table = catalog.get_table('default.t')
-
- def write(rows, buffered):
- builder = table.new_postpone_fixed_bucket_write_builder()
- writer, commit = builder.new_write(), builder.new_commit()
- try:
- assert isinstance(writer, NativePostponeFixedBucketTableWrite)
- for identifier, partition in rows:
- writer.write_row(GenericRow(
- [identifier, partition,
Blob.from_data(str(identifier).encode())], table.fields))
- assert len(writer._blob_rows.readers) == buffered
- messages = writer.prepare_commit()
- assert not writer._blob_rows.readers
- commit.commit(messages)
- finally:
- writer.close()
- commit.close()
-
- write([(1, 'known')], 1)
- write([(2, 'known'), (3, 'new')], 1)
- rows = _rows(table)
- assert [row['id'] for row in rows] == [1, 2, 3]
- assert [Blob.from_descriptor_bytes(row['payload'],
table.file_io).to_data() for row in rows] == [b'1', b'2', b'3']
- assert all(not
BlobDescriptor.deserialize(row['payload']).uri.startswith('pypaimon-blob-row:')
for row in rows)
+ table = SimpleNamespace(
+ table_schema=SimpleNamespace(fields=fields), fields=fields,
+ field_dict={field.name: field for field in fields},
partition_keys=['pt'],
+ is_primary_key_table=True,
+ options=CoreOptions(Options({'bucket': '-2',
'postpone.target-row-num-per-bucket': '1'})))
+ backend = Mock()
+ staged, payloads = [], []
+ backend.write_arrow.side_effect = lambda batch, bucket:
staged.append(batch)
+ writer = NativePostponeFixedBucketTableWrite(table, 'job', None, None,
backend)
+ factory = backend.with_blob_uri_reader_factory.call_args[0][0]
+ error = RuntimeError('prepare failed')
+
+ def prepare():
+ for batch in staged:
+ for encoded in batch.column('payload').to_pylist():
+ uri = BlobDescriptor.deserialize(encoded).uri
+ with factory.create(uri).new_input_stream(uri) as stream:
+ payloads.append(stream.read())
+ if fail:
+ raise error
+ return []
+
+ backend.prepare_commit.side_effect = prepare
+ try:
+ for identifier in [1, 2]:
+ writer.write_row(GenericRow(
+ [identifier, 'new', Blob.from_data(str(identifier).encode())],
fields))
+ assert len(factory.readers) == 2
+ assert not payloads
+ if fail:
+ with pytest.raises(RuntimeError) as raised:
+ writer.prepare_commit()
+ assert raised.value is error
+ else:
+ assert writer.prepare_commit() == []
+ assert payloads == [b'1', b'2']
+ assert not factory.readers
+ backend.abort.assert_not_called()
+ finally:
+ writer.close()
diff --git a/paimon-python/pypaimon/tests/native_write_routing_test.py
b/paimon-python/pypaimon/tests/native_write_routing_test.py
new file mode 100644
index 0000000000..b2871d506f
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_write_routing_test.py
@@ -0,0 +1,326 @@
+# 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.
+
+"""Existing writer APIs stay native without changing query-read policies."""
+
+import json
+from contextlib import ExitStack
+from types import ModuleType
+from unittest.mock import Mock, patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.api.api_response import AuthTableQueryResponse, ErrorResponse
+from pypaimon.catalog.catalog_exception import TableNoPermissionException
+from pypaimon.read.native_plan import native_runtime_available
+from pypaimon.tests import native_plan_rest_test
+from pypaimon.write.native_commit import create_native_write_table
+from pypaimon.write.native_write import NativePostponeFixedBucketTableWrite,
NativeTableWrite
+from pypaimon.write.postpone_bucket import PostponeBucketPlan
+
+rest_catalog = native_plan_rest_test.rest_catalog
+pytestmark = [pytest.mark.python_write, pytest.mark.python_commit]
+native = pytest.mark.skipif(not native_runtime_available(),
reason='pypaimon-rust not installed')
+SCHEMA = pa.schema([('id', pa.int64()), ('pt', pa.string()), ('value',
pa.string())])
+
+
+def _table(tmp_path, mode='append', catalog=None, use_native=False,
guarded=False):
+ catalog = catalog or CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ options = {
+ 'write.native.enabled': str(use_native).lower(),
'commit.native.enabled': str(use_native).lower(),
+ 'query-auth.enabled': str(guarded).lower(), 'read.native.enabled':
'false',
+ 'scan.native-plan.enabled': 'false',
+ }
+ if mode == 'data-evolution':
+ options.update({'row-tracking.enabled': 'true',
'data-evolution.enabled': 'true'})
+ if mode == 'postpone':
+ options.update({'bucket': '-2', 'postpone.default-bucket-num': '2'})
+ elif mode in ('fixed', 'dynamic'):
+ options['bucket'] = '1' if mode == 'fixed' else '-1'
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ SCHEMA, partition_keys=['pt'], primary_keys=['pt', 'id']
+ if mode in ('fixed', 'dynamic', 'postpone') else [], options=options),
False)
+ return catalog.get_table('default.t')
+
+
+def _writer(table, native_writer, partition=None):
+ return NativeTableWrite(table, 'job', partition, False, native_writer)
+
+
[email protected]('mode', ['append', 'fixed'])
[email protected]('staged', [False, True])
[email protected]('partition', [None, {'pt': 'a'}, {}])
+def
test_write_ray_reuses_coordination_without_switching_local_writer(tmp_path,
mode, staged, partition):
+ table = _table(tmp_path, mode)
+ underlying = Mock()
+ writer = _writer(table, underlying, {'pt': 'builder'})
+ if staged:
+ writer.write_arrow_batch(pa.RecordBatch.from_pylist(
+ [{'id': 1, 'pt': 'a', 'value': 'local'}], schema=SCHEMA))
+ module = ModuleType('pypaimon.write.ray_datasink')
+ module.write_paimon_dataset = Mock()
+ dataset = object()
+ with patch.dict('sys.modules', {'pypaimon.write.ray_datasink': module}), \
+ patch.object(writer, '_new_python_writer',
side_effect=AssertionError('Python writer')):
+ assert writer.write_ray(dataset, overwrite=True, concurrency=3,
+ ray_remote_args={'num_cpus': 2},
hash_fixed_precluster='map_groups',
+ static_partition=partition) is None
+ module.write_paimon_dataset.assert_called_once_with(
+ dataset, table, overwrite=True, concurrency=3,
ray_remote_args={'num_cpus': 2},
+ hash_fixed_precluster='map_groups', static_partition={'pt': 'builder'}
if partition is None else partition)
+ assert writer._python_writer is None
+ underlying.close.assert_not_called()
+
+
[email protected]('plan', [None, PostponeBucketPlan({('a',): 4})])
[email protected]('overwrite', [False, True])
+def test_native_postpone_ray_preserves_fixed_bucket_planner(tmp_path, plan,
overwrite):
+ table = _table(tmp_path, 'postpone')
+ writer = NativePostponeFixedBucketTableWrite(table, 'job', {} if overwrite
else None, plan, Mock())
+ module = ModuleType('pypaimon.write.ray_datasink')
+ module.write_paimon_dataset = Mock()
+ with patch.dict('sys.modules', {'pypaimon.write.ray_datasink': module}):
+ writer.write_ray(object(), hash_fixed_precluster='off')
+ options = module.write_paimon_dataset.call_args[1]
+ planner = options['postpone_bucket_planner']
+ assert planner.current_plan().as_dict() == ({} if plan is None else
{('a',): 4})
+ assert options['static_partition'] == ({} if overwrite else None)
+ assert writer._python_writer is None
+
+
[email protected]('native_writer', [False, True])
[email protected]('mode', ['append', 'builder-overwrite',
'call-overwrite', 'call-partition', 'call-empty'])
[email protected]('explicit_plan', [False, True])
+def test_ray_postpone_plans_use_effective_overwrite(tmp_path, native_writer,
mode, explicit_plan):
+ from pypaimon.write.postpone_batch_table_write import
PostponeFixedBucketBatchTableWrite
+ from pypaimon.write.postpone_bucket import PostponeBucketPlanner
+
+ table = _table(tmp_path, 'postpone')
+ plan = PostponeBucketPlan({('a',): 8}) if explicit_plan else None
+ partition = {} if mode == 'builder-overwrite' else None
+ module = ModuleType('pypaimon.write.ray_datasink')
+ module.write_paimon_dataset = Mock()
+ with patch.object(PostponeBucketPlanner, '_load_bucket_metadata',
return_value={('a',): 4}), \
+ patch.dict('sys.modules', {'pypaimon.write.ray_datasink': module}):
+ writer = (NativePostponeFixedBucketTableWrite(table, 'job', partition,
plan, Mock()) if native_writer
+ else PostponeFixedBucketBatchTableWrite(table, 'job',
partition, plan))
+ try:
+ writer.write_ray(object(), overwrite=mode == 'call-overwrite',
static_partition=(
+ {'pt': 'a'} if mode == 'call-partition' else {} if mode ==
'call-empty' else None))
+ finally:
+ writer.close()
+ actual =
module.write_paimon_dataset.call_args[1]['postpone_bucket_planner'].current_plan().as_dict()
+ assert actual == ({('a',): 8} if explicit_plan else {('a',): 4} if mode ==
'append' else {})
+
+
[email protected]('mode', ['append', 'fixed', 'data-evolution'])
+def test_non_video_group_boundaries_validate_without_switching(tmp_path, mode):
+ underlying = Mock()
+ writer = _writer(_table(tmp_path, mode), underlying)
+ for staged in (False, True):
+ if staged:
+ writer.write_arrow_batch(pa.RecordBatch.from_pylist(
+ [{'id': 1, 'pt': 'a', 'value': 'x'}], schema=SCHEMA))
+ for count in (1, 1000):
+ assert writer.roll_before_group_if_needed(count) is None
+ for count in (True, False, 0, -1, 1.0, '1', None):
+ with pytest.raises(ValueError, match='positive integer'):
+ writer.roll_before_group_if_needed(count)
+ assert writer._python_writer is None
+ underlying.close.assert_not_called()
+
+
[email protected]('names', [SCHEMA.names, list(reversed(SCHEMA.names))])
+def test_pk_complete_write_type_keeps_table_order_and_native_writer(tmp_path,
names):
+ underlying = Mock()
+ writer = _writer(_table(tmp_path, 'fixed'), underlying)
+ assert writer.with_write_type(iter(names)) is writer
+ writer.write_arrow_batch(pa.RecordBatch.from_pylist(
+ [{'id': 1, 'pt': 'a', 'value': 'x'}], schema=SCHEMA))
+ assert underlying.write_arrow.call_args[0][0].schema.names == SCHEMA.names
+ underlying.with_write_type.assert_not_called()
+ assert writer._python_writer is None
+
+
+def
test_standard_rest_query_auth_write_table_keeps_native_publication(rest_catalog):
+ catalog, _ = rest_catalog
+ table = _table(None, catalog=catalog, guarded=True)
+ module = ModuleType('pypaimon_rust.datafusion')
+ module.Table = Mock()
+ module.PaimonCatalog = Mock()
+ resolved = Mock(spec=['location', 'rest_table_uuid', 'branch',
'copy_with_resolved_schema'])
+ module.PaimonCatalog.return_value.get_table.return_value = resolved
+ resolved.location.return_value = table.table_path
+ resolved.rest_table_uuid.return_value = table.catalog_environment.uuid
+ resolved.branch.return_value = 'main'
+ resolved.copy_with_resolved_schema.return_value = resolved
+ with patch.dict('sys.modules', {'pypaimon_rust.datafusion': module}):
+ assert create_native_write_table(table) is resolved
+ resolved.copy_with_resolved_schema.assert_called_once()
+ schema = json.loads(resolved.copy_with_resolved_schema.call_args[0][0])
+ assert schema['options']['query-auth.enabled'] == 'true'
+
+
+def test_read_based_updates_keep_existing_guarded_selection(rest_catalog):
+ from pypaimon.write.native_update import _native_row_id_table
+
+ catalog, _ = rest_catalog
+ table = _table(None, 'data-evolution', catalog, use_native=True,
guarded=True)
+ with patch('pypaimon.write.native_update.native_runtime_available',
return_value=True), \
+ patch('pypaimon.write.native_update.create_native_write_table') as
create:
+ assert _native_row_id_table(table) is None
+ create.assert_not_called()
+
+
+def _rows(table):
+ builder = table.new_read_builder()
+ return
builder.new_read().to_arrow(builder.new_scan().plan().splits()).sort_by(
+ [('id', 'ascending'), ('value', 'ascending')]).to_pylist()
+
+
[email protected]('use_native,mode', [
+ (False, 'append'), (False, 'fixed'), (False, 'dynamic'), (False,
'postpone'),
+ pytest.param(True, 'append', marks=native), pytest.param(True, 'fixed',
marks=native),
+ pytest.param(True, 'dynamic', marks=native), pytest.param(True,
'data-evolution', marks=native),
+ pytest.param(True, 'postpone', marks=native),
+])
+def
test_query_auth_writes_do_not_require_select_and_reads_still_authorize(rest_catalog,
use_native, mode):
+ catalog, server = rest_catalog
+ table = _table(None, mode, catalog, use_native, guarded=True)
+ denied = server._mock_response(ErrorResponse('TABLE', 't', 'SELECT
denied', 403), 403)
+ with patch.object(server, '_table_query_auth_handle', return_value=denied)
as query:
+ for rows in ([{'id': 1, 'pt': 'a', 'value': 'before'}, {'id': 2, 'pt':
'b', 'value': 'hidden'}],
+ [{'id': 1, 'pt': 'a', 'value': 'after'}]):
+ builder = (table.new_postpone_fixed_bucket_write_builder()
+ if mode == 'postpone' else
table.new_batch_write_builder())
+ writer, commit = builder.new_write(), builder.new_commit()
+ with ExitStack() as stack:
+ if use_native:
+ assert isinstance(writer, NativeTableWrite)
+ stack.enter_context(patch.object(commit.file_store_commit,
'commit',
+
side_effect=AssertionError('Python commit')))
+ try:
+ writer.write_arrow(pa.Table.from_pylist(rows,
schema=SCHEMA))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+ query.assert_not_called()
+ with pytest.raises(TableNoPermissionException):
+ _rows(table)
+
+ # Row policies apply only when the newly stored rows are read, including
+ # masking stored values; hidden rows were still written normally.
+ field = {'name': 'id', 'type': 'BIGINT', 'index': 0}
+ server.table_query_auth_store[table.identifier.get_full_name()] =
AuthTableQueryResponse(
+ filter=[json.dumps({'kind': 'LEAF', 'transform': {'name': 'FIELD_REF',
'fieldRef': field},
+ 'function': 'EQUAL', 'literals': [1]})],
+ column_masking={'value': json.dumps({'name': 'CONCAT', 'inputs':
['masked']})})
+ visible = _rows(table)
+ count = 2 if mode in ('append', 'data-evolution') else 1
+ assert visible == [{'id': 1, 'pt': 'a', 'value': 'masked'}] * count
+
+
[email protected]('restore_id', [0, 1])
[email protected]('use_native', [False, pytest.param(True,
marks=native)])
+def
test_guarded_dynamic_restore_keeps_explicit_snapshot_baseline(rest_catalog,
restore_id, use_native):
+ catalog, server = rest_catalog
+ table = _table(None, 'dynamic', catalog,
guarded=True).copy({'dynamic-bucket.target-row-num': '1'})
+ denied = server._mock_response(ErrorResponse('TABLE', 't', 'SELECT
denied', 403), 403)
+ with patch.object(server, '_table_query_auth_handle', return_value=denied)
as query:
+ for rows in ([{'id': 1, 'pt': 'a', 'value': 'base'}], [
+ {'id': 1, 'pt': 'a', 'value': 'future-1'}, {'id': 1, 'pt':
'a', 'value': 'future-2'},
+ {'id': 2, 'pt': 'a', 'value': 'new-bucket'}]):
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(pa.Table.from_pylist(rows, schema=SCHEMA))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+ assert table.snapshot_manager().get_latest_snapshot().id == 2
+ restored = table.copy({'write.native.enabled':
str(use_native).lower()})
+ builder =
restored.new_batch_write_builder().with_restore_snapshot(restore_id)
+ writer = builder.new_write()
+ assert isinstance(writer, NativeTableWrite) == use_native
+ try:
+ writer.write_arrow(pa.Table.from_pylist([
+ {'id': 1, 'pt': 'a', 'value': 'restored'}, {'id': 3, 'pt':
'a', 'value': 'another'}], schema=SCHEMA))
+ messages = writer.prepare_commit()
+ assert {message.bucket for message in messages} == {0, 1}
+ files = {message.bucket: message.new_files[0] for message in
messages}
+ assert files[0].min_sequence_number == restore_id
+ assert files[1].min_sequence_number == 0
+ # These messages are explicitly abandoned, never submitted for
+ # publication. The old snapshots and their HASH files stay intact.
+ commit = builder.new_commit()
+ try:
+ commit.abort(messages)
+ finally:
+ commit.close()
+ finally:
+ writer.close()
+ query.assert_not_called()
+ assert table.snapshot_manager().get_latest_snapshot().id == 2
+
+
[email protected]('use_native', [False, pytest.param(True,
marks=native)])
+def test_guarded_overwrite_and_truncate_keep_rest_publication(rest_catalog,
use_native):
+ catalog, server = rest_catalog
+ table = _table(None, catalog=catalog, use_native=use_native, guarded=True)
+ denied = server._mock_response(ErrorResponse('TABLE', 't', 'SELECT
denied', 403), 403)
+ with patch.object(server, '_table_query_auth_handle', return_value=denied)
as query:
+ for partition, rows in [(None, [
+ {'id': 1, 'pt': 'a', 'value': 'old'}, {'id': 2, 'pt': 'b',
'value': 'keep'}]),
+ ({'pt': 'a'}, [{'id': 3, 'pt': 'a', 'value': 'replace'}])]:
+ builder = table.new_batch_write_builder()
+ if partition is not None:
+ builder.overwrite(partition)
+ writer, commit = builder.new_write(), builder.new_commit()
+ with ExitStack() as stack:
+ if use_native:
+ assert isinstance(writer, NativeTableWrite)
+ stack.enter_context(patch.object(commit.file_store_commit,
'commit',
+
side_effect=AssertionError('Python commit')))
+ stack.enter_context(patch.object(commit.file_store_commit,
'overwrite',
+
side_effect=AssertionError('Python overwrite')))
+ try:
+ writer.write_arrow(pa.Table.from_pylist(rows,
schema=SCHEMA))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+ assert
table.snapshot_manager().get_latest_snapshot().total_record_count == 2
+ commit = table.new_batch_write_builder().new_commit()
+ with ExitStack() as stack:
+ if use_native:
+ stack.enter_context(patch.object(commit.file_store_commit,
'drop_partitions',
+
side_effect=AssertionError('Python drop')))
+ stack.enter_context(patch.object(commit.file_store_commit,
'truncate_table',
+
side_effect=AssertionError('Python truncate')))
+ try:
+ commit.truncate_partitions([{'pt': 'b'}])
+ assert
table.snapshot_manager().get_latest_snapshot().total_record_count == 1
+ commit.truncate_table()
+ assert
table.snapshot_manager().get_latest_snapshot().total_record_count == 0
+ finally:
+ commit.close()
+ query.assert_not_called()
diff --git a/paimon-python/pypaimon/tests/native_write_test.py
b/paimon-python/pypaimon/tests/native_write_test.py
index 5816909a14..f5d0e95923 100644
--- a/paimon-python/pypaimon/tests/native_write_test.py
+++ b/paimon-python/pypaimon/tests/native_write_test.py
@@ -492,12 +492,13 @@ def
test_blob_only_partial_write_uses_native_and_updates_existing_range(tmp_path
@requires_native
@pytest.mark.parametrize('streaming', [False, True])
-def test_primary_key_full_write_type_selects_writer_before_staging(tmp_path,
streaming):
+def test_primary_key_full_write_type_keeps_native_writer(tmp_path, streaming):
table = _table(tmp_path, primary_key=True)
builder = table.new_stream_write_builder() if streaming else
table.new_batch_write_builder()
writer = builder.new_write()
try:
- assert writer.with_write_type(table.field_names) is
writer._python_writer
+ assert writer.with_write_type(table.field_names) is writer
+ assert writer._python_writer is None
writer.write_arrow_batch(_batch([1], ['a']))
messages = writer.prepare_commit(7) if streaming else
writer.prepare_commit()
commit = builder.new_commit()
diff --git a/paimon-python/pypaimon/tests/query_auth_validation_test.py
b/paimon-python/pypaimon/tests/query_auth_validation_test.py
index 6aaf1a9c6c..66a99b47c3 100644
--- a/paimon-python/pypaimon/tests/query_auth_validation_test.py
+++ b/paimon-python/pypaimon/tests/query_auth_validation_test.py
@@ -792,6 +792,7 @@ def _exploding_schema_manager():
return SimpleNamespace(latest=boom)
[email protected]_read
class TestSchemaRefreshAsksTheCatalog(unittest.TestCase):
def _table(self, cached, latest):
@@ -865,6 +866,7 @@ class TestSchemaRefreshAsksTheCatalog(unittest.TestCase):
pickle.dumps(read._auth_schema)
[email protected]_read
class TestReaderAppliesTheRules(unittest.TestCase):
def _read(self, fields, data, latest=None, projected=None):
diff --git a/paimon-python/pypaimon/tests/ray_vector_filter_exactness_test.py
b/paimon-python/pypaimon/tests/ray_vector_filter_exactness_test.py
index 983ba96ef8..91761c2bd3 100644
--- a/paimon-python/pypaimon/tests/ray_vector_filter_exactness_test.py
+++ b/paimon-python/pypaimon/tests/ray_vector_filter_exactness_test.py
@@ -141,6 +141,7 @@ def
test_worker_filters_overlapping_scalar_shards_at_planned_snapshot(tmp_path,
@pytest.mark.parametrize("batch", [False, True])
@pytest.mark.parametrize("mode", ["exact", "candidate", "unavailable"])
[email protected]_plan
def test_shared_scalar_index_is_evaluated_once_on_ray(table, tmp_path,
ray_cluster, batch, mode):
import os
from pathlib import Path
diff --git
a/paimon-python/pypaimon/tests/scanner/file_scanner_schema_fields_test.py
b/paimon-python/pypaimon/tests/scanner/file_scanner_schema_fields_test.py
index b75e7505d0..67b807b2c8 100644
--- a/paimon-python/pypaimon/tests/scanner/file_scanner_schema_fields_test.py
+++ b/paimon-python/pypaimon/tests/scanner/file_scanner_schema_fields_test.py
@@ -18,6 +18,11 @@
import unittest
from unittest.mock import MagicMock
+import pyarrow as pa
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.manifest.manifest_list_manager import ManifestListManager
from pypaimon.read.scanner.file_scanner import FileScanner
from pypaimon.schema.data_types import AtomicType, DataField
@@ -90,5 +95,40 @@ class
FileScannerSchemaFieldsShortCircuitTest(unittest.TestCase):
scanner.table.schema_manager.get_schema.assert_not_called()
[email protected]_plan
[email protected]_read
[email protected]_write
[email protected]_commit
[email protected]('guarded', [False, True])
+def test_transported_query_auth_files_retain_snapshot_statistics(tmp_path,
guarded):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ schema = pa.schema([('id', pa.int32()), ('value', pa.string())])
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(schema,
options={
+ 'metadata.stats-mode': 'full',
+ 'write.native.enabled': 'false', 'commit.native.enabled': 'false',
+ 'scan.native-plan.enabled': 'false', 'read.native.enabled': 'false',
+ }), False)
+ table = catalog.get_table('default.t')
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(pa.table({'id': [1, 2], 'value': ['a', 'z']},
schema=schema))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+ table = table.copy({'query-auth.enabled': str(guarded).lower()})
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ manifests = ManifestListManager(table).read_all(snapshot)
+ scanner = table.new_read_builder().new_scan().file_scanner
+ entries = scanner.read_manifest_entries(manifests)
+ assert len(entries) == 1
+ # Core validates the complete transported metadata against the snapshot.
+ # Ordinary scans retain their existing reduced-statistics behavior.
+ assert entries[0].file.value_stats.null_counts == ([0, 0] if guarded else
[])
+ assert entries[0].file.value_stats_cols == (None if guarded else [])
+
+
if __name__ == "__main__":
unittest.main()
diff --git a/paimon-python/pypaimon/tests/table_update_test.py
b/paimon-python/pypaimon/tests/table_update_test.py
index 83f971c2e2..552d3d0d29 100644
--- a/paimon-python/pypaimon/tests/table_update_test.py
+++ b/paimon-python/pypaimon/tests/table_update_test.py
@@ -366,6 +366,7 @@ class _TableUpdateTestBase(DataEvolutionTestBase):
result['name'].to_pylist(),
)
+ @pytest.mark.python_plan
@pytest.mark.python_write
def test_update_by_predicate_processes_one_file_group_at_a_time(self):
from pypaimon.write.table_update_by_row_id import TableUpdateByRowId
diff --git a/paimon-python/pypaimon/tests/write/table_write_test.py
b/paimon-python/pypaimon/tests/write/table_write_test.py
index 13a94e9c68..072d050489 100644
--- a/paimon-python/pypaimon/tests/write/table_write_test.py
+++ b/paimon-python/pypaimon/tests/write/table_write_test.py
@@ -803,10 +803,11 @@ class TableWriteTest(unittest.TestCase):
PostponeFixedBucketBatchTableWrite,
PostponeFixedBucketWriteBuilder,
)
+ from pypaimon.write.native_write import
NativePostponeFixedBucketTableWrite
self.assertIsInstance(
write_builder, PostponeFixedBucketWriteBuilder)
self.assertIsInstance(
- table_write, PostponeFixedBucketBatchTableWrite)
+ table_write, (PostponeFixedBucketBatchTableWrite,
NativePostponeFixedBucketTableWrite))
table_commit = write_builder.new_commit()
table_write.write_arrow(expect)
commit_messages = table_write.prepare_commit()
@@ -1128,6 +1129,7 @@ class TableWriteTest(unittest.TestCase):
self.assertEqual(1, result.num_rows)
@parameterized.expand([('mod',), ('hive',)])
+ @pytest.mark.python_write
def test_postpone_rejects_unsupported_bucket_function(
self, bucket_function):
pa_schema = pa.schema([
diff --git a/paimon-python/pypaimon/write/file_store_write.py
b/paimon-python/pypaimon/write/file_store_write.py
index 7a75291334..151c038154 100644
--- a/paimon-python/pypaimon/write/file_store_write.py
+++ b/paimon-python/pypaimon/write/file_store_write.py
@@ -386,7 +386,7 @@ class FileStoreWrite:
if snapshot is None:
raise ValueError("Snapshot id '{}' doesn't
exist".format(self.restore_snapshot_id))
# Replace inherited selectors and mode, including an explicit empty
- # view. plan_all_files still validates row filters and column
masks.
+ # view for the writer's internal manifest lookup.
return self.table._copy_with_snapshot(snapshot)
return self.table
@@ -401,15 +401,18 @@ class FileStoreWrite:
else predicate_builder.equal(key, value))
partition_filter = predicate_builder.and_predicates(sub_predicates)
- scan = read_builder.with_filter(partition_filter).new_scan()
- splits = scan.plan_all_files().splits()
+ scanner =
read_builder.with_filter(partition_filter).new_scan().file_scanner
+ # Sequence restoration reads file metadata, not query results. As in
+ # Java's writer scan, include L0 and do not require SELECT or apply row
+ # filters/masks to internal state. Public plan_all_files stays guarded.
+ scanner.skip_level0 = False
max_seq_numbers = {}
- for split in splits:
- current_seq_num = max([file.max_sequence_number for file in
split.files])
- existing_max = max_seq_numbers.get(split.bucket, -1)
+ for entry in scanner.plan_files():
+ current_seq_num = entry.file.max_sequence_number
+ existing_max = max_seq_numbers.get(entry.bucket, -1)
if current_seq_num > existing_max:
- max_seq_numbers[split.bucket] = current_seq_num
+ max_seq_numbers[entry.bucket] = current_seq_num
return max_seq_numbers
diff --git a/paimon-python/pypaimon/write/native_commit.py
b/paimon-python/pypaimon/write/native_commit.py
index 56d6d3db64..d3cb5235dd 100644
--- a/paimon-python/pypaimon/write/native_commit.py
+++ b/paimon-python/pypaimon/write/native_commit.py
@@ -144,8 +144,7 @@ def create_native_write_table(table):
environment = table.catalog_environment
if (type(table) is not FileStoreTable
- or type(environment) is not CatalogEnvironment
- or table.options.query_auth_enabled):
+ or type(environment) is not CatalogEnvironment):
return None
from pypaimon_rust.datafusion import Table as NativeTable
diff --git a/paimon-python/pypaimon/write/native_update.py
b/paimon-python/pypaimon/write/native_update.py
index da7fb72c90..610eaedd54 100644
--- a/paimon-python/pypaimon/write/native_update.py
+++ b/paimon-python/pypaimon/write/native_update.py
@@ -45,6 +45,9 @@ def _native_update_table(table):
"""Resolve common native update prerequisites before preparing changes."""
if (type(table) is not FileStoreTable
or not table.options.native_write_enabled()
+ # These operations read target rows before staging changes. Pure
+ # write/commit support must not select an unsupported guarded read.
+ or table.options.query_auth_enabled
or not _native_update_formats_supported(table)
or table.options.changelog_producer() != ChangelogProducer.NONE
or any(table.options.options.contains_key(key) for key in
SCAN_KEYS)
diff --git a/paimon-python/pypaimon/write/native_write.py
b/paimon-python/pypaimon/write/native_write.py
index e26a4d87c3..d7aef5ebfd 100644
--- a/paimon-python/pypaimon/write/native_write.py
+++ b/paimon-python/pypaimon/write/native_write.py
@@ -150,11 +150,15 @@ class NativeTableWrite:
return self._python_writer.with_write_type(write_cols)
if self._written:
raise RuntimeError('with_write_type must be called before writing;
after native data it cannot change')
+ names = list(write_cols)
if self.table.is_primary_key_table:
+ if len(names) == len(self._table_schema.names) and set(names) ==
set(self._table_schema.names):
+ # Python treats a complete PK write type as the unchanged table
+ # schema. It does not configure a PK column-subset writer.
+ return self
# Core's write type is an append-table operation. Keep the existing
# Python writer contract for PK setters before any data is staged.
- return self._switch_to_python().with_write_type(write_cols)
- names = list(write_cols)
+ return self._switch_to_python().with_write_type(names)
self._native_writer.with_write_type(names)
self._schema = pa.schema([self._table_schema.field(name) for name in
names])
self._write_cols = names
@@ -241,6 +245,26 @@ class NativeTableWrite:
return self._python_writer.write_pandas(dataframe)
self.write_arrow_batch(pa.RecordBatch.from_pandas(dataframe,
schema=self._schema))
+ def write_ray(self, dataset, overwrite=False, concurrency=None,
ray_remote_args=None,
+ hash_fixed_precluster='auto', static_partition=None):
+ if self._python_writer is not None:
+ return self._python_writer.write_ray(
+ dataset, overwrite, concurrency, ray_remote_args,
hash_fixed_precluster, static_partition)
+ from pypaimon.write.table_write import _write_ray
+ _write_ray(self, dataset, overwrite, concurrency, ray_remote_args,
+ hash_fixed_precluster, static_partition)
+
+ def _distributed_write_options(self, *, overwrite=False):
+ return {}
+
+ def roll_before_group_if_needed(self, row_count):
+ if self._python_writer is not None:
+ return self._python_writer.roll_before_group_if_needed(row_count)
+ if isinstance(row_count, bool) or not isinstance(row_count, int) or
row_count <= 0:
+ raise ValueError('Group row count must be a positive integer.')
+ # Python's group boundary only affects packed video writers. Native
+ # selection excludes video, so ordinary Parquet/BLOB writes do not
roll.
+
def write_row(self, row):
if self._python_writer is not None:
return self._python_writer.write_row(row)
@@ -321,3 +345,10 @@ class
NativePostponeFixedBucketTableWrite(NativeTableWrite):
from pypaimon.write.postpone_batch_table_write import
PostponeFixedBucketBatchTableWrite
return
self._configure_python_writer(PostponeFixedBucketBatchTableWrite(
self.table, self.commit_user, self.static_partition,
self._bucket_plan))
+
+ def _distributed_write_options(self, *, overwrite=False):
+ from pypaimon.write.postpone_bucket import PostponeBucketPlanner
+ return {'postpone_bucket_planner': PostponeBucketPlanner(
+ self.table, known_num_buckets=(
+ self._bucket_plan.as_dict() if self._bucket_plan is not None
+ else {} if overwrite else None))}
diff --git a/paimon-python/pypaimon/write/postpone_batch_table_write.py
b/paimon-python/pypaimon/write/postpone_batch_table_write.py
index e1171f76ea..4486d697d3 100644
--- a/paimon-python/pypaimon/write/postpone_batch_table_write.py
+++ b/paimon-python/pypaimon/write/postpone_batch_table_write.py
@@ -204,7 +204,9 @@ class PostponeFixedBucketBatchTableWrite(BatchTableWrite):
self._flush_pending_inputs()
return self._prepare_commit(BATCH_COMMIT_IDENTIFIER)
- def _distributed_write_options(self) -> Dict[str, Any]:
+ def _distributed_write_options(self, *, overwrite=False) -> Dict[str, Any]:
+ if overwrite and not self._plan_provided:
+ return {"postpone_bucket_planner":
PostponeBucketPlanner(self.table, known_num_buckets={})}
return {"postpone_bucket_planner": self._planner}
def close(self):
diff --git a/paimon-python/pypaimon/write/table_write.py
b/paimon-python/pypaimon/write/table_write.py
index c78ea60801..63b175e168 100644
--- a/paimon-python/pypaimon/write/table_write.py
+++ b/paimon-python/pypaimon/write/table_write.py
@@ -36,6 +36,19 @@ if TYPE_CHECKING:
from ray.data import Dataset
+def _write_ray(writer, dataset, overwrite=False, concurrency=None,
ray_remote_args=None,
+ hash_fixed_precluster='auto', static_partition=None):
+ """Share Ray coordination without constructing another local data
writer."""
+ from pypaimon.write.ray_datasink import write_paimon_dataset
+
+ overwrite_partition = writer.static_partition if static_partition is None
else static_partition
+ return write_paimon_dataset(
+ dataset, writer.table, overwrite=overwrite,
static_partition=overwrite_partition,
+ concurrency=concurrency, ray_remote_args=ray_remote_args,
+ hash_fixed_precluster=hash_fixed_precluster,
+ **writer._distributed_write_options(overwrite=overwrite or
overwrite_partition is not None))
+
+
class TableWrite:
def __init__(self, table, commit_user, static_partition: Optional[dict] =
None,
*, restore_snapshot_id=None):
@@ -242,24 +255,10 @@ class TableWrite:
the Ray write runs in overwrite mode for this partition and
overrides any builder-level partition spec.
"""
- from pypaimon.write.ray_datasink import write_paimon_dataset
-
- overwrite_partition = self.static_partition
- if static_partition is not None:
- overwrite_partition = static_partition
-
- write_paimon_dataset(
- dataset,
- self.table,
- overwrite=overwrite,
- static_partition=overwrite_partition,
- concurrency=concurrency,
- ray_remote_args=ray_remote_args,
- hash_fixed_precluster=hash_fixed_precluster,
- **self._distributed_write_options(),
- )
+ _write_ray(self, dataset, overwrite, concurrency, ray_remote_args,
+ hash_fixed_precluster, static_partition)
- def _distributed_write_options(self) -> Dict[str, Any]:
+ def _distributed_write_options(self, *, overwrite=False) -> Dict[str, Any]:
"""Return options forwarded by ``write_ray`` to the Ray writer."""
return {}
