This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit b946d37d8a29d7fd8f3160da7bc9231b1c536a33 Author: chaoyang <[email protected]> AuthorDate: Thu Sep 17 16:24:55 2026 +0800 [python] Keep local search and lookup on one snapshot (#9843) --- docs/docs/pypaimon/multimodal-search.md | 8 + .../data_evolution_global_index_scanner.py | 2 + paimon-python/pypaimon/multimodal/query.py | 23 +- paimon-python/pypaimon/read/table_scan.py | 6 +- .../pypaimon/snapshot/time_travel_util.py | 10 + paimon-python/pypaimon/table/file_store_table.py | 25 ++- .../pypaimon/table/source/full_text_read.py | 6 + .../pypaimon/table/source/full_text_scan.py | 17 +- .../table/source/global_index_live_row_filter.py | 13 +- .../pypaimon/table/source/hybrid_search_builder.py | 8 +- .../table/source/primary_key_full_text_scan.py | 20 +- .../table/source/primary_key_vector_scan.py | 20 +- .../pypaimon/table/source/vector_search_scan.py | 9 +- .../pypaimon/tests/multimodal_table_test.py | 38 +++- .../tests/primary_key_index_definitions_test.py | 12 +- .../pypaimon/tests/search_snapshot_test.py | 233 +++++++++++++++++++++ .../pypaimon/tests/vector_search_filter_test.py | 13 +- 17 files changed, 375 insertions(+), 88 deletions(-) diff --git a/docs/docs/pypaimon/multimodal-search.md b/docs/docs/pypaimon/multimodal-search.md index c2685f5b29..2a2940fe1c 100644 --- a/docs/docs/pypaimon/multimodal-search.md +++ b/docs/docs/pypaimon/multimodal-search.md @@ -63,6 +63,14 @@ filter the rows read from the search result. Both `pre_filter` and `where()` accept SQL-like predicate strings. For full-text search, `pre_filter` must only reference partition columns. +Each execution of `search`, `search_vectors`, or `search_hybrid` reads one +snapshot across candidate search, filtering, reranking, and result lookup. +Concurrent commits become visible on the next execution, including when reusing +the same query object. Explicit snapshot, tag, and timestamp selectors are +honored. All routes in a hybrid search and all vectors in a batch share that +execution's snapshot. Keep the snapshot's data files available for the duration +of the query; capturing a read view does not prevent snapshot expiration. + ```python neighbors = ( docs.search( diff --git a/paimon-python/pypaimon/globalindex/data_evolution_global_index_scanner.py b/paimon-python/pypaimon/globalindex/data_evolution_global_index_scanner.py index dde7f24f57..e7acdee35c 100644 --- a/paimon-python/pypaimon/globalindex/data_evolution_global_index_scanner.py +++ b/paimon-python/pypaimon/globalindex/data_evolution_global_index_scanner.py @@ -344,6 +344,8 @@ class _PaddingGlobalIndexReader(GlobalIndexReader): def _resolve_snapshot(table, snapshot): if snapshot is not None: return snapshot + if hasattr(table, "_read_snapshot"): + return table._read_snapshot snapshot_manager = table.snapshot_manager() if snapshot_manager is None: return None diff --git a/paimon-python/pypaimon/multimodal/query.py b/paimon-python/pypaimon/multimodal/query.py index e21edb2f1e..9f5d92c19e 100644 --- a/paimon-python/pypaimon/multimodal/query.py +++ b/paimon-python/pypaimon/multimodal/query.py @@ -15,6 +15,7 @@ # specific language governing permissions and limitations # under the License. +from copy import copy from typing import Any, Callable, Dict, List, Optional, Tuple import pyarrow as pa @@ -467,6 +468,15 @@ class ScanQuery: class _PreFilterQuery(ScanQuery): + def _for_execution(self): + from pypaimon.snapshot.time_travel_util import TimeTravelUtil + query = copy(self) + query._table = self._table._copy_with_snapshot(TimeTravelUtil.resolve_snapshot(self._table)) + return query + + def to_arrow(self): + return ScanQuery.to_arrow(self._for_execution()) + def __init__( self, table, @@ -525,7 +535,7 @@ class VectorQuery(_PreFilterQuery): def _execute_vector(self, query): limit = query._limit if query._limit is not None else 10 builder = ( - self._table.new_vector_search_builder() + query._table.new_vector_search_builder() .with_vector_column(self._vector_column) .with_query_vector(self._vector) .with_limit(limit) @@ -547,7 +557,7 @@ class TextQuery(_PreFilterQuery): def _execute_fts(self, query): limit = query._limit if query._limit is not None else 10 builder = ( - self._table.new_full_text_search_builder() + query._table.new_full_text_search_builder() .with_query(self._text_query["column"], self._text_query["query"]) .with_limit(limit) ) @@ -582,7 +592,7 @@ class HybridQuery(_PreFilterQuery): final_limit = query._limit if query._limit is not None else 10 route_limit = self._route_limit or final_limit builder = ( - self._table.new_hybrid_search_builder() + query._table.new_hybrid_search_builder() .with_limit(final_limit) .with_ranker(self._ranker) ) @@ -623,9 +633,10 @@ class BatchVectorQuery(_PreFilterQuery): super().__init__(table, pre_filter=pre_filter) def to_arrow(self): + query = self._for_execution() return [ - self._read_global_index_result(result) - for result in self._execute_batch_vector(self) + query._read_global_index_result(result) + for result in query._execute_batch_vector(query) ] def to_pandas(self): @@ -637,7 +648,7 @@ class BatchVectorQuery(_PreFilterQuery): def _execute_batch_vector(self, query): limit = query._limit if query._limit is not None else 10 builder = ( - self._table.new_batch_vector_search_builder() + query._table.new_batch_vector_search_builder() .with_vector_column(self._vector_column) .with_query_vectors(self._vectors) .with_limit(limit) diff --git a/paimon-python/pypaimon/read/table_scan.py b/paimon-python/pypaimon/read/table_scan.py index 74a5be1e6b..2a21527242 100755 --- a/paimon-python/pypaimon/read/table_scan.py +++ b/paimon-python/pypaimon/read/table_scan.py @@ -491,9 +491,7 @@ class TableScan: if has_time_travel: def time_travel_manifest_scanner(): - snapshot = TimeTravelUtil.try_travel_to_snapshot( - options, self.table.tag_manager(), snapshot_manager - ) + snapshot = TimeTravelUtil.resolve_snapshot(self.table) if snapshot is None: raise ValueError( "Could not resolve time travel snapshot from scan options." @@ -510,7 +508,7 @@ class TableScan: ) def all_manifests(): - snapshot = snapshot_manager.get_latest_snapshot() + snapshot = TimeTravelUtil.resolve_snapshot(self.table) return manifest_list_manager.read_all(snapshot), snapshot return FileScanner( diff --git a/paimon-python/pypaimon/snapshot/time_travel_util.py b/paimon-python/pypaimon/snapshot/time_travel_util.py index 99bfd3a1bf..8f6de35061 100644 --- a/paimon-python/pypaimon/snapshot/time_travel_util.py +++ b/paimon-python/pypaimon/snapshot/time_travel_util.py @@ -66,6 +66,16 @@ def _parse_timestamp_to_millis(timestamp_str: str) -> int: class TimeTravelUtil: """The util class of resolve snapshot from scan params for time travel.""" + @staticmethod + def resolve_snapshot(table): + """Resolve the table's read view, including an explicitly pinned empty view.""" + if hasattr(table, "_read_snapshot"): + return table._read_snapshot + manager = table.snapshot_manager() + snapshot = TimeTravelUtil.try_travel_to_snapshot( + table.options.options, table.tag_manager(), manager) + return snapshot if snapshot is not None else manager.get_latest_snapshot() + @staticmethod def try_travel_to_snapshot( options: Options, diff --git a/paimon-python/pypaimon/table/file_store_table.py b/paimon-python/pypaimon/table/file_store_table.py index c2962236b9..c939a34586 100644 --- a/paimon-python/pypaimon/table/file_store_table.py +++ b/paimon-python/pypaimon/table/file_store_table.py @@ -517,6 +517,22 @@ class FileStoreTable(Table): """Copy this table while preserving its already resolved schema.""" return self._copy(options, resolve_time_travel=False) + def _copy_with_snapshot(self, snapshot): + """Keep one resolved read view, including tag metadata and empty tables.""" + from pypaimon.snapshot.time_travel_util import SCAN_KEYS + options = {key: None for key in SCAN_KEYS if key in self.table_schema.options} + 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 + def _copy(self, options: dict, resolve_time_travel: bool) -> 'FileStoreTable': if CoreOptions.BUCKET.key() in options and int(options.get(CoreOptions.BUCKET.key())) != self.options.bucket(): raise ValueError("Cannot change bucket number") @@ -532,7 +548,12 @@ class FileStoreTable(Table): # Cumulative copy() overrides (removals kept as None) vs the on-disk schema. applied_options = {**getattr(self, '_applied_dynamic_options', {}), **options} - if resolve_time_travel: + from pypaimon.snapshot.time_travel_util import SCAN_KEYS + preserve_snapshot = hasattr(self, "_read_snapshot") and not any( + key in options for key in SCAN_KEYS + [ + CoreOptions.SCAN_MODE.key(), CoreOptions.BRANCH.key(), + CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP.key()]) + if resolve_time_travel and not preserve_snapshot: time_travel_schema = self._try_time_travel(Options(new_options), set(applied_options)) if time_travel_schema is not None: new_table_schema = time_travel_schema @@ -555,6 +576,8 @@ class FileStoreTable(Table): new_table = FileStoreTable(self.file_io, new_identifier, self.table_path, new_table_schema, catalog_env) new_table._applied_dynamic_options = applied_options + if preserve_snapshot: + new_table._read_snapshot = self._read_snapshot return new_table def _try_time_travel(self, options: Options, dynamic_option_keys: Set[str]) -> Optional[TableSchema]: diff --git a/paimon-python/pypaimon/table/source/full_text_read.py b/paimon-python/pypaimon/table/source/full_text_read.py index 036a6d9d3d..e6ddbbcd52 100644 --- a/paimon-python/pypaimon/table/source/full_text_read.py +++ b/paimon-python/pypaimon/table/source/full_text_read.py @@ -18,6 +18,7 @@ """Full-text read to read index files.""" from abc import ABC, abstractmethod +from copy import copy from concurrent.futures import wait from io import BytesIO from typing import Dict, List @@ -72,6 +73,11 @@ class DataEvolutionFullTextRead(FullTextRead): self._query = query self._partition_filter = partition_filter + def read_plan(self, plan: FullTextScanPlan) -> GlobalIndexResult: + reader = copy(self) + reader._table = global_index_live_row_filter.table_at_snapshot(self._table, plan.snapshot()) + return reader.read(plan.splits()) + def read(self, splits: List[FullTextSearchSplit]) -> GlobalIndexResult: index_splits, raw_splits = _split_search_splits(splits) if not index_splits and not raw_splits: diff --git a/paimon-python/pypaimon/table/source/full_text_scan.py b/paimon-python/pypaimon/table/source/full_text_scan.py index 1387b02536..29b44ee5b0 100644 --- a/paimon-python/pypaimon/table/source/full_text_scan.py +++ b/paimon-python/pypaimon/table/source/full_text_scan.py @@ -36,8 +36,12 @@ from pypaimon.utils.range import Range class FullTextScanPlan: """Plan of full-text scan.""" - def __init__(self, splits: List[FullTextSearchSplit]): + def __init__(self, splits: List[FullTextSearchSplit], snapshot=None): self._splits = splits + self._snapshot = snapshot + + def snapshot(self): + return self._snapshot def splits(self) -> List[FullTextSearchSplit]: return self._splits @@ -73,14 +77,9 @@ class DataEvolutionFullTextScan(FullTextScan): id_to_column = {field.id: field.name for field in self._text_columns} from pypaimon.snapshot.time_travel_util import TimeTravelUtil - from pypaimon.common.options.options import Options - snapshot = TimeTravelUtil.try_travel_to_snapshot( - Options(self._table.table_schema.options), - self._table.tag_manager(), - self._table.snapshot_manager(), - ) + snapshot = TimeTravelUtil.resolve_snapshot(self._table) if snapshot is None: - snapshot = self._table.snapshot_manager().get_latest_snapshot() + return FullTextScanPlan([]) index_file_handler = IndexFileHandler(table=self._table) partition_filter = self._partition_filter @@ -129,7 +128,7 @@ class DataEvolutionFullTextScan(FullTextScan): if raw_row_ranges: splits.append(RawFullTextSearchSplit(raw_row_ranges)) - return FullTextScanPlan(splits) + return FullTextScanPlan(splits, snapshot) def _supports_full_text_search(index_type): diff --git a/paimon-python/pypaimon/table/source/global_index_live_row_filter.py b/paimon-python/pypaimon/table/source/global_index_live_row_filter.py index af96b85e5b..fe59df49c2 100644 --- a/paimon-python/pypaimon/table/source/global_index_live_row_filter.py +++ b/paimon-python/pypaimon/table/source/global_index_live_row_filter.py @@ -19,7 +19,6 @@ from typing import Optional -from pypaimon.common.options.core_options import CoreOptions from pypaimon.deletionvectors.deletion_vector import DeletionVector from pypaimon.read.query_auth_split import QueryAuthSplit from pypaimon.read.split import DataSplit @@ -67,17 +66,7 @@ def table_at_snapshot(table, snapshot): if snapshot is None: return table - pin_options = { - CoreOptions.SCAN_MODE.key(): "from-snapshot", - CoreOptions.SCAN_SNAPSHOT_ID.key(): str(snapshot.id), - } - for option in (CoreOptions.SCAN_TAG_NAME, - CoreOptions.SCAN_WATERMARK, - CoreOptions.SCAN_TIMESTAMP, - CoreOptions.SCAN_TIMESTAMP_MILLIS): - if option.key() in table.table_schema.options: - pin_options[option.key()] = None - return table.copy_without_time_travel(pin_options) + return table._copy_with_snapshot(snapshot) def for_range(live_row_ids: Optional[RoaringBitmap64], diff --git a/paimon-python/pypaimon/table/source/hybrid_search_builder.py b/paimon-python/pypaimon/table/source/hybrid_search_builder.py index e05c248a63..a3e43c3fb6 100644 --- a/paimon-python/pypaimon/table/source/hybrid_search_builder.py +++ b/paimon-python/pypaimon/table/source/hybrid_search_builder.py @@ -19,6 +19,7 @@ import heapq import math +from copy import copy from abc import ABC, abstractmethod from concurrent.futures import FIRST_EXCEPTION, ThreadPoolExecutor, wait from dataclasses import dataclass, field @@ -312,16 +313,19 @@ class HybridSearchBuilderImpl(HybridSearchBuilder): def route_builders(self) -> List[HybridSearchRouteBuilder]: self._validate_search() + from pypaimon.snapshot.time_travel_util import TimeTravelUtil + execution = copy(self) + execution._table = self._table._copy_with_snapshot(TimeTravelUtil.resolve_snapshot(self._table)) builders = [] for route in self._routes: if route.is_vector(): builders.append( HybridSearchRouteBuilder( - route, self._new_vector_search_builder(route))) + route, execution._new_vector_search_builder(route))) else: builders.append( HybridSearchRouteBuilder( - route, self._new_full_text_search_builder(route))) + route, execution._new_full_text_search_builder(route))) return builders def to_route_result( diff --git a/paimon-python/pypaimon/table/source/primary_key_full_text_scan.py b/paimon-python/pypaimon/table/source/primary_key_full_text_scan.py index dab5f0e66e..4a7d7c515e 100644 --- a/paimon-python/pypaimon/table/source/primary_key_full_text_scan.py +++ b/paimon-python/pypaimon/table/source/primary_key_full_text_scan.py @@ -17,8 +17,6 @@ from dataclasses import dataclass -from pypaimon.common.options.core_options import CoreOptions -from pypaimon.common.options.options import Options from pypaimon.globalindex.indexed_split import IndexedSplit from pypaimon.index.index_file_handler import IndexFileHandler from pypaimon.index.pk.primary_key_index_source_meta import ( @@ -28,6 +26,7 @@ from pypaimon.index.pk.primary_key_index_source_policy import ( from pypaimon.read.query_auth_split import QueryAuthSplit from pypaimon.read.split import DataSplit from pypaimon.snapshot.time_travel_util import TimeTravelUtil +from pypaimon.table.source.global_index_live_row_filter import table_at_snapshot from pypaimon.table.source.full_text_scan import FullTextScan, FullTextScanPlan @@ -48,24 +47,11 @@ class PrimaryKeyFullTextScan(FullTextScan): self._partition_filter = partition_filter def scan(self): - snapshot = TimeTravelUtil.try_travel_to_snapshot( - Options(self._table.table_schema.options), self._table.tag_manager(), - self._table.snapshot_manager()) - if snapshot is None: - snapshot = self._table.snapshot_manager().get_latest_snapshot() + snapshot = TimeTravelUtil.resolve_snapshot(self._table) if snapshot is None: return PrimaryKeyFullTextScanPlan(0, []) - pin_options = { - CoreOptions.SCAN_MODE.key(): "from-snapshot", - CoreOptions.SCAN_SNAPSHOT_ID.key(): str(snapshot.id)} - for option in (CoreOptions.SCAN_TAG_NAME, - CoreOptions.SCAN_WATERMARK, - CoreOptions.SCAN_TIMESTAMP, - CoreOptions.SCAN_TIMESTAMP_MILLIS): - if option.key() in self._table.table_schema.options: - pin_options[option.key()] = None - scan_table = self._table.copy(pin_options) + scan_table = table_at_snapshot(self._table, snapshot) builder = scan_table.new_read_builder() if self._partition_filter is not None: builder = builder.with_partition_filter(self._partition_filter) diff --git a/paimon-python/pypaimon/table/source/primary_key_vector_scan.py b/paimon-python/pypaimon/table/source/primary_key_vector_scan.py index 99156528f0..0ccdf9bf3f 100644 --- a/paimon-python/pypaimon/table/source/primary_key_vector_scan.py +++ b/paimon-python/pypaimon/table/source/primary_key_vector_scan.py @@ -17,8 +17,8 @@ from dataclasses import dataclass -from pypaimon.common.options.options import Options from pypaimon.common.options.core_options import CoreOptions +from pypaimon.common.options.options import Options from pypaimon.index.index_file_handler import IndexFileHandler from pypaimon.index.pk.primary_key_index_source_meta import PrimaryKeyIndexSourceMeta from pypaimon.index.pk.primary_key_index_source_policy import ( @@ -28,6 +28,7 @@ from pypaimon.read.split import DataSplit from pypaimon.globalindex.indexed_split import IndexedSplit from pypaimon.deletionvectors.deletion_vector import DeletionVector from pypaimon.snapshot.time_travel_util import TimeTravelUtil +from pypaimon.table.source.global_index_live_row_filter import table_at_snapshot from pypaimon.table.source.vector_search_scan import VectorSearchScan, VectorSearchScanPlan from pypaimon.table.row.generic_row import GenericRow from pypaimon.utils.range import Range @@ -54,24 +55,11 @@ class PrimaryKeyVectorScan(VectorSearchScan): self._index_type = index_type def scan(self): - snapshot = TimeTravelUtil.try_travel_to_snapshot( - Options(self._table.table_schema.options), self._table.tag_manager(), - self._table.snapshot_manager()) - if snapshot is None: - snapshot = self._table.snapshot_manager().get_latest_snapshot() + snapshot = TimeTravelUtil.resolve_snapshot(self._table) if snapshot is None: return PrimaryKeyVectorScanPlan(0, []) - pin_options = { - CoreOptions.SCAN_MODE.key(): "from-snapshot", - CoreOptions.SCAN_SNAPSHOT_ID.key(): str(snapshot.id)} - for option in (CoreOptions.SCAN_TAG_NAME, - CoreOptions.SCAN_WATERMARK, - CoreOptions.SCAN_TIMESTAMP, - CoreOptions.SCAN_TIMESTAMP_MILLIS): - if option.key() in self._table.table_schema.options: - pin_options[option.key()] = None - scan_table = self._table.copy(pin_options) + scan_table = table_at_snapshot(self._table, snapshot) builder = scan_table.new_read_builder() if self._partition_filter is not None: builder = builder.with_partition_filter(self._partition_filter) diff --git a/paimon-python/pypaimon/table/source/vector_search_scan.py b/paimon-python/pypaimon/table/source/vector_search_scan.py index 8c94390b51..8484588f35 100644 --- a/paimon-python/pypaimon/table/source/vector_search_scan.py +++ b/paimon-python/pypaimon/table/source/vector_search_scan.py @@ -77,7 +77,6 @@ class DataEvolutionVectorScan(VectorSearchScan): def scan(self): # type: () -> VectorSearchScanPlan - from pypaimon.common.options.options import Options from pypaimon.index.index_file_handler import IndexFileHandler from pypaimon.read.push_down_utils import _get_all_fields from pypaimon.snapshot.time_travel_util import TimeTravelUtil @@ -95,13 +94,9 @@ class DataEvolutionVectorScan(VectorSearchScan): if field is not None: filter_field_ids.add(field.id) - snapshot = TimeTravelUtil.try_travel_to_snapshot( - Options(self._table.table_schema.options), - self._table.tag_manager(), - self._table.snapshot_manager(), - ) + snapshot = TimeTravelUtil.resolve_snapshot(self._table) if snapshot is None: - snapshot = self._table.snapshot_manager().get_latest_snapshot() + return VectorSearchScanPlan([]) index_file_handler = IndexFileHandler(table=self._table) diff --git a/paimon-python/pypaimon/tests/multimodal_table_test.py b/paimon-python/pypaimon/tests/multimodal_table_test.py index 50975a3892..e4264ae914 100644 --- a/paimon-python/pypaimon/tests/multimodal_table_test.py +++ b/paimon-python/pypaimon/tests/multimodal_table_test.py @@ -2199,7 +2199,9 @@ class MultimodalTableTest(unittest.TestCase): def execute_local(self): return GlobalIndexResult.from_range(Range(1, 1)) - docs.raw_table.new_vector_search_builder = lambda: FakeVectorBuilder() + patcher = patch.object(type(docs.raw_table), "new_vector_search_builder", return_value=FakeVectorBuilder()) + patcher.start() + self.addCleanup(patcher.stop) result = ( docs.search([0.0, 1.0, 0.0]) @@ -2250,7 +2252,9 @@ class MultimodalTableTest(unittest.TestCase): def execute_local(self): return GlobalIndexResult.from_range(Range(0, 0)) - docs.raw_table.new_vector_search_builder = lambda: FakeVectorBuilder() + patcher = patch.object(type(docs.raw_table), "new_vector_search_builder", return_value=FakeVectorBuilder()) + patcher.start() + self.addCleanup(patcher.stop) docs.search( [1.0, 0.0, 0.0], @@ -2400,7 +2404,9 @@ class MultimodalTableTest(unittest.TestCase): def execute_local(self): return GlobalIndexResult.from_range(Range(0, 0)) - docs.raw_table.new_vector_search_builder = lambda: FakeVectorBuilder() + patcher = patch.object(type(docs.raw_table), "new_vector_search_builder", return_value=FakeVectorBuilder()) + patcher.start() + self.addCleanup(patcher.stop) result = docs.search((v for v in [1.0, 0.0, 0.0])).limit(1).to_list() @@ -2435,7 +2441,9 @@ class MultimodalTableTest(unittest.TestCase): def execute_local(self): return GlobalIndexResult.from_range(Range(0, 0)) - docs.raw_table.new_vector_search_builder = lambda: FakeVectorBuilder() + patcher = patch.object(type(docs.raw_table), "new_vector_search_builder", return_value=FakeVectorBuilder()) + patcher.start() + self.addCleanup(patcher.stop) result = ( docs.search([1.0, 0.0, 0.0], column="embedding") @@ -2505,8 +2513,10 @@ class MultimodalTableTest(unittest.TestCase): GlobalIndexResult.from_range(Range(2, 2)), ] - docs.raw_table.new_batch_vector_search_builder = ( - lambda: FakeBatchVectorBuilder()) + patcher = patch.object(type(docs.raw_table), "new_batch_vector_search_builder", + return_value=FakeBatchVectorBuilder()) + patcher.start() + self.addCleanup(patcher.stop) result = ( docs.search_vectors( @@ -2573,7 +2583,9 @@ class MultimodalTableTest(unittest.TestCase): def execute_local(self): return GlobalIndexResult.from_range(Range(0, 0)) - docs.raw_table.new_full_text_search_builder = lambda: FakeFullTextBuilder() + patcher = patch.object(type(docs.raw_table), "new_full_text_search_builder", return_value=FakeFullTextBuilder()) + patcher.start() + self.addCleanup(patcher.stop) result = ( docs.search("paimon vector") @@ -2664,7 +2676,9 @@ class MultimodalTableTest(unittest.TestCase): def execute_local(self): return GlobalIndexResult.from_range(Range(0, 1)) - docs.raw_table.new_hybrid_search_builder = lambda: FakeHybridBuilder() + patcher = patch.object(type(docs.raw_table), "new_hybrid_search_builder", return_value=FakeHybridBuilder()) + patcher.start() + self.addCleanup(patcher.stop) result = ( docs.search_hybrid( @@ -2726,7 +2740,9 @@ class MultimodalTableTest(unittest.TestCase): def execute_local(self): return GlobalIndexResult.from_range(Range(0, 0)) - docs.raw_table.new_hybrid_search_builder = lambda: FakeHybridBuilder() + patcher = patch.object(type(docs.raw_table), "new_hybrid_search_builder", return_value=FakeHybridBuilder()) + patcher.start() + self.addCleanup(patcher.stop) result = ( docs.search_hybrid( @@ -2805,7 +2821,9 @@ class MultimodalTableTest(unittest.TestCase): def execute_local(self): return GlobalIndexResult.from_range(Range(0, 0)) - docs.raw_table.new_hybrid_search_builder = lambda: FakeHybridBuilder() + patcher = patch.object(type(docs.raw_table), "new_hybrid_search_builder", return_value=FakeHybridBuilder()) + patcher.start() + self.addCleanup(patcher.stop) result = ( docs.search_hybrid( 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 5a599f2699..2b9329dcd2 100644 --- a/paimon-python/pypaimon/tests/primary_key_index_definitions_test.py +++ b/paimon-python/pypaimon/tests/primary_key_index_definitions_test.py @@ -18,12 +18,14 @@ import unittest from unittest import mock from types import SimpleNamespace -from pypaimon.common.options.core_options import GlobalIndexSearchMode +from pypaimon.common.options.core_options import CoreOptions, GlobalIndexSearchMode +from pypaimon.common.options.options import Options from pypaimon.index.pk.primary_key_index_definitions import PrimaryKeyIndexDefinitions from pypaimon.index.pk.primary_key_index_source_file import PrimaryKeyIndexSourceFile from pypaimon.index.pk.primary_key_index_source_meta import PrimaryKeyIndexSourceMeta from pypaimon.schema.data_types import AtomicType, DataField from pypaimon.schema.table_schema import TableSchema +from pypaimon.table.file_store_table import FileStoreTable from pypaimon.table.source.full_text_search_builder import FullTextSearchBuilderImpl from pypaimon.table.source.primary_key_full_text_read import PrimaryKeyFullTextRead from pypaimon.table.source.primary_key_full_text_scan import PrimaryKeyFullTextScan @@ -311,6 +313,7 @@ class PrimaryKeyIndexDefinitionsTest(unittest.TestCase): "scan.tag-name": "old-tag"}) snapshot = SimpleNamespace(id=17) copied_options = [] + source_snapshots = [] index_snapshots = [] class _ScanTable: @@ -318,6 +321,7 @@ class PrimaryKeyIndexDefinitionsTest(unittest.TestCase): fields = schema.fields def new_read_builder(self): + source_snapshots.append(self._read_snapshot) plan = SimpleNamespace(splits=lambda: []) scan = SimpleNamespace(plan=lambda: plan) return SimpleNamespace(new_scan=lambda: scan) @@ -325,6 +329,8 @@ class PrimaryKeyIndexDefinitionsTest(unittest.TestCase): class _Table: table_schema = schema fields = schema.fields + options = CoreOptions(Options(schema.options)) + _copy_with_snapshot = FileStoreTable._copy_with_snapshot def tag_manager(self): return None @@ -332,7 +338,7 @@ class PrimaryKeyIndexDefinitionsTest(unittest.TestCase): def snapshot_manager(self): return SimpleNamespace(get_latest_snapshot=lambda: snapshot) - def copy(self, options): + def copy_without_time_travel(self, options): copied_options.append(options) return _ScanTable() @@ -354,10 +360,12 @@ class PrimaryKeyIndexDefinitionsTest(unittest.TestCase): _Table(), schema.fields[0], index_type="vindex").scan() self.assertEqual(17, plan.snapshot_id) + self.assertEqual([snapshot], source_snapshots) self.assertEqual([snapshot], index_snapshots) 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"]) 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 new file mode 100644 index 0000000000..39bd387c9c --- /dev/null +++ b/paimon-python/pypaimon/tests/search_snapshot_test.py @@ -0,0 +1,233 @@ +# 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. + +import threading +from unittest.mock import patch + +import pyarrow as pa +import pytest + +import pypaimon.multimodal as pm +from pypaimon.multimodal.query import ScanQuery +from pypaimon.globalindex.vector_search_result import DictBasedScoredIndexResult +from pypaimon.table.source.full_text_scan import FullTextScanPlan +from pypaimon.table.source.full_text_search_split import RawFullTextSearchSplit +from pypaimon.table.source.vector_search_scan import DataEvolutionVectorScan +from pypaimon.utils.range import Range + +SCHEMA = pa.schema([ + ("id", pa.int64()), ("category", pa.string()), ("embedding", pa.list_(pa.float32(), 2))]) +DATA = pa.table({"id": [1, 2], "category": ["allowed", "allowed"], + "embedding": [[0, 0], [10, 0]]}, schema=SCHEMA) + + [email protected] +def table(tmp_path): + return pm.connect(options={"warehouse": str(tmp_path)}).create_table( + "vectors", schema=SCHEMA, + options={"file.format": "parquet", "vector.file.format": "parquet", + "vector-index.search-mode": "full"}) + + +def search(table, kind="vector", **options): + if kind == "batch": + query = table.search_vectors([[0, 0], [0, 0]], **options) + elif kind == "hybrid": + query = table.search_hybrid([ + pm.vector_route("embedding", [0, 0]), pm.vector_route("embedding", [1, 0])], **options) + else: + query = table.search([0, 0], **options) + return query.pre_filter("category = 'allowed'").select(["id", "category"]).limit(1) + + [email protected]("kind", ["vector", "batch", "hybrid"]) [email protected]("change", ["update", "delete"]) +def test_search_and_lookup_share_snapshot_and_query_is_reusable(table, kind, change): + table.add(DATA) + query = search(table, kind) + original_table = query._table + expected = query.to_list() + expected_row = [{"id": 1, "category": "allowed"}] + assert expected == ([expected_row, expected_row] if kind == "batch" else expected_row) + lookup = ScanQuery._read_global_index_result + lookups = [] + + def commit_before_lookup(execution, result): + if not lookups: + if change == "delete": + table.delete("id = 1") + else: + table.update("id = 1", {"category": "blocked"}) + lookups.append(execution._table) + return lookup(execution, result) + + with patch.object(ScanQuery, "_read_global_index_result", commit_before_lookup): + assert query.to_list() == expected + assert query._table is original_table + assert all(t is lookups[0] for t in lookups) + latest = [{"id": 2, "category": "allowed"}] + assert query.to_list() == ([latest, latest] if kind == "batch" else latest) + + +def test_hybrid_builder_pins_snapshot_before_routes_start(table): + table.add(DATA) + builder = table.raw_table.new_hybrid_search_builder().with_limit(1) + builder.add_vector_route("embedding", [0, 0], limit=1) + builder.add_vector_route("embedding", [1, 0], limit=1) + original_scan = DataEvolutionVectorScan.scan + snapshots = [] + lock = threading.Lock() + + def scan_and_commit(scan): + with lock: + plan = original_scan(scan) + if not snapshots: + table.delete("id = 1") + snapshots.append(plan.snapshot().id) + return plan + + with patch.object(DataEvolutionVectorScan, "scan", scan_and_commit): + result = builder.execute_local() + assert snapshots == [1, 1] + assert list(result.results()) == [0] + + +def test_empty_search_does_not_follow_first_commit(table): + query = search(table) + original_scan = DataEvolutionVectorScan.scan + + def append_before_scan(scan): + table.add(DATA) + return original_scan(scan) + + with patch.object(DataEvolutionVectorScan, "scan", append_before_scan): + assert query.to_list() == [] + assert query.to_list() == [{"id": 1, "category": "allowed"}] + + [email protected]("selector", ["snapshot", "tag", "version", "timestamp"]) +def test_time_travel_survives_concurrent_writes_and_retained_tag_snapshot(table, selector): + from pypaimon.multimodal.query import VectorQuery + + table.add(DATA) + source = table.raw_table.snapshot_manager().get_latest_snapshot() + if selector in ("tag", "version"): + table.raw_table.create_tag("training", snapshot_id=source.id) + if selector == "tag": + query = search(table, tag_name="training") + else: + read_table = table.raw_table.copy({"scan.version": "training"}) + query = VectorQuery(read_table, [0, 0], "embedding").select(["id", "category"]).limit(1) + # Tags retain manifests even after their original snapshot JSON expires. + table.add(pa.table({"id": [3], "category": ["allowed"], + "embedding": [[20, 0]]}, schema=SCHEMA)) + table.raw_table.file_io.delete(table.raw_table.snapshot_manager().get_snapshot_path(source.id)) + elif selector == "timestamp": + read_table = table.raw_table.copy({"scan.timestamp-millis": str(source.time_millis)}) + query = VectorQuery(read_table, [0, 0], "embedding").select(["id", "category"]).limit(1) + else: + query = search(table, snapshot_id=source.id) + lookup = ScanQuery._read_global_index_result + + def update_before_lookup(execution, result): + 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) + # Read-option copies must retain the captured snapshot too. + execution._table = execution._table.copy({"read.batch-size": "2"}) + return lookup(execution, result) + + with patch.object(ScanQuery, "_read_global_index_result", update_before_lookup): + assert query.to_list() == [{"id": 1, "category": "allowed"}] + + +def test_full_text_plan_pins_live_rows_and_raw_fallback_without_leaking_to_read(table): + table.add(DATA) + snapshot = table.raw_table.snapshot_manager().get_latest_snapshot() + splits = [RawFullTextSearchSplit([Range(0, 1)])] + plan = FullTextScanPlan(splits, snapshot) + reader = (table.raw_table.new_full_text_search_builder() + .with_query("category", "allowed").with_limit(2).new_full_text_read()) + table.delete("id = 1") + live, raw = [], [] + + def indexed_query(unused, rows): + live.append(list(rows)) + return DictBasedScoredIndexResult({}) + + def raw_index(row_ids, texts, offset): + raw.append((row_ids, texts)) + return None + + with patch.object(reader, "_eval_column_query", indexed_query), \ + patch.object(reader, "_build_raw_index", raw_index): + reader.read_plan(plan) + reader.read(splits) + assert live == [[0, 1], [1]] + assert raw == [([0, 1], ["allowed", "allowed"]), ([1], ["allowed"])] + + +def test_read_view_copy_retains_snapshot_unless_selector_changes(table): + from pypaimon.multimodal.query import VectorQuery + + table.add(DATA) + view = search(table)._for_execution()._table + table.update("id = 1", {"category": "blocked"}) + + def read(read_table): + return VectorQuery(read_table, [0, 0], "embedding").select(["category"]).limit(1).to_list() + + assert read(view.copy({"read.batch-size": "2"})) == [{"category": "allowed"}] + latest = table.raw_table.snapshot_manager().get_latest_snapshot() + assert read(view.copy({"scan.snapshot-id": str(latest.id)})) == [{"category": "blocked"}] + assert read(view.copy({"scan.snapshot-id": None, "scan.mode": "default"})) == [{"category": "blocked"}] + table.raw_table.branch_manager().create_branch("empty") + assert read(view.copy({"branch": "empty", "scan.snapshot-id": None, "scan.mode": "default"})) == [] + + +def test_indexed_search_lookup_uses_planned_snapshot(tmp_path): + pytest.importorskip("paimon_vindex") + from pypaimon.read.table_scan import TableScan + + table = pm.connect(options={"warehouse": str(tmp_path)}).create_table( + "indexed", schema=SCHEMA, + options={"file.format": "parquet", "vector.file.format": "parquet", + "deletion-vectors.enabled": "false"}) + table.add(DATA) + table.create_index("embedding", index_type="ivf-flat", options={ + "ivf-flat.dimension": "2", "ivf-flat.nlist": "1", "ivf-flat.distance.metric": "l2"}) + query = table.search([0, 0]).select(["id"]).limit(1) + snapshot = table.raw_table.snapshot_manager().get_latest_snapshot() + lookup = ScanQuery._read_global_index_result + plan = TableScan.plan + lookup_snapshots = [] + + def record_plan(scan): + result = plan(scan) + lookup_snapshots.append(result.snapshot_id) + return result + + def append_before_lookup(execution, result): + table.add(DATA) + with patch.object(TableScan, "plan", record_plan): + return lookup(execution, result) + + with patch.object(ScanQuery, "_read_global_index_result", append_before_lookup): + assert query.to_list() == [{"id": 1}] + assert lookup_snapshots == [snapshot.id] + assert table.raw_table.snapshot_manager().get_latest_snapshot().id > snapshot.id diff --git a/paimon-python/pypaimon/tests/vector_search_filter_test.py b/paimon-python/pypaimon/tests/vector_search_filter_test.py index 39daa2bbd5..1d2731666f 100644 --- a/paimon-python/pypaimon/tests/vector_search_filter_test.py +++ b/paimon-python/pypaimon/tests/vector_search_filter_test.py @@ -44,6 +44,7 @@ from pypaimon.globalindex.vector_search_result import ScoredGlobalIndexResult from pypaimon.index.index_file_meta import IndexFileMeta from pypaimon.manifest.index_manifest_entry import IndexManifestEntry from pypaimon.schema.data_types import AtomicType, DataField +from pypaimon.table.file_store_table import FileStoreTable from pypaimon.table.row.generic_row import GenericRow from pypaimon.table.source.vector_search_builder import VectorSearchBuilderImpl from pypaimon.utils.roaring_bitmap import RoaringBitmap64 @@ -106,7 +107,7 @@ class _StubTable: return None def snapshot_manager(self): - return None + return types.SimpleNamespace(get_latest_snapshot=lambda: None) def path_factory(self): class _P: @@ -121,7 +122,10 @@ class _StubTable: return self def copy_without_time_travel(self, options): - return self + from copy import copy + return copy(self) + + _copy_with_snapshot = FileStoreTable._copy_with_snapshot def new_vector_search_builder(self): from pypaimon.table.source.vector_search_builder import ( @@ -361,6 +365,7 @@ class GlobalIndexLiveRowFilterTest(unittest.TestCase): return False class _Table: + _copy_with_snapshot = FileStoreTable._copy_with_snapshot options = _Options() def new_read_builder(self_inner): @@ -412,6 +417,7 @@ class GlobalIndexLiveRowFilterTest(unittest.TestCase): return _Scan() class _Table: + _copy_with_snapshot = FileStoreTable._copy_with_snapshot options = _Options() table_schema = _StubSchema() file_io = object() @@ -491,6 +497,7 @@ class GlobalIndexLiveRowFilterTest(unittest.TestCase): return True class _Table: + _copy_with_snapshot = FileStoreTable._copy_with_snapshot options = _Options() file_io = object() @@ -3146,7 +3153,9 @@ 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) def tearDown(self): mock.patch.stopall()
