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 0c24e922d7 [python] Keep standalone helpers shared and row key
factories internal (#10781)
0c24e922d7 is described below
commit 0c24e922d73decaf2131b7246c51dada19e1250e
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Oct 23 22:33:38 2026 +0800
[python] Keep standalone helpers shared and row key factories internal
(#10781)
---
paimon-python/dev/test_packaging.py | 9 ++--
paimon-python/pypaimon/catalog/table_query_auth.py | 6 +--
.../arrow_predicate.py} | 2 +-
.../write => common}/external_path_provider.py | 2 +-
paimon-python/pypaimon/execution/README.md | 30 ++++++-----
.../python/read/reader/auth_masking_reader.py | 2 +-
.../execution/python/write/merge_condition.py | 62 ----------------------
.../execution/python/write/merge_transform.py | 6 +--
.../execution/python/write/row_key_extractor.py | 20 +++++++
.../pypaimon/execution/python/write/table_write.py | 5 +-
.../execution/python/write/writer/data_writer.py | 2 +-
paimon-python/pypaimon/ray/merge_condition.py | 6 +--
paimon-python/pypaimon/read/read_dispatch.py | 5 --
paimon-python/pypaimon/table/file_store_table.py | 11 +---
.../pypaimon/tests/execution_layout_test.py | 9 ++++
.../pypaimon/tests/external_paths_test.py | 2 +-
.../tests/multimodal_temporal_boundary_test.py | 2 +-
.../pypaimon/tests/native_cast_completion_test.py | 2 +-
.../pypaimon/tests/native_write_type_test.py | 5 +-
.../pypaimon/tests/predicate_json_parser_test.py | 2 +-
.../public_execution_factory_boundary_test.py | 39 +++++++++-----
.../tests/query_auth_execution_boundary_test.py | 23 +++++++-
.../pypaimon/tests/test_aggregation_e2e.py | 2 +-
.../pypaimon/tests/test_sequence_field_write.py | 2 +-
.../pypaimon/tests/write/table_write_test.py | 3 +-
.../pypaimon/utils/file_store_path_factory.py | 6 +--
paimon-python/pypaimon/write/backend_dispatch.py | 42 ---------------
paimon-python/pypaimon/write/merge_condition.py | 39 ++++++++++++++
28 files changed, 168 insertions(+), 178 deletions(-)
diff --git a/paimon-python/dev/test_packaging.py
b/paimon-python/dev/test_packaging.py
index ddf4877ec2..56e1725533 100644
--- a/paimon-python/dev/test_packaging.py
+++ b/paimon-python/dev/test_packaging.py
@@ -131,14 +131,13 @@ class PackagingTest(unittest.TestCase):
"pypaimon/execution/python/read/search/distributed_full_text_search.py",
"pypaimon/execution/python/map_blob_key_serializer.py",
"pypaimon/execution/python/globalindex/ray_index_build.py",
- "pypaimon/execution/python/write/external_path_provider.py",
- "pypaimon/execution/python/read/predicate_json_parser.py",
+ "pypaimon/common/external_path_provider.py",
+ "pypaimon/common/arrow_predicate.py",
"pypaimon/execution/python/read/blob_view_lookup.py",
"pypaimon/execution/python/read/interval_partition.py",
"pypaimon/execution/python/read/map_shared_shredding.py",
"pypaimon/execution/python/read/variant_shredding.py",
"pypaimon/execution/python/write/dynamic_bucket.py",
- "pypaimon/execution/python/write/merge_condition.py",
"pypaimon/execution/python/write/merge_transform.py",
"pypaimon/schema/data_type_casts.py",
"pypaimon/utils/numeric_utils.py",
@@ -221,7 +220,9 @@ class PackagingTest(unittest.TestCase):
if name.startswith("pypaimon/execution/") and name.endswith(".py"):
self.assertIn(name.split("/")[2], ("python", "native",
"__init__.py"))
internal_aliases = (
- "pypaimon/common/external_path_provider.py",
+ "pypaimon/execution/python/write/external_path_provider.py",
+ "pypaimon/execution/python/read/predicate_json_parser.py",
+ "pypaimon/execution/python/write/merge_condition.py",
"pypaimon/common/map_blob_key_serializer.py",
"pypaimon/utils/blob_view_lookup.py",
"pypaimon/read/interval_partition.py",
diff --git a/paimon-python/pypaimon/catalog/table_query_auth.py
b/paimon-python/pypaimon/catalog/table_query_auth.py
index 4a486cb6d0..8803475853 100644
--- a/paimon-python/pypaimon/catalog/table_query_auth.py
+++ b/paimon-python/pypaimon/catalog/table_query_auth.py
@@ -136,12 +136,12 @@ class TableQueryAuthResult:
def extract_row_filter(self) -> Optional[Callable[[pa.RecordBatch],
pa.Array]]:
if not self.filter:
return None
- from pypaimon.read.read_dispatch import create_python_row_filter
+ from pypaimon.common.arrow_predicate import
parse_predicate_to_batch_filter
if len(self.filter) == 1:
- return create_python_row_filter(self.filter[0])
+ return parse_predicate_to_batch_filter(self.filter[0])
# Java PredicateBuilder.and evaluates entries in their original order.
- return create_python_row_filter(json.dumps({
+ return parse_predicate_to_batch_filter(json.dumps({
'kind': 'COMPOUND', 'function': 'AND',
'children': [json.loads(rule) for rule in self.filter],
}))
diff --git
a/paimon-python/pypaimon/execution/python/read/predicate_json_parser.py
b/paimon-python/pypaimon/common/arrow_predicate.py
similarity index 99%
rename from
paimon-python/pypaimon/execution/python/read/predicate_json_parser.py
rename to paimon-python/pypaimon/common/arrow_predicate.py
index f3bde3c2e9..916a849232 100644
--- a/paimon-python/pypaimon/execution/python/read/predicate_json_parser.py
+++ b/paimon-python/pypaimon/common/arrow_predicate.py
@@ -16,7 +16,7 @@
# limitations under the License.
################################################################################
-"""Execute REST row filters and column-mask transforms on Python Arrow
batches."""
+"""Compile REST predicates and column transforms for standalone Arrow
batches."""
import json
import re
diff --git
a/paimon-python/pypaimon/execution/python/write/external_path_provider.py
b/paimon-python/pypaimon/common/external_path_provider.py
similarity index 99%
rename from
paimon-python/pypaimon/execution/python/write/external_path_provider.py
rename to paimon-python/pypaimon/common/external_path_provider.py
index 8ded4dba5d..8457850113 100644
--- a/paimon-python/pypaimon/execution/python/write/external_path_provider.py
+++ b/paimon-python/pypaimon/common/external_path_provider.py
@@ -15,7 +15,7 @@
# specific language governing permissions and limitations
# under the License.
-"""Python writer strategies for selecting the next external file location."""
+"""Shared strategies for selecting the next external file location."""
import bisect
import ctypes
diff --git a/paimon-python/pypaimon/execution/README.md
b/paimon-python/pypaimon/execution/README.md
index 88845fe0aa..7f40826489 100644
--- a/paimon-python/pypaimon/execution/README.md
+++ b/paimon-python/pypaimon/execution/README.md
@@ -41,6 +41,9 @@ Keep this boundary when adding or moving code:
```text
pypaimon/
+ common/
+ arrow_predicate.py Standalone Arrow predicate and transform
helpers
+ external_path_provider.py Shared external file placement strategies
read/ Public read/scan APIs, Arrow output and explain
read_dispatch.py Read/scan selection across the two backends
read_support.py Shared output, schema and reader ownership
@@ -51,7 +54,7 @@ pypaimon/
write_dispatch.py Writer coordination and safe pre-write
switching
update_dispatch.py Update/upsert/delete/MERGE coordination
input_adapter.py Pandas transport and distributed input
conversion
- merge_condition.py Shared MERGE condition parsing and validation
+ merge_condition.py Shared MERGE condition parsing and Arrow batch
filtering
merge_transform.py Shared Arrow assignments and callable inputs
ray_datasink.py Ray worker writes and coordinator commits
globalindex/ Public index contracts and backend coordination
@@ -71,12 +74,10 @@ pypaimon/
search/ Vector/text/hybrid search and Python factories
blob_view_lookup.py Upstream Blob descriptor planning, loading and
resolution
interval_partition.py Primary-key sorted-run grouping
- predicate_json_parser.py Authorization row-filter and masking execution
map_shared_shredding.py Physical MAP decoding and selected-column
pruning
variant_shredding.py Physical VARIANT reconstruction and shredding
deletionvectors/ DV bitmap codecs, file I/O and deleted-row
filtering
write/ File writers, committers and update execution
- external_path_provider.py Per-writer external file placement strategies
globalindex/ Index file codecs, readers, writers and build
algorithms
index/ Primary-key index source selection and
sorted-index grouping
merge_engine_dispatch.py Python merge-function selection
@@ -149,9 +150,9 @@ executor; Native commits pass the spec to Rust.
Physical MAP/VARIANT codecs and Blob View resolution belong to Python
execution. Public value objects, schema/spec parsing and selected-key metadata
-stay in `data`. MERGE condition evaluation and first-matching-clause selection
-also belong to Python execution; callable transport, SQL rewriting and shared
-Arrow assignment adapters stay in `write`.
+stay in `data`. Table MERGE action selection belongs to the selected backend.
+Standalone Arrow condition filtering, SQL rewriting, callable transport and
+shared Arrow assignment adapters stay in `write`.
Dynamic bucket HASH index publication and writer-owned pending state belong
to the Python writer. The shared input bucket assigner remains in `index`:
@@ -159,18 +160,19 @@ distributed producers use it to route inputs before
either backend writes.
`input_routing` and `bucket_planning` similarly serve both backends' input
protocols without retaining a file writer or publishing data.
-External-path round-robin, weighted selection and entropy directories belong
-to Python write execution. Shared path factories retain configuration and
-existing-file resolution, loading a provider only when Python allocates a new
-file. MAP BLOB key serializers implement the physical file format and are
+External-path round-robin, weighted selection and entropy directories are
shared
+path utilities in `common`. Shared path factories retain configuration and
+existing-file resolution, loading a provider when a caller allocates a new
file.
+MAP BLOB key serializers implement the physical file format and are
shared within the Python backend. Generic URI/FileIO contracts, resource
caches and integer codecs stay in `common`; Native Blob inputs also use the
URI reader contract.
-Authorization row-filter and column-mask evaluation belongs to Python read
-execution. JSON field-reference extraction and rule/schema validation remain
-shared. The public authorization result creates a Python filter lazily when
-explicitly requested; Native reads pass the original rules to Rust.
+Standalone Arrow predicate and transform helpers live in
`common/arrow_predicate`.
+JSON field-reference extraction and rule/schema validation remain independent
+of that Arrow evaluator. The public authorization result creates an Arrow
+filter lazily when explicitly requested without selecting either backend.
+Python readers use the shared evaluator; Native reads pass the original rules
to Rust.
Framework adapters keep their actual implementation in their original modules,
including their public class identities, serialization and worker closures.
diff --git
a/paimon-python/pypaimon/execution/python/read/reader/auth_masking_reader.py
b/paimon-python/pypaimon/execution/python/read/reader/auth_masking_reader.py
index 3c8e12b7d8..6ad5df55f0 100644
--- a/paimon-python/pypaimon/execution/python/read/reader/auth_masking_reader.py
+++ b/paimon-python/pypaimon/execution/python/read/reader/auth_masking_reader.py
@@ -22,7 +22,7 @@ from typing import Callable, Dict, List, Optional
import pyarrow as pa
from pypaimon.common.predicate_json_parser import
_collect_all_field_refs_from_transform
-from pypaimon.execution.python.read.predicate_json_parser import
_apply_predicate_transform
+from pypaimon.common.arrow_predicate import _apply_predicate_transform
from pypaimon.read.reader.iface.record_batch_reader import RecordBatchReader
from pypaimon.read.reader.iface.record_iterator import RecordIterator
from pypaimon.read.reader.iface.record_reader import RecordReader
diff --git a/paimon-python/pypaimon/execution/python/write/merge_condition.py
b/paimon-python/pypaimon/execution/python/write/merge_condition.py
deleted file mode 100644
index f0c66cea11..0000000000
--- a/paimon-python/pypaimon/execution/python/write/merge_condition.py
+++ /dev/null
@@ -1,62 +0,0 @@
-################################################################################
-# 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.
-################################################################################
-
-"""Evaluate MERGE conditions with the Python DataFusion engine."""
-
-import pyarrow as pa
-
-
-def filter_batch(
- batch: pa.Table, condition: str, _pre_rewritten: bool = False,
-) -> pa.Table:
- if batch.num_rows == 0:
- return batch
- from pypaimon.write.merge_condition import _load_datafusion,
rewrite_condition
-
- datafusion = _load_datafusion()
- rewritten = condition if _pre_rewritten else rewrite_condition(condition)
- config = datafusion.SessionConfig().set(
- "datafusion.optimizer.enable_round_robin_repartition", "false"
- )
- ctx = datafusion.SessionContext(config)
- input_batches = batch.to_batches()
- # Use one batch per partition and rebuild from partitioned batches so
- # DataFusion neither concatenates 32-bit offsets nor reorders the input.
- ctx.register_record_batches(
- "_batch", [[record_batch] for record_batch in input_batches]
- )
- result = ctx.sql(
- f'SELECT * FROM _batch WHERE {rewritten}'
- )
- output_batches = [
- record_batch
- for partition in result.collect_partitioned()
- for record_batch in partition
- ]
- if not output_batches:
- return batch.schema.empty_table()
- return pa.Table.from_batches(output_batches)
-
-
-def apply_condition(
- batch: pa.Table, rewritten: str, empty_schema: pa.Schema,
-) -> pa.Table:
- batch = filter_batch(batch, rewritten, _pre_rewritten=True)
- if batch.num_rows == 0:
- return empty_schema.empty_table()
- return batch
diff --git a/paimon-python/pypaimon/execution/python/write/merge_transform.py
b/paimon-python/pypaimon/execution/python/write/merge_transform.py
index d009eccc1b..6819b41126 100644
--- a/paimon-python/pypaimon/execution/python/write/merge_transform.py
+++ b/paimon-python/pypaimon/execution/python/write/merge_transform.py
@@ -56,7 +56,7 @@ def _build_matched_transform(
_filter_batch = None
if any(r is not None for _, r, _ in prepared_clauses):
_load_datafusion()
- from pypaimon.execution.python.write.merge_condition import
filter_batch as _filter_batch
+ from pypaimon.write.merge_condition import filter_batch as
_filter_batch
def _transform(batch: pa.Table) -> pa.Table:
remaining = batch
@@ -123,7 +123,7 @@ def _build_matched_delete_transform(
_filter_batch = None
if any(r is not None for r, _ in prepared_clauses):
_load_datafusion()
- from pypaimon.execution.python.write.merge_condition import
filter_batch as _filter_batch
+ from pypaimon.write.merge_condition import filter_batch as
_filter_batch
def _transform(batch: pa.Table) -> pa.Table:
remaining = batch
@@ -174,7 +174,7 @@ def _build_insert_transform(
_filter_batch = None
if any(r is not None for _, r in prepared_clauses):
_load_datafusion()
- from pypaimon.execution.python.write.merge_condition import
filter_batch as _filter_batch
+ from pypaimon.write.merge_condition import filter_batch as
_filter_batch
def _transform(batch: pa.Table) -> pa.Table:
remaining = batch
diff --git a/paimon-python/pypaimon/execution/python/write/row_key_extractor.py
b/paimon-python/pypaimon/execution/python/write/row_key_extractor.py
index 81589e0926..eb34e3304e 100644
--- a/paimon-python/pypaimon/execution/python/write/row_key_extractor.py
+++ b/paimon-python/pypaimon/execution/python/write/row_key_extractor.py
@@ -520,3 +520,23 @@ class PostponeBucketRowKeyExtractor(RowKeyExtractor):
class PostponeFixedBucketRowKeyExtractor(PostponeFixedBucketInputRouter,
RowKeyExtractor):
"""Route Python file writes with the shared distributed input router."""
+
+
+def create_row_key_extractor(table, ignore_existing=False):
+ """Construct a per-row extractor for Python file writing."""
+ bucket_mode = table.bucket_mode()
+ if bucket_mode == BucketMode.HASH_FIXED:
+ return FixedBucketRowKeyExtractor(table.table_schema)
+ elif bucket_mode == BucketMode.BUCKET_UNAWARE:
+ return UnawareBucketRowKeyExtractor(table.table_schema)
+ elif bucket_mode == BucketMode.POSTPONE_MODE:
+ return PostponeBucketRowKeyExtractor(table.table_schema)
+ elif bucket_mode == BucketMode.HASH_DYNAMIC:
+ return DynamicBucketRowKeyExtractor(
+ table.table_schema, table=table, ignore_existing=ignore_existing)
+ elif bucket_mode == BucketMode.CROSS_PARTITION:
+ raise ValueError(
+ "CROSS_PARTITION primary-key writes require a persistent "
+ "global primary-key index, which PyPaimon does not yet support"
+ )
+ raise ValueError(f"Unsupported bucket mode: {bucket_mode}")
diff --git a/paimon-python/pypaimon/execution/python/write/table_write.py
b/paimon-python/pypaimon/execution/python/write/table_write.py
index ac0d892000..61a53562fd 100644
--- a/paimon-python/pypaimon/execution/python/write/table_write.py
+++ b/paimon-python/pypaimon/execution/python/write/table_write.py
@@ -78,9 +78,8 @@ class PythonTableWrite:
writer.table.table_schema, table=writer.table,
ignore_existing=static_partition is not None,
base_snapshot_id=writer.restore_snapshot_id)
- return writer.table.create_row_key_extractor(
- ignore_existing=static_partition is not None
- )
+ from pypaimon.execution.python.write.row_key_extractor import
create_row_key_extractor
+ return create_row_key_extractor(writer.table,
ignore_existing=static_partition is not None)
def write_arrow(self, data, bucket: Optional[int] = None):
"""Write Arrow data, optionally to a bucket assigned upstream."""
diff --git
a/paimon-python/pypaimon/execution/python/write/writer/data_writer.py
b/paimon-python/pypaimon/execution/python/write/writer/data_writer.py
index 548743d3be..98a3136abc 100644
--- a/paimon-python/pypaimon/execution/python/write/writer/data_writer.py
+++ b/paimon-python/pypaimon/execution/python/write/writer/data_writer.py
@@ -22,7 +22,7 @@ from abc import ABC, abstractmethod
from typing import Dict, List, Optional, Tuple
from pypaimon.common.options.core_options import CoreOptions, ChangelogProducer
-from pypaimon.execution.python.write.external_path_provider import
ExternalPathProvider
+from pypaimon.common.external_path_provider import ExternalPathProvider
from pypaimon.data.timestamp import Timestamp
from pypaimon.manifest.schema.data_file_meta import DataFileMeta
from pypaimon.manifest.schema.simple_stats import SimpleStats
diff --git a/paimon-python/pypaimon/ray/merge_condition.py
b/paimon-python/pypaimon/ray/merge_condition.py
index 7afcda4c7a..0985557668 100644
--- a/paimon-python/pypaimon/ray/merge_condition.py
+++ b/paimon-python/pypaimon/ray/merge_condition.py
@@ -17,7 +17,7 @@
################################################################################
-"""Public MERGE condition helpers retain their shared and Python delegates."""
+"""Public MERGE condition helpers delegate to the shared Arrow utilities."""
from typing import Mapping, Optional, Set
@@ -51,14 +51,14 @@ def remap_source_on_keys(
def filter_batch(
batch: pa.Table, condition: str, _pre_rewritten: bool = False,
) -> pa.Table:
- from pypaimon.write.backend_dispatch import filter_merge_condition_batch
as execute
+ from pypaimon.write.merge_condition import filter_batch as execute
return execute(batch, condition, _pre_rewritten)
def apply_condition(
batch: pa.Table, rewritten: str, empty_schema: pa.Schema,
) -> pa.Table:
- from pypaimon.write.backend_dispatch import apply_merge_condition as
execute
+ from pypaimon.write.merge_condition import apply_condition as execute
return execute(batch, rewritten, empty_schema)
diff --git a/paimon-python/pypaimon/read/read_dispatch.py
b/paimon-python/pypaimon/read/read_dispatch.py
index 2e2867fedb..a78356f2b6 100644
--- a/paimon-python/pypaimon/read/read_dispatch.py
+++ b/paimon-python/pypaimon/read/read_dispatch.py
@@ -55,11 +55,6 @@ def python_row_group_cache(request):
return request._cache
-def create_python_row_filter(rule_json):
- from pypaimon.execution.python.read.predicate_json_parser import
parse_predicate_to_batch_filter
- return parse_predicate_to_batch_filter(rule_json)
-
-
def validate_read_backend(read):
"""Validate the explicitly selected Python backend without loading
Native."""
if not read.table.options.native_read_enabled():
diff --git a/paimon-python/pypaimon/table/file_store_table.py
b/paimon-python/pypaimon/table/file_store_table.py
index 40e0dcb816..4ca3e6bde4 100644
--- a/paimon-python/pypaimon/table/file_store_table.py
+++ b/paimon-python/pypaimon/table/file_store_table.py
@@ -15,7 +15,7 @@
# specific language governing permissions and limitations
# under the License.
-from typing import List, Optional, Set, TYPE_CHECKING
+from typing import List, Optional, Set
from pypaimon.catalog.catalog_environment import CatalogEnvironment
from pypaimon.common.file_io import FileIO
@@ -30,9 +30,6 @@ from pypaimon.table.bucket_mode import BucketMode
from pypaimon.table.table import Table
from pypaimon.write.write_builder import BatchWriteBuilder, StreamWriteBuilder
-if TYPE_CHECKING:
- from pypaimon.execution.python.write.row_key_extractor import
RowKeyExtractor
-
class FileStoreTable(Table):
def __init__(self, file_io: FileIO, identifier: Identifier, table_path:
str,
@@ -486,12 +483,6 @@ class FileStoreTable(Table):
dry_run=dry_run,
)
- def create_row_key_extractor(
- self, ignore_existing: bool = False
- ) -> 'RowKeyExtractor':
- from pypaimon.write.backend_dispatch import create_row_key_extractor
- return create_row_key_extractor(self, ignore_existing)
-
def copy(self, options: dict) -> 'FileStoreTable':
return self._copy(options, resolve_time_travel=True)
diff --git a/paimon-python/pypaimon/tests/execution_layout_test.py
b/paimon-python/pypaimon/tests/execution_layout_test.py
index 8c20b61fb5..beed1f371c 100644
--- a/paimon-python/pypaimon/tests/execution_layout_test.py
+++ b/paimon-python/pypaimon/tests/execution_layout_test.py
@@ -61,11 +61,20 @@ modules = [
'pypaimon.schema.arrow_schema', 'pypaimon.schema.data_type_casts',
'pypaimon.schema.schema_manager', 'pypaimon.utils.file_store_path_factory',
'pypaimon.table.system.files_table',
'pypaimon.common.predicate_json_parser',
+ 'pypaimon.common.transform_schema', 'pypaimon.common.arrow_predicate',
+ 'pypaimon.common.external_path_provider',
'pypaimon.common.uri_reader', 'pypaimon.common.blob_index_cache',
'pypaimon.common.delta_varint_compressor', 'pypaimon.common.file_io',
]
for name in modules:
importlib.import_module(name)
+
+from pypaimon.utils.file_store_path_factory import FileStorePathFactory
+factory = FileStorePathFactory(
+ '/warehouse', ['p'], '__DEFAULT_PARTITION__', 'parquet', 'data-',
'changelog-',
+ False, False, 'zstd', external_paths=['/external'],
external_path_strategy='round-robin')
+provider = factory.create_external_path_provider(('x',), 0,
canonical_partition=True)
+assert provider.get_next_external_data_path('data.parquet') ==
'/external/p=x/bucket-0/data.parquet'
assert not any(name == 'pypaimon.execution' or
name.startswith('pypaimon.execution.')
for name in sys.modules)
"""
diff --git a/paimon-python/pypaimon/tests/external_paths_test.py
b/paimon-python/pypaimon/tests/external_paths_test.py
index 0a6457f12d..a77d3506a9 100644
--- a/paimon-python/pypaimon/tests/external_paths_test.py
+++ b/paimon-python/pypaimon/tests/external_paths_test.py
@@ -24,7 +24,7 @@ import pyarrow as pa
from pypaimon import CatalogFactory, Schema
from pypaimon.catalog.catalog import Identifier
-from pypaimon.execution.python.write.external_path_provider import (
+from pypaimon.common.external_path_provider import (
EntropyInjectExternalPathProvider, ExternalPathProvider,
RoundRobinExternalPathProvider, WeightedExternalPathProvider, _murmur3_32)
from pypaimon.common.options.core_options import (CoreOptions,
diff --git a/paimon-python/pypaimon/tests/multimodal_temporal_boundary_test.py
b/paimon-python/pypaimon/tests/multimodal_temporal_boundary_test.py
index 4dc94cec86..5fc030a059 100644
--- a/paimon-python/pypaimon/tests/multimodal_temporal_boundary_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_temporal_boundary_test.py
@@ -81,7 +81,7 @@ assert not loaded, loaded
])
def test_temporal_mask_schema_matches_actual_python_mask(data_type, transform):
from pypaimon.common.transform_schema import effective_masked_schema
- from pypaimon.execution.python.read.predicate_json_parser import
_apply_predicate_transform
+ from pypaimon.common.arrow_predicate import _apply_predicate_transform
schema = pa.schema([pa.field('value', data_type, nullable=False,
metadata={b'field': b'value'})],
metadata={b'schema': b'value'})
diff --git a/paimon-python/pypaimon/tests/native_cast_completion_test.py
b/paimon-python/pypaimon/tests/native_cast_completion_test.py
index 845ad33f48..bcbe4042ea 100644
--- a/paimon-python/pypaimon/tests/native_cast_completion_test.py
+++ b/paimon-python/pypaimon/tests/native_cast_completion_test.py
@@ -55,7 +55,7 @@ def native_only():
'pypaimon.execution.python.read.table_read.PythonTableRead._create_split_read',
side_effect=AssertionError('Native read fell back')
), \
-
patch('pypaimon.execution.python.read.predicate_json_parser._apply_predicate_transform',
+ patch('pypaimon.common.arrow_predicate._apply_predicate_transform',
side_effect=AssertionError('Python executed an authorization
transform')), \
patch('pypaimon.execution.python.read.reader.auth_masking_reader._apply_predicate_transform',
side_effect=AssertionError('Python executed an authorization
mask')):
diff --git a/paimon-python/pypaimon/tests/native_write_type_test.py
b/paimon-python/pypaimon/tests/native_write_type_test.py
index 7d081dea09..59fca4d726 100644
--- a/paimon-python/pypaimon/tests/native_write_type_test.py
+++ b/paimon-python/pypaimon/tests/native_write_type_test.py
@@ -27,6 +27,7 @@ import pytest
from pypaimon import Schema
from pypaimon.execution.native.runtime import (native_runtime_available)
+from pypaimon.execution.python.write.row_key_extractor import
create_row_key_extractor
from pypaimon.table.row.generic_row import GenericRow
from pypaimon.tests import native_plan_rest_test
from pypaimon.write.write_dispatch import NativeTableWrite
@@ -173,7 +174,7 @@ def
test_full_input_routes_before_projecting_unwritten_routing_fields(
data = pa.Table.from_pylist([
dict(id=1, pt=7, value=42), dict(id=None, pt=None, value=43),
], schema=_SCHEMA)
- partitions, bucket_ids =
table.create_row_key_extractor().extract_partition_bucket_batch(data.to_batches()[0])
+ partitions, bucket_ids =
create_row_key_extractor(table).extract_partition_bucket_batch(data.to_batches()[0])
expected_buckets = dict(zip(partitions, bucket_ids))
builder = table.new_batch_write_builder()
writer = builder.new_write()
@@ -210,7 +211,7 @@ def
test_rows_route_with_unwritten_columns_without_requiring_other_fields(
values = dict(id=1, pt=7, value=42, unwritten=blob if unwritten_blob else
999)
row = GenericRow([values[name] for name in names], [fields[name] for name
in names])
full_input = pa.Table.from_pylist([dict(values, unwritten=None)],
schema=schema).to_batches()[0]
- _, buckets =
table.create_row_key_extractor().extract_partition_bucket_batch(full_input)
+ _, buckets =
create_row_key_extractor(table).extract_partition_bucket_batch(full_input)
builder = table.new_batch_write_builder()
writer = builder.new_write()
try:
diff --git a/paimon-python/pypaimon/tests/predicate_json_parser_test.py
b/paimon-python/pypaimon/tests/predicate_json_parser_test.py
index 66cbc9bf37..ea3d94d9bb 100644
--- a/paimon-python/pypaimon/tests/predicate_json_parser_test.py
+++ b/paimon-python/pypaimon/tests/predicate_json_parser_test.py
@@ -24,7 +24,7 @@ from pathlib import Path
import pyarrow as pa
from pypaimon.common.predicate_json_parser import extract_referenced_fields
-from pypaimon.execution.python.read.predicate_json_parser import (
+from pypaimon.common.arrow_predicate import (
_apply_predicate_transform,
_convert_literal,
_paimon_type_to_arrow,
diff --git
a/paimon-python/pypaimon/tests/public_execution_factory_boundary_test.py
b/paimon-python/pypaimon/tests/public_execution_factory_boundary_test.py
index 7869fedb20..6f4b928dbb 100644
--- a/paimon-python/pypaimon/tests/public_execution_factory_boundary_test.py
+++ b/paimon-python/pypaimon/tests/public_execution_factory_boundary_test.py
@@ -15,7 +15,7 @@
# specific language governing permissions and limitations
# under the License.
-"""Public factories retain captured inputs and enter their selected dispatch
boundary."""
+"""Public factories retain captured inputs and keep standalone helpers
independent."""
import subprocess
import sys
@@ -172,7 +172,9 @@ assert not any(name.startswith('pypaimon.execution.python')
for name in sys.modu
('-2', ['id', 'p'], 'PostponeBucketRowKeyExtractor'),
('-1', ['id', 'p'], 'DynamicBucketRowKeyExtractor'),
])
-def test_public_row_key_factory_preserves_per_row_contract(tmp_path, bucket,
primary_keys, expected):
+def test_python_row_key_factory_preserves_per_row_contract(tmp_path, bucket,
primary_keys, expected):
+ from pypaimon.execution.python.write.row_key_extractor import
create_row_key_extractor
+
catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
catalog.create_database('default', True)
arrow_schema = pa.schema([('id', pa.int32()), ('p', pa.string())])
@@ -184,7 +186,7 @@ def
test_public_row_key_factory_preserves_per_row_contract(tmp_path, bucket, pri
rows = pa.record_batch([pa.array([1, 2], pa.int32()), pa.array(['a',
'a'])], schema=arrow_schema)
with patch('pypaimon.execution.native.runtime.native_runtime_available',
side_effect=AssertionError('Per-row helper rediscovered the
Native writer')):
- extractor = table.create_row_key_extractor(ignore_existing=True)
+ extractor = create_row_key_extractor(table, ignore_existing=True)
partitions, buckets = extractor.extract_partition_bucket_batch(rows)
assert type(extractor).__name__ == expected
assert partitions == [('a',), ('a',)]
@@ -194,16 +196,29 @@ def
test_public_row_key_factory_preserves_per_row_contract(tmp_path, bucket, pri
assert buckets == [-2 if bucket == '-2' else 0] * 2
-def test_ray_condition_compatibility_helpers_keep_python_execution():
+def test_shared_merge_condition_helpers_do_not_load_execution():
if sys.version_info < (3, 10):
pytest.skip('Python MERGE conditions require Python 3.10')
pytest.importorskip('datafusion')
- from pypaimon.ray.merge_condition import apply_condition, filter_batch,
rewrite_condition
+ source = r'''
+import importlib.abc
+import sys
+import pyarrow as pa
- rows = pa.table({'s.id': pa.chunked_array([[1, 2], [3]]), 't.id': [2, 1,
2]})
- with patch('pypaimon.execution.native.runtime.native_runtime_available',
- side_effect=AssertionError('Python condition helper discovered
Native execution')):
- assert filter_batch(rows, 's.id > t.id').column('s.id').to_pylist() ==
[2, 3]
- schema = pa.schema([('s.id', pa.int64())])
- empty = apply_condition(rows, rewrite_condition('s.id < 0'), schema)
- assert empty.num_rows == 0 and empty.schema == schema
+class RejectExecution(importlib.abc.MetaPathFinder):
+ def find_spec(self, fullname, path=None, target=None):
+ if fullname.startswith(('pypaimon.execution', 'pypaimon_rust')):
+ raise AssertionError('Shared condition helper selected execution:
' + fullname)
+
+sys.meta_path.insert(0, RejectExecution())
+from pypaimon.ray.merge_condition import apply_condition, filter_batch,
rewrite_condition
+rows = pa.table({'s.id': pa.chunked_array([[1, 2], [3]]), 't.id': [2, 1, 2]})
+assert filter_batch(rows, 's.id > t.id').column('s.id').to_pylist() == [2, 3]
+schema = pa.schema([('s.id', pa.int64())])
+empty = apply_condition(rows, rewrite_condition('s.id < 0'), schema)
+assert empty.num_rows == 0 and empty.schema == schema
+assert not any(name.startswith(('pypaimon.execution', 'pypaimon_rust')) for
name in sys.modules)
+'''
+ result = subprocess.run([sys.executable, '-c', source],
stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE, universal_newlines=True,
timeout=30)
+ assert result.returncode == 0, result.stdout + result.stderr
diff --git a/paimon-python/pypaimon/tests/query_auth_execution_boundary_test.py
b/paimon-python/pypaimon/tests/query_auth_execution_boundary_test.py
index 0284b763ef..9bc4a3471e 100644
--- a/paimon-python/pypaimon/tests/query_auth_execution_boundary_test.py
+++ b/paimon-python/pypaimon/tests/query_auth_execution_boundary_test.py
@@ -15,7 +15,7 @@
# specific language governing permissions and limitations
# under the License.
-"""Authorization metadata validation does not select a read executor."""
+"""Authorization metadata and standalone Arrow filters do not select an
executor."""
import subprocess
import sys
@@ -26,6 +26,8 @@ def
test_authorization_fields_validate_without_loading_either_execution_backend(
import importlib.abc
import json
import sys
+from datetime import datetime
+import pyarrow as pa
class RejectExecution(importlib.abc.MetaPathFinder):
def find_spec(self, fullname, path=None, target=None):
@@ -59,6 +61,25 @@ result.validate_read_type(fields, fields)
assert result.get_extra_fields([fields[1]], fields) == [fields[0]]
assert TableQueryAuthResult(None, None).extract_row_filter() is None
assert TableQueryAuthResult([], None).extract_row_filter() is None
+assert 'pypaimon.common.arrow_predicate' not in sys.modules
+
+batch = pa.RecordBatch.from_pydict({
+ 'event_time': pa.array([datetime(2026, 1, 2), datetime(2025, 1, 2)],
type=pa.timestamp('ms')),
+ 'secret': ['first', 'second'],
+})
+assert result.extract_row_filter()(batch).to_pylist() == [True, False]
+cast = json.dumps({
+ 'kind': 'LEAF', 'function': 'EQUAL', 'literals': ['42'],
+ 'transform': {'name': 'CAST', 'fieldRef': {'name': 'value', 'type':
'INT'}, 'type': 'STRING'},
+})
+assert TableQueryAuthResult([cast], None).extract_row_filter()(
+ pa.RecordBatch.from_pydict({'value': [42, 43]})).to_pylist() == [True,
False]
+like = json.dumps({
+ 'kind': 'LEAF', 'function': 'LIKE', 'literals': ['a%'],
+ 'transform': {'name': 'FIELD_REF', 'fieldRef': {'name': 'value', 'type':
'STRING'}},
+})
+assert TableQueryAuthResult([like], None).extract_row_filter()(
+ pa.RecordBatch.from_pydict({'value': ['abc', 'xyz']})).to_pylist() ==
[True, False]
try:
TableQueryAuthResult([year_filter('missing')],
None).validate_against_schema(fields)
diff --git a/paimon-python/pypaimon/tests/test_aggregation_e2e.py
b/paimon-python/pypaimon/tests/test_aggregation_e2e.py
index 31968d0d78..3eee1c327b 100644
--- a/paimon-python/pypaimon/tests/test_aggregation_e2e.py
+++ b/paimon-python/pypaimon/tests/test_aggregation_e2e.py
@@ -297,7 +297,7 @@ class AggregationMergeEngineE2ETest(unittest.TestCase):
writer_class = StreamTableWrite if streaming else BatchTableWrite
# A rejected writer must never create the bucket index maintainer,
# including callers that construct TableWrite directly.
- with patch.object(table, 'create_row_key_extractor') as extractor:
+ with
patch('pypaimon.execution.python.write.row_key_extractor.create_row_key_extractor')
as extractor:
for create in (builder.new_write, lambda: writer_class(table,
'test')):
with self.subTest(streaming=streaming, create=create):
with self.assertRaisesRegex(
diff --git a/paimon-python/pypaimon/tests/test_sequence_field_write.py
b/paimon-python/pypaimon/tests/test_sequence_field_write.py
index 5a3231189f..47130214f6 100644
--- a/paimon-python/pypaimon/tests/test_sequence_field_write.py
+++ b/paimon-python/pypaimon/tests/test_sequence_field_write.py
@@ -221,7 +221,7 @@ def
test_dynamic_bucket_sequence_rejected_before_index_creation(
builder = (table.new_stream_write_builder() if streaming
else table.new_batch_write_builder())
writer_class = StreamTableWrite if streaming else BatchTableWrite
- with patch.object(table, 'create_row_key_extractor') as extractor:
+ with
patch('pypaimon.execution.python.write.row_key_extractor.create_row_key_extractor')
as extractor:
with pytest.raises(error):
if direct:
writer_class(table, 'test')
diff --git a/paimon-python/pypaimon/tests/write/table_write_test.py
b/paimon-python/pypaimon/tests/write/table_write_test.py
index 8ab45cb3ac..5685f0253f 100644
--- a/paimon-python/pypaimon/tests/write/table_write_test.py
+++ b/paimon-python/pypaimon/tests/write/table_write_test.py
@@ -38,6 +38,7 @@ from pypaimon.common.options.options import Options
from pypaimon.manifest.manifest_list_manager import ManifestListManager
from pypaimon.write.table_write import TableWrite
from pypaimon.execution.python.write.table_write import PythonTableWrite
+from pypaimon.execution.python.write.row_key_extractor import
create_row_key_extractor
from pypaimon.execution.python.write.writer.append_only_data_writer import
AppendOnlyDataWriter
@@ -363,7 +364,7 @@ class TableWriteTest(unittest.TestCase):
self.pa_schema, partition_keys=partition_keys,
options={'bucket': '-1'})
self.catalog.create_table(name, schema, False)
- return self.catalog.get_table(name).create_row_key_extractor()
+ return create_row_key_extractor(self.catalog.get_table(name))
def test_extract_partition_bucket_groups_multi_partition(self):
ex = self._unaware_partitioned_extractor(
diff --git a/paimon-python/pypaimon/utils/file_store_path_factory.py
b/paimon-python/pypaimon/utils/file_store_path_factory.py
index 7ba5342b7c..b66b54f406 100644
--- a/paimon-python/pypaimon/utils/file_store_path_factory.py
+++ b/paimon-python/pypaimon/utils/file_store_path_factory.py
@@ -26,7 +26,7 @@ from pypaimon.table.bucket_mode import BucketMode
from pypaimon.table.row.generic_row import _is_ltz_type, _normalize_ltz,
_parse_type_precision_scale
if TYPE_CHECKING:
- from pypaimon.execution.python.write.external_path_provider import
ExternalPathProvider
+ from pypaimon.common.external_path_provider import ExternalPathProvider
def _is_null_or_whitespace_only(value) -> bool:
@@ -246,10 +246,10 @@ class FileStorePathFactory:
if not self.external_paths:
return None
- from pypaimon.write.backend_dispatch import
create_external_path_provider
+ from pypaimon.common.external_path_provider import ExternalPathProvider
relative_bucket_path = self.relative_bucket_path(partition, bucket,
canonical_partition)
- return create_external_path_provider(
+ return ExternalPathProvider.create(
self.external_path_strategy,
self.external_paths,
relative_bucket_path,
diff --git a/paimon-python/pypaimon/write/backend_dispatch.py
b/paimon-python/pypaimon/write/backend_dispatch.py
index ab773ac5af..d72931ac08 100644
--- a/paimon-python/pypaimon/write/backend_dispatch.py
+++ b/paimon-python/pypaimon/write/backend_dispatch.py
@@ -565,48 +565,6 @@ def close_table_commit(owner):
pass
-def create_row_key_extractor(table, ignore_existing=False):
- """Construct the public per-row extractor through the Python write
boundary."""
- from pypaimon.table.bucket_mode import BucketMode
- from pypaimon.execution.python.write.row_key_extractor import (
- DynamicBucketRowKeyExtractor, FixedBucketRowKeyExtractor,
- PostponeBucketRowKeyExtractor, UnawareBucketRowKeyExtractor,
- )
-
- bucket_mode = table.bucket_mode()
- if bucket_mode == BucketMode.HASH_FIXED:
- return FixedBucketRowKeyExtractor(table.table_schema)
- elif bucket_mode == BucketMode.BUCKET_UNAWARE:
- return UnawareBucketRowKeyExtractor(table.table_schema)
- elif bucket_mode == BucketMode.POSTPONE_MODE:
- return PostponeBucketRowKeyExtractor(table.table_schema)
- elif bucket_mode == BucketMode.HASH_DYNAMIC:
- return DynamicBucketRowKeyExtractor(
- table.table_schema, table=table, ignore_existing=ignore_existing)
- elif bucket_mode == BucketMode.CROSS_PARTITION:
- raise ValueError(
- "CROSS_PARTITION primary-key writes require a persistent "
- "global primary-key index, which PyPaimon does not yet support"
- )
- raise ValueError(f"Unsupported bucket mode: {bucket_mode}")
-
-
-def filter_merge_condition_batch(batch, condition, pre_rewritten=False):
- """Execute the public Python condition helper without selecting a table
writer."""
- from pypaimon.execution.python.write.merge_condition import filter_batch
- return filter_batch(batch, condition, pre_rewritten)
-
-
-def apply_merge_condition(batch, rewritten, empty_schema):
- from pypaimon.execution.python.write.merge_condition import apply_condition
- return apply_condition(batch, rewritten, empty_schema)
-
-
-def create_external_path_provider(strategy, paths, relative_bucket_path,
weights):
- from pypaimon.execution.python.write.external_path_provider import
ExternalPathProvider
- return ExternalPathProvider.create(strategy, paths, relative_bucket_path,
weights)
-
-
def execute_row_id_conflict_check(table, messages, latest_snapshot_id,
check_from_snapshot):
if table.options.native_commit_enabled():
from pypaimon.execution.native.write.native_commit import
prepare_native_row_id_conflict_check
diff --git a/paimon-python/pypaimon/write/merge_condition.py
b/paimon-python/pypaimon/write/merge_condition.py
index 884c369dfc..ae96b0f073 100644
--- a/paimon-python/pypaimon/write/merge_condition.py
+++ b/paimon-python/pypaimon/write/merge_condition.py
@@ -250,3 +250,42 @@ def _safe_literals(field, literals) -> bool:
if type_name in {'STRING', 'CHAR', 'VARCHAR'}:
return all(isinstance(literal, str) for literal in literals)
return False
+
+
+def filter_batch(
+ batch: pa.Table, condition: str, _pre_rewritten: bool = False,
+) -> pa.Table:
+ if batch.num_rows == 0:
+ return batch
+ datafusion = _load_datafusion()
+ rewritten = condition if _pre_rewritten else rewrite_condition(condition)
+ config = datafusion.SessionConfig().set(
+ "datafusion.optimizer.enable_round_robin_repartition", "false"
+ )
+ ctx = datafusion.SessionContext(config)
+ input_batches = batch.to_batches()
+ # Use one batch per partition and rebuild from partitioned batches so
+ # DataFusion neither concatenates 32-bit offsets nor reorders the input.
+ ctx.register_record_batches(
+ "_batch", [[record_batch] for record_batch in input_batches]
+ )
+ result = ctx.sql(
+ f'SELECT * FROM _batch WHERE {rewritten}'
+ )
+ output_batches = [
+ record_batch
+ for partition in result.collect_partitioned()
+ for record_batch in partition
+ ]
+ if not output_batches:
+ return batch.schema.empty_table()
+ return pa.Table.from_batches(output_batches)
+
+
+def apply_condition(
+ batch: pa.Table, rewritten: str, empty_schema: pa.Schema,
+) -> pa.Table:
+ batch = filter_batch(batch, rewritten, _pre_rewritten=True)
+ if batch.num_rows == 0:
+ return empty_schema.empty_table()
+ return batch
