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 40b011ddba [Python] Fix execution boundary regressions and restore
Native CI (#10782)
40b011ddba is described below
commit 40b011ddbab8254e1f82687aa3a09c2c24fb1545
Author: Jingsong Lee <[email protected]>
AuthorDate: Sat Oct 24 08:53:16 2026 +0800
[Python] Fix execution boundary regressions and restore Native CI (#10782)
---
.github/workflows/ci-python.yml | 1 +
.../execution/native/write/native_commit.py | 5 ++-
.../vindex/vindex_vector_global_index_reader.py | 6 +---
paimon-python/pypaimon/read/read_dispatch.py | 10 ++++--
.../pypaimon/table/source/search_dispatch.py | 42 +++++++++++++---------
.../tests/multimodal_lance_descriptor_test.py | 3 +-
.../pypaimon/tests/native_blob_update_test.py | 37 +++++++++----------
.../tests/native_map_shredding_write_test.py | 8 ++---
.../pypaimon/tests/native_nested_update_test.py | 23 ++++++------
.../tests/native_parquet_vector_update_test.py | 25 ++++++-------
.../pypaimon/tests/native_ray_index_build_test.py | 9 +++--
.../pypaimon/tests/native_row_sidecar_test.py | 2 +-
.../tests/native_update_container_casts_test.py | 25 ++++++-------
.../native_update_inactive_vector_option_test.py | 29 +++++++--------
.../pypaimon/tests/native_upsert_layout_test.py | 21 +++++------
.../pypaimon/tests/native_variant_update_test.py | 21 +++++------
.../tests/native_vector_index_build_test.py | 3 +-
paimon-python/pypaimon/tests/native_write_test.py | 2 +-
.../tests/raw_file_scan_auth_scope_test.py | 14 ++++++--
.../pypaimon/tests/ray_bucket_join_test.py | 2 +-
.../pypaimon/tests/read_execution_boundary_test.py | 6 +++-
.../pypaimon/tests/vindex_vector_index_test.py | 23 +++++++++++-
.../tests/write/execution_lifecycle_test.py | 4 +++
.../tests/write/row_id_conflict_protocol_test.py | 5 +--
24 files changed, 195 insertions(+), 131 deletions(-)
diff --git a/.github/workflows/ci-python.yml b/.github/workflows/ci-python.yml
index 1dd9742be0..580a6a2a78 100644
--- a/.github/workflows/ci-python.yml
+++ b/.github/workflows/ci-python.yml
@@ -239,6 +239,7 @@ jobs:
- name: Run Python tests with native planning, reading, writing,
updating and committing
shell: bash
env:
+ DAFT_RUNNER: 'native'
PYTEST_ADDOPTS: '--durations=20'
PYPAIMON_TEST_NATIVE_PLAN: '1'
PYPAIMON_TEST_NATIVE_READ: '1'
diff --git a/paimon-python/pypaimon/execution/native/write/native_commit.py
b/paimon-python/pypaimon/execution/native/write/native_commit.py
index adc819e1aa..9cf739cf59 100644
--- a/paimon-python/pypaimon/execution/native/write/native_commit.py
+++ b/paimon-python/pypaimon/execution/native/write/native_commit.py
@@ -204,7 +204,10 @@ def from_native_commit_messages(table, messages):
def native_commit_error(error):
"""Keep retry classification structured; never infer it from error text."""
- from pypaimon_rust import RowIdExistenceConflict as NativeRowIdConflict
+ try:
+ from pypaimon_rust import RowIdExistenceConflict as NativeRowIdConflict
+ except ImportError:
+ return error
from pypaimon.write.row_id_conflict import RowIdExistenceConflict
if not isinstance(error, NativeRowIdConflict):
diff --git
a/paimon-python/pypaimon/execution/python/globalindex/vindex/vindex_vector_global_index_reader.py
b/paimon-python/pypaimon/execution/python/globalindex/vindex/vindex_vector_global_index_reader.py
index ee97f68579..7a36decc16 100644
---
a/paimon-python/pypaimon/execution/python/globalindex/vindex/vindex_vector_global_index_reader.py
+++
b/paimon-python/pypaimon/execution/python/globalindex/vindex/vindex_vector_global_index_reader.py
@@ -59,12 +59,8 @@ class
PythonVindexVectorGlobalIndexReader(VindexReaderResources, GlobalIndexRead
params = _search_params(self._search_params_type, options, effective_k)
filter_bytes = _filter_bytes(include_row_ids)
- if n == 1:
- ids, distances = self._reader.search(
- queries[0], params, filter_bytes=filter_bytes)
- return [_result_from_scores(ids, distances, self._metadata.metric)]
-
if hasattr(self._reader, "search_batch"):
+ # The batch ABI pads short results; the scalar ABI requires
exactly K.
batch_result = self._reader.search_batch(
np.ascontiguousarray(queries),
params,
diff --git a/paimon-python/pypaimon/read/read_dispatch.py
b/paimon-python/pypaimon/read/read_dispatch.py
index a78356f2b6..f91f46c57b 100644
--- a/paimon-python/pypaimon/read/read_dispatch.py
+++ b/paimon-python/pypaimon/read/read_dispatch.py
@@ -223,6 +223,9 @@ def plan_files(scanner):
native_supported = _native_query_auth_supported(scan.table)
else:
native_supported = False
+ native_auth = native_supported and scan.table.options.query_auth_enabled
+ if not native_auth and scan._TableScan__auth_query(all_columns=True) is
not None:
+ raise TableNoPermissionException(scan.table.identifier)
if native_supported:
from pypaimon.execution.native.read.plan import NativeTableScan
result = NativeTableScan(scan).plan_files(scanner.skip_level0,
scanner.only_read_real_buckets)
@@ -231,7 +234,7 @@ def plan_files(scanner):
scanner._scanned_snapshot = None
scanner._scanned_snapshot_id = snapshot_id
return entries
- if scan._TableScan__auth_query(all_columns=True) is not None:
+ if native_auth and scan._TableScan__auth_query(all_columns=True) is not
None:
raise TableNoPermissionException(scan.table.identifier)
from pypaimon.execution.python.read.table_scan import PythonTableScan
# Let the selected executor refresh cached rules for this unrestricted
scope.
@@ -253,11 +256,14 @@ def plan_all(self) -> Plan:
native_supported = _native_query_auth_supported(self.table) and
self._native_plan_supported()
else:
native_supported = False
+ native_auth = native_supported and self.table.options.query_auth_enabled
+ if not native_auth and self._TableScan__auth_query(all_columns=True) is
not None:
+ raise TableNoPermissionException(self.table.identifier)
if native_supported:
plan = self._try_native_plan(all_files=True)
if plan is not None:
return plan
- if self._TableScan__auth_query(all_columns=True) is not None:
+ if native_auth and self._TableScan__auth_query(all_columns=True) is not
None:
raise TableNoPermissionException(self.table.identifier)
from pypaimon.execution.python.read.table_scan import PythonTableScan
return PythonTableScan(self).plan_all()
diff --git a/paimon-python/pypaimon/table/source/search_dispatch.py
b/paimon-python/pypaimon/table/source/search_dispatch.py
index 963d96ab18..c1e2cad4fa 100644
--- a/paimon-python/pypaimon/table/source/search_dispatch.py
+++ b/paimon-python/pypaimon/table/source/search_dispatch.py
@@ -21,6 +21,28 @@ The selected backend consumes ordinary snapshot-and-split
plans, including
transported plans. An execution error never triggers another search.
"""
+from functools import partial
+
+
+def _python_vector_scan(execution):
+ from pypaimon.execution.python.read.search.factories import new_vector_scan
+ return new_vector_scan(execution)
+
+
+def _python_vector_read(builder, batch):
+ from pypaimon.execution.python.read.search.factories import
new_batch_vector_read, new_vector_read
+ return new_batch_vector_read(builder) if batch else
new_vector_read(builder)
+
+
+def _python_full_text_scan(execution):
+ from pypaimon.execution.python.read.search.factories import
new_full_text_scan
+ return new_full_text_scan(execution)
+
+
+def _python_full_text_read(execution):
+ from pypaimon.execution.python.read.search.factories import
new_full_text_read
+ return new_full_text_read(execution)
+
class _SearchScan:
def __init__(self, native, python_factory):
@@ -78,10 +100,7 @@ def _native_requested(builder):
def create_vector_scan(builder, execution, batch=False):
- def python_factory():
- from pypaimon.execution.python.read.search.factories import
new_vector_scan
- return new_vector_scan(execution)
-
+ python_factory = partial(_python_vector_scan, execution)
if not _native_requested(builder):
return python_factory()
from pypaimon.execution.native.read.native_search_plan import
native_vector_scan
@@ -90,10 +109,7 @@ def create_vector_scan(builder, execution, batch=False):
def create_vector_read(builder, batch=False):
- def python_factory():
- from pypaimon.execution.python.read.search.factories import
new_batch_vector_read, new_vector_read
- return new_batch_vector_read(builder) if batch else
new_vector_read(builder)
-
+ python_factory = partial(_python_vector_read, builder, batch)
if not _native_requested(builder):
return python_factory()
from pypaimon.execution.native.read.native_search_plan import
native_vector_read
@@ -105,10 +121,7 @@ def create_vector_read(builder, batch=False):
def create_full_text_scan(builder, execution):
- def python_factory():
- from pypaimon.execution.python.read.search.factories import
new_full_text_scan
- return new_full_text_scan(execution)
-
+ python_factory = partial(_python_full_text_scan, execution)
if not _native_requested(builder):
return python_factory()
from pypaimon.execution.native.read.native_search_plan import
native_full_text_scan
@@ -117,10 +130,7 @@ def create_full_text_scan(builder, execution):
def create_full_text_read(builder, execution):
- def python_factory():
- from pypaimon.execution.python.read.search.factories import
new_full_text_read
- return new_full_text_read(execution)
-
+ python_factory = partial(_python_full_text_read, execution)
if not _native_requested(builder):
return python_factory()
from pypaimon.execution.native.read.native_search_plan import
native_full_text_read
diff --git a/paimon-python/pypaimon/tests/multimodal_lance_descriptor_test.py
b/paimon-python/pypaimon/tests/multimodal_lance_descriptor_test.py
index 46c508406e..5ea0948419 100644
--- a/paimon-python/pypaimon/tests/multimodal_lance_descriptor_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_lance_descriptor_test.py
@@ -119,9 +119,10 @@ def
test_export_lance_rejects_video_frame_in_ordinary_descriptor_field(conn, tmp
video = tmp_path / "video.bin"
video.write_bytes(b"encoded-video")
descriptor = VideoFrameDescriptor(str(video), 0, video.stat().st_size, 2,
-1, 0)
+ # Seed the invalid legacy value through Python; Native rejects it before
export.
table = lance_tests._table(conn, "frame_reference", {"id": pa.int32(),
"payload": pa.large_binary()}, [
{"id": 1, "payload": descriptor},
- ], options={"blob-descriptor-field": "payload"})
+ ], options={"blob-descriptor-field": "payload", "write.native.enabled":
"false"})
path = tmp_path / "frame_reference.lance"
if mode == "overwrite":
lance.write_dataset(pa.table({"id": [99]}), str(path))
diff --git a/paimon-python/pypaimon/tests/native_blob_update_test.py
b/paimon-python/pypaimon/tests/native_blob_update_test.py
index 19e2968f2b..b434fe5015 100644
--- a/paimon-python/pypaimon/tests/native_blob_update_test.py
+++ b/paimon-python/pypaimon/tests/native_blob_update_test.py
@@ -18,7 +18,9 @@
"""Native updates share the Java Blob placeholder and row-id layout."""
from pypaimon.execution.python.write.table_update_by_row_id import
PythonTableUpdateByRowId
-from contextlib import ExitStack
+from pypaimon.execution.python.write.table_update import PythonTableUpdate
+from pypaimon.execution.python.write.table_upsert_by_key import
PythonTableUpsert
+from contextlib import ExitStack, contextmanager
from unittest.mock import patch
from urllib.parse import urlparse
@@ -31,11 +33,8 @@ from pypaimon.schema.data_types import ArrayType,
AtomicType, DataField, MapType
from pypaimon.table.row.blob import Blob, BlobDescriptor, BlobViewStruct
from pypaimon.table.row.generic_row import GenericRow
from pypaimon.table.data_evolution_merge_into import WhenMatched,
WhenNotMatched, source_col
-from pypaimon.write.table_delete import TableDeleteByRowId
from pypaimon.execution.native.write.native_update import
NativeTableUpdateByRowId
from pypaimon.write.write_dispatch import NativeTableWrite
-from pypaimon.write.table_update import TableUpdate
-from pypaimon.write.table_upsert_by_key import TableUpsertByKey
pytestmark = pytest.mark.native_plan
@@ -85,15 +84,17 @@ def _read(table, descriptors=False, projection=None,
resolve_views=True):
return read.to_arrow(splits).sort_by('id').to_pylist()
+@contextmanager
def _native_only():
- stack = ExitStack()
- for cls, method in [(PythonTableUpdateByRowId,
'_load_existing_files_info'),
- (TableUpdate, '_build_predicate_update_table'),
- (TableUpsertByKey, '_upsert_partition'),
- (TableUpsertByKey, '_upsert_row_partition'),
- (TableDeleteByRowId, 'delete')]:
- stack.enter_context(patch.object(cls, method,
side_effect=AssertionError('Python update fallback')))
- return stack
+ with ExitStack() as stack:
+ for cls, method in [(PythonTableUpdateByRowId,
'_load_existing_files_info'),
+ (PythonTableUpdate,
'_build_predicate_update_table'),
+ (PythonTableUpsert, '_upsert_partition'),
+ (PythonTableUpsert, '_upsert_row_partition'),
+ (PythonTableUpdate, 'delete'),
+ (PythonTableUpdate, 'delete_by_row_id')]:
+ stack.enter_context(patch.object(cls, method,
side_effect=AssertionError('Python update fallback')))
+ yield
def _commit(builder, messages, stream):
@@ -238,13 +239,13 @@ def test_native_inline_blob_updates_preserve_references(
else update.update_by_arrow_with_row_id(data))
elif operation == 'incremental':
updater = update.new_update_by_row_id(7) if stream else
update.new_update_by_row_id()
- assert isinstance(updater, NativeTableUpdateByRowId) == native
+ assert isinstance(updater._execution, NativeTableUpdateByRowId) ==
native
updater.update_columns(data.slice(0, 1), ['payload'])
updater.update_columns(data.slice(1), ['payload'])
messages = updater.commit_messages
elif operation in ('row', 'rows'):
updater = update.new_update_by_row_id(7) if stream else
update.new_update_by_row_id()
- assert isinstance(updater, NativeTableUpdateByRowId) == native
+ assert isinstance(updater._execution, NativeTableUpdateByRowId) ==
native
value = (Blob.from_view(BlobViewStruct.deserialize(replacement))
if option == 'blob-view-field'
else Blob.from_descriptor_bytes(replacement,
file_io=table.file_io))
rows = [GenericRow([1, 'a', 999, value], table.fields),
@@ -347,13 +348,13 @@ def
test_native_raw_blob_updates_across_existing_update_apis(tmp_path, kind, str
messages = update.update_by_arrow_batches_with_row_id(batches)
elif operation == 'incremental':
updater = update.new_update_by_row_id(7) if stream else
update.new_update_by_row_id()
- assert isinstance(updater, NativeTableUpdateByRowId)
+ assert isinstance(updater._execution, NativeTableUpdateByRowId)
updater.update_columns(data.slice(0, 1), ['payload'])
updater.update_columns(data.slice(1), ['payload'])
messages = updater.commit_messages
elif operation in ('row', 'rows'):
updater = update.new_update_by_row_id(7) if stream else
update.new_update_by_row_id()
- assert isinstance(updater, NativeTableUpdateByRowId)
+ assert isinstance(updater._execution, NativeTableUpdateByRowId)
rows = [GenericRow([1, 'a', 999, replacement], table.fields),
GenericRow([3, 'b', 999, None], table.fields)]
if operation == 'row':
@@ -499,8 +500,8 @@ def test_raw_blob_abort_preserves_prepared_files(tmp_path,
published):
paths = set(tmp_path.rglob('*.blob'))
if published:
_commit(builder, messages, False)
- updater.writer._abort()
- updater.writer._abort()
+ updater._execution.writer._abort()
+ updater._execution.writer._abort()
assert set(tmp_path.rglob('*.blob')) == paths
if not published:
assert _read(table)[0]['payload'] == b'large' * 10
diff --git a/paimon-python/pypaimon/tests/native_map_shredding_write_test.py
b/paimon-python/pypaimon/tests/native_map_shredding_write_test.py
index df647e6361..52fe2f5ea2 100644
--- a/paimon-python/pypaimon/tests/native_map_shredding_write_test.py
+++ b/paimon-python/pypaimon/tests/native_map_shredding_write_test.py
@@ -278,8 +278,8 @@ def test_map_row_id_update_keeps_physical_layout(tmp_path,
native_update):
def test_native_map_predicate_update_and_upsert(tmp_path, stream, operation):
from contextlib import ExitStack
from pypaimon.common.predicate_builder import PredicateBuilder
- from pypaimon.write.table_update import TableUpdate
- from pypaimon.write.table_upsert_by_key import TableUpsertByKey
+ from pypaimon.execution.python.write.table_update import PythonTableUpdate
+ from pypaimon.execution.python.write.table_upsert_by_key import
PythonTableUpsert
table, schema = _table(tmp_path, 'evolution')
_write(table, schema, [
@@ -291,8 +291,8 @@ def test_native_map_predicate_update_and_upsert(tmp_path,
stream, operation):
replacement = [('new', 7), ('a', None)]
with ExitStack() as stack:
for cls, method in [(PythonTableUpdateByRowId,
'_load_existing_files_info'),
- (TableUpdate, '_build_predicate_update_table'),
- (TableUpsertByKey, '_upsert_partition')]:
+ (PythonTableUpdate,
'_build_predicate_update_table'),
+ (PythonTableUpsert, '_upsert_partition')]:
stack.enter_context(patch.object(cls, method,
side_effect=AssertionError('Python update fallback')))
if operation == 'upsert':
data = pa.Table.from_pylist([
diff --git a/paimon-python/pypaimon/tests/native_nested_update_test.py
b/paimon-python/pypaimon/tests/native_nested_update_test.py
index ed5df8a02d..1eea346dc0 100644
--- a/paimon-python/pypaimon/tests/native_nested_update_test.py
+++ b/paimon-python/pypaimon/tests/native_nested_update_test.py
@@ -18,7 +18,7 @@
"""Nested Native update dispatch and parity with the Python implementation."""
from pypaimon.execution.python.write.table_update_by_row_id import
PythonTableUpdateByRowId
-from contextlib import ExitStack
+from contextlib import ExitStack, contextmanager
from unittest.mock import patch
import pyarrow as pa
@@ -27,8 +27,8 @@ import pytest
from pypaimon import CatalogFactory, Schema
from pypaimon.common.predicate_builder import PredicateBuilder
from pypaimon.table.row.generic_row import GenericRow
-from pypaimon.write.table_update import TableUpdate
-from pypaimon.write.table_upsert_by_key import TableUpsertByKey
+from pypaimon.execution.python.write.table_update import PythonTableUpdate
+from pypaimon.execution.python.write.table_upsert_by_key import
PythonTableUpsert
pytestmark = pytest.mark.native_plan
@@ -66,15 +66,16 @@ def _table(tmp_path, data_type, native):
return table, schema
+@contextmanager
def _native_only(native):
- stack = ExitStack()
- if native:
- for cls, method in [(PythonTableUpdateByRowId,
'_load_existing_files_info'),
- (TableUpdate, '_build_predicate_update_table'),
- (TableUpsertByKey, '_upsert_partition'),
- (TableUpsertByKey, '_upsert_row_partition')]:
- stack.enter_context(patch.object(cls, method,
side_effect=AssertionError('Python fallback')))
- return stack
+ with ExitStack() as stack:
+ if native:
+ for cls, method in [(PythonTableUpdateByRowId,
'_load_existing_files_info'),
+ (PythonTableUpdate,
'_build_predicate_update_table'),
+ (PythonTableUpsert, '_upsert_partition'),
+ (PythonTableUpsert, '_upsert_row_partition')]:
+ stack.enter_context(patch.object(cls, method,
side_effect=AssertionError('Python fallback')))
+ yield
def _commit(builder, messages, stream):
diff --git a/paimon-python/pypaimon/tests/native_parquet_vector_update_test.py
b/paimon-python/pypaimon/tests/native_parquet_vector_update_test.py
index 06f3c8dc7d..43f1bf4227 100644
--- a/paimon-python/pypaimon/tests/native_parquet_vector_update_test.py
+++ b/paimon-python/pypaimon/tests/native_parquet_vector_update_test.py
@@ -18,7 +18,7 @@
"""Parquet vector sidecars must not force row-level updates back to Python."""
from pypaimon.execution.python.write.table_update_by_row_id import
PythonTableUpdateByRowId
-from contextlib import ExitStack
+from contextlib import ExitStack, contextmanager
from unittest.mock import patch
import pyarrow as pa
@@ -29,8 +29,8 @@ from pypaimon import CatalogFactory, Schema
from pypaimon.execution.native.read.table_read import (native_read as
read_native_splits)
from pypaimon.schema.data_types import PyarrowFieldParser
from pypaimon.table.data_evolution_merge_into import WhenMatched,
WhenNotMatched, source_col
-from pypaimon.write.table_update import TableUpdate
-from pypaimon.write.table_upsert_by_key import TableUpsertByKey
+from pypaimon.execution.python.write.table_update import PythonTableUpdate
+from pypaimon.execution.python.write.table_upsert_by_key import
PythonTableUpsert
pytestmark = pytest.mark.native_plan
@@ -73,16 +73,17 @@ def _rows(table):
yield sorted(rows, key=lambda row: row['id'])
+@contextmanager
def _native_only(native):
- stack = ExitStack()
- if native:
- for owner, method in [(PythonTableUpdateByRowId,
'_load_existing_files_info'),
- (TableUpsertByKey, '_upsert_partition'),
- (TableUpdate, '_build_predicate_update_table')]:
- stack.enter_context(patch.object(owner, method,
side_effect=AssertionError('Python update fallback')))
-
stack.enter_context(patch('pypaimon.execution.python.write.merge_into._build_tables',
- side_effect=AssertionError('Python merge
fallback')))
- return stack
+ with ExitStack() as stack:
+ if native:
+ for owner, method in [(PythonTableUpdateByRowId,
'_load_existing_files_info'),
+ (PythonTableUpsert, '_upsert_partition'),
+ (PythonTableUpdate,
'_build_predicate_update_table')]:
+ stack.enter_context(patch.object(owner, method,
side_effect=AssertionError('Python update fallback')))
+
stack.enter_context(patch('pypaimon.execution.python.write.merge_into._build_tables',
+ side_effect=AssertionError('Python merge
fallback')))
+ yield
def _commit(builder, messages, stream):
diff --git a/paimon-python/pypaimon/tests/native_ray_index_build_test.py
b/paimon-python/pypaimon/tests/native_ray_index_build_test.py
index 5f6fd1b626..d7de610781 100644
--- a/paimon-python/pypaimon/tests/native_ray_index_build_test.py
+++ b/paimon-python/pypaimon/tests/native_ray_index_build_test.py
@@ -96,7 +96,7 @@ def _request(tmp_path):
@pytest.fixture
-def native_protocol():
+def native_protocol(monkeypatch):
module = 'pypaimon.execution.native.write.native_index_build'
core = Mock()
shards = [SimpleNamespace(ordinal=ordinal) for ordinal in range(3)]
@@ -106,8 +106,11 @@ def native_protocol():
sdk = ModuleType('pypaimon_rust.datafusion')
sdk.GlobalIndexBuildBuilder = SimpleNamespace(supports_index_type=lambda
kind: True)
scheduler = _Ray()
- with patch.dict(sys.modules, {'pypaimon_rust.datafusion': sdk, 'ray':
scheduler}), \
- patch(module + '.native_runtime_available', return_value=True), \
+ # Retain lazy imports between cases instead of restoring all of
sys.modules,
+ # which leaves stale package attributes for Python 3.10's mock resolver.
+ monkeypatch.setitem(sys.modules, 'pypaimon_rust.datafusion', sdk)
+ monkeypatch.setitem(sys.modules, 'ray', scheduler)
+ with patch(module + '.native_runtime_available', return_value=True), \
patch(module + '._rest_catalog_supported', return_value=True), \
patch(module + '._native_table', return_value=native_table), \
patch(module + '.from_native_commit_messages', side_effect=lambda
table, messages: messages), \
diff --git a/paimon-python/pypaimon/tests/native_row_sidecar_test.py
b/paimon-python/pypaimon/tests/native_row_sidecar_test.py
index 7e91ccea0d..5f514b9cf8 100644
--- a/paimon-python/pypaimon/tests/native_row_sidecar_test.py
+++ b/paimon-python/pypaimon/tests/native_row_sidecar_test.py
@@ -203,7 +203,7 @@ def
test_sidecars_keep_updates_native_and_apply_deletion_vectors(tmp_path, strea
data = pa.table({'_ROW_ID': [5, 7], 'value': ['changed-5', 'changed-7']})
if incremental:
updater = update.new_update_by_row_id(1) if stream else
update.new_update_by_row_id()
- assert isinstance(updater, NativeTableUpdateByRowId)
+ assert isinstance(updater._execution, NativeTableUpdateByRowId)
messages = updater.update_columns(data, ['value'])
else:
with patch.object(TableUpdateByRowId, 'update_columns',
diff --git a/paimon-python/pypaimon/tests/native_update_container_casts_test.py
b/paimon-python/pypaimon/tests/native_update_container_casts_test.py
index 3b2be5e984..fc043a9cf2 100644
--- a/paimon-python/pypaimon/tests/native_update_container_casts_test.py
+++ b/paimon-python/pypaimon/tests/native_update_container_casts_test.py
@@ -18,7 +18,7 @@
"""Equivalent Arrow container inputs must work on the existing update APIs."""
from pypaimon.execution.python.write.table_update_by_row_id import
PythonTableUpdateByRowId
-from contextlib import ExitStack
+from contextlib import ExitStack, contextmanager
from decimal import Decimal
from unittest.mock import patch
@@ -28,8 +28,8 @@ import pytest
from pypaimon import CatalogFactory, Schema
from pypaimon.common.predicate_builder import PredicateBuilder
from pypaimon.table.data_evolution_merge_into import WhenMatched, source_col
-from pypaimon.write.table_update import TableUpdate
-from pypaimon.write.table_upsert_by_key import TableUpsertByKey
+from pypaimon.execution.python.write.table_update import PythonTableUpdate
+from pypaimon.execution.python.write.table_upsert_by_key import
PythonTableUpsert
pytestmark = pytest.mark.native_plan
@@ -92,16 +92,17 @@ def _table(tmp_path, value_type, native):
return table
+@contextmanager
def _native_only(native):
- stack = ExitStack()
- if native:
- for cls, method in [(PythonTableUpdateByRowId,
'_load_existing_files_info'),
- (TableUpdate, '_build_predicate_update_table'),
- (TableUpsertByKey, '_upsert_partition')]:
- stack.enter_context(patch.object(cls, method,
side_effect=AssertionError('Python update fallback')))
-
stack.enter_context(patch('pypaimon.execution.python.write.merge_into._build_tables',
- side_effect=AssertionError('Python MERGE
fallback')))
- return stack
+ with ExitStack() as stack:
+ if native:
+ for cls, method in [(PythonTableUpdateByRowId,
'_load_existing_files_info'),
+ (PythonTableUpdate,
'_build_predicate_update_table'),
+ (PythonTableUpsert, '_upsert_partition')]:
+ stack.enter_context(patch.object(cls, method,
side_effect=AssertionError('Python update fallback')))
+
stack.enter_context(patch('pypaimon.execution.python.write.merge_into._build_tables',
+ side_effect=AssertionError('Python MERGE
fallback')))
+ yield
def _apply(table, source, native, stream, operation):
diff --git
a/paimon-python/pypaimon/tests/native_update_inactive_vector_option_test.py
b/paimon-python/pypaimon/tests/native_update_inactive_vector_option_test.py
index 083845654f..e12920c829 100644
--- a/paimon-python/pypaimon/tests/native_update_inactive_vector_option_test.py
+++ b/paimon-python/pypaimon/tests/native_update_inactive_vector_option_test.py
@@ -19,7 +19,7 @@
from pypaimon.execution.python.write.table_update_by_row_id import
PythonTableUpdateByRowId
from pypaimon.execution.python.write.table_delete import
PythonTableDeleteByRowId
-from contextlib import ExitStack
+from contextlib import ExitStack, contextmanager
from unittest.mock import patch
import pyarrow as pa
@@ -32,8 +32,8 @@ from pypaimon.table.data_evolution_merge_into import
WhenMatched, WhenNotMatched
from pypaimon.table.row.generic_row import GenericRow
from pypaimon.tests import native_plan_rest_test
from pypaimon.execution.native.write.native_update import
_native_update_formats_supported
-from pypaimon.write.table_update import TableUpdate
-from pypaimon.write.table_upsert_by_key import TableUpsertByKey
+from pypaimon.execution.python.write.table_update import PythonTableUpdate
+from pypaimon.execution.python.write.table_upsert_by_key import
PythonTableUpsert
rest_catalog = native_plan_rest_test.rest_catalog
@@ -80,18 +80,19 @@ def _read(table):
return
builder.new_read().to_arrow(builder.new_scan().plan().splits()).sort_by('id').to_pylist()
+@contextmanager
def _native_only(native):
- stack = ExitStack()
- if native:
- for owner, method in [(PythonTableUpdateByRowId,
'_load_existing_files_info'),
- (TableUpsertByKey, '_upsert_partition'),
- (TableUpsertByKey, '_upsert_row_partition'),
- (TableUpdate, '_build_predicate_update_table'),
- (PythonTableDeleteByRowId,
'_scan_anchor_ranges')]:
- stack.enter_context(patch.object(owner, method,
side_effect=AssertionError('Python mutation fallback')))
-
stack.enter_context(patch('pypaimon.execution.python.write.merge_into._build_tables',
- side_effect=AssertionError('Python MERGE
fallback')))
- return stack
+ with ExitStack() as stack:
+ if native:
+ for owner, method in [(PythonTableUpdateByRowId,
'_load_existing_files_info'),
+ (PythonTableUpsert, '_upsert_partition'),
+ (PythonTableUpsert, '_upsert_row_partition'),
+ (PythonTableUpdate,
'_build_predicate_update_table'),
+ (PythonTableDeleteByRowId,
'_scan_anchor_ranges')]:
+ stack.enter_context(patch.object(owner, method,
side_effect=AssertionError('Python mutation fallback')))
+
stack.enter_context(patch('pypaimon.execution.python.write.merge_into._build_tables',
+ side_effect=AssertionError('Python MERGE
fallback')))
+ yield
@pytest.mark.parametrize('native', _ENGINES, ids=['python', 'native'])
diff --git a/paimon-python/pypaimon/tests/native_upsert_layout_test.py
b/paimon-python/pypaimon/tests/native_upsert_layout_test.py
index 223fda6079..189080d8a9 100644
--- a/paimon-python/pypaimon/tests/native_upsert_layout_test.py
+++ b/paimon-python/pypaimon/tests/native_upsert_layout_test.py
@@ -18,7 +18,7 @@
"""Equivalent Arrow layouts must not send upsert matching back to Python."""
from pypaimon.execution.python.write.table_update_by_row_id import
PythonTableUpdateByRowId
-from contextlib import ExitStack
+from contextlib import ExitStack, contextmanager
from unittest.mock import patch
import pyarrow as pa
@@ -26,8 +26,8 @@ import pytest
from pypaimon import CatalogFactory, Schema
from pypaimon.schema.data_types import PyarrowFieldParser
-from pypaimon.write.table_update import TableUpdate
-from pypaimon.write.table_upsert_by_key import TableUpsertByKey
+from pypaimon.execution.python.write.table_update import PythonTableUpdate
+from pypaimon.execution.python.write.table_upsert_by_key import
PythonTableUpsert
pytestmark = pytest.mark.native_plan
@@ -76,14 +76,15 @@ def _seed(table, schema, keys, values):
writer.close()
+@contextmanager
def _native_only(native):
- stack = ExitStack()
- if native:
- for owner, method in [(TableUpsertByKey, '_upsert_partition'),
- (PythonTableUpdateByRowId,
'_load_existing_files_info'),
- (TableUpdate, '_build_predicate_update_table')]:
- stack.enter_context(patch.object(owner, method,
side_effect=AssertionError('Python update fallback')))
- return stack
+ with ExitStack() as stack:
+ if native:
+ for owner, method in [(PythonTableUpsert, '_upsert_partition'),
+ (PythonTableUpdateByRowId,
'_load_existing_files_info'),
+ (PythonTableUpdate,
'_build_predicate_update_table')]:
+ stack.enter_context(patch.object(owner, method,
side_effect=AssertionError('Python update fallback')))
+ yield
def _check(table, values, key, new_key='new'):
diff --git a/paimon-python/pypaimon/tests/native_variant_update_test.py
b/paimon-python/pypaimon/tests/native_variant_update_test.py
index 0335cf9956..6f0f5497c5 100644
--- a/paimon-python/pypaimon/tests/native_variant_update_test.py
+++ b/paimon-python/pypaimon/tests/native_variant_update_test.py
@@ -19,7 +19,7 @@
from pypaimon.execution.python.write.table_update_by_row_id import
PythonTableUpdateByRowId
import json
-from contextlib import ExitStack
+from contextlib import ExitStack, contextmanager
from unittest.mock import patch
import pyarrow as pa
@@ -32,8 +32,8 @@ from pypaimon.data.generic_variant import GenericVariant
from pypaimon.schema.data_types import PyarrowFieldParser
from pypaimon.execution.native.write.native_update import
NativeTableUpdateByRowId
from pypaimon.write.write_dispatch import NativeTableWrite
-from pypaimon.write.table_update import TableUpdate
-from pypaimon.write.table_upsert_by_key import TableUpsertByKey
+from pypaimon.execution.python.write.table_update import PythonTableUpdate
+from pypaimon.execution.python.write.table_upsert_by_key import
PythonTableUpsert
pytestmark = pytest.mark.native_plan
@@ -104,13 +104,14 @@ def _read(table, native, projection=None):
return rows
+@contextmanager
def _native_only():
- stack = ExitStack()
- for cls, method in [(PythonTableUpdateByRowId,
'_load_existing_files_info'),
- (TableUpdate, '_build_predicate_update_table'),
- (TableUpsertByKey, '_upsert_partition')]:
- stack.enter_context(patch.object(cls, method,
side_effect=AssertionError('Python update fallback')))
- return stack
+ with ExitStack() as stack:
+ for cls, method in [(PythonTableUpdateByRowId,
'_load_existing_files_info'),
+ (PythonTableUpdate,
'_build_predicate_update_table'),
+ (PythonTableUpsert, '_upsert_partition')]:
+ stack.enter_context(patch.object(cls, method,
side_effect=AssertionError('Python update fallback')))
+ yield
def test_python_configured_variant_schema_uses_java_semantics(tmp_path):
@@ -145,7 +146,7 @@ def test_native_variant_partial_updates(tmp_path, stream,
key, enabled, operatio
else update.update_by_arrow_with_row_id(data))
elif operation == 'incremental':
updater = update.new_update_by_row_id(7) if stream else
update.new_update_by_row_id()
- assert isinstance(updater, NativeTableUpdateByRowId)
+ assert isinstance(updater._execution, NativeTableUpdateByRowId)
updater.update_columns(data.slice(0, 1), ['payload'])
updater.update_columns(data.slice(1), ['payload'])
messages = updater.commit_messages
diff --git a/paimon-python/pypaimon/tests/native_vector_index_build_test.py
b/paimon-python/pypaimon/tests/native_vector_index_build_test.py
index 04fd3e5d98..5392d27ad0 100644
--- a/paimon-python/pypaimon/tests/native_vector_index_build_test.py
+++ b/paimon-python/pypaimon/tests/native_vector_index_build_test.py
@@ -137,8 +137,7 @@ def
test_native_search_filters_live_null_vectors_with_deletion_vectors(rest_cata
result =
(_native_table(table).new_vector_search_builder().with_vector_column('embedding')
.with_query_vector([1.,
1.]).with_limit(64).with_options(options).execute())
assert sorted(result.row_ids()) == [1, 4]
- # Python table-level planning also uses the core low-level index reader,
- # avoiding the third-party scalar FFI's fixed-result-length restriction.
+ # Python table-level reads use the batch ABI for short filtered results.
_assert_ids(table, [1, 4], kind)
diff --git a/paimon-python/pypaimon/tests/native_write_test.py
b/paimon-python/pypaimon/tests/native_write_test.py
index 48d3b7765c..ecc2f1420f 100644
--- a/paimon-python/pypaimon/tests/native_write_test.py
+++ b/paimon-python/pypaimon/tests/native_write_test.py
@@ -528,7 +528,7 @@ def
test_uri_reader_factory_none_keeps_native_and_rejects_late_change(tmp_path):
native.write_arrow_batch(_batch([2], ['a']))
with pytest.raises(RuntimeError, match='before any write'):
native.with_blob_uri_reader_factory(None)
- with pytest.raises(ValueError, match='before any data is written'):
+ with pytest.raises(RuntimeError, match='before any write'):
native.with_write_type(['id'])
native.abort()
diff --git a/paimon-python/pypaimon/tests/raw_file_scan_auth_scope_test.py
b/paimon-python/pypaimon/tests/raw_file_scan_auth_scope_test.py
index d1cbe980f8..83566f6e67 100644
--- a/paimon-python/pypaimon/tests/raw_file_scan_auth_scope_test.py
+++ b/paimon-python/pypaimon/tests/raw_file_scan_auth_scope_test.py
@@ -92,14 +92,22 @@ def test_raw_scope_clears_cached_query_rules(table,
restriction, native_fallback
assert native.call_count == int(native_fallback)
-def test_restricted_raw_scope_does_not_clear_cached_rules(table):
[email protected]('operation', ['raw-files', 'all-files'])
+def test_restricted_raw_scope_does_not_clear_cached_rules(table, operation):
scan = table.new_read_builder().new_scan()
scan._query_auth_fn = lambda select:
TableQueryAuthResult([_partition_filter()], None)
scan.plan()
source = scan._python_file_scanner
prior_partition_filter = source.auth_partition_predicate
assert prior_partition_filter is not None
- with patch.object(source, 'plan_files',
side_effect=AssertionError('Unauthorized file access')):
+ plan = scan.file_scanner.plan_files if operation == 'raw-files' else
scan.plan_all
+ with patch.object(source, 'plan_files',
side_effect=AssertionError('Unauthorized file access')), \
+ patch.object(NativeTableScan, 'plan_files',
+ side_effect=AssertionError('Unauthorized Native file
access')) as native, \
+ patch.object(scan, '_try_native_plan',
+ side_effect=AssertionError('Unauthorized Native
planning')) as native_plan:
with pytest.raises(TableNoPermissionException):
- scan.file_scanner.plan_files()
+ plan()
+ native.assert_not_called()
+ native_plan.assert_not_called()
assert source.auth_partition_predicate is prior_partition_filter
diff --git a/paimon-python/pypaimon/tests/ray_bucket_join_test.py
b/paimon-python/pypaimon/tests/ray_bucket_join_test.py
index 292712361d..115be8aae5 100644
--- a/paimon-python/pypaimon/tests/ray_bucket_join_test.py
+++ b/paimon-python/pypaimon/tests/ray_bucket_join_test.py
@@ -309,7 +309,7 @@ class RayBucketJoinTest(unittest.TestCase):
# Files left under an old bucket count (rescale not yet rewritten)
would carry a
# different total_buckets; the same bucket id must not be treated as
co-located.
import types
- from pypaimon.execution.python.read.scanner.file_scanner import
(FileScanner)
+ from pypaimon.read.file_scanner import FileScanner
sch = pa.schema([("url", pa.string())])
self._bucketed_table("default.rs_a", sch, "url",
pa.Table.from_pydict({"url": ["u0", "u1"]},
schema=sch))
diff --git a/paimon-python/pypaimon/tests/read_execution_boundary_test.py
b/paimon-python/pypaimon/tests/read_execution_boundary_test.py
index 4f1236e7bc..878a2b05f8 100644
--- a/paimon-python/pypaimon/tests/read_execution_boundary_test.py
+++ b/paimon-python/pypaimon/tests/read_execution_boundary_test.py
@@ -17,6 +17,7 @@
"""Read entry points load only the selected execution backend."""
+import importlib
import subprocess
import sys
import textwrap
@@ -259,7 +260,8 @@ def
test_native_row_conversion_keeps_its_failure_when_closing_batches_also_fails
read.output_projection = None
read._output_arrow_schema = lambda splits: pa.schema([('id', pa.int32())])
read._try_native_batches = Mock(return_value=source)
- with patch('pypaimon.execution.native.read.row_iterator.native_rows',
side_effect=primary), \
+ row_iterator =
importlib.import_module('pypaimon.execution.native.read.row_iterator')
+ with patch.object(row_iterator, 'native_rows', side_effect=primary), \
patch.dict(sys.modules,
{'pypaimon.execution.python.read.table_read': None}):
rows = read._try_native_rows([])
with pytest.raises(ValueError) as caught:
@@ -272,6 +274,7 @@ def _python_read_source(primary=None, cleanup=None):
from types import SimpleNamespace
import pyarrow as pa
from pypaimon.read.reader.iface.record_batch_reader import
RecordBatchReader
+ from pypaimon.read.read_support import _try_to_pad_batch_by_schema
from pypaimon.schema.data_types import AtomicType, DataField
from pypaimon.table.row.offset_row import OffsetRow
@@ -310,6 +313,7 @@ def _python_read_source(primary=None, cleanup=None):
_blob_as_descriptor=False,
_blob_view_output_indices=(),
_project_batch_to_output=lambda batch: batch,
+ _try_to_pad_batch_by_schema=_try_to_pad_batch_by_schema,
output_projection=None,
_adapter_read_type=[DataField(0, 'id', AtomicType('INT'))],
include_row_kind=False,
diff --git a/paimon-python/pypaimon/tests/vindex_vector_index_test.py
b/paimon-python/pypaimon/tests/vindex_vector_index_test.py
index c2d67bec7d..46a83ae283 100644
--- a/paimon-python/pypaimon/tests/vindex_vector_index_test.py
+++ b/paimon-python/pypaimon/tests/vindex_vector_index_test.py
@@ -19,17 +19,21 @@ import io
import sys
import types
import unittest
-from unittest.mock import patch
+from unittest.mock import Mock, patch
+
+import numpy as np
from pypaimon.globalindex.batch_vector_search import BatchVectorSearch
from pypaimon.globalindex.global_index_meta import GlobalIndexIOMeta
from pypaimon.globalindex.vector_search_result import
DictBasedScoredIndexResult
+from pypaimon.globalindex.vector_search import VectorSearch
from pypaimon.globalindex.vindex_reader import (
VindexVectorGlobalIndexReader,
)
from
pypaimon.execution.python.globalindex.vindex.vindex_vector_global_index_reader
import (
_build_scores,
_search_params,
+ PythonVindexVectorGlobalIndexReader,
)
from pypaimon.utils.roaring_bitmap import RoaringBitmap64
@@ -113,6 +117,23 @@ class _BytesFileIO:
class VindexVectorIndexTest(unittest.TestCase):
+ def test_single_query_preserves_short_filtered_results(self):
+ reader = PythonVindexVectorGlobalIndexReader(
+ _BytesFileIO(), '/tmp', [GlobalIndexIOMeta('index', 1)])
+ reader._metadata = _FakeMetadata()
+ reader._search_params_type = _FakeSearchParams
+ reader._reader = Mock()
+ reader._reader.search.side_effect = RuntimeError('scalar ABI requires
exactly K results')
+ reader._reader.search_batch.return_value = (
+ np.array([[10, 11, -1]]), np.array([[0., 1., np.nan]]))
+ with patch.object(reader, '_ensure_loaded'):
+ result = reader.visit_vector_search(VectorSearch([1., 0.], 3,
'embedding')).result()
+ self.assertEqual([10, 11], result.results().to_list())
+ self.assertEqual(1., result.score_getter()(10))
+ self.assertEqual(.5, result.score_getter()(11))
+ reader._reader.search.assert_not_called()
+ reader.close()
+
def test_query_widths_match_java_integer_text_and_bounds(self):
for key in ('ivf.nprobe', 'diskann.l_search'):
for value, expected in (('+12', 12), ('１２', 12), ('٢', 2),
('2147483647', 2147483647), (12, 12)):
diff --git a/paimon-python/pypaimon/tests/write/execution_lifecycle_test.py
b/paimon-python/pypaimon/tests/write/execution_lifecycle_test.py
index 1c3938cfb5..3272eab174 100644
--- a/paimon-python/pypaimon/tests/write/execution_lifecycle_test.py
+++ b/paimon-python/pypaimon/tests/write/execution_lifecycle_test.py
@@ -267,7 +267,10 @@ def
test_native_merge_does_not_construct_python_writer_or_file_index(tmp_path):
table = _data_evolution_table(tmp_path)
native_table, builder = _core_builder()
expected = [_message()]
+ encoded_clause = object()
with
patch('pypaimon.execution.native.write.native_merge_into._native_row_id_table',
return_value=native_table), \
+
patch('pypaimon.execution.native.write.native_merge_into.encode_merge_clause',
+ return_value=encoded_clause), \
patch('pypaimon.execution.native.write.native_merge_into.from_native_commit_messages',
return_value=expected), \
patch('pypaimon.execution.python.write.table_write.PythonTableWrite.__init__',
@@ -279,6 +282,7 @@ def
test_native_merge_does_not_construct_python_writer_or_file_index(tmp_path):
when_matched=[WhenMatched.update({'value': source_col('value')})])
assert result is expected
builder.new_update.return_value.merge_into.assert_called_once()
+ assert
builder.new_update.return_value.merge_into.call_args[1]['when_matched'] ==
[encoded_clause]
def
test_failed_python_construction_closes_snapshot_and_preserves_original_error():
diff --git
a/paimon-python/pypaimon/tests/write/row_id_conflict_protocol_test.py
b/paimon-python/pypaimon/tests/write/row_id_conflict_protocol_test.py
index 091cc57d20..4c22822587 100644
--- a/paimon-python/pypaimon/tests/write/row_id_conflict_protocol_test.py
+++ b/paimon-python/pypaimon/tests/write/row_id_conflict_protocol_test.py
@@ -91,7 +91,8 @@ def
test_native_commit_translates_only_typed_row_id_conflicts(batch):
@pytest.mark.parametrize('opaque_message', ['unknown commit result', 'Row ID
existence conflict: not a typed error'])
-def test_native_unknown_outcome_is_not_a_rewrite_signal(opaque_message):
[email protected]('native_available', [False, True])
+def test_native_unknown_outcome_is_not_a_rewrite_signal(opaque_message,
native_available):
from pypaimon.execution.native.write.commit import NativeTableCommit
from pypaimon.ray.row_id_conflict_rewriter import _find_row_id_conflict
@@ -100,7 +101,7 @@ def
test_native_unknown_outcome_is_not_a_rewrite_signal(opaque_message):
failure.__cause__ = _native_conflict(module)
core = Mock()
core.commit.side_effect = failure
- with patch.dict(sys.modules, {'pypaimon_rust': module}):
+ with patch.dict(sys.modules, {'pypaimon_rust': module if native_available
else None}):
with pytest.raises(ValueError) as raised:
NativeTableCommit(core, is_batch=True).commit([], 7)
assert raised.value is failure
