[python] Keep local search and lookup on one snapshot

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(

paimon-python/pypaimon/globalindex/data_evolution_global_index_scanner.py

@@ -344,6 +344,8 @@ def close(self):
 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

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
@@ -446,6 +447,15 @@ def _and_predicate(self, left, right):
 
 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,
@@ -504,7 +514,7 @@ def __init__(
     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)
@@ -526,7 +536,7 @@ def __init__(self, table, text_query, pre_filter=None):
     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)
         )
@@ -561,7 +571,7 @@ def _execute_hybrid(self, query):
         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)
         )
@@ -602,9 +612,10 @@ def __init__(
         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):
@@ -616,7 +627,7 @@ def to_list(self) -> List[List[dict]]:
     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)

paimon-python/pypaimon/read/table_scan.py

@@ -447,9 +447,7 @@ def incremental_manifest():
 
         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."
@@ -466,7 +464,7 @@ def time_travel_manifest_scanner():
             )
 
         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(

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,

paimon-python/pypaimon/table/file_store_table.py

@@ -516,6 +516,22 @@ def copy_without_time_travel(self, options: dict) -> 'FileStoreTable':
         """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")
@@ -531,7 +547,12 @@ def _copy(self, options: dict, resolve_time_travel: bool) -> 'FileStoreTable':
         # 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
@@ -554,6 +575,8 @@ def _copy(self, options: dict, resolve_time_travel: bool) -> 'FileStoreTable':
         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]:

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 @@ def __init__(
         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:

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

@@ -36,8 +36,12 @@
 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 @@ def scan(self) -> FullTextScanPlan:
         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 @@ def index_file_filter(entry):
             if raw_row_ranges:
                 splits.append(RawFullTextSearchSplit(raw_row_ranges))
 
-        return FullTextScanPlan(splits)
+        return FullTextScanPlan(splits, snapshot)
 
 
 def _supports_full_text_search(index_type):

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

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 @@ def add_route(self, route: HybridSearchRoute) -> '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(

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.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 @@ def __init__(self, table, definition, partition_filter=None):
         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)

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.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 @@ def __init__(self, table, vector_column, filter_=None,
         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)