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 3b9a9e55a4 [python] Run Paimon table-source MERGE in native core
(#10419)
3b9a9e55a4 is described below
commit 3b9a9e55a4a79bdb2f2f94f1130f1f4cc2839fed
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Oct 8 13:22:41 2026 +0800
[python] Run Paimon table-source MERGE in native core (#10419)
---
paimon-python/README.md | 19 +-
paimon-python/pypaimon/read/native_plan.py | 43 +--
.../pypaimon/read/scanner/split_generator.py | 22 +-
paimon-python/pypaimon/read/table_read.py | 14 +-
.../pypaimon/table/data_evolution_merge_into.py | 11 +-
.../pypaimon/table/system/file_key_ranges_table.py | 7 +-
.../tests/data_evolution_group_stats_test.py | 6 +
.../tests/data_evolution_split_generator_test.py | 6 +
.../pypaimon/tests/interval_partition_test.py | 1 +
.../pypaimon/tests/mosaic_writer_options_test.py | 15 +-
paimon-python/pypaimon/tests/native_commit_test.py | 17 +-
.../pypaimon/tests/native_merge_into_test.py | 302 +++++++++++++++++++++
.../pypaimon/tests/native_plan_integration_test.py | 2 +-
paimon-python/pypaimon/tests/native_plan_test.py | 82 ------
.../tests/native_precomputed_bucket_test.py | 11 +-
.../pypaimon/tests/native_row_sidecar_test.py | 61 ++++-
paimon-python/pypaimon/tests/native_update_test.py | 4 +-
.../pypaimon/tests/write/commit_callback_test.py | 6 +-
paimon-python/pypaimon/write/file_store_commit.py | 18 +-
paimon-python/pypaimon/write/native_commit.py | 7 -
paimon-python/pypaimon/write/native_merge_into.py | 37 ++-
paimon-python/pypaimon/write/native_update.py | 30 --
paimon-python/pypaimon/write/writer/data_writer.py | 8 +-
23 files changed, 491 insertions(+), 238 deletions(-)
diff --git a/paimon-python/README.md b/paimon-python/README.md
index a6cbfcbd75..dc2459e3c0 100644
--- a/paimon-python/README.md
+++ b/paimon-python/README.md
@@ -170,7 +170,8 @@ propagate without retrying that batch through Python.
Batch and stream `merge_into` also use Rust core for eligible data-evolution
Parquet tables when `write.native.enabled=true`. Existing `WhenMatched` and
-`WhenNotMatched` clauses accept Arrow/pandas input or self-merge on `_ROW_ID`.
+`WhenNotMatched` clauses accept Arrow/pandas input, Paimon table sources, or
+self-merge on `_ROW_ID`.
Core pins the target snapshot, matches keys, selects the first satisfied clause
and prepares updates, deletion vectors and inserts. SQL conditions, including
subqueries, execute in the Rust DataFusion adapter; Python transports the
@@ -180,9 +181,19 @@ DataFusion and `pypaimon-rust`. Shared clause validation
checks that dependency
before native execution. NULL keys do not match.
Multiple source rows matching a target are rejected before action conditions,
except for a sole unconditional DELETE, following Paimon Spark MERGE.
-Non-self table sources and packed-video inserts use the Python path selected
-before native execution starts. Native failures propagate without Python retry
-or deleting files from earlier prepared actions.
+For table sources, core independently pins the selected source branch/snapshot
+and reads only keys and referenced columns. Append and primary-key sources do
+not require data evolution. Source scan options select the same point in time
+on Python and native paths. Incremental source scans, non-REST Blob views
+requiring catalog resolution, and packed-video inserts use the Python path
+selected before native execution starts. Native failures propagate without
+Python retry or deleting files from earlier prepared actions.
+
+Python and native writers use Java's escaped partition directories for data,
+changelog, Blob and row sidecar files, including external locations. Both
+planners, file-range metadata, commit callbacks and explicit aborts use the
same
+path rules. Row sidecar reads resolve that path even when the primary file is
+unavailable.
MAP columns configured with `fields.<name>.map.storage-layout=shared-shredding`
also use native Parquet writes and data-evolution updates, including predicate
diff --git a/paimon-python/pypaimon/read/native_plan.py
b/paimon-python/pypaimon/read/native_plan.py
index fbad1d8c37..60ece775c9 100644
--- a/paimon-python/pypaimon/read/native_plan.py
+++ b/paimon-python/pypaimon/read/native_plan.py
@@ -34,7 +34,6 @@ from pypaimon.common.options.config import CatalogOptions,
OssOptions
from pypaimon.common.options.options_utils import OptionsUtils
from pypaimon.common.predicate import Predicate
from pypaimon.read.plan import Plan
-from pypaimon.read.split import Split
from pypaimon.read.split_serializer import (
deserialize_split_v1, serialize_split_v1)
from pypaimon.schema.data_types import DataField, RowType
@@ -77,6 +76,16 @@ def native_reader_available() -> bool:
and native_method_available('TableRead', 'read'))
+def _native_blob_view_supported(table, read_names) -> bool:
+ """Resolve Blob views natively only with the REST catalog environment."""
+ if (not (table.options.blob_view_fields() & set(read_names))
+ or not table.options.blob_view_resolve_enabled()):
+ return True
+ loader = getattr(getattr(table, 'catalog_environment', None),
'catalog_loader', None)
+ # Python also leaves view structs unresolved without a loader.
+ return loader is None or _catalog_metastore(loader) == 'rest'
+
+
def native_split_bridge_available() -> bool:
"""Whether Rust accepts Java-compatible split bytes from Python plans."""
return native_reader_available() and native_method_available(
@@ -241,37 +250,6 @@ def _predicate_to_native(predicate: Predicate) -> dict:
}
-def _restore_python_partition_paths(table, splits: List[Split]) -> None:
- """Restore legacy PyPaimon paths with one listing per bucket."""
- if not table.partition_keys:
- return
- path_factory = table.path_factory()
- bucket_files = {}
- for split in splits:
- bucket_path = path_factory.bucket_path(
- tuple(split.partition.values), split.bucket)
- candidates = []
- for data_file in split.files:
- python_path = "%s/%s" % (
- bucket_path.rstrip('/'), data_file.file_name)
- if (not data_file.external_path
- and python_path != data_file.file_path):
- candidates.append((data_file, python_path))
- if not candidates:
- continue
- if bucket_path not in bucket_files:
- bucket_files[bucket_path] = {
- status.base_name
- for status in table.file_io.list_status(bucket_path)
- }
- for data_file, python_path in candidates:
- if data_file.file_name in bucket_files[bucket_path]:
- data_file.file_path = python_path
- # The retained Rust split still points at its canonical path.
- # Invalidate it so native reading cannot bypass this repair.
- split._native_split = None
-
-
def _resolved_rest_table_response(table):
"""Reuse REST metadata only when the standard loader can be reproduced."""
from pypaimon.catalog.catalog_environment import CatalogEnvironment
@@ -488,7 +466,6 @@ def native_plan(
# time from its current fields.
for split, rust_split in zip(splits, rust_splits):
split._native_split = rust_split
- _restore_python_partition_paths(table, splits)
snapshot_id = getattr(rust_plan, 'snapshot_id', None)
if callable(snapshot_id):
snapshot_id = snapshot_id()
diff --git a/paimon-python/pypaimon/read/scanner/split_generator.py
b/paimon-python/pypaimon/read/scanner/split_generator.py
index adbccdf123..2aa8f2ffb9 100644
--- a/paimon-python/pypaimon/read/scanner/split_generator.py
+++ b/paimon-python/pypaimon/read/scanner/split_generator.py
@@ -130,23 +130,15 @@ class AbstractSplitGenerator(ABC):
return splits
def _set_data_file_paths(self, files, partition: GenericRow, bucket: int):
- """Resolve Java/Rust escaped paths and legacy Python partition
paths."""
+ """Resolve Java's partition paths without probing the primary file.
+
+ A row-sidecar read may not open the primary file at all. Its aligned
+ path must still be correct when that primary file is unavailable.
+ """
values = tuple(partition.values)
- escaped_partition = False
- if values:
- path_factory = self.table.path_factory()
- escaped_partition = (
- path_factory.bucket_path(values, bucket,
canonical_partition=True)
- != path_factory.bucket_path(values, bucket))
for data_file in files:
- data_file.set_file_path(
- self.table.table_path, partition, bucket,
self.default_part_value,
- self.table.options.data_file_path_directory())
- if escaped_partition and not data_file.external_path:
- canonical_path = canonical_data_file_path(
- self.table, values, bucket, data_file.file_name)
- if self.table.file_io.exists(canonical_path):
- data_file.file_path = canonical_path
+ data_file.file_path = data_file.physical_path() if
data_file.external_path else canonical_data_file_path(
+ self.table, values, bucket, data_file.file_name)
def _get_deletion_files_for_split(
self,
diff --git a/paimon-python/pypaimon/read/table_read.py
b/paimon-python/pypaimon/read/table_read.py
index ee8f8e9f8d..e0e683d973 100644
--- a/paimon-python/pypaimon/read/table_read.py
+++ b/paimon-python/pypaimon/read/table_read.py
@@ -976,18 +976,8 @@ class TableRead:
or self._deferred_blob_limit_may_prune(splits))
def _native_blob_view_supported(self) -> bool:
- """Use native view resolution only with the REST catalog
environment."""
- read_names = {field.name for field in self._scan_read_type}
- view_fields = self.table.options.blob_view_fields() & read_names
- if not view_fields or not
self.table.options.blob_view_resolve_enabled():
- return True
- loader = getattr(getattr(self.table, 'catalog_environment', None),
- 'catalog_loader', None)
- if loader is None:
- # Python also leaves view structs unresolved without a loader.
- return True
- from pypaimon.read.native_plan import _catalog_metastore
- return _catalog_metastore(loader) == 'rest'
+ from pypaimon.read.native_plan import _native_blob_view_supported
+ return _native_blob_view_supported(self.table, (field.name for field
in self._scan_read_type))
def _native_resolved_blob_fields(self) -> set:
"""BLOB fields Rust resolves before emitting a batch."""
diff --git a/paimon-python/pypaimon/table/data_evolution_merge_into.py
b/paimon-python/pypaimon/table/data_evolution_merge_into.py
index d5baff21bc..cdf4feeb01 100644
--- a/paimon-python/pypaimon/table/data_evolution_merge_into.py
+++ b/paimon-python/pypaimon/table/data_evolution_merge_into.py
@@ -136,6 +136,8 @@ def _prepare(
when_matched,
when_not_matched,
on,
+ *,
+ source_schema=None,
):
if not when_matched and not when_not_matched:
raise ValueError(
@@ -249,7 +251,8 @@ def _prepare(
source_table = None
source_col_names = set(full_target_field_names) | set(source_on_cols)
else:
- source_table = _normalize_source(source)
+ # Native table input needs schema validation without Python I/O.
+ source_table = source_schema.empty_table() if source_schema is not
None else _normalize_source(source)
_validate_source_on_cols(source_table, source_on_cols)
source_col_names = set(source_table.schema.names)
@@ -751,9 +754,9 @@ def _normalize_source(source: Any) -> pa.Table:
if isinstance(source, pa.Table):
return source
if _is_table_like(source):
- snapshot = source.snapshot_manager().get_latest_snapshot()
- snapshot_id = snapshot.id if snapshot is not None else None
- return _read_table(source, snapshot_id=snapshot_id)
+ # Let the read resolve its branch and point-in-time selectors once.
+ # Pinning the latest snapshot here would override the source's options.
+ return _read_table(source)
try:
import pandas as pd
except ImportError:
diff --git a/paimon-python/pypaimon/table/system/file_key_ranges_table.py
b/paimon-python/pypaimon/table/system/file_key_ranges_table.py
index 67b60ff5bd..290caae252 100644
--- a/paimon-python/pypaimon/table/system/file_key_ranges_table.py
+++ b/paimon-python/pypaimon/table/system/file_key_ranges_table.py
@@ -30,6 +30,7 @@ from pypaimon.table.system.files_table import (
_stringify_path,
)
from pypaimon.table.system.system_table import SystemTable
+from pypaimon.utils.file_store_path_factory import canonical_data_file_path
TABLE_TYPE = RowType(False, [
@@ -71,7 +72,6 @@ class FileKeyRangesTable(SystemTable):
manifest_files, drop_stats=False)
file_format = self.base_table.options.file_format()
- path_factory = self.base_table.path_factory()
rows = {
"partition": [],
"bucket": [],
@@ -88,14 +88,13 @@ class FileKeyRangesTable(SystemTable):
for entry in entries:
meta = entry.file
- bucket_path = path_factory.bucket_path(
- tuple(entry.partition.values), int(entry.bucket))
rows["partition"].append(_render_partition(entry.partition))
rows["bucket"].append(int(entry.bucket))
rows["file_path"].append(_stringify_path(
meta.external_path
or meta.file_path
- or "%s/%s" % (bucket_path, meta.file_name)))
+ or canonical_data_file_path(
+ self.base_table, tuple(entry.partition.values),
int(entry.bucket), meta.file_name)))
rows["file_format"].append(file_format)
rows["schema_id"].append(int(meta.schema_id))
rows["level"].append(int(meta.level))
diff --git a/paimon-python/pypaimon/tests/data_evolution_group_stats_test.py
b/paimon-python/pypaimon/tests/data_evolution_group_stats_test.py
index 80826a7f0d..a49d540872 100644
--- a/paimon-python/pypaimon/tests/data_evolution_group_stats_test.py
+++ b/paimon-python/pypaimon/tests/data_evolution_group_stats_test.py
@@ -32,6 +32,7 @@ from pypaimon.read.scanner.data_evolution_stats import \
from pypaimon.schema.data_types import AtomicType, DataField
from pypaimon.table.row.generic_row import GenericRow
from pypaimon.table.special_fields import SpecialFields
+from pypaimon.utils.file_store_path_factory import FileStorePathFactory
def _empty_stats():
@@ -399,6 +400,11 @@ class
DataEvolutionGroupStatsPlanningTest(unittest.TestCase):
table_path = '/tmp/table'
options = _Options()
+ def path_factory(self):
+ return FileStorePathFactory(
+ self.table_path, [], '__DEFAULT_PARTITION__', 'parquet',
+ 'data-', 'changelog-', False, False, 'zstd')
+
predicate = PredicateBuilder(fields).equal('id', 5)
group_filter = DataEvolutionGroupStatsFilter(
predicate,
diff --git
a/paimon-python/pypaimon/tests/data_evolution_split_generator_test.py
b/paimon-python/pypaimon/tests/data_evolution_split_generator_test.py
index 4b3c739070..fe9b1edc30 100644
--- a/paimon-python/pypaimon/tests/data_evolution_split_generator_test.py
+++ b/paimon-python/pypaimon/tests/data_evolution_split_generator_test.py
@@ -23,6 +23,7 @@ from pypaimon.manifest.schema.manifest_entry import
ManifestEntry
from pypaimon.manifest.schema.simple_stats import SimpleStats
from pypaimon.read.scanner.data_evolution_split_generator import
DataEvolutionSplitGenerator
from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.utils.file_store_path_factory import FileStorePathFactory
from pypaimon.utils.range import Range
@@ -127,6 +128,11 @@ class SplitOrderTest(unittest.TestCase):
table_path = '/table'
options = None
+ def path_factory(self):
+ return FileStorePathFactory(
+ self.table_path, [], '__DEFAULT_PARTITION__', 'parquet',
+ 'data-', 'changelog-', False, False, 'zstd')
+
_Table.options = _Options()
@staticmethod
diff --git a/paimon-python/pypaimon/tests/interval_partition_test.py
b/paimon-python/pypaimon/tests/interval_partition_test.py
index 822d3e64d1..592ba589b3 100644
--- a/paimon-python/pypaimon/tests/interval_partition_test.py
+++ b/paimon-python/pypaimon/tests/interval_partition_test.py
@@ -63,6 +63,7 @@ def
test_signed_zero_key_ranges_keep_versions_in_one_split(type_name):
path_factory = Mock()
path_factory.bucket_path.return_value = '/tmp/interval-test/bucket-0'
+ path_factory.data_file_path.return_value = '/tmp/interval-test'
table = SimpleNamespace(table_path='/tmp/interval-test',
options=CoreOptions(Options({})),
path_factory=lambda: path_factory)
entries = [ManifestEntry(0, GenericRow([], []), 0, 1, file) for file in
files]
diff --git a/paimon-python/pypaimon/tests/mosaic_writer_options_test.py
b/paimon-python/pypaimon/tests/mosaic_writer_options_test.py
index 0eac7c3849..88b3e5ad2e 100644
--- a/paimon-python/pypaimon/tests/mosaic_writer_options_test.py
+++ b/paimon-python/pypaimon/tests/mosaic_writer_options_test.py
@@ -24,6 +24,7 @@ import pytest
from pypaimon.common.options.core_options import CoreOptions
from pypaimon.filesystem.local_file_io import LocalFileIO
from pypaimon.schema.data_types import AtomicType, DataField
+from pypaimon.utils.file_store_path_factory import FileStorePathFactory
from pypaimon.write.writer.append_only_data_writer import AppendOnlyDataWriter
from pypaimon.write.writer.mosaic_writer_options import
create_mosaic_writer_options
@@ -33,17 +34,6 @@ class _FakeWriterOptions:
self.kwargs = kwargs
-class _PathFactory:
- def __init__(self, bucket_path):
- self._bucket_path = bucket_path
-
- def create_external_path_provider(self, partition, bucket):
- return None
-
- def bucket_path(self, partition, bucket):
- return self._bucket_path
-
-
class _FileIO:
def __init__(self):
self.mosaic_options = None
@@ -121,7 +111,8 @@ def
test_data_writer_passes_mosaic_writer_options(monkeypatch, tmp_path):
trimmed_primary_keys=["id"],
trimmed_primary_keys_fields=[DataField(0, "id", AtomicType("INT"))],
fields=[DataField(0, "id", AtomicType("INT"))],
- path_factory=lambda: _PathFactory(str(tmp_path)),
+ path_factory=lambda: FileStorePathFactory(
+ str(tmp_path), [], '__DEFAULT_PARTITION__', 'mosaic', 'data-',
'changelog-', False, False, 'zstd'),
)
writer = AppendOnlyDataWriter(table, (), 0, 0, options)
diff --git a/paimon-python/pypaimon/tests/native_commit_test.py
b/paimon-python/pypaimon/tests/native_commit_test.py
index 85b72f3ec7..3b62aa0ac5 100644
--- a/paimon-python/pypaimon/tests/native_commit_test.py
+++ b/paimon-python/pypaimon/tests/native_commit_test.py
@@ -332,22 +332,29 @@ def test_native_abort_removes_uncommitted_files(tmp_path,
native_rest_catalog, o
@requires_native
-def test_python_file_in_legacy_partition_uses_python_abort(
- tmp_path, native_rest_catalog):
[email protected]('serialized', [False, True])
+def test_python_written_escaped_partition_supports_native_abort(
+ tmp_path, native_rest_catalog, serialized):
+ from pypaimon.write.commit_message_serializer import
deserialize_commit_message, serialize_commit_message
+
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)
+ assert '/pt=a%2Fb/' in file.file_path
+ if serialized:
+ messages =
[deserialize_commit_message(serialize_commit_message(message,
table.partition_keys_fields),
+ table.partition_keys_fields)
for message in messages]
+ assert 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:
+ side_effect=AssertionError('Python fallback')):
commit.abort(messages)
- python_abort.assert_called_once_with(messages)
assert not table.file_io.exists(file.file_path)
+ assert table.snapshot_manager().get_latest_snapshot() is None
finally:
commit.close()
diff --git a/paimon-python/pypaimon/tests/native_merge_into_test.py
b/paimon-python/pypaimon/tests/native_merge_into_test.py
index d7713a70a5..1272f129ae 100644
--- a/paimon-python/pypaimon/tests/native_merge_into_test.py
+++ b/paimon-python/pypaimon/tests/native_merge_into_test.py
@@ -211,6 +211,7 @@ def
test_native_self_merge_does_not_read_unreferenced_blob(tmp_path, condition):
catalog.create_table('db.t', Schema.from_pyarrow_schema(schema, options={
'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true',
'blob-descriptor-field': 'payload', 'blob-as-descriptor': 'false',
'write.native.enabled': 'true',
+ 'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true',
}), False)
table = catalog.get_table('db.t')
builder = table.new_batch_write_builder()
@@ -228,3 +229,304 @@ def
test_native_self_merge_does_not_read_unreferenced_blob(tmp_path, condition):
builder.new_commit().commit(messages)
reader = table.new_read_builder().with_projection(['id', 'name'])
assert
reader.new_read().to_arrow(reader.new_scan().plan().splits()).to_pylist() ==
[dict(id=1, name='after')]
+
+
[email protected]('stream', [False, True])
+def test_native_merge_reads_paimon_source_in_core(tmp_path, stream):
+ target = _table(tmp_path / 'target', [dict(id=1, value=10, name='old')])
+ source = _table(tmp_path / 'source', [dict(id=1, value=11, name='new'),
dict(id=2, value=22, name='insert')])
+ builder = target.new_stream_write_builder() if stream else
target.new_batch_write_builder()
+ kwargs = dict(commit_identifier=57) if stream else {}
+ with patch('pypaimon.table.data_evolution_merge_into._normalize_source',
+ side_effect=AssertionError('Python source materialization')), \
+ patch('pypaimon.table.data_evolution_merge_into._build_tables',
+ side_effect=AssertionError('Python merge orchestration')):
+ messages = builder.new_update().merge_into(source, on=['id'],
when_matched=[
+ WhenMatched.update({'value': source_col('value')},
condition='s.value > t.value')],
+ when_not_matched=[WhenNotMatched('*')], **kwargs)
+ commit = builder.new_commit()
+ commit.commit(messages, 57) if stream else commit.commit(messages)
+ assert _rows(target) == [dict(id=1, value=11, name='old'), dict(id=2,
value=22, name='insert')]
+ assert _rows(source) == [dict(id=1, value=11, name='new'), dict(id=2,
value=22, name='insert')]
+
+
[email protected]('native', [False, True])
+def test_table_source_respects_point_in_time_read_options(tmp_path, native):
+ target = _table(tmp_path / 'target', [dict(id=1, value=10, name='old')])
+ source = _table(tmp_path / 'source', [dict(id=1, value=11, name='new')])
+ snapshot = source.snapshot_manager().get_latest_snapshot().id
+ builder = source.new_batch_write_builder()
+ writer = builder.new_write()
+ try:
+ writer.write_arrow(pa.Table.from_pylist([dict(id=2, value=22,
name='later')], schema=_SCHEMA))
+ builder.new_commit().commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ target = target.copy({'write.native.enabled': str(native).lower()})
+ historical = source.copy({'scan.snapshot-id': str(snapshot)})
+ builder = target.new_batch_write_builder()
+ messages = builder.new_update().merge_into(historical, on=['id'],
when_matched=[
+ WhenMatched.update({'value': source_col('value')})],
when_not_matched=[WhenNotMatched('*')])
+ builder.new_commit().commit(messages)
+ assert _rows(target) == [dict(id=1, value=11, name='old')]
+ assert _rows(source) == [dict(id=1, value=11, name='new'), dict(id=2,
value=22, name='later')]
+
+
+def test_native_table_source_projects_away_unreadable_blob(tmp_path):
+ from pypaimon.table.row.blob import BlobDescriptor
+
+ target = _table(tmp_path / 'target', [dict(id=1, value=10, name='old')])
+ schema = pa.schema(list(_SCHEMA) + [pa.field('payload',
pa.large_binary())])
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path / 'source')})
+ catalog.create_database('db', True)
+ catalog.create_table('db.source', Schema.from_pyarrow_schema(schema,
options={
+ 'blob-descriptor-field': 'payload', 'blob-as-descriptor': 'false',
'write.native.enabled': 'true',
+ 'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true',
+ }), False)
+ source = catalog.get_table('db.source')
+ builder = source.new_batch_write_builder()
+ writer = builder.new_write()
+ try:
+ values = dict(id=1, value=11, name='new',
+ payload=BlobDescriptor(str(tmp_path / 'unreadable.bin'),
0, 1).serialize())
+ writer.write_arrow(pa.Table.from_pylist([values], schema=schema))
+ builder.new_commit().commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ builder = target.new_batch_write_builder()
+ with patch('pypaimon.table.data_evolution_merge_into._normalize_source',
+ side_effect=AssertionError('Python source materialization')):
+ messages = builder.new_update().merge_into(source, on=['id'],
when_matched=[
+ WhenMatched.update({'value': source_col('value')},
condition='s.value > t.value')])
+ builder.new_commit().commit(messages)
+ assert _rows(target) == [dict(id=1, value=11, name='old')]
+
+
+def test_native_table_source_pk_reads_logical_rows_and_needs_no_de(tmp_path):
+ target = _table(tmp_path / 'target', [dict(id=1, value=10, name='old')])
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path / 'source')})
+ catalog.create_database('db', True)
+ source_schema = Schema.from_pyarrow_schema(
+ _SCHEMA, primary_keys=['id'], options={'bucket': '1',
'write.native.enabled': 'true'})
+ catalog.create_table('db.source', source_schema, False)
+ source = catalog.get_table('db.source')
+ for values in [dict(id=1, value=11, name='first'), dict(id=1, value=12,
name='latest'),
+ dict(id=2, value=22, name='new')]:
+ builder = source.new_batch_write_builder()
+ writer = builder.new_write()
+ try:
+ writer.write_arrow(pa.Table.from_pylist([values], schema=_SCHEMA))
+ builder.new_commit().commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ builder = target.new_batch_write_builder()
+ with patch('pypaimon.table.data_evolution_merge_into._normalize_source',
+ side_effect=AssertionError('Python source materialization')):
+ messages = builder.new_update().merge_into(source, on=['id'],
when_matched=[
+ WhenMatched.update({'value': source_col('value')})],
when_not_matched=[WhenNotMatched('*')])
+ builder.new_commit().commit(messages)
+ assert _rows(target) == [dict(id=1, value=12, name='old'), dict(id=2,
value=22, name='new')]
+
+
+def _write_rows(table, schema, rows):
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ try:
+ writer.write_arrow(pa.Table.from_pylist(rows, schema=schema))
+ builder.new_commit().commit(writer.prepare_commit())
+ finally:
+ writer.close()
+
+
+def
test_native_table_source_preserves_nested_values_and_null_keys_across_files(tmp_path):
+ schema = pa.schema([
+ ('id', pa.int32()), ('items', pa.list_(pa.int32())), ('lookup',
pa.map_(pa.string(), pa.int32())),
+ ])
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('db', True)
+ common = {'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true',
+ 'write.native.enabled': 'true'}
+ for name in ['source', 'target']:
+ catalog.create_table('db.' + name, Schema.from_pyarrow_schema(schema,
options=common), False)
+ source, target = catalog.get_table('db.source'),
catalog.get_table('db.target')
+ _write_rows(target, schema, [dict(id=1, items=[0], lookup=[('old', 0)]),
+ dict(id=None, items=[9], lookup=[])])
+ values = [dict(id=2, items=[], lookup=[('two', None)]),
+ dict(id=1, items=[1, None], lookup=[('one', 1)]),
+ dict(id=None, items=None, lookup=None)]
+ for value in values:
+ _write_rows(source, schema, [value])
+ builder = target.new_batch_write_builder()
+ with patch('pypaimon.table.data_evolution_merge_into._normalize_source',
+ side_effect=AssertionError('Python source materialization')):
+ messages = builder.new_update().merge_into(
+ source, on=['id'], when_matched=[WhenMatched.update('*')],
+ when_not_matched=[WhenNotMatched('*')])
+ builder.new_commit().commit(messages)
+ reader = target.new_read_builder()
+ actual =
reader.new_read().to_arrow(reader.new_scan().plan().splits()).to_pylist()
+ assert sorted([row for row in actual if row['id'] is not None], key=lambda
row: row['id']) == sorted(
+ [row for row in values if row['id'] is not None], key=lambda row:
row['id'])
+ assert sorted([row for row in actual if row['id'] is None], key=lambda
row: row['items'] is None) == [
+ dict(id=None, items=[9], lookup=[]), dict(id=None, items=None,
lookup=None)]
+
+
[email protected]('native', [False, True])
+def test_table_source_resolves_filesystem_blob_views_before_merge(tmp_path,
native):
+ from pypaimon.table.row.blob import BlobViewStruct
+
+ schema = pa.schema([('id', pa.int32()), ('payload', pa.large_binary())])
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('db', True)
+ common = {'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true'}
+ for name, extra in [('upstream', {}), ('source', {'blob-view-field':
'payload'}),
+ ('target', {'write.native.enabled':
str(native).lower()})]:
+ catalog.create_table('db.' + name, Schema.from_pyarrow_schema(schema,
options=dict(common, **extra)), False)
+ upstream, source = catalog.get_table('db.upstream'),
catalog.get_table('db.source')
+ _write_rows(upstream, schema, [dict(id=1, payload=b'actual-payload')])
+ reference = BlobViewStruct('db.upstream',
upstream.field_dict['payload'].id, 0).serialize()
+ _write_rows(source, schema, [dict(id=1, payload=reference)])
+ target = catalog.get_table('db.target')
+ builder = target.new_batch_write_builder()
+ messages = builder.new_update().merge_into(source, on=['id'],
when_not_matched=[WhenNotMatched('*')])
+ builder.new_commit().commit(messages)
+ reader = target.new_read_builder()
+ assert
reader.new_read().to_arrow(reader.new_scan().plan().splits()).to_pylist() == [
+ dict(id=1, payload=b'actual-payload')]
+
+
+def test_native_table_source_rejects_invalid_scan_options(tmp_path):
+ target = _table(tmp_path / 'target', [dict(id=1, value=10, name='old')])
+ source = _table(tmp_path / 'source', [dict(id=1, value=11, name='new')])
+ source = source.copy({'scan.mode': 'default',
'scan.file-creation-time-millis': '1'})
+ before = set(tmp_path.rglob('*.parquet'))
+ with pytest.raises(ValueError, match='not yet supported|conflicts'):
+ target.new_batch_write_builder().new_update().merge_into(
+ source, on=['id'], when_matched=[WhenMatched.update({'value':
source_col('value')})])
+ assert set(tmp_path.rglob('*.parquet')) == before
+ assert _rows(target) == [dict(id=1, value=10, name='old')]
+
+
+def test_native_table_source_keeps_its_selected_branch(tmp_path):
+ from pypaimon.common.identifier import Identifier
+
+ target = _table(tmp_path / 'target', [dict(id=1, value=10, name='old')])
+ source = _table(tmp_path / 'source', [dict(id=1, value=11, name='base')])
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path / 'source')})
+ catalog.create_tag(source.identifier, 'base')
+ catalog.create_branch(source.identifier, 'dev', tag_name='base')
+ branch = catalog.get_table(Identifier('db', 't', branch='dev'))
+ _write_rows(branch, _SCHEMA, [dict(id=2, value=22, name='branch')])
+ _write_rows(source, _SCHEMA, [dict(id=3, value=33, name='main')])
+ builder = target.new_batch_write_builder()
+ with patch('pypaimon.table.data_evolution_merge_into._normalize_source',
+ side_effect=AssertionError('Python source materialization')):
+ messages = builder.new_update().merge_into(
+ branch, on=['id'], when_matched=[WhenMatched.update({'value':
source_col('value')})],
+ when_not_matched=[WhenNotMatched('*')])
+ builder.new_commit().commit(messages)
+ assert _rows(target) == [dict(id=1, value=11, name='old'), dict(id=2,
value=22, name='branch')]
+ assert _rows(source) == [dict(id=1, value=11, name='base'), dict(id=3,
value=33, name='main')]
+
+
[email protected]_write
[email protected]('stream', [False, True])
[email protected]('source_mode', ['append', 'pk', 'evolution'])
[email protected]('location', ['local', 'file-uri', 'directory',
'external'])
+def
test_python_written_partition_source_uses_java_paths_for_native_merge(tmp_path,
stream, source_mode, location):
+ from pathlib import Path
+ from pypaimon.write.native_write import NativeTableWrite
+
+ target = _table(tmp_path / 'target', [dict(id=1, value=10, name='old')])
+ warehouse = tmp_path / 'source'
+ catalog = CatalogFactory.create({'warehouse': warehouse.as_uri() if
location == 'file-uri' else str(warehouse)})
+ catalog.create_database('db', True)
+ schema = pa.schema(list(_SCHEMA) + [pa.field('pa#rt', pa.string())])
+ options = {'write.native.enabled': 'false'}
+ if source_mode == 'pk':
+ options.update({'bucket': '1', 'changelog-producer': 'input'})
+ if source_mode == 'evolution':
+ schema = pa.schema(list(schema) + [pa.field('payload',
pa.large_binary())])
+ options.update({'row-tracking.enabled': 'true',
'data-evolution.enabled': 'true',
+ 'data-evolution.row-sidecar.enabled': 'true'})
+ if location == 'directory':
+ options['data-file.path-directory'] = 'data/nested'
+ if location == 'external':
+ options.update({'data-file.external-paths': (tmp_path /
'external').as_uri(),
+ 'data-file.external-paths.strategy': 'round-robin'})
+ catalog.create_table('db.source', Schema.from_pyarrow_schema(
+ schema, primary_keys=['id', 'pa#rt'] if source_mode == 'pk' else [],
+ partition_keys=['pa#rt'], options=options), False)
+ source = catalog.get_table('db.source')
+ builder = source.new_batch_write_builder()
+ writer = builder.new_write()
+ assert not isinstance(writer, NativeTableWrite)
+ values = [dict(id=1, value=11, name='new', **{'pa#rt': 'a/b'}),
+ dict(id=2, value=22, name='insert', **{'pa#rt': 'a%b'})]
+ if source_mode == 'evolution':
+ for value in values:
+ value['payload'] = b'blob-payload'
+ try:
+ writer.write_arrow(pa.Table.from_pylist(values, schema=schema))
+ messages = writer.prepare_commit()
+ for message in messages:
+ component = 'pa%23rt=' + ('a%2Fb' if message.partition == ('a/b',)
else 'a%25b')
+ for file in message.new_files + message.changelog_files:
+ assert component in file.file_path.split('/'), file.file_path
+ assert Path(file.file_path).is_file(), file.file_path
+ for extra_file in file.extra_files:
+ assert Path(file.file_path).with_name(extra_file).is_file()
+ builder.new_commit().commit(messages)
+ finally:
+ writer.close()
+ # Java's path wins even if an incorrectly laid-out file has the same name.
+ for message in messages:
+ wrong_bucket =
source.path_factory().bucket_path(tuple(message.partition), message.bucket)
+ for file in message.new_files:
+ if not file.external_path:
+ with source.file_io.new_output_stream(wrong_bucket + '/' +
file.file_name) as output:
+ output.write(b'wrong partition directory')
+ assert _rows(source) == values
+ ranges = catalog.get_table('db.source$file_key_ranges').new_read_builder()
+ reported =
ranges.new_read().to_arrow(ranges.new_scan().plan().splits())['file_path'].to_pylist()
+ assert reported and all(source.file_io.exists(path) for path in reported)
+ assert all('pa%23rt=' in path for path in reported)
+ builder = target.new_stream_write_builder() if stream else
target.new_batch_write_builder()
+ kwargs = dict(commit_identifier=57) if stream else {}
+ with patch('pypaimon.table.data_evolution_merge_into._normalize_source',
+ side_effect=AssertionError('Python source materialization')), \
+ patch('pypaimon.table.data_evolution_merge_into._build_tables',
+ side_effect=AssertionError('Python merge orchestration')):
+ messages = builder.new_update().merge_into(
+ source, on=['id'], when_matched=[WhenMatched.update({'value':
source_col('value')})],
+ when_not_matched=[WhenNotMatched('*')], **kwargs)
+ commit = builder.new_commit()
+ commit.commit(messages, 57) if stream else commit.commit(messages)
+ assert _rows(target) == [dict(id=1, value=11, name='old'), dict(id=2,
value=22, name='insert')]
+
+
[email protected]('native', [False, True])
[email protected]('stream', [False, True])
[email protected]('partition_type', [pa.float32(), pa.float64()])
+def test_float_partition_source_selects_python_before_native_merge(tmp_path,
native, stream, partition_type):
+ target = _table(tmp_path / 'target', [dict(id=1, value=10,
name='old')]).copy({
+ 'write.native.enabled': str(native).lower()})
+ schema = pa.schema(list(_SCHEMA) + [pa.field('part', partition_type)])
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path / 'source')})
+ catalog.create_database('db', True)
+ catalog.create_table('db.source', Schema.from_pyarrow_schema(schema,
partition_keys=['part'], options={
+ 'write.native.enabled': 'false', 'scan.native-plan.enabled': 'false',
'read.native.enabled': 'false',
+ }), False)
+ source = catalog.get_table('db.source')
+ _write_rows(source, schema, [dict(id=1, value=11, name='new', part=1.5),
+ dict(id=2, value=22, name='insert',
part=2.25)])
+ builder = target.new_stream_write_builder() if stream else
target.new_batch_write_builder()
+ kwargs = dict(commit_identifier=57) if stream else {}
+ with
patch('pypaimon.write.native_merge_into.NativeTableMergeInto.prepare_commit',
+ side_effect=AssertionError('Unsupported source entered native
MERGE')):
+ messages = builder.new_update().merge_into(
+ source, on=['id'], when_matched=[WhenMatched.update({'value':
source_col('value')})],
+ when_not_matched=[WhenNotMatched('*')], **kwargs)
+ commit = builder.new_commit()
+ commit.commit(messages, 57) if stream else commit.commit(messages)
+ assert _rows(target) == [dict(id=1, value=11, name='old'), dict(id=2,
value=22, name='insert')]
diff --git a/paimon-python/pypaimon/tests/native_plan_integration_test.py
b/paimon-python/pypaimon/tests/native_plan_integration_test.py
index 3f6cf52fe1..7f0647b492 100644
--- a/paimon-python/pypaimon/tests/native_plan_integration_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_integration_test.py
@@ -1501,7 +1501,7 @@ class NativePlanIntegrationTest(unittest.TestCase):
@pytest.mark.python_write
def test_partitioned_table_matches_normal_plan(self):
- # Native decoding restores PyPaimon's legacy unescaped partition path.
+ # Python writes and both planners use Java's escaped partition path.
schema = pa.schema([('k', pa.int64()), ('p', pa.string())])
self.cat.create_table('default.pt_t', Schema.from_pyarrow_schema(
schema, partition_keys=['p']), False)
diff --git a/paimon-python/pypaimon/tests/native_plan_test.py
b/paimon-python/pypaimon/tests/native_plan_test.py
index 3621bfdbae..63075bb1b7 100644
--- a/paimon-python/pypaimon/tests/native_plan_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_test.py
@@ -38,7 +38,6 @@ from pypaimon.read.native_plan import (
_native_read_builder,
_predicate_to_native,
_resolved_schema_json,
- _restore_python_partition_paths,
native_family_search_modes_available,
native_plan,
native_version_at_least,
@@ -812,87 +811,6 @@ class NativePlanTest(unittest.TestCase):
self.assertEqual(
native_family_search_modes_available(), expected)
- def test_partition_path_prefers_existing_python_legacy_path(self):
- table = Mock(partition_keys=['p'])
- table.path_factory.return_value.bucket_path.return_value = (
- '/warehouse/t/p=a/b/bucket-0')
- table.file_io.list_status.return_value =
[Mock(base_name='data.parquet')]
- data_file = Mock(
- external_path=None,
- file_name='data.parquet',
- file_path='/warehouse/t/p=a%2Fb/bucket-0/data.parquet',
- )
- split = Mock(
- partition=Mock(values=['a/b']), bucket=0, files=[data_file])
- split._native_split = object()
-
- _restore_python_partition_paths(table, [split])
-
- self.assertEqual(
- data_file.file_path,
- '/warehouse/t/p=a/b/bucket-0/data.parquet',
- )
- self.assertIsNone(split._native_split)
-
- def test_partition_path_keeps_existing_rust_path(self):
- table = Mock(partition_keys=['p'])
- table.path_factory.return_value.bucket_path.return_value = (
- '/warehouse/t/p=a/b/bucket-0')
- table.file_io.list_status.return_value = []
- rust_path = '/warehouse/t/p=a%2Fb/bucket-0/data.parquet'
- data_file = Mock(
- external_path=None,
- file_name='data.parquet',
- file_path=rust_path,
- )
- split = Mock(
- partition=Mock(values=['a/b']), bucket=0, files=[data_file])
-
- _restore_python_partition_paths(table, [split])
-
- self.assertEqual(data_file.file_path, rust_path)
-
- def test_partition_path_lists_each_bucket_once(self):
- table = Mock(partition_keys=['p'])
- table.path_factory.return_value.bucket_path.return_value = (
- '/warehouse/t/p=a/b/bucket-0')
- table.file_io.list_status.return_value = [
- Mock(base_name='a.parquet'), Mock(base_name='b.parquet')]
- splits = [
- Mock(partition=Mock(values=['a/b']), bucket=0, files=[Mock(
- external_path=None,
- file_name=name,
- file_path='/warehouse/t/p=a%%2Fb/bucket-0/%s' % name,
- )])
- for name in ('a.parquet', 'b.parquet')
- ]
-
- _restore_python_partition_paths(table, splits)
-
- table.file_io.list_status.assert_called_once_with(
- '/warehouse/t/p=a/b/bucket-0')
- self.assertEqual(
- [split.files[0].file_path for split in splits],
- [
- '/warehouse/t/p=a/b/bucket-0/a.parquet',
- '/warehouse/t/p=a/b/bucket-0/b.parquet',
- ],
- )
-
- def test_partition_path_listing_failure_is_not_hidden(self):
- table = Mock(partition_keys=['p'])
- table.path_factory.return_value.bucket_path.return_value = (
- '/warehouse/t/p=a/b/bucket-0')
- table.file_io.list_status.side_effect = PermissionError('denied')
- split = Mock(partition=Mock(values=['a/b']), bucket=0, files=[Mock(
- external_path=None,
- file_name='data.parquet',
- file_path='/warehouse/t/p=a%2Fb/bucket-0/data.parquet',
- )])
-
- with self.assertRaises(PermissionError):
- _restore_python_partition_paths(table, [split])
-
def test_rest_catalog_retains_response_for_native_reads(self):
from pypaimon.api.api_response import GetTableResponse
from pypaimon.catalog.rest.rest_catalog import RESTCatalog
diff --git a/paimon-python/pypaimon/tests/native_precomputed_bucket_test.py
b/paimon-python/pypaimon/tests/native_precomputed_bucket_test.py
index 8b62c6b221..5d070bccec 100644
--- a/paimon-python/pypaimon/tests/native_precomputed_bucket_test.py
+++ b/paimon-python/pypaimon/tests/native_precomputed_bucket_test.py
@@ -201,8 +201,9 @@ def
test_pinned_dynamic_configuration_survives_early_python_selection(tmp_path,
second.write_arrow(_batch([2], ['later']), bucket=0)
later.new_commit().commit(second.prepare_commit())
second.close()
- # This existing advanced API still selects Python before any write.
- writer.with_blob_consumer(lambda field, blob: blob)
+ # BlobConsumer is native now; explicitly select the Python branch whose
+ # pinned snapshot transport this test exercises before any write.
+ writer._switch_to_python()
assert writer._python_writer.row_key_extractor.base_snapshot_id ==
pinned_id
writer.write_arrow(_batch([1], ['restored']), bucket=0)
messages = writer.prepare_commit()
@@ -362,7 +363,7 @@ def
test_empty_restore_ignores_existing_sequences_and_hashes(tmp_path, python_fa
writer = builder.new_write()
try:
if python_fallback:
- writer.with_blob_consumer(lambda field, blob: blob)
+ writer._switch_to_python()
writer.write_arrow(_batch([3], ['empty-base']), bucket=0)
messages = writer.prepare_commit()
assert messages[0].new_files[0].min_sequence_number == 0
@@ -386,7 +387,7 @@ def
test_empty_restore_preserves_python_write_authorization(tmp_path, overwrite)
builder.overwrite()
writer = builder.new_write()
# Exercise the ordinary writer selected before any Native data is staged.
- writer.with_blob_consumer(lambda field, blob: blob)
+ writer._switch_to_python()
auth = TableQueryAuthResult(None, {'value': '{"name":"NULL"}'})
try:
with patch.object(type(table.catalog_environment), 'table_query_auth',
@@ -405,7 +406,7 @@ def
test_python_pk_file_sequence_range_matches_deduplicated_rows(tmp_path):
builder = table.new_batch_write_builder().with_restore_snapshot(0)
writer = builder.new_write()
try:
- writer.with_blob_consumer(lambda field, blob: blob)
+ writer._switch_to_python()
writer.write_arrow(_batch([1, 1], ['old', 'new']), bucket=0)
messages = writer.prepare_commit()
file = messages[0].new_files[0]
diff --git a/paimon-python/pypaimon/tests/native_row_sidecar_test.py
b/paimon-python/pypaimon/tests/native_row_sidecar_test.py
index 510bd6bd05..28dc2f19ec 100644
--- a/paimon-python/pypaimon/tests/native_row_sidecar_test.py
+++ b/paimon-python/pypaimon/tests/native_row_sidecar_test.py
@@ -62,7 +62,7 @@ def
test_row_sidecar_options_keep_batch_and_stream_writes_native(tmp_path, strea
writer.close()
-def _table(tmp_path, schema=None, options=None):
+def _table(tmp_path, schema=None, options=None, partition_keys=None):
catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
catalog.create_database('default', True)
if schema is None:
@@ -72,7 +72,8 @@ def _table(tmp_path, schema=None, options=None):
'data-evolution.enabled': 'true', 'deletion-vectors.enabled':
'true',
'data-evolution.row-sidecar.enabled': 'true'}
settings.update(options or {})
- catalog.create_table('default.t', Schema.from_pyarrow_schema(schema,
options=settings), False)
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ schema, partition_keys=partition_keys or [], options=settings), False)
return catalog.get_table('default.t')
@@ -132,6 +133,62 @@ def
test_row_sidecars_interoperate_between_python_and_rust(tmp_path, native_writ
'id': [5], 'value': ['v5'], '_ROW_ID': [25]}
[email protected]_write
[email protected]('native_read', [False, True])
+def
test_escaped_partition_sidecar_reads_do_not_require_the_primary_file(tmp_path,
native_read):
+ schema = pa.schema([('id', pa.int32()), ('value', pa.string()), ('p',
pa.string())])
+ table = _table(tmp_path, schema=schema, partition_keys=['p'], options={
+ 'write.native.enabled': 'false', 'read.native.enabled':
str(native_read).lower(),
+ 'scan.native-plan.enabled': 'false'})
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ try:
+ writer.write_arrow(pa.table({'id': list(range(100)), 'value': ['v%d' %
i for i in range(100)],
+ 'p': ['a/b'] * 100}, schema=schema))
+ messages = writer.prepare_commit()
+ _commit(builder, messages)
+ finally:
+ writer.close()
+ _remove_primaries(messages)
+ assert _read(table, ['id', 'value', '_ROW_ID'], 25,
native_read).to_pydict() == {
+ 'id': [25], 'value': ['v25'], '_ROW_ID': [25]}
+
+
[email protected]_write
[email protected]('file_uri', [False, True])
+def
test_serialized_abort_deletes_escaped_data_and_sidecars_but_preserves_committed_files(tmp_path,
file_uri):
+ from pypaimon.write.commit_message_serializer import
deserialize_commit_message, serialize_commit_message
+
+ schema = pa.schema([('id', pa.int32()), ('value', pa.string()), ('p',
pa.string())])
+ table = _table(tmp_path.as_uri() if file_uri else tmp_path, schema=schema,
partition_keys=['p'], options={
+ 'write.native.enabled': 'false', 'commit.native.enabled': 'false'})
+ paths = []
+ for row_id in [1, 2]:
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ try:
+ writer.write_arrow(pa.table({'id': [row_id], 'value': ['v'], 'p':
['a/b']}, schema=schema))
+ messages = writer.prepare_commit()
+ paths.append([path for message in messages for file in
message.new_files for path in file.collect_files()])
+ if row_id == 1:
+ _commit(builder, messages)
+ finally:
+ writer.close()
+ partition_fields = [table.field_dict[name] for name in
table.partition_keys]
+ restored = [deserialize_commit_message(serialize_commit_message(message,
partition_fields), partition_fields)
+ for message in messages]
+ assert all(file.file_path is None for message in restored for file in
message.new_files)
+ assert all(table.file_io.exists(path) for path in paths[0] + paths[1])
+ commit = builder.new_commit()
+ try:
+ # Only these never-submitted messages are explicitly abandoned.
+ commit.abort(restored)
+ finally:
+ commit.close()
+ assert all(table.file_io.exists(path) for path in paths[0])
+ assert all(not table.file_io.exists(path) for path in paths[1])
+
+
@pytest.mark.parametrize('stream', [False, True])
@pytest.mark.parametrize('incremental', [False, True])
def test_sidecars_keep_updates_native_and_apply_deletion_vectors(tmp_path,
stream, incremental):
diff --git a/paimon-python/pypaimon/tests/native_update_test.py
b/paimon-python/pypaimon/tests/native_update_test.py
index e0d6ceca12..e59e81813a 100644
--- a/paimon-python/pypaimon/tests/native_update_test.py
+++ b/paimon-python/pypaimon/tests/native_update_test.py
@@ -967,7 +967,7 @@ def
test_core_nested_row_update_uses_whole_column_constructor_fallback(tmp_path,
@pytest.mark.parametrize('layout', [
- 'legacy', 'canonical', 'ordinary', 'directory', 'external', 'float',
'double', 'external-binary',
+ 'python-escaped', 'canonical', 'ordinary', 'directory', 'external',
'float', 'double', 'external-binary',
])
@pytest.mark.parametrize('operation', ['row', 'predicate'])
def test_update_dispatch_preserves_existing_partition_paths(tmp_path, layout,
operation):
@@ -1027,5 +1027,5 @@ def
test_update_dispatch_preserves_existing_partition_paths(tmp_path, layout, op
read = table.new_read_builder()
actual =
read.new_read().to_arrow(read.new_scan().plan().splits()).sort_by('id')
assert actual.to_pydict() == {'id': [1, 2], 'part': parts,
'value': [value, value]}
- expected_calls = 2 if layout in ('canonical', 'ordinary', 'external') else 0
+ expected_calls = 2 if layout in ('python-escaped', 'canonical',
'ordinary', 'directory', 'external') else 0
assert len(native_calls) == expected_calls
diff --git a/paimon-python/pypaimon/tests/write/commit_callback_test.py
b/paimon-python/pypaimon/tests/write/commit_callback_test.py
index 55ba65dd4d..033dec989a 100644
--- a/paimon-python/pypaimon/tests/write/commit_callback_test.py
+++ b/paimon-python/pypaimon/tests/write/commit_callback_test.py
@@ -135,6 +135,7 @@ class CommitCallbackTest(unittest.TestCase):
def test_callback_invoked_after_lost_commit_response(self):
table = self._create_table(
'test_callback_response_loss',
+ partition_keys=['dt'],
options={
'row-tracking.enabled': 'true',
'data-evolution.enabled': 'true',
@@ -150,7 +151,7 @@ class CommitCallbackTest(unittest.TestCase):
table_write.write_arrow(pa.Table.from_pydict({
'id': [1, 2],
'name': ['a', 'b'],
- 'dt': ['p1', 'p1'],
+ 'dt': ['a/b', 'a%b'],
}, schema=self.pa_schema))
messages = table_write.prepare_commit()
expected_paths = sorted(
@@ -171,6 +172,9 @@ class CommitCallbackTest(unittest.TestCase):
for entry in callback.contexts[0].commit_entries
if entry.kind == 0
))
+ self.assertTrue(all(table.file_io.exists(path) for path in
expected_paths))
+ self.assertTrue(any('/dt=a%2Fb/' in path for path in expected_paths))
+ self.assertTrue(any('/dt=a%25b/' in path for path in expected_paths))
table_write.close()
table_commit.close()
diff --git a/paimon-python/pypaimon/write/file_store_commit.py
b/paimon-python/pypaimon/write/file_store_commit.py
index 03e718e707..10d8ce23c6 100644
--- a/paimon-python/pypaimon/write/file_store_commit.py
+++ b/paimon-python/pypaimon/write/file_store_commit.py
@@ -35,6 +35,7 @@ from pypaimon.snapshot.snapshot_commit import
(PartitionStatistics,
SnapshotCommit)
from pypaimon.table.row.generic_row import GenericRow
from pypaimon.table.row.offset_row import OffsetRow
+from pypaimon.utils.file_store_path_factory import canonical_data_file_path
from pypaimon.write.commit.commit_rollback import CommitRollback
from pypaimon.write.commit.commit_scanner import CommitScanner
from pypaimon.write.commit.conflict_detection import (
@@ -861,16 +862,10 @@ class FileStoreCommit:
entries.extend(self.manifest_file_manager.read(
manifest.file_name, drop_stats=False,
file_size=manifest.file_size))
- path_factory = self.table.path_factory()
for entry in entries:
file = entry.file
- file.file_path = file.physical_path() if
file.external_path else "%s/%s" % (
- path_factory.bucket_path(
- tuple(entry.partition.values),
- entry.bucket,
- ).rstrip("/"),
- file.file_name,
- )
+ file.file_path = file.physical_path() if
file.external_path else canonical_data_file_path(
+ self.table, tuple(entry.partition.values),
entry.bucket, file.file_name)
self._notify_commit_callbacks(
snapshot, entries, commit_identifier)
return True
@@ -1033,9 +1028,10 @@ class FileStoreCommit:
+ list(message.compact_changelog_files)):
path = None
try:
- bucket_path = None if file.physical_path() else
self.table.path_factory().bucket_path(
- tuple(message.partition), message.bucket)
- for path in file.collect_files(bucket_path):
+ if not file.physical_path():
+ file.file_path = canonical_data_file_path(
+ self.table, tuple(message.partition),
message.bucket, file.file_name)
+ for path in file.collect_files():
self.table.file_io.delete_quietly(path)
except Exception as error:
logger.warning(
diff --git a/paimon-python/pypaimon/write/native_commit.py
b/paimon-python/pypaimon/write/native_commit.py
index acbb1d2081..91716cc357 100644
--- a/paimon-python/pypaimon/write/native_commit.py
+++ b/paimon-python/pypaimon/write/native_commit.py
@@ -39,15 +39,12 @@ def native_commit_available() -> bool:
def native_messages_supported(table, messages) -> bool:
- 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
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
@@ -56,10 +53,6 @@ def native_messages_supported(table, messages) -> bool:
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
diff --git a/paimon-python/pypaimon/write/native_merge_into.py
b/paimon-python/pypaimon/write/native_merge_into.py
index 1e12185f9e..7d535ce1dd 100644
--- a/paimon-python/pypaimon/write/native_merge_into.py
+++ b/paimon-python/pypaimon/write/native_merge_into.py
@@ -20,10 +20,11 @@
import pyarrow as pa
from pypaimon.snapshot.snapshot import BATCH_COMMIT_IDENTIFIER
-from pypaimon.write.native_commit import from_native_commit_messages
+from pypaimon.write.native_commit import create_native_write_table,
from_native_commit_messages
from pypaimon.write.native_update import (
_native_row_id_table, _native_update_columns_supported,
_supported_upsert_key_type,
)
+from pypaimon.write.native_write import _native_partition_types_supported
def create_native_merge_into(table, source, on, matched, not_matched,
commit_user, commit_identifier):
@@ -36,12 +37,37 @@ def create_native_merge_into(table, source, on, matched,
not_matched, commit_use
self_merge = _is_self_merge(table, source, target_keys, source_keys)
if not_matched and table.options.video_frame_fields():
return None
- if _is_table_like(source) and not self_merge:
- return None
native_table = _native_row_id_table(table)
if native_table is None:
return None
- source, matched, not_matched, context = _prepare(table, source,
list(matched), list(not_matched), on)
+ source_schema = None
+ native_source = None
+ if _is_table_like(source) and not self_merge:
+ # The source can be an append or PK table; only the target requires DE.
+ from pypaimon.read.native_plan import _native_blob_view_supported
+ from pypaimon.table.file_store_table import FileStoreTable
+ if (type(source) is not FileStoreTable
+ or source.options.file_format() != 'parquet'
+ or source.options.video_frame_fields() or
source.options.with_vector_format()
+ or (source.is_primary_key_table and not
source.trimmed_primary_keys)):
+ return None
+ source.new_read_builder().new_scan()._validate_scan_mode()
+ if not _native_blob_view_supported(source, source.field_names):
+ return None
+ # This operation reads one full source snapshot. Incremental scans stay
+ # on the Python path, whose scanner applies the requested delta range.
+ if (source.options.scan_mode() not in ('default', 'from-snapshot',
'from-timestamp')
+ or any(str(key).startswith('incremental-') for key in
source.table_schema.options)):
+ return None
+ from pypaimon.schema.data_types import PyarrowFieldParser
+ source_schema =
PyarrowFieldParser.from_paimon_schema(source.table_schema.fields)
+ if not _native_partition_types_supported(source_schema,
source.partition_keys):
+ return None
+ native_source = create_native_write_table(source)
+ if native_source is None:
+ return None
+ source, matched, not_matched, context = _prepare(
+ table, source, list(matched), list(not_matched), on,
source_schema=source_schema)
if any(callable(value) for clause in matched + not_matched for value in
clause.spec.values()):
return None
if not _native_update_columns_supported(table,
_union_update_cols(matched)):
@@ -74,7 +100,8 @@ def create_native_merge_into(table, source, on, matched,
not_matched, commit_use
if commit_identifier == BATCH_COMMIT_IDENTIFIER else
native_table.new_stream_write_builder().with_commit_user(commit_user))
writer = builder.new_update()
- return NativeTableMergeInto(table, writer, source, list(zip(target_keys,
source_keys)),
+ return NativeTableMergeInto(table, writer, native_source if native_source
is not None else source,
+ list(zip(target_keys, source_keys)),
encoded_matched, encoded_not_matched,
commit_identifier)
diff --git a/paimon-python/pypaimon/write/native_update.py
b/paimon-python/pypaimon/write/native_update.py
index 9b19e49b1f..3175d30226 100644
--- a/paimon-python/pypaimon/write/native_update.py
+++ b/paimon-python/pypaimon/write/native_update.py
@@ -48,39 +48,9 @@ def _native_row_id_table(table):
schema = PyarrowFieldParser.from_paimon_schema(table.table_schema.fields)
if not _native_partition_types_supported(schema, table.partition_keys):
return None
- if not _native_update_paths_supported(table):
- return None
return create_native_write_table(table)
-def _native_update_paths_supported(table):
- """Keep legacy Python partition directories on the path-aware Python
updater."""
- if not table.partition_keys:
- return True
- factory = table.path_factory()
- # Core update scans for itself, unlike native reads which receive splits
- # with repaired file paths. Decide before callbacks or staging can start.
- splits = table.new_read_builder().new_scan().plan_for_write().splits()
- bucket_files = {}
- for split in splits:
- partition = tuple(split.partition.values)
- bucket_path = factory.bucket_path(partition, split.bucket)
- if bucket_path == factory.bucket_path(partition, split.bucket,
canonical_partition=True):
- continue
- candidates = [file for file in split.files if not file.external_path]
- if not candidates:
- continue
- if bucket_path not in bucket_files:
- bucket_files[bucket_path] = {
- status.base_name for status in
table.file_io.list_status(bucket_path)
- }
- # Python reads prefer an existing legacy path even if a canonical copy
- # also exists. Canonical-only files and explicit external paths are
safe.
- if any(file.file_name in bucket_files[bucket_path] for file in
candidates):
- return False
- return True
-
-
def create_native_update(table, commit_user, columns):
"""Use the public core updater for direct and grouped row-ID updates."""
if not _native_update_columns_supported(table, columns):
diff --git a/paimon-python/pypaimon/write/writer/data_writer.py
b/paimon-python/pypaimon/write/writer/data_writer.py
index 8c7e7fa45a..57d1dbbbe7 100644
--- a/paimon-python/pypaimon/write/writer/data_writer.py
+++ b/paimon-python/pypaimon/write/writer/data_writer.py
@@ -29,6 +29,7 @@ from pypaimon.manifest.schema.simple_stats import SimpleStats
from pypaimon.schema.data_types import PyarrowFieldParser
from pypaimon.table.bucket_mode import BucketMode
from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.utils.file_store_path_factory import canonical_data_file_path
from pypaimon.write.map_shared_shredding_writer import MapSharedShreddingWriter
from pypaimon.write.writer.mosaic_writer_options import
create_mosaic_writer_options
from pypaimon.write.writer.parquet_writer_options import
create_parquet_writer_options
@@ -95,7 +96,7 @@ class DataWriter(ABC):
self.path_factory = self.table.path_factory()
self.external_path_provider: Optional[ExternalPathProvider] =
self.path_factory.create_external_path_provider(
- self.partition, self.bucket
+ self.partition, self.bucket, canonical_partition=True
)
# Variant shredding (static mode) — col_name → (obj_fields,
target_arrow_type)
self._variant_shredding: Dict[str, Tuple] = {}
@@ -498,8 +499,9 @@ class DataWriter(ABC):
if self.external_path_provider:
return
self.external_path_provider.get_next_external_data_path(file_name)
- bucket_path = self.path_factory.bucket_path(self.partition,
self.bucket)
- return f"{bucket_path.rstrip('/')}/{file_name}"
+ # New files use Java's escaped partition names, including on the pure
+ # Python writer path. Never introduce another Python-specific layout.
+ return canonical_data_file_path(self.table, self.partition,
self.bucket, file_name)
def _should_write_row_sidecar(self) -> bool:
return (
