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 b7931efc06 [python] Keep local search and lookup on one snapshot
(#9843)
b7931efc06 is described below
commit b7931efc0643d1b52ef3d440a0bee61a422e64e8
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()
