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 5f6a19b699 [Python] Preserve captured read views through Native Table
(#10497)
5f6a19b699 is described below
commit 5f6a19b6997e85fe5412e534af0b107bf42389b4
Author: Jingsong Lee <[email protected]>
AuthorDate: Sun Oct 11 11:26:39 2026 +0800
[Python] Preserve captured read views through Native Table (#10497)
---
paimon-python/pypaimon/read/native_plan.py | 8 +-
paimon-python/pypaimon/table/file_store_table.py | 5 -
.../pypaimon/table/source/native_hybrid_search.py | 4 -
.../pypaimon/table/source/native_vector_search.py | 77 +++---
paimon-python/pypaimon/tests/native_plan_test.py | 15 +-
.../pypaimon/tests/native_read_snapshot_test.py | 262 +++++++++++++++++++++
.../pypaimon/tests/native_vector_search_test.py | 18 +-
.../tests/primary_key_index_definitions_test.py | 2 +-
.../pypaimon/tests/search_snapshot_test.py | 2 +-
.../pypaimon/tests/vector_search_filter_test.py | 1 -
10 files changed, 334 insertions(+), 60 deletions(-)
diff --git a/paimon-python/pypaimon/read/native_plan.py
b/paimon-python/pypaimon/read/native_plan.py
index 0a923cd701..a024b44c41 100644
--- a/paimon-python/pypaimon/read/native_plan.py
+++ b/paimon-python/pypaimon/read/native_plan.py
@@ -331,7 +331,7 @@ def _rest_data_token(table):
def _native_table(table):
- """Return the current schema and shared REST FileIO for Native
operations."""
+ """Transport the resolved schema, shared FileIO and captured read view."""
rest_response = _resolved_rest_table_response(table)
file_io_options = _resolved_schema_file_io_options(table)
if rest_response is not None:
@@ -365,6 +365,12 @@ def _native_table(table):
branch = getattr(rt, 'branch', None)
if not callable(branch) or branch() != table.current_branch():
raise RuntimeError("Native table did not resolve the requested
branch")
+ # Only explicitly stored state captures a view; None means a fixed empty
view.
+ table_state = vars(table)
+ if '_read_snapshot' in table_state:
+ from pypaimon.common.json_util import JSON
+ snapshot = table_state['_read_snapshot']
+ rt = rt.copy_with_pinned_snapshot(JSON.to_json(snapshot) if snapshot
is not None else None)
return rt
diff --git a/paimon-python/pypaimon/table/file_store_table.py
b/paimon-python/pypaimon/table/file_store_table.py
index 8dea054608..220f3b56f2 100644
--- a/paimon-python/pypaimon/table/file_store_table.py
+++ b/paimon-python/pypaimon/table/file_store_table.py
@@ -531,11 +531,6 @@ class FileStoreTable(Table):
options[CoreOptions.SCAN_MODE.key()] = "from-snapshot" if snapshot is
not None else "default"
if snapshot is not None:
options[CoreOptions.SCAN_SNAPSHOT_ID.key()] = str(snapshot.id)
- # Native planning cannot consume retained tag metadata or a pinned
empty view.
- # scan.version can resolve to a tag as well.
- if snapshot is None or any(option.key() in self.table_schema.options
for option in (
- CoreOptions.SCAN_TAG_NAME, CoreOptions.SCAN_VERSION)):
- options[CoreOptions.SCAN_NATIVE_PLAN_ENABLED.key()] = "false"
table = self.copy_without_time_travel(options)
table._read_snapshot = snapshot
return table
diff --git a/paimon-python/pypaimon/table/source/native_hybrid_search.py
b/paimon-python/pypaimon/table/source/native_hybrid_search.py
index a19391628f..a820660ab0 100644
--- a/paimon-python/pypaimon/table/source/native_hybrid_search.py
+++ b/paimon-python/pypaimon/table/source/native_hybrid_search.py
@@ -18,7 +18,6 @@
"""Transport an existing local hybrid query to one Rust core operation."""
from pypaimon.catalog.table_query_auth import reject_search_under_query_auth
-from pypaimon.common.json_util import JSON
from pypaimon.globalindex.full_text.native_full_text_global_index_reader
import NativeFullTextIndexOptions
from pypaimon.globalindex.vector_search_result import
DictBasedScoredIndexResult
from pypaimon.read.native_plan import _native_table, _predicate_to_native,
native_method_available
@@ -47,9 +46,6 @@ def try_native_hybrid_search(builder):
table.options.options.to_map()).to_native_options().items()}
table = table.copy(options)
native = _native_table(table).new_hybrid_search_builder()
- if hasattr(table, '_read_snapshot'):
- snapshot = table._read_snapshot
- native.with_snapshot(JSON.to_json(snapshot) if snapshot is not None
else None)
native.with_limit(builder._limit).with_ranker(builder._ranker)
for route in builder._routes:
if route.is_vector():
diff --git a/paimon-python/pypaimon/table/source/native_vector_search.py
b/paimon-python/pypaimon/table/source/native_vector_search.py
index ce61d41b53..b41d8f0ca7 100644
--- a/paimon-python/pypaimon/table/source/native_vector_search.py
+++ b/paimon-python/pypaimon/table/source/native_vector_search.py
@@ -17,19 +17,15 @@
"""Local vector search through the Rust core Scan -> Plan -> Read path."""
-import logging
-
from pypaimon.catalog.table_query_auth import reject_search_under_query_auth
from pypaimon.globalindex.vector_search_result import
DictBasedScoredIndexResult
from pypaimon.read.native_plan import (
_catalog_metastore, _native_table, _partition_fields, _predicate_to_native,
- _raise_if_native_fork_safety_error, native_method_available)
+ native_method_available)
from pypaimon.read.split_serializer import deserialize_split_v1
from pypaimon.table.source.primary_key_scored_result import (
PrimaryKeyScoredResult, PrimaryKeySearchPosition)
-logger = logging.getLogger(__name__)
-
def try_native_vector_search(builder, queries, batch=False):
"""Return existing Python result types, or None to use the Python reader.
@@ -39,47 +35,48 @@ def try_native_vector_search(builder, queries, batch=False):
"""
table = builder._table
reject_search_under_query_auth(table)
+ native_enabled = getattr(table.options, 'native_read_enabled', None)
+ if not callable(native_enabled) or not native_enabled():
+ return None
+ loader = getattr(getattr(table, 'catalog_environment', None),
'catalog_loader', None)
+ if loader is None or _catalog_metastore(loader) != 'rest':
+ return None
+ context_fn = getattr(loader, 'context', None)
+ if not callable(context_fn):
+ return None
+ context = context_fn()
+ if any(getattr(context, attr, None) is not None for attr in (
+ 'hadoop_conf', 'prefer_io_loader', 'fallback_io_loader')):
+ return None
+ # Python's batch reader currently supports only global row-ID results.
+ if batch and builder._is_primary_key_vector_search():
+ return None
+ method = 'new_batch_vector_search_builder' if batch else
'new_vector_search_builder'
+ if not native_method_available('Table', method):
+ return None
+ native = getattr(_native_table(table), method)()
+ if builder._vector_column is not None:
+ native.with_vector_column(builder._vector_column.name)
+ native.with_limit(builder._limit).with_options(builder._options)
try:
- native_enabled = getattr(table.options, 'native_read_enabled', None)
- if not callable(native_enabled) or not native_enabled():
- return None
- loader = getattr(getattr(table, 'catalog_environment', None),
'catalog_loader', None)
- if loader is None or _catalog_metastore(loader) != 'rest':
- return None
- context_fn = getattr(loader, 'context', None)
- if not callable(context_fn):
- return None
- context = context_fn()
- if any(getattr(context, attr, None) is not None for attr in (
- 'hadoop_conf', 'prefer_io_loader', 'fallback_io_loader')):
- return None
- # Python's batch reader currently supports only global row-ID results.
- if batch and builder._is_primary_key_vector_search():
- return None
- method = 'new_batch_vector_search_builder' if batch else
'new_vector_search_builder'
- if not native_method_available('Table', method):
- return None
- native = getattr(_native_table(table), method)()
- if builder._vector_column is not None:
- native.with_vector_column(builder._vector_column.name)
- native.with_limit(builder._limit).with_options(builder._options)
if builder._filter is not None:
native.with_filter(_predicate_to_native(builder._filter))
if builder._partition_filter is not None:
native.with_partition_filter(_predicate_to_native(builder._partition_filter))
- if batch:
- native.with_query_vectors(queries)
- results = native.execute_batch_local()
- return [DictBasedScoredIndexResult(result.row_ids()) for result in
results]
- native.with_query_vector(queries)
- result = native.execute_local()
- if builder._is_primary_key_vector_search():
- return _NativePrimaryKeyScoredResult(table, result)
- return DictBasedScoredIndexResult(result.row_ids())
- except Exception as error:
- _raise_if_native_fork_safety_error(error)
- logger.warning('Native vector search failed, using the Python reader:
%s', error)
+ except ValueError:
+ # Qualify unsupported typed literals before searching.
return None
+ # Once the Native route is chosen, errors must not trigger another search
+ # against a potentially newer snapshot.
+ if batch:
+ native.with_query_vectors(queries)
+ results = native.execute_batch_local()
+ return [DictBasedScoredIndexResult(result.row_ids()) for result in
results]
+ native.with_query_vector(queries)
+ result = native.execute_local()
+ if builder._is_primary_key_vector_search():
+ return _NativePrimaryKeyScoredResult(table, result)
+ return DictBasedScoredIndexResult(result.row_ids())
class _NativePrimaryKeyScoredResult(PrimaryKeyScoredResult):
diff --git a/paimon-python/pypaimon/tests/native_plan_test.py
b/paimon-python/pypaimon/tests/native_plan_test.py
index 7a56edac8d..92378ec1fd 100644
--- a/paimon-python/pypaimon/tests/native_plan_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_test.py
@@ -902,13 +902,23 @@ class NativePlanTest(unittest.TestCase):
patch('pypaimon.read.native_plan._resolved_schema_json',
return_value=resolved):
for _ in range(3):
self.assertIs(_native_read_builder(table),
native_table.new_read_builder.return_value)
+ native_table.copy_with_pinned_snapshot.assert_not_called()
+ # Capturing empty uses a read-view copy, keeping the cached base
+ # reusable when a later caller has no captured snapshot.
+ pinned_table = native_table.copy_with_pinned_snapshot.return_value
+ table._read_snapshot = None
+ self.assertIs(_native_read_builder(table),
pinned_table.new_read_builder.return_value)
+
native_table.copy_with_pinned_snapshot.assert_called_once_with(None)
+ del table._read_snapshot
+ self.assertIs(_native_read_builder(table),
native_table.new_read_builder.return_value)
fake_df.PaimonCatalog.assert_not_called()
fake_df.Table.from_rest_response.assert_called_once_with(
response, database='db', table='t$branch_dev',
rest_options=_catalog_options(table))
self.assertEqual(native_table.copy_with_resolved_schema.call_args_list,
- [call(resolved, branch='dev')] * 3)
- self.assertEqual(native_table.new_read_builder.call_count, 3)
+ [call(resolved, branch='dev')] * 5)
+ self.assertEqual(native_table.new_read_builder.call_count, 4)
+ pinned_table.new_read_builder.assert_called_once_with()
def test_native_rest_cache_invalidation_and_serialization(self):
import pickle
@@ -1207,6 +1217,7 @@ class NativePlanTest(unittest.TestCase):
plan = native_plan(table)
self.assertEqual(plan.snapshot_id, 7)
scan.plan.assert_called_once_with()
+ rt.copy_with_pinned_snapshot.assert_not_called()
rt.branch.return_value = 'main'
with self.assertRaisesRegex(RuntimeError, 'requested branch'):
native_plan(table)
diff --git a/paimon-python/pypaimon/tests/native_read_snapshot_test.py
b/paimon-python/pypaimon/tests/native_read_snapshot_test.py
new file mode 100644
index 0000000000..74370736d7
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_read_snapshot_test.py
@@ -0,0 +1,262 @@
+# 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.
+
+"""Resolved read views travel through Native planning and every local search
API."""
+
+from contextlib import contextmanager
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon.read.scanner.file_scanner import FileScanner
+from pypaimon.table.source.full_text_search_builder import
FullTextSearchBuilderImpl
+from pypaimon.table.source.vector_search_read import DataEvolutionVectorRead,
BatchVectorSearchReadImpl
+from pypaimon.tests import native_hybrid_search_test as hybrid,
native_vector_search_test as vectors
+from pypaimon.tests.native_full_text_search_test import _scores
+from pypaimon.tests.vector_search_filter_test import match_query
+
+rest_catalog = hybrid.rest_catalog
+golden_catalog = vectors.golden_catalog
+pk_vector_table = vectors.pk_vector_table
+pytestmark = pytest.mark.native_plan
+
+
+@contextmanager
+def _native_only():
+ with patch.object(FileScanner, 'scan', side_effect=AssertionError('Python
file scan')), \
+ patch.object(DataEvolutionVectorRead, 'read_plan',
side_effect=AssertionError('Python vector read')), \
+ patch.object(BatchVectorSearchReadImpl, 'read_batch_plan',
+ side_effect=AssertionError('Python batch read')), \
+ patch.object(FullTextSearchBuilderImpl, 'new_full_text_scan',
+ side_effect=AssertionError('Python text scan')):
+ yield
+
+
+def _search(table, family):
+ if family == 'scan':
+ return hybrid.indexes._ids(table)
+ if family == 'text':
+ return
_scores(table.new_full_text_search_builder().with_query('content',
match_query('paimon'))
+ .with_limit(16).execute_local())
+ if family == 'batch':
+ return [_scores(result) for result in
table.new_batch_vector_search_builder()
+ .with_vector_column('embedding').with_query_vectors([[0., 0.],
[6., 0.], [0., 0.]])
+ .with_limit(16).execute_batch_local()]
+ return
_scores(table.new_vector_search_builder().with_vector_column('embedding')
+ .with_query_vector([0., 0.]).with_limit(16).execute_local())
+
+
[email protected]('family', ['scan', 'vector', 'batch', 'text'])
+def test_fixed_empty_view_does_not_follow_latest(rest_catalog, family):
+ catalog, _ = rest_catalog
+ table = hybrid._table(catalog)
+ fixed = table._copy_with_snapshot(None)
+ with _native_only():
+ result = _search(fixed, family)
+ assert result == ([{}, {}, {}] if family == 'batch' else [] if family ==
'scan' else {})
+ assert hybrid.indexes._ids(table) == list(range(8))
+
+
[email protected]('family', ['scan', 'vector', 'batch', 'text'])
+def
test_retained_tag_metadata_is_used_without_original_snapshot_file(rest_catalog,
family):
+ catalog, _ = rest_catalog
+ table = hybrid._table(catalog)
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ table.tag_manager().create_tag(snapshot, 'before')
+ fixed = table.copy({'scan.tag-name':
'before'})._copy_with_snapshot(snapshot)
+ hybrid.indexes._append(table, hybrid._rows(8, 4), schema=hybrid.SCHEMA)
+
table.file_io.delete(table.snapshot_manager().get_snapshot_path(snapshot.id))
+ classic = fixed.copy({'read.native.enabled': 'false',
'scan.native-plan.enabled': 'false'})
+ expected = _search(classic, family)
+ with _native_only():
+ result = _search(fixed, family)
+ if family == 'scan':
+ assert result == expected == list(range(8))
+ elif family == 'batch':
+ assert len(result) == 3
+ for actual, reference in zip(result, expected):
+ assert actual == pytest.approx(reference, abs=1e-6)
+ else:
+ assert result == pytest.approx(expected, abs=1e-6)
+ assert hybrid.indexes._ids(table) == list(range(12))
+
+
+def _query(catalog, table, family):
+ from pypaimon.multimodal.table import MultimodalTable, text_route,
vector_route
+ multimodal = MultimodalTable(catalog, table.identifier, table)
+ if family == 'batch':
+ query = multimodal.search_vectors([[0., 0.], [6., 0.]],
column='embedding')
+ elif family == 'text':
+ query = multimodal.search('paimon', column='content')
+ elif family == 'hybrid':
+ query = multimodal.search_hybrid([
+ vector_route('embedding', [0., 0.]),
text_route(match_query('paimon'), column='content')])
+ else:
+ query = multimodal.search([0., 0.], column='embedding')
+ return query.pre_filter("label = 'keep'").select(['id', 'label']).limit(16)
+
+
[email protected]('family', ['vector', 'batch', 'text', 'hybrid'])
[email protected]('change', ['update', 'delete'])
+def test_native_search_and_lookup_use_one_view_across_commit(rest_catalog,
family, change):
+ from pypaimon.multimodal.query import ScanQuery
+ catalog, _ = rest_catalog
+ schema = pa.schema([pa.field('embedding', pa.list_(pa.float32(), 2))
+ if field.name == 'embedding' else field for field in
hybrid.SCHEMA])
+ table = hybrid.indexes._create(catalog, schema=schema, partitioned=True,
options={
+ 'read.native.enabled': 'true', 'vector-index.search-mode': 'full',
+ 'full-text-index.search-mode': 'full'})
+ hybrid.indexes._append(table, hybrid._rows(), schema=schema)
+ table.create_global_index('content', index_type='full-text')
+ table.create_global_index('embedding', index_type='ivf-flat', options={
+ 'ivf-flat.dimension': '2', 'ivf-flat.nlist': '1',
'ivf-flat.distance.metric': 'l2'})
+ table = table.copy({'deletion-vectors.enabled': 'true'})
+ query = _query(catalog, table, family)
+ expected = query.to_list()
+ lookup = ScanQuery._read_global_index_result
+ views = []
+ write_builder = table.new_batch_write_builder()
+ updater = write_builder.new_update()
+ predicate = table.new_read_builder().new_predicate_builder().equal('id', 1)
+ messages = (updater.update_by_predicate(predicate, {'label': 'drop'}) if
change == 'update'
+ else updater.delete_by_predicate(predicate))
+ commit = write_builder.new_commit()
+
+ def commit_before_lookup(execution, result):
+ if not views:
+ commit.commit(messages)
+ views.append(execution._table)
+ return lookup(execution, result)
+
+ try:
+ with _native_only(), hybrid._core_only(), \
+ patch.object(ScanQuery, '_read_global_index_result',
commit_before_lookup):
+ assert query.to_list() == expected
+ finally:
+ commit.close()
+ assert all(view._read_snapshot.id == views[0]._read_snapshot.id for view
in views)
+ with _native_only(), hybrid._core_only():
+ latest = query.to_list()
+ assert latest != expected
+ for rows in (latest if family == 'batch' else [latest]):
+ assert 1 not in {row['id'] for row in rows}
+ assert not hasattr(query._table, '_read_snapshot')
+
+
[email protected]('family', ['scan', 'vector', 'batch', 'text'])
+def
test_captured_view_skips_rest_snapshot_lookup_and_keeps_cache_isolated(rest_catalog,
family):
+ from pypaimon.api.api_response import ErrorResponse
+ catalog, server = rest_catalog
+ table = hybrid._table(catalog)
+ first = table.snapshot_manager().get_latest_snapshot()
+ fixed = table._copy_with_snapshot(first).copy({'read.batch-size': '1'})
+ expected = _search(fixed.copy({'read.native.enabled': 'false',
'scan.native-plan.enabled': 'false'}), family)
+ hybrid.indexes._append(table, hybrid._rows(8, 4), schema=hybrid.SCHEMA)
+ reply = ErrorResponse('TABLE', 'indexes', 'latest must not be loaded', 503)
+ with patch.object(server, '_table_snapshot_handle',
return_value=server._mock_response(reply, 503)) as latest:
+ with _native_only():
+ actual = _search(fixed, family)
+ empty = _search(table._copy_with_snapshot(None), family)
+ assert latest.call_count == 0
+ if family == 'batch':
+ for scores, reference in zip(actual, expected):
+ assert scores == pytest.approx(reference, abs=1e-6)
+ assert empty == [{}, {}, {}]
+ else:
+ assert actual == (expected if family == 'scan' else
pytest.approx(expected, abs=1e-6))
+ assert empty == ([] if family == 'scan' else {})
+ # The cached REST Table is a base view. Pinning copies cannot change it.
+ assert hybrid.indexes._ids(table) == list(range(12))
+ changed = fixed.copy({'scan.snapshot-id': None, 'scan.mode': 'default'})
+ assert not hasattr(changed, '_read_snapshot')
+ assert hybrid.indexes._ids(changed) == list(range(12))
+
+
[email protected]('family', ['vector', 'batch', 'text'])
[email protected]('response', ['empty', 'error'])
+def
test_rest_snapshot_response_is_authoritative_without_search_retry(rest_catalog,
family, response):
+ from pypaimon.api.api_response import ErrorResponse,
GetTableSnapshotResponse
+ catalog, server = rest_catalog
+ table = hybrid._table(catalog)
+ reply = GetTableSnapshotResponse() if response == 'empty' else
ErrorResponse(
+ 'TABLE', 'indexes', 'snapshot unavailable', 503)
+ code = 200 if response == 'empty' else 503
+ with patch.object(server, '_table_snapshot_handle',
return_value=server._mock_response(reply, code)) as latest:
+ with _native_only():
+ if response == 'error':
+ with pytest.raises(Exception, match='snapshot unavailable'):
+ _search(table, family)
+ else:
+ assert _search(table, family) == ([{}, {}, {}] if family ==
'batch' else {})
+ assert latest.call_count == 1
+ assert hybrid.indexes._ids(table) == list(range(8))
+
+
[email protected]('family', ['vector', 'batch', 'text'])
+def test_missing_index_preserves_native_error_without_retry(rest_catalog,
family):
+ catalog, _ = rest_catalog
+ table = hybrid._table(catalog)
+ index_type = 'full-text' if family == 'text' else 'ivf-flat'
+ index = next(file for file in hybrid.indexes._files(table) if
file.index_type == index_type)
+ table.file_io.delete(hybrid.indexes._path(table, index))
+ with _native_only(), pytest.raises(Exception, match='not found|No such
file|does not exist|NotFound'):
+ _search(table, family)
+ assert hybrid.indexes._ids(table) == list(range(8))
+
+
+def test_snapshot_json_binding_requires_java_snapshot_metadata(rest_catalog):
+ from pypaimon.read.native_plan import _native_table
+ catalog, _ = rest_catalog
+ table = hybrid._table(catalog)
+ with pytest.raises(ValueError, match='Invalid snapshot JSON'):
+ _native_table(table).copy_with_pinned_snapshot('{}')
+ with pytest.raises(ValueError, match='Invalid snapshot JSON'):
+ _native_table(table).copy_with_pinned_snapshot('not json')
+ assert hybrid.indexes._ids(table) == list(range(8))
+
+
[email protected]('view', ['empty', 'tag'])
+def test_primary_key_vector_search_uses_captured_metadata(pk_vector_table,
view):
+ from pypaimon.table.source.primary_key_vector_read import
PrimaryKeyVectorRead
+ table = pk_vector_table
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ if view == 'tag':
+ table.tag_manager().create_tag(snapshot, 'before')
+ fixed = table._copy_with_snapshot(snapshot if view == 'tag' else None)
+ if view == 'tag':
+
table.file_io.delete(table.snapshot_manager().get_snapshot_path(snapshot.id))
+
+ def search(read_table):
+ return
(read_table.new_vector_search_builder().with_vector_column('embedding')
+ .with_query_vector([1., 0., 0.,
0.]).with_limit(2).execute_local())
+
+ classic = fixed.copy({'read.native.enabled': 'false',
'scan.native-plan.enabled': 'false'})
+ expected = search(classic)
+ native = fixed.copy({'read.native.enabled': 'true',
'scan.native-plan.enabled': 'true'})
+ with _native_only(), patch.object(PrimaryKeyVectorRead, 'read_plan',
+ side_effect=AssertionError('Python
primary-key vector read')):
+ result = search(native)
+ read = native.new_read_builder()
+ assert read.new_scan().plan().snapshot_id == (snapshot.id if view ==
'tag' else None)
+ rows = read.new_read().to_arrow(result.splits)
+ assert result.snapshot_id == expected.snapshot_id
+ assert [(position.bucket, position.data_file_name, position.row_position)
for position in result.positions] == [
+ (position.bucket, position.data_file_name, position.row_position) for
position in expected.positions]
+ assert [position.score for position in result.positions] == pytest.approx([
+ position.score for position in expected.positions])
+ assert sorted(rows.column('id').to_pylist()) == ([1, 2] if view == 'tag'
else [])
diff --git a/paimon-python/pypaimon/tests/native_vector_search_test.py
b/paimon-python/pypaimon/tests/native_vector_search_test.py
index cecd4ca927..5f078bf2cb 100644
--- a/paimon-python/pypaimon/tests/native_vector_search_test.py
+++ b/paimon-python/pypaimon/tests/native_vector_search_test.py
@@ -215,12 +215,20 @@ def
test_java_pk_index_results_are_physical_selections(pk_vector_table, filtered
assert split.scores is not None
-def test_native_search_failure_falls_back(vector_table):
- native = vector_table.copy({'read.native.enabled': 'true'})
- expected = _scores(_single(native.copy({'read.native.enabled':
'false'})).execute_local())
[email protected]('batch', [False, True])
+def test_native_search_failure_is_propagated_without_retry(vector_table,
batch):
from pypaimon.table.source import native_vector_search
- with patch.object(native_vector_search, '_native_table',
side_effect=RuntimeError('unavailable')):
- _assert_scores(_scores(_single(native).execute_local()), expected)
+ from pypaimon.table.source.vector_search_read import
DataEvolutionVectorRead, BatchVectorSearchReadImpl
+ table = vector_table.copy({'read.native.enabled': 'true'})
+ builder =
(table.new_batch_vector_search_builder().with_vector_column('embedding')
+ .with_query_vectors([[1, 0], [0, 1]]).with_limit(3)) if batch
else _single(table)
+ method = 'execute_batch_local' if batch else 'execute_local'
+ with patch.object(native_vector_search, '_native_table',
side_effect=RuntimeError('unavailable')), \
+ patch.object(DataEvolutionVectorRead, 'read_plan',
side_effect=AssertionError('Python search retried')), \
+ patch.object(BatchVectorSearchReadImpl, 'read_batch_plan',
+ side_effect=AssertionError('Python batch search
retried')):
+ with pytest.raises(RuntimeError, match='unavailable'):
+ getattr(builder, method)()
def test_explicit_disable_uses_python(vector_table):
diff --git a/paimon-python/pypaimon/tests/primary_key_index_definitions_test.py
b/paimon-python/pypaimon/tests/primary_key_index_definitions_test.py
index 28b67863c1..1f22cd7d65 100644
--- a/paimon-python/pypaimon/tests/primary_key_index_definitions_test.py
+++ b/paimon-python/pypaimon/tests/primary_key_index_definitions_test.py
@@ -366,7 +366,7 @@ class PrimaryKeyIndexDefinitionsTest(unittest.TestCase):
self.assertEqual("from-snapshot", copied_options[0]["scan.mode"])
self.assertEqual("17", copied_options[0]["scan.snapshot-id"])
self.assertIsNone(copied_options[0]["scan.tag-name"])
- self.assertEqual("false",
copied_options[0]["scan.native-plan.enabled"])
+ self.assertNotIn("scan.native-plan.enabled", copied_options[0])
def test_pk_index_source_policy_matches_java(self):
compact = SimpleNamespace(file_source=1, level=1)
diff --git a/paimon-python/pypaimon/tests/search_snapshot_test.py
b/paimon-python/pypaimon/tests/search_snapshot_test.py
index 39bd387c9c..b19e9eda1e 100644
--- a/paimon-python/pypaimon/tests/search_snapshot_test.py
+++ b/paimon-python/pypaimon/tests/search_snapshot_test.py
@@ -147,7 +147,7 @@ def
test_time_travel_survives_concurrent_writes_and_retained_tag_snapshot(table,
table.update("id = 1", {"category": "blocked"})
if selector in ("tag", "version"):
table.raw_table.replace_tag("training")
- assert not
execution._table.options.native_plan_enabled(default=True)
+ assert execution._table._read_snapshot.id == source.id
# Read-option copies must retain the captured snapshot too.
execution._table = execution._table.copy({"read.batch-size": "2"})
return lookup(execution, result)
diff --git a/paimon-python/pypaimon/tests/vector_search_filter_test.py
b/paimon-python/pypaimon/tests/vector_search_filter_test.py
index 655080f8bc..578f252b83 100644
--- a/paimon-python/pypaimon/tests/vector_search_filter_test.py
+++ b/paimon-python/pypaimon/tests/vector_search_filter_test.py
@@ -3231,7 +3231,6 @@ class VectorSearchManySplitsTest(unittest.TestCase):
CoreOptions.SCAN_SNAPSHOT_ID.key(): "7",
CoreOptions.SCAN_TAG_NAME.key(): None,
CoreOptions.SCAN_TIMESTAMP.key(): None,
- CoreOptions.SCAN_NATIVE_PLAN_ENABLED.key(): "false",
})
self.assertIs(read_table._read_snapshot, snapshot)
