[Python] Fix execution boundary regressions and restore Native CI

.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'

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):

paimon-python/pypaimon/execution/python/globalindex/vindex/vindex_vector_global_index_reader.py

@@ -59,12 +59,8 @@ def _run_search(self, vectors, limit, include_row_ids, query_options):
         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,

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()

paimon-python/pypaimon/table/source/search_dispatch.py

@@ -21,6 +21,28 @@
 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 python_factory():
 
 
 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 python_factory():
 
 
 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 python_factory():
 
 
 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

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))

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.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

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([

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 @@
 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):

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.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):

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), \

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',