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 af4b262962 [Python] Delegate local full-text search to Rust core
(#10494)
af4b262962 is described below
commit af4b26296257cc89e64fb6883217e2a522113436
Author: Jingsong Lee <[email protected]>
AuthorDate: Sat Oct 10 23:11:05 2026 +0800
[Python] Delegate local full-text search to Rust core (#10494)
---
paimon-python/README.md | 35 ++
paimon-python/pypaimon/common/like_optimization.py | 37 ++
.../globalindex/btree/btree_index_reader.py | 61 +--
.../globalindex/sorted_file_global_index_reader.py | 21 +-
.../globalindex/sorted_file_meta_selector.py | 11 +-
.../pypaimon/table/source/full_text_read.py | 76 +++-
.../pypaimon/table/source/full_text_scan.py | 31 +-
.../table/source/full_text_search_builder.py | 9 +
.../table/source/full_text_search_split.py | 3 +-
.../table/source/global_index_row_filter.py | 25 ++
.../table/source/native_full_text_search.py | 61 +++
.../pypaimon/table/source/vector_search_read.py | 4 +-
.../pypaimon/tests/btree_index_reader_test.py | 69 +++
.../pypaimon/tests/full_text_scalar_filter_test.py | 53 ++-
.../pypaimon/tests/native_full_text_search_test.py | 466 +++++++++++++++++++++
.../pypaimon/tests/vector_filter_exactness_test.py | 45 +-
16 files changed, 909 insertions(+), 98 deletions(-)
diff --git a/paimon-python/README.md b/paimon-python/README.md
index 3702f659ee..b0f5a2eb3e 100644
--- a/paimon-python/README.md
+++ b/paimon-python/README.md
@@ -789,6 +789,41 @@ falls back to the Python search reader. Distributed Ray
searches keep their
existing scan/read path.
+# Native local full-text search
+
+For REST Data Evolution tables with row tracking and Parquet files,
+`read.native.enabled=true` delegates local full-text search to Rust core.
+The Python builder and scored result API stay the same:
+
+```python
+import json
+
+native_table = table.copy({'read.native.enabled': 'true'})
+predicate =
native_table.new_read_builder().new_predicate_builder().equal('category',
'news')
+result = (native_table.new_full_text_search_builder()
+ .with_query('content', json.dumps({'match': {'query': 'paimon'}}))
+ .with_filter(predicate)
+ .with_limit(10)
+ .execute_local())
+```
+
+Rust selects one snapshot, prunes partitions, evaluates scalar filters before
+ranking, and searches indexed and uncovered rows. Full-text and scalar
+`fast`, `full`, and `detail` search modes follow Java coverage semantics.
+Scalar candidate refinement obeys `global-index.filter.refine-from-data`.
+Search predicates `contains`, `endsWith`, and residual `LIKE` require
+refinement for both BTree and bitmap indexes, as in Java. Simple `LIKE`
+patterns are first rewritten to equality, prefix, suffix or substring
+predicates; equality and prefix retain the underlying index's exactness.
+Filtering keeps the complete shard or raw corpus for BM25 scoring.
+
+A full-text index definition must exist even in `full` or `detail` mode.
+Tables outside the supported route, or Rust versions without the full-text
+builder, use the Python implementation. Once native execution starts,
+errors propagate without retrying the search through Python. PK and distributed
+searches retain their existing Python scan/read path.
+
+
# Vector index range reads
Native vector indexes (`ivf-flat`, `ivf-pq`, `ivf-sq`, `ivf-rq`, and `diskann`)
diff --git a/paimon-python/pypaimon/common/like_optimization.py
b/paimon-python/pypaimon/common/like_optimization.py
new file mode 100644
index 0000000000..d90fb1641c
--- /dev/null
+++ b/paimon-python/pypaimon/common/like_optimization.py
@@ -0,0 +1,37 @@
+# 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.
+
+"""Simple LIKE rewrites shared by scalar readers and search filtering."""
+
+
+def try_optimize_like(literal):
+ """Follow Java LikeOptimization, leaving escaped/underscore patterns
intact."""
+ if literal is None:
+ return None
+ pattern = str(literal)
+ if "_" in pattern or "\\" in pattern:
+ return None
+ if "%" not in pattern and pattern:
+ return "equal", pattern
+ if (pattern.startswith("%") and pattern.endswith("%")
+ and pattern.count("%") == 2 and pattern[1:-1]):
+ return "contains", pattern[1:-1]
+ if pattern.startswith("%") and pattern.count("%") == 1 and pattern[1:]:
+ return "ends_with", pattern[1:]
+ if pattern.endswith("%") and pattern.count("%") == 1 and pattern[:-1]:
+ return "starts_with", pattern[:-1]
+ return None
diff --git a/paimon-python/pypaimon/globalindex/btree/btree_index_reader.py
b/paimon-python/pypaimon/globalindex/btree/btree_index_reader.py
index f45e8e9e9d..8e9f21e6e3 100644
--- a/paimon-python/pypaimon/globalindex/btree/btree_index_reader.py
+++ b/paimon-python/pypaimon/globalindex/btree/btree_index_reader.py
@@ -36,6 +36,7 @@ from pypaimon.utils.roaring_bitmap import RoaringBitmap64
from pypaimon.globalindex.btree.btree_file_footer import BTreeFileFooter
from pypaimon.globalindex.btree.sst_file_reader import SstFileReader
from pypaimon.globalindex.memory_slice_input import MemorySliceInput
+from pypaimon.globalindex.sorted_file_meta_selector import
SortedFileMetaSelector
def _deserialize_row_ids(data: bytes) -> List[int]:
@@ -163,43 +164,39 @@ class BTreeIndexReader:
return RoaringBitmap64()
return self._range_query(self.min_key, self.max_key, True, True)
- def _range_query(
- self,
- from_key: object,
- to_key: object,
- from_inclusive: bool,
- to_inclusive: bool
- ) -> RoaringBitmap64:
- result = RoaringBitmap64()
-
+ def _entries_from(self, serialized_key: bytes):
file_iter = self.reader.create_iterator()
- file_iter.seek_to(self.key_serializer.serialize(from_key))
-
+ file_iter.seek_to(serialized_key)
while True:
data_iter = file_iter.read_batch()
if data_iter is None:
- break
-
+ return
while data_iter.has_next():
entry = data_iter.__next__()
if entry is None:
break
+ yield entry
- key_bytes = entry.key
- value_bytes = entry.value
-
- key = self.key_serializer.deserialize(key_bytes)
+ def _range_query(
+ self,
+ from_key: object,
+ to_key: object,
+ from_inclusive: bool,
+ to_inclusive: bool
+ ) -> RoaringBitmap64:
+ result = RoaringBitmap64()
+ for entry in
self._entries_from(self.key_serializer.serialize(from_key)):
+ key = self.key_serializer.deserialize(entry.key)
- if not from_inclusive and self.comparator(key, from_key) == 0:
- continue
+ if not from_inclusive and self.comparator(key, from_key) == 0:
+ continue
- difference = self.comparator(key, to_key)
- if difference > 0 or (not to_inclusive and difference == 0):
- return result
+ difference = self.comparator(key, to_key)
+ if difference > 0 or (not to_inclusive and difference == 0):
+ return result
- row_ids = _deserialize_row_ids(value_bytes)
- for row_id in row_ids:
- result.add(row_id)
+ for row_id in _deserialize_row_ids(entry.value):
+ result.add(row_id)
return result
@@ -248,7 +245,19 @@ class BTreeIndexReader:
return GlobalIndexResult.create(result)
def visit_starts_with(self, literal: object) ->
Optional[GlobalIndexResult]:
- return GlobalIndexResult.create(self._all_non_null_rows(),
is_exact=False)
+ result = RoaringBitmap64()
+ if self.min_key is None:
+ return GlobalIndexResult.create(result)
+ prefix = self.key_serializer.serialize(literal)
+ upper_bound = SortedFileMetaSelector._prefix_upper_bound(prefix)
+ # STRING keys follow Java's unsigned UTF-8 order. The exclusive byte
+ # boundary need not be a valid Unicode string, so do not deserialize
it.
+ for entry in self._entries_from(prefix):
+ if upper_bound is not None and entry.key >= upper_bound:
+ break
+ for row_id in _deserialize_row_ids(entry.value):
+ result.add(row_id)
+ return GlobalIndexResult.create(result)
def visit_ends_with(self, literal: object) -> Optional[GlobalIndexResult]:
return GlobalIndexResult.create(self._all_non_null_rows(),
is_exact=False)
diff --git
a/paimon-python/pypaimon/globalindex/sorted_file_global_index_reader.py
b/paimon-python/pypaimon/globalindex/sorted_file_global_index_reader.py
index 273a701c74..77a9594cb9 100644
--- a/paimon-python/pypaimon/globalindex/sorted_file_global_index_reader.py
+++ b/paimon-python/pypaimon/globalindex/sorted_file_global_index_reader.py
@@ -21,6 +21,7 @@ import threading
from concurrent.futures import Executor, Future
from typing import Callable, Dict, List, Optional
+from pypaimon.common.like_optimization import try_optimize_like
from pypaimon.globalindex.global_index_meta import GlobalIndexIOMeta
from pypaimon.globalindex.global_index_reader import FieldRef,
GlobalIndexReader, _completed_future
from pypaimon.globalindex.global_index_result import GlobalIndexResult
@@ -225,7 +226,7 @@ class SortedFileGlobalIndexReader(GlobalIndexReader):
if not self._is_string_type(field_ref) or literal is None:
return _completed_future(None)
- optimized = self._try_optimize_like(literal)
+ optimized = try_optimize_like(literal)
if optimized is not None:
method, optimized_literal = optimized
if method == "equal":
@@ -282,24 +283,6 @@ class SortedFileGlobalIndexReader(GlobalIndexReader):
or data_type == "VARCHAR"
or data_type.startswith("VARCHAR("))
- @staticmethod
- def _try_optimize_like(literal):
- pattern = str(literal)
- if "_" in pattern:
- return None
- if "%" not in pattern and pattern:
- return "equal", pattern
- if (pattern.startswith("%")
- and pattern.endswith("%")
- and pattern.count("%") == 2
- and pattern[1:-1]):
- return "contains", pattern[1:-1]
- if pattern.startswith("%") and pattern.count("%") == 1 and pattern[1:]:
- return "ends_with", pattern[1:]
- if pattern.endswith("%") and pattern.count("%") == 1 and pattern[:-1]:
- return "starts_with", pattern[:-1]
- return None
-
def close(self) -> None:
with self._cache_lock:
for reader in self._reader_cache.values():
diff --git a/paimon-python/pypaimon/globalindex/sorted_file_meta_selector.py
b/paimon-python/pypaimon/globalindex/sorted_file_meta_selector.py
index f486d5900f..4fdc776bbd 100644
--- a/paimon-python/pypaimon/globalindex/sorted_file_meta_selector.py
+++ b/paimon-python/pypaimon/globalindex/sorted_file_meta_selector.py
@@ -53,17 +53,14 @@ class SortedFileMetaSelector:
if len(prefix) == 0:
return self._filter(lambda meta: not meta.only_nulls())
- prefix_key = self._key_serializer.deserialize(prefix)
upper_bound_bytes = self._prefix_upper_bound(prefix)
- upper_bound = (None if upper_bound_bytes is None
- else
self._key_serializer.deserialize(upper_bound_bytes))
-
+ # Prefix selection applies to STRING keys, whose serialized order is
+ # Java's unsigned UTF-8 order. A byte upper bound may not decode to
text.
return self._filter(
lambda meta: (
not meta.only_nulls()
- and self._compare_last_key(meta, prefix_key) >= 0
- and (upper_bound is None
- or self._compare_first_key(meta, upper_bound) < 0)))
+ and meta.last_key >= prefix
+ and (upper_bound_bytes is None or meta.first_key <
upper_bound_bytes)))
def select_ends_with(self, literal) -> List[GlobalIndexIOMeta]:
return self._filter(lambda meta: literal is not None and not
meta.only_nulls())
diff --git a/paimon-python/pypaimon/table/source/full_text_read.py
b/paimon-python/pypaimon/table/source/full_text_read.py
index c4236b6646..022e448c71 100644
--- a/paimon-python/pypaimon/table/source/full_text_read.py
+++ b/paimon-python/pypaimon/table/source/full_text_read.py
@@ -79,6 +79,7 @@ class DataEvolutionFullTextRead(FullTextRead):
def read_plan(self, plan: FullTextScanPlan) -> GlobalIndexResult:
reader = copy(self)
reader._table =
global_index_live_row_filter.table_at_snapshot(self._table, plan.snapshot())
+ reader._snapshot = plan.snapshot()
return reader.read(plan.splits())
def read(self, splits: List[FullTextSearchSplit]) -> GlobalIndexResult:
@@ -91,27 +92,72 @@ class DataEvolutionFullTextRead(FullTextRead):
splits_by_column.setdefault(split.column_name, []).append(split)
live_rows = global_index_live_row_filter.live_rows(
self._table, self._partition_filter)
- if self._filter is not None:
- from pypaimon.table.source.global_index_row_filter import
matching_rows
-
- candidates = RoaringBitmap64()
- for split in index_splits:
- candidates.add_range(split.row_range_start,
split.row_range_end)
- for row_range in _raw_row_ranges(raw_splits):
- candidates.add_range(row_range.from_, row_range.to)
- # Scalar indexes may prune this read, but incomplete scalar
coverage
- # must not exclude rows covered by the full-text search plan.
- table = self._table.copy({"scalar-index.search-mode": "full"})
- matched = matching_rows(table, self._filter, candidates,
self._partition_filter)
- live_rows = matched if live_rows is None else
RoaringBitmap64.and_(live_rows, matched)
- indexed_result = self._eval_column_query(splits_by_column, live_rows)
+ matched = self._matched_indexed_rows(index_splits, live_rows)
+ indexed_result = self._eval_column_query(splits_by_column, matched if
self._filter is not None else live_rows)
raw_ranges = _raw_row_ranges(raw_splits)
if self._filter is None:
raw_result = self._read_raw_search(raw_ranges,
_index_type(index_splits))
else:
- raw_result = self._read_raw_search(raw_ranges,
_index_type(index_splits), live_rows)
+ candidates = GlobalIndexResult.from_ranges(raw_ranges).results()
+ if live_rows is not None:
+ candidates = RoaringBitmap64.and_(candidates, live_rows)
+ matched = self._matching_rows(candidates)
+ raw_result = self._read_raw_search(raw_ranges,
_index_type(index_splits), matched)
return indexed_result.or_(raw_result).top_k(self._limit)
+ def _matching_rows(self, candidates):
+ from pypaimon.table.source.global_index_row_filter import matching_rows
+
+ table = self._table.copy({'scalar-index.search-mode': 'full'})
+ return matching_rows(table, self._filter, candidates,
self._partition_filter)
+
+ def _matched_indexed_rows(self, splits, live_rows):
+ if self._filter is None:
+ return live_rows
+ from pypaimon.common.options.core_options import GlobalIndexSearchMode
+ from pypaimon.globalindex.data_evolution_global_index_coverage import
DataEvolutionGlobalIndexCoverage
+ from pypaimon.globalindex.data_evolution_global_index_scanner import
DataEvolutionGlobalIndexScanner
+ from pypaimon.snapshot.time_travel_util import TimeTravelUtil
+ from pypaimon.table.source.global_index_row_filter import is_exact
+
+ covered = RoaringBitmap64()
+ files = {}
+ for split in splits:
+ covered.add_range(split.row_range_start, split.row_range_end)
+ files.update((file.file_name, file) for file in
split.scalar_index_files)
+ if live_rows is not None:
+ covered = RoaringBitmap64.and_(covered, live_rows)
+ if covered.is_empty():
+ return covered
+ snapshot = getattr(self, '_snapshot', None)
+ if snapshot is None:
+ snapshot = TimeTravelUtil.resolve_snapshot(self._table)
+ mode = self._table.options.scalar_index_search_mode()
+ scalar_files = list(files.values())
+ coverage = DataEvolutionGlobalIndexCoverage(self._table, snapshot,
self._partition_filter, scalar_files)
+ raw_ranges = coverage.unindexed_ranges(self._table.fields,
self._filter, mode)
+ unindexed =
RoaringBitmap64.and_(GlobalIndexResult.from_ranges(raw_ranges).results(),
covered)
+ decided = RoaringBitmap64.remove_all(covered, unindexed)
+ matched = RoaringBitmap64()
+ if not decided.is_empty():
+ scanner = DataEvolutionGlobalIndexScanner.create(self._table,
index_files=scalar_files,
+
snapshot=snapshot, partition_filter=self._partition_filter)
+ evaluation = None
+ if scanner is not None:
+ try:
+ evaluation = scanner.scan_with_coverage(self._filter)
+ finally:
+ scanner.close()
+ if evaluation is not None:
+ candidates = RoaringBitmap64.and_(evaluation.result.results(),
decided)
+ if is_exact(self._filter, evaluation.result):
+ matched = candidates
+ elif
self._table.options.global_index_filter_refine_from_data():
+ matched = self._matching_rows(candidates)
+ elif mode != GlobalIndexSearchMode.FAST:
+ unindexed = RoaringBitmap64.or_(unindexed, decided)
+ return RoaringBitmap64.or_(matched, self._matching_rows(unindexed))
+
def _eval_column_query(
self,
splits_by_column: Dict[str, List[IndexFullTextSearchSplit]],
diff --git a/paimon-python/pypaimon/table/source/full_text_scan.py
b/paimon-python/pypaimon/table/source/full_text_scan.py
index ae5d089d0b..57b7d3ea30 100644
--- a/paimon-python/pypaimon/table/source/full_text_scan.py
+++ b/paimon-python/pypaimon/table/source/full_text_scan.py
@@ -62,10 +62,12 @@ class DataEvolutionFullTextScan(FullTextScan):
self,
table: 'FileStoreTable',
text_columns,
- partition_filter=None):
+ partition_filter=None,
+ filter_=None):
self._table = table
self._text_columns = list(text_columns)
self._partition_filter = partition_filter
+ self._filter = filter_
def scan(self) -> FullTextScanPlan:
from pypaimon.index.index_file_handler import IndexFileHandler
@@ -96,8 +98,26 @@ class DataEvolutionFullTextScan(FullTextScan):
and _supports_full_text_search(entry.index_file.index_type)
)
- entries = index_file_handler.scan(snapshot, index_file_filter)
- all_index_files = [entry.index_file for entry in entries]
+ from pypaimon.read.push_down_utils import _get_all_fields
+ filter_names = _get_all_fields(self._filter) if self._filter is not
None else set()
+ filter_ids = {field.id for field in self._table.fields if field.name
in filter_names}
+
+ def selected(entry):
+ if index_file_filter(entry):
+ return True
+ if partition_filter is not None and not
partition_filter.test(entry.partition):
+ return False
+ file = entry.index_file
+ meta = file.global_index_meta
+ return (meta is not None and file.index_type in ('btree', 'bitmap')
+ and bool(filter_ids.intersection([meta.index_field_id] +
list(meta.extra_field_ids or []))))
+
+ entries = index_file_handler.scan(snapshot, selected)
+ all_index_files = [entry.index_file for entry in entries if
index_file_filter(entry)]
+ scalar_files = [entry.index_file for entry in entries if
entry.index_file.index_type in ('btree', 'bitmap')]
+ # Java requires an existing full-text definition before admitting raw
ranges.
+ if not all_index_files:
+ return FullTextScanPlan([], snapshot)
# Group full-text index files by column and (rowRangeStart,
rowRangeEnd).
by_column_and_range = defaultdict(lambda: defaultdict(list))
@@ -113,7 +133,10 @@ class DataEvolutionFullTextScan(FullTextScan):
for range_key, files in by_range.items():
splits.append(
IndexFullTextSearchSplit(
- column_name, range_key.from_, range_key.to, files))
+ column_name, range_key.from_, range_key.to, files,
+ [file for file in scalar_files
+ if file.global_index_meta.row_range_start <=
range_key.to
+ and file.global_index_meta.row_range_end >=
range_key.from_]))
raw_row_ranges = DataEvolutionGlobalIndexCoverage(
self._table,
diff --git a/paimon-python/pypaimon/table/source/full_text_search_builder.py
b/paimon-python/pypaimon/table/source/full_text_search_builder.py
index 72b34a92dd..3d2d4dd802 100644
--- a/paimon-python/pypaimon/table/source/full_text_search_builder.py
+++ b/paimon-python/pypaimon/table/source/full_text_search_builder.py
@@ -84,6 +84,14 @@ class FullTextSearchBuilderImpl(FullTextSearchBuilder):
self._limit = limit
return self
+ def execute_local(self) -> GlobalIndexResult:
+ from pypaimon.table.source.native_full_text_search import
try_native_full_text_search
+
+ result = try_native_full_text_search(self)
+ if result is not None:
+ return result
+ return super().execute_local()
+
def with_query(self, field_name: str, query: str) ->
'FullTextSearchBuilder':
self._field_name = field_name
self._query = query
@@ -149,6 +157,7 @@ class FullTextSearchBuilderImpl(FullTextSearchBuilder):
self._table,
self._text_columns(),
partition_filter=self._partition_filter,
+ filter_=self._filter,
)
def new_full_text_read(self) -> FullTextRead:
diff --git a/paimon-python/pypaimon/table/source/full_text_search_split.py
b/paimon-python/pypaimon/table/source/full_text_search_split.py
index bb2cdeda9e..e06d485a85 100644
--- a/paimon-python/pypaimon/table/source/full_text_search_split.py
+++ b/paimon-python/pypaimon/table/source/full_text_search_split.py
@@ -17,7 +17,7 @@
"""Splits of full-text search."""
-from dataclasses import dataclass
+from dataclasses import dataclass, field
from typing import List
from pypaimon.index.index_file_meta import IndexFileMeta
@@ -36,6 +36,7 @@ class IndexFullTextSearchSplit(FullTextSearchSplit):
row_range_start: int
row_range_end: int
full_text_index_files: List[IndexFileMeta]
+ scalar_index_files: List[IndexFileMeta] = field(default_factory=list)
@dataclass
diff --git a/paimon-python/pypaimon/table/source/global_index_row_filter.py
b/paimon-python/pypaimon/table/source/global_index_row_filter.py
index 0dc087f56a..2e9820c3e4 100644
--- a/paimon-python/pypaimon/table/source/global_index_row_filter.py
+++ b/paimon-python/pypaimon/table/source/global_index_row_filter.py
@@ -17,12 +17,37 @@
"""Exact row filtering for data-evolution search candidates."""
+from pypaimon.common.like_optimization import try_optimize_like
from pypaimon.read.table_read import _ClosableArrowBatchReader
from pypaimon.table.special_fields import SpecialFields
from pypaimon.table.source.global_index_live_row_filter import
table_at_snapshot
from pypaimon.utils.roaring_bitmap import RoaringBitmap64
+def is_exact(predicate, result):
+ """Apply Java FilteredRowIdReader's predicate contract before search Top-K.
+
+ Reader exactness alone is insufficient: contains, endsWith and residual
+ LIKE must be refined even when a bitmap dictionary yields exact rows.
+ Simple LIKE rewrites to equality/prefix retain their usual exactness.
+ """
+ if result is None or not result.is_exact():
+ return False
+ pending = [predicate] if predicate is not None else []
+ while pending:
+ leaf = pending.pop()
+ if leaf.method in ('and', 'or'):
+ pending.extend(leaf.literals)
+ continue
+ method = leaf.method
+ if method == 'like':
+ optimized = try_optimize_like(leaf.literals[0])
+ method = optimized[0] if optimized is not None else 'like'
+ if method in ('contains', 'endsWith', 'ends_with', 'like'):
+ return False
+ return True
+
+
def matching_rows(table, predicate, candidates, partition_filter=None,
snapshot=None):
"""Read only filter dependencies and row IDs, closing both reader and
source."""
matched = RoaringBitmap64()
diff --git a/paimon-python/pypaimon/table/source/native_full_text_search.py
b/paimon-python/pypaimon/table/source/native_full_text_search.py
new file mode 100644
index 0000000000..97a1484352
--- /dev/null
+++ b/paimon-python/pypaimon/table/source/native_full_text_search.py
@@ -0,0 +1,61 @@
+# 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.
+
+"""Local REST full-text search delegated to Rust core."""
+
+from pypaimon.catalog.table_query_auth import reject_search_under_query_auth
+from pypaimon.globalindex.full_text.native_full_text_global_index_reader
import NativeFullTextIndexOptions
+from pypaimon.globalindex.vector_search_result import
DictBasedScoredIndexResult
+from pypaimon.read.native_plan import _native_table, _predicate_to_native,
native_method_available
+from pypaimon.table.file_store_table import FileStoreTable
+from pypaimon.write.native_commit import _rest_catalog_supported
+
+
+def try_native_full_text_search(builder):
+ """Qualify the route before execution; Rust owns scan, filtering and
ranking."""
+ table = builder._table
+ reject_search_under_query_auth(table)
+ if (type(table) is not FileStoreTable
+ or not table.options.native_read_enabled()
+ or not table.options.data_evolution_enabled()
+ or not table.options.row_tracking_enabled()
+ or table.is_primary_key_table
+ or table.options.file_format() != 'parquet'
+ or not native_method_available('Table',
'new_full_text_search_builder')
+ or not _rest_catalog_supported(table)):
+ return None
+ if builder._limit is None or builder._limit <= 0:
+ raise ValueError('Limit must be positive, set via with_limit()')
+ # The full-text adapter accepts typed lists and booleans. Reuse its mapping
+ # before the generic table bridge stringifies options for Rust.
+ options = {'full-text.' + key: value for key, value in
NativeFullTextIndexOptions.from_options(
+ table.options.options.to_map()).to_native_options().items()}
+ native = _native_table(table.copy(options)).new_full_text_search_builder()
+ if builder._field_name is not None and builder._query is not None:
+ native.with_query(builder._field_name, builder._query)
+ native.with_limit(builder._limit)
+ try:
+ if builder._partition_filter is not None:
+
native.with_partition_filter(_predicate_to_native(builder._partition_filter))
+ if builder._filter is not None:
+ native.with_filter(_predicate_to_native(builder._filter))
+ except ValueError:
+ # Qualify literal support before searching. Some Python literals cannot
+ # be represented by the strict typed predicate FFI.
+ return None
+ # Once execution starts, propagate errors instead of running a second
search.
+ return DictBasedScoredIndexResult(native.execute_local().row_ids())
diff --git a/paimon-python/pypaimon/table/source/vector_search_read.py
b/paimon-python/pypaimon/table/source/vector_search_read.py
index 839993fbed..20d661f709 100644
--- a/paimon-python/pypaimon/table/source/vector_search_read.py
+++ b/paimon-python/pypaimon/table/source/vector_search_read.py
@@ -224,7 +224,9 @@ class AbstractVectorSearchReadImpl:
scanner.close()
def _refine_scalar_result(self, splits, result, snapshot=None):
- if result is not None and result.is_exact():
+ from pypaimon.table.source.global_index_row_filter import is_exact
+
+ if is_exact(self._filter, result):
return result.results()
if not self._table.options.global_index_filter_refine_from_data():
logging.getLogger(__name__).warning(
diff --git a/paimon-python/pypaimon/tests/btree_index_reader_test.py
b/paimon-python/pypaimon/tests/btree_index_reader_test.py
index 45184640d1..c1e396467e 100644
--- a/paimon-python/pypaimon/tests/btree_index_reader_test.py
+++ b/paimon-python/pypaimon/tests/btree_index_reader_test.py
@@ -29,6 +29,8 @@ from pypaimon.globalindex.btree.btree_index_reader import
BTreeIndexReader
from pypaimon.globalindex.btree.btree_index_writer import BTreeIndexWriter
from pypaimon.globalindex.global_index_meta import GlobalIndexIOMeta
from pypaimon.globalindex.key_serializer import create_serializer
+from pypaimon.globalindex.sorted_file_meta_selector import
SortedFileMetaSelector
+from pypaimon.globalindex.sorted_index_file_meta import SortedIndexFileMeta
from pypaimon.schema.data_types import AtomicType
@@ -111,6 +113,73 @@ class BTreeIndexReaderTest(unittest.TestCase):
reader.close()
self.assertTrue(file_io.input_stream.closed)
+ def test_string_prefix_intervals_are_exact(self):
+ values = [None, '', '\x00', 'ke', 'keep', 'keep', 'keeper', 'kf',
'missing-after',
+ '\x7f', '\x7fa', '\u0080', '\u00ff', '\u00ffa', '\u0100',
+ '\u07ff', '\u07ffa', '\u0800', '\uffff', '\uffffa',
'\U00010000',
+ '\U0010ffff', '\U0010ffffa']
+ serializer = create_serializer(AtomicType('STRING'))
+ with tempfile.TemporaryDirectory() as directory:
+ file_io = LocalFileIO()
+ writer = BTreeIndexWriter(file_io, directory, serializer,
block_size=32)
+ for row_id, value in sorted(enumerate(values), key=lambda item:
(item[1] is not None, item[1])):
+ writer.write(value, row_id)
+ entry = writer.finish()[0]
+ with open(os.path.join(directory, entry.file_name), 'rb') as
stream:
+ data = stream.read()
+ io_meta = GlobalIndexIOMeta(entry.file_name, len(data), entry.meta)
+ for input_type in (_TrackingInput, _PreadInput):
+ reader = BTreeIndexReader(serializer, self._file_io(data,
input_type), '/unused', io_meta)
+ try:
+ for prefix in ('', '\x00', 'ke', 'keep', 'missing', 'absent',
'zz',
+ '\x7f', '\u00ff', '\u07ff', '\uffff',
'\U0010ffff'):
+ with self.subTest(input_type=input_type, prefix=prefix):
+ result = reader.visit_starts_with(prefix)
+ self.assertTrue(result.is_exact())
+ self.assertEqual(
+ [row_id for row_id, value in enumerate(values)
+ if value is not None and
value.startswith(prefix)],
+ result.results().to_list())
+ finally:
+ reader.close()
+
+ def test_prefix_query_on_null_only_file_is_exact_and_empty(self):
+ serializer = create_serializer(AtomicType('STRING'))
+ with tempfile.TemporaryDirectory() as directory:
+ file_io = LocalFileIO()
+ writer = BTreeIndexWriter(file_io, directory, serializer,
block_size=32)
+ writer.write(None, 0)
+ writer.write(None, 1)
+ entry = writer.finish()[0]
+ file_size = os.path.getsize(os.path.join(directory,
entry.file_name))
+ io_meta = GlobalIndexIOMeta(entry.file_name, file_size, entry.meta)
+ reader = BTreeIndexReader(serializer, file_io, directory, io_meta)
+ try:
+ for prefix in ('', 'ke'):
+ with self.subTest(prefix=prefix):
+ result = reader.visit_starts_with(prefix)
+ self.assertTrue(result.is_exact())
+ self.assertTrue(result.results().is_empty())
+ finally:
+ reader.close()
+
+ def test_string_prefix_file_selection_uses_utf8_byte_bounds(self):
+ values = ['', '\x00', 'ke', 'kf', '\x7f', '\x7fa', '\u0080', '\u00ff',
'\u00ffa', '\u0100',
+ '\u07ff', '\u07ffa', '\u0800', '\uffff', '\uffffa',
'\U00010000', '\U0010ffff', None]
+ serializer = create_serializer(AtomicType('STRING'))
+ files = []
+ for row_id, value in enumerate(values):
+ key = None if value is None else serializer.serialize(value)
+ meta = SortedIndexFileMeta(key, key, value is None)
+ files.append((GlobalIndexIOMeta(str(row_id), 0, meta.serialize()),
meta))
+ selector = SortedFileMetaSelector(files, serializer)
+ for prefix in ('', '\x00', 'ke', 'keep', '\x7f', '\u00ff', '\u07ff',
'\uffff', '\U0010ffff'):
+ with self.subTest(prefix=prefix):
+ self.assertEqual(
+ [str(row_id) for row_id, value in enumerate(values)
+ if value is not None and value.startswith(prefix)],
+ [file.file_name for file in
selector.select_starts_with(prefix)])
+
def test_unsupported_versions_rejected_before_sst_read(self):
for input_type in (_TrackingInput, _PreadInput):
for version in (0, 2, 3, 0x80000000, 0xffffffff):
diff --git a/paimon-python/pypaimon/tests/full_text_scalar_filter_test.py
b/paimon-python/pypaimon/tests/full_text_scalar_filter_test.py
index 523f573dea..5de9a03795 100644
--- a/paimon-python/pypaimon/tests/full_text_scalar_filter_test.py
+++ b/paimon-python/pypaimon/tests/full_text_scalar_filter_test.py
@@ -41,10 +41,12 @@ def docs(tmp_path):
return table
-def append_rows(docs):
- docs.add(pa.table({"id": [3, 4], "text": ["paimon", "paimon another long
document"],
- "label": ["other", "target"], "pt": ["b"] * 2,
- "embedding": [[3., 1.], [4., 1.]]},
schema=docs.scan().to_arrow().schema))
+def append_rows(docs, row_ids=(3, 4)):
+ rows = [{"id": 3, "text": "paimon", "label": "other", "pt": "b",
"embedding": [3., 1.]},
+ {"id": 4, "text": "paimon another long document", "label":
"target", "pt": "b", "embedding": [4., 1.]}]
+ batch = pa.Table.from_pylist(
+ [row for row in rows if row["id"] in row_ids],
schema=docs.scan().to_arrow().schema)
+ docs.add(batch)
def builder(docs, predicate=None, limit=10):
@@ -62,27 +64,34 @@ def scores(result):
@pytest.mark.parametrize("kind", [None, "btree", "bitmap"])
@pytest.mark.parametrize("raw", [False, True])
-def test_full_text_data_filters_precede_top_k_and_preserve_scores(docs, kind,
raw):
[email protected]("refine", [False, True])
+def test_full_text_data_filters_precede_top_k_and_preserve_scores(docs, kind,
raw, refine):
+ docs.raw_table =
docs.raw_table.copy({"global-index.filter.refine-from-data":
str(refine).lower()})
if kind:
docs.raw_table.copy({"deletion-vectors.enabled":
"false"}).create_global_index("label", kind)
if raw:
append_rows(docs)
expected = scores(builder(docs).execute_local())
actual = scores(builder(docs, "label LIKE '%target%'",
limit=1).execute_local())
- candidates = [2, 4] if raw else [2]
- best = max(candidates, key=lambda row_id: (expected[row_id], -row_id))
- assert actual == {best: expected[best]}
+ # Java requires LIKE/contains refinement for either scalar index family.
+ # Raw rows are evaluated from the data independently of this option.
+ candidates = [] if kind is not None and not refine else [2]
+ if raw:
+ candidates.append(4)
+ best = max(candidates, key=lambda row_id: (expected[row_id], -row_id)) if
candidates else None
+ assert actual == ({best: expected[best]} if best is not None else {})
public = docs.search("paimon", column="text", pre_filter="label LIKE
'%target%'").select(["id"]).limit(1)
- assert public.to_list() == [{"id": best}]
+ assert public.to_list() == ([{"id": best}] if best is not None else [])
-def test_partial_scalar_coverage_does_not_drop_full_text_matches(docs):
[email protected]("mode, expected", [("fast", [2]), ("full", [2, 4]),
("detail", [2, 4])])
+def test_partial_scalar_coverage_follows_search_mode(docs, mode, expected):
docs.raw_table.copy({"deletion-vectors.enabled":
"false"}).create_global_index("label", "btree")
append_rows(docs)
docs.raw_table.copy({"deletion-vectors.enabled":
"false"}).create_global_index("text", "full-text")
- docs.raw_table = docs.raw_table.copy({"scalar-index.search-mode": "fast"})
- assert docs.search("paimon", column="text", pre_filter="label =
'target'").select(["id"]).limit(10).to_list() == [
- {"id": 2}, {"id": 4}]
+ docs.raw_table = docs.raw_table.copy({"scalar-index.search-mode": mode})
+ actual = docs.search("paimon", column="text", pre_filter="label =
'target'").select(["id"]).limit(10).to_list()
+ assert sorted(row["id"] for row in actual) == expected
@pytest.mark.parametrize("predicate, expected", [
@@ -91,7 +100,9 @@ def
test_partial_scalar_coverage_does_not_drop_full_text_matches(docs):
("id = 2", [2]),
])
def test_partition_and_data_predicates_keep_boolean_semantics(docs, predicate,
expected):
- append_rows(docs)
+ append_rows(docs, row_ids=[3])
+ docs.raw_table.copy({"deletion-vectors.enabled":
"false"}).create_global_index("text", "full-text")
+ append_rows(docs, row_ids=[4])
actual = docs.search("paimon", column="text",
pre_filter=predicate).select(["id"]).limit(10).to_list()
assert sorted(row["id"] for row in actual) == expected
@@ -102,11 +113,21 @@ def test_partition_only_filter_keeps_existing_path(docs):
assert len(docs.search("paimon", column="text", pre_filter="pt =
'a'").limit(10).to_list()) == 3
[email protected]("mode, expected", [("fast", []), ("full", [{"id":
4}])])
-def test_partition_without_full_text_index_respects_search_mode(docs, mode,
expected):
[email protected]("mode", ["fast", "full", "detail"])
+def test_partition_without_full_text_definition_is_empty(docs, mode):
append_rows(docs)
docs.raw_table = docs.raw_table.copy({"full-text-index.search-mode": mode})
search = docs.search("paimon", column="text", pre_filter="pt = 'b' AND
label = 'target'")
+ assert search.select(["id"]).limit(1).to_list() == []
+
+
[email protected]("mode, expected", [("fast", []), ("full", [{"id":
4}]), ("detail", [{"id": 4}])])
+def test_partition_with_full_text_definition_reads_unindexed_tail(docs, mode,
expected):
+ append_rows(docs, row_ids=[3])
+ docs.raw_table.copy({"deletion-vectors.enabled":
"false"}).create_global_index("text", "full-text")
+ append_rows(docs, row_ids=[4])
+ docs.raw_table = docs.raw_table.copy({"full-text-index.search-mode": mode})
+ search = docs.search("paimon", column="text", pre_filter="pt = 'b' AND
label = 'target'")
assert search.select(["id"]).limit(1).to_list() == expected
diff --git a/paimon-python/pypaimon/tests/native_full_text_search_test.py
b/paimon-python/pypaimon/tests/native_full_text_search_test.py
new file mode 100644
index 0000000000..2f578ce47f
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_full_text_search_test.py
@@ -0,0 +1,466 @@
+# 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.
+
+"""REST full-text queries retain Java coverage, filtering and snapshot
semantics."""
+
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon.api.api_response import GetTableSnapshotResponse, ErrorResponse
+from pypaimon.read.native_plan import _native_table, native_method_available,
_predicate_to_native
+from pypaimon.snapshot.table_snapshot import TableSnapshot
+from pypaimon.table.source.full_text_search_builder import
FullTextSearchBuilderImpl
+from pypaimon.tests import native_sorted_index_build_test as indexes
+from pypaimon.tests.vector_search_filter_test import match_query, boost_query
+
+rest_catalog = indexes.rest_catalog
+pytestmark = [pytest.mark.native_plan, pytest.mark.skipif(
+ not native_method_available('Table', 'new_full_text_search_builder'),
reason='Rust full-text search API required')]
+SCHEMA = pa.schema([('id', pa.int32()), ('content', pa.string()), ('label',
pa.string()), ('pt', pa.int32())])
+
+
+def _rows(start, count):
+ return [{'id': i, 'content': None if i == 3 else ('paimon lake' if i % 2
else 'paimon paimon paimon'),
+ 'label': 'keep' if i % 2 else 'drop', 'pt': i // 4} for i in
range(start, start + count)]
+
+
+def _table(catalog, options=None, *, indexed=True, partial=False):
+ settings = {'read.native.enabled': 'true', 'full-text-index.search-mode':
'full',
+ 'scalar-index.search-mode': 'full',
'global-index.row-count-per-shard': '4'}
+ settings.update(options or {})
+ table = indexes._create(catalog, schema=SCHEMA, partitioned=True,
options=settings)
+ indexes._append(table, _rows(0, 8), schema=SCHEMA)
+ if indexed:
+ table.create_global_index('content', index_type='full-text')
+ if partial:
+ indexes._append(table, _rows(8, 4), schema=SCHEMA)
+ return table
+
+
+def _builder(table, query=None, limit=64, predicate=None, partition=None,
column='content'):
+ builder = (table.new_full_text_search_builder().with_query(column, query
or match_query('paimon'))
+ .with_limit(limit))
+ if predicate is not None:
+ builder.with_filter(predicate)
+ if partition is not None:
+ builder.with_partition_filter(partition)
+ return builder
+
+
+def _scores(result):
+ if result.is_empty():
+ return {}
+ score = result.score_getter()
+ return {row_id: score(row_id) for row_id in result.results()}
+
+
+def _compare(table, **kwargs):
+ classic = table.copy({'read.native.enabled': 'false',
'scan.native-plan.enabled': 'false'})
+ expected = _scores(_builder(classic, **kwargs).execute_local())
+ with patch.object(FullTextSearchBuilderImpl, 'new_full_text_scan',
+ side_effect=AssertionError('Python full-text scan')):
+ actual = _scores(_builder(table, **kwargs).execute_local())
+ assert set(actual) == set(expected)
+ assert actual == pytest.approx(expected, rel=1e-6, abs=1e-6)
+ return actual
+
+
[email protected]('mode', ['fast', 'full', 'detail'])
[email protected]('query', [
+ match_query('paimon'), match_query('paimon lake', 'And'),
+ boost_query(match_query('paimon'), match_query('lake'), 0.1)])
[email protected]('limit', [1, 64])
+def test_native_full_text_coverage_and_structured_queries(rest_catalog, mode,
query, limit):
+ catalog, _ = rest_catalog
+ table = _table(catalog, {'full-text-index.search-mode': mode},
partial=True)
+ scores = _compare(table, query=query, limit=limit)
+ if limit == 64 and query == match_query('paimon'):
+ assert set(scores) == ({0, 1, 2, 4, 5, 6, 7} if mode == 'fast' else
{0, 1, 2, 4, 5, 6, 7, 8, 9, 10, 11})
+ assert indexes._ids(table) == list(range(12))
+
+
[email protected]('text_mode', ['fast', 'full', 'detail'])
[email protected]('scalar_mode', ['fast', 'full', 'detail'])
[email protected]('scalar_index', [None, 'btree', 'bitmap'])
+def test_scalar_filter_modes_before_full_text_top_k(rest_catalog, text_mode,
scalar_mode, scalar_index):
+ catalog, _ = rest_catalog
+ table = _table(catalog, {'full-text-index.search-mode': text_mode,
'scalar-index.search-mode': scalar_mode})
+ if scalar_index is not None:
+ table.create_global_index('label', index_type=scalar_index)
+ indexes._append(table, _rows(8, 4), schema=SCHEMA)
+ predicate =
table.new_read_builder().new_predicate_builder().equal('label', 'keep')
+ scores = _compare(table, predicate=predicate)
+ indexed = set() if scalar_mode == 'fast' and scalar_index is None else {1,
5, 7}
+ assert set(scores) == indexed.union(set() if text_mode == 'fast' else {9,
11})
+ top = _compare(table, predicate=predicate, limit=1)
+ assert len(top) == min(len(scores), 1)
+ assert all(row_id % 2 for row_id in top)
+ assert indexes._ids(table) == list(range(12))
+
+
[email protected]('refine', [False, True])
[email protected]('scalar_mode', ['fast', 'full', 'detail'])
[email protected]('index_type', ['btree', 'bitmap'])
[email protected]('method, literal', [('contains', 'eep'), ('endswith',
'eep'), ('like', 'k%p')])
+def test_candidate_filter_refinement_retains_full_corpus_scores(
+ rest_catalog, refine, scalar_mode, index_type, method, literal):
+ catalog, _ = rest_catalog
+ table = _table(catalog, {'global-index.filter.refine-from-data':
str(refine).lower(),
+ 'scalar-index.search-mode': scalar_mode})
+ table.create_global_index('label', index_type=index_type)
+ indexes._append(table, _rows(8, 4), schema=SCHEMA)
+ predicate = getattr(table.new_read_builder().new_predicate_builder(),
method)('label', literal)
+ scores = _compare(table, predicate=predicate)
+ assert set(scores) == ({1, 5, 7, 9, 11} if refine else {9, 11})
+ full = _compare(table)
+ assert scores == pytest.approx({row_id: full[row_id] for row_id in
scores}, abs=1e-6)
+ assert indexes._ids(table) == list(range(12))
+
+
[email protected]('refine', [False, True])
[email protected]('compound', ['and', 'or'])
[email protected]('method, literal', [('contains', 'eep'), ('endswith',
'eep'), ('like', '%eep%')])
+def test_bitmap_candidate_leaves_require_refinement_in_compounds(rest_catalog,
refine, compound, method, literal):
+ catalog, _ = rest_catalog
+ table = _table(catalog, {'global-index.filter.refine-from-data':
str(refine).lower(),
+ 'scalar-index.search-mode': 'fast'})
+ table.create_global_index('label', index_type='bitmap')
+ indexes._append(table, _rows(8, 4), schema=SCHEMA)
+ pb = table.new_read_builder().new_predicate_builder()
+ candidate = getattr(pb, method)('label', literal)
+ predicate = (pb.and_predicates([pb.is_not_null('label'), candidate]) if
compound == 'and'
+ else pb.or_predicates([pb.equal('label', 'drop'), candidate]))
+ rows = set(_compare(table, predicate=predicate))
+ indexed = ({1, 5, 7} if compound == 'and' else {0, 1, 2, 4, 5, 6, 7}) if
refine else set()
+ assert rows == indexed.union({9, 11} if compound == 'and' else {8, 9, 10,
11})
+ assert indexes._ids(table) == list(range(12))
+
+
[email protected]('pattern', ['keep', 'ke%'])
[email protected]('index_type', ['btree', 'bitmap'])
+def test_optimized_like_filters_stay_exact_without_refinement(rest_catalog,
pattern, index_type):
+ catalog, _ = rest_catalog
+ table = _table(catalog, {'global-index.filter.refine-from-data': 'false',
+ 'scalar-index.search-mode': 'fast'})
+ table.create_global_index('label', index_type=index_type)
+ predicate = table.new_read_builder().new_predicate_builder().like('label',
pattern)
+ assert set(_compare(table, predicate=predicate)) == {1, 5, 7}
+ assert indexes._ids(table) == list(range(8))
+
+
[email protected]('refine', [False, True])
[email protected]('partial', [False, True])
[email protected]('index_type', ['btree', 'bitmap'])
[email protected]('method, literal', [
+ ('startswith', 'ke'), ('like', 'ke%'), ('startswith', 'missing'), ('like',
'missing%')])
+def test_prefix_filters_are_exact_before_refinement(rest_catalog, refine,
partial, index_type, method, literal):
+ catalog, _ = rest_catalog
+ table = _table(catalog, {'global-index.filter.refine-from-data':
str(refine).lower()})
+ table.create_global_index('label', index_type=index_type)
+ if partial:
+ indexes._append(table, _rows(8, 4), schema=SCHEMA)
+ predicate = getattr(table.new_read_builder().new_predicate_builder(),
method)('label', literal)
+ expected = ({1, 5, 7, 9, 11} if partial else {1, 5, 7}) if
literal.startswith('ke') else set()
+ assert set(_compare(table, predicate=predicate)) == expected
+ assert indexes._ids(table) == list(range(12 if partial else 8))
+
+
[email protected]('prefix', ['\x7f', '\u00ff', '\uffff', '\U0010ffff'])
[email protected]('method', ['startswith', 'like'])
+def
test_btree_unicode_prefix_bounds_keep_indexed_and_raw_matches(rest_catalog,
prefix, method):
+ catalog, _ = rest_catalog
+ table = indexes._create(catalog, schema=SCHEMA, options={
+ 'read.native.enabled': 'true', 'global-index.filter.refine-from-data':
'false'})
+ indexes._append(table, [
+ {'id': i, 'content': 'paimon', 'label': label, 'pt': 0}
+ for i, label in enumerate([prefix, prefix + 'a', None, 'keep'])],
schema=SCHEMA)
+ table.create_global_index('content', index_type='full-text')
+ table.create_global_index('label', index_type='btree')
+ indexes._append(table, [
+ {'id': 4, 'content': 'paimon', 'label': prefix + 'raw', 'pt': 0},
+ {'id': 5, 'content': 'paimon', 'label': 'drop', 'pt': 0}],
schema=SCHEMA)
+ literal = prefix + '%' if method == 'like' else prefix
+ predicate = getattr(table.new_read_builder().new_predicate_builder(),
method)('label', literal)
+ assert set(_compare(table, predicate=predicate)) == {0, 1, 4}
+ assert indexes._ids(table) == list(range(6))
+
+
[email protected]('refine', [False, True])
[email protected]('partial', [False, True])
[email protected]('index_type', ['btree', 'bitmap'])
[email protected]('method, literal', [
+ ('like', ''), ('like', '%'), ('like', '%%'), ('contains', ''),
('endswith', ''), ('startswith', '')])
+def test_empty_string_filters_preserve_java_refinement(rest_catalog, refine,
partial, index_type, method, literal):
+ catalog, _ = rest_catalog
+ table = indexes._create(catalog, schema=SCHEMA, partitioned=True, options={
+ 'read.native.enabled': 'true', 'full-text-index.search-mode': 'full',
+ 'scalar-index.search-mode': 'full',
'global-index.filter.refine-from-data': str(refine).lower()})
+ labels = ['', None, 'keep', 'drop', '', 'keep', 'drop', None]
+ indexes._append(table, [
+ {'id': i, 'content': 'paimon', 'label': label, 'pt': 0}
+ for i, label in enumerate(labels)], schema=SCHEMA)
+ table.create_global_index('content', index_type='full-text')
+ table.create_global_index('label', index_type=index_type)
+ if partial:
+ labels += ['', None, 'keep', 'drop']
+ indexes._append(table, [
+ {'id': i, 'content': 'paimon', 'label': label, 'pt': 0}
+ for i, label in enumerate(labels[8:], 8)], schema=SCHEMA)
+ predicate = getattr(table.new_read_builder().new_predicate_builder(),
method)('label', literal)
+ indexed = {0, 4} if method == 'like' and literal == '' else {0, 2, 3, 4,
5, 6}
+ raw = ({8} if method == 'like' and literal == '' else {8, 10, 11}) if
partial else set()
+ expected = (indexed if refine or method == 'startswith' else
set()).union(raw)
+ assert set(_compare(table, predicate=predicate)) == expected
+ assert indexes._ids(table) == list(range(len(labels)))
+
+
[email protected]('refine', [False, True])
+def test_escaped_bitmap_like_is_refined_without_prefix_rewrite(rest_catalog,
refine):
+ catalog, _ = rest_catalog
+ table = indexes._create(catalog, schema=SCHEMA, partitioned=True, options={
+ 'read.native.enabled': 'true', 'scalar-index.search-mode': 'fast',
+ 'global-index.filter.refine-from-data': str(refine).lower()})
+ indexes._append(table, [
+ {'id': 0, 'content': 'paimon', 'label': 'ke%', 'pt': 0},
+ {'id': 1, 'content': 'paimon', 'label': r'ke\suffix', 'pt': 0},
+ {'id': 2, 'content': 'paimon', 'label': 'keep', 'pt': 0}],
schema=SCHEMA)
+ table.create_global_index('content', index_type='full-text')
+ table.create_global_index('label', index_type='bitmap')
+ predicate = table.new_read_builder().new_predicate_builder().like('label',
r'ke\%')
+ assert set(_compare(table, predicate=predicate)) == ({0} if refine else
set())
+ assert indexes._ids(table) == list(range(3))
+
+
[email protected]('selection', ['partition', 'mixed', 'repeat'])
[email protected]('mode', ['full', 'detail'])
+def test_partition_filter_applies_to_index_and_raw_reads(rest_catalog,
selection, mode):
+ catalog, _ = rest_catalog
+ table = _table(catalog, {'full-text-index.search-mode': mode},
partial=True)
+ pb = table.new_read_builder().new_predicate_builder()
+ partition = pb.greater_or_equal('pt', 1)
+ if selection == 'partition':
+ scores = _compare(table, partition=partition)
+ assert set(scores) == set(range(4, 12))
+ elif selection == 'mixed':
+ scores = _compare(table, predicate=pb.and_predicates([partition,
pb.equal('label', 'keep')]))
+ assert set(scores) == {5, 7, 9, 11}
+ else:
+ builder = _builder(table,
partition=partition).with_partition_filter(pb.less_than('pt', 2))
+ assert set(_scores(builder.execute_local())) == set(range(4, 8))
+ assert indexes._ids(table) == list(range(12))
+
+
[email protected]('mode', ['fast', 'full', 'detail'])
+def test_no_full_text_definition_is_empty_in_every_mode(rest_catalog, mode):
+ catalog, _ = rest_catalog
+ table = _table(catalog, {'full-text-index.search-mode': mode},
indexed=False)
+ assert _compare(table) == {}
+ assert indexes._ids(table) == list(range(8))
+
+
+def test_scan_read_plan_pins_snapshot_and_rejects_other_filters(rest_catalog):
+ catalog, _ = rest_catalog
+ table = _table(catalog)
+ native = (_native_table(table).new_full_text_search_builder()
+ .with_query('content', match_query('paimon')).with_limit(64))
+ plan = native.new_full_text_scan().scan()
+ before = table.snapshot_manager().get_latest_snapshot().id
+ assert plan.snapshot_id() == before
+ reader = native.new_full_text_read()
+ indexes._append(table, _rows(8, 4), schema=SCHEMA)
+ assert set(reader.read_plan(plan).row_ids()) == {0, 1, 2, 4, 5, 6, 7}
+ assert set(native.execute_local().row_ids()) == {0, 1, 2, 4, 5, 6, 7, 8,
9, 10, 11}
+ predicate = table.new_read_builder().new_predicate_builder().equal('id', 0)
+ changed = (_native_table(table).new_full_text_search_builder()
+ .with_query('content', match_query('paimon')).with_limit(64))
+ changed.with_filter(_predicate_to_native(predicate))
+ with pytest.raises(Exception, match='different table, column or filter'):
+ changed.new_full_text_read().read_plan(plan)
+ assert indexes._ids(table) == list(range(12))
+
+
[email protected]('response', ['first', 'empty', 'error'])
+def test_rest_snapshot_response_is_authoritative(rest_catalog, response):
+ catalog, server = rest_catalog
+ table = _table(catalog)
+ first = table.snapshot_manager().get_latest_snapshot()
+ indexes._append(table, _rows(8, 4), schema=SCHEMA)
+ if response == 'error':
+ reply = ErrorResponse('TABLE', 'indexes', 'snapshot unavailable', 503)
+ code = 503
+ else:
+ reply = (GetTableSnapshotResponse(TableSnapshot(first, 1, 0, 7,
first.time_millis))
+ if response == 'first' else GetTableSnapshotResponse())
+ code = 200
+ with patch.object(server, '_table_snapshot_handle',
return_value=server._mock_response(reply, code)):
+ if response == 'error':
+ with patch.object(FullTextSearchBuilderImpl, 'new_full_text_scan',
+ side_effect=AssertionError('Unexpected Python
retry')):
+ with pytest.raises(Exception, match='snapshot unavailable'):
+ _builder(table).execute_local()
+ else:
+ assert set(_compare(table)) == ({0, 1, 2, 4, 5, 6, 7} if response
== 'first' else set())
+ assert indexes._ids(table) == list(range(12))
+
+
[email protected]('native', [False, True])
+def test_missing_index_errors_propagate_without_search_retry(rest_catalog,
native):
+ catalog, _ = rest_catalog
+ table = _table(catalog, {'read.native.enabled': str(native).lower()})
+ file = indexes._files(table)[0]
+ table.file_io.delete(indexes._path(table, file))
+ if native:
+ with patch.object(FullTextSearchBuilderImpl, 'new_full_text_scan',
+ side_effect=AssertionError('Unexpected Python
retry')):
+ with pytest.raises(Exception, match='NotFound|not found|No such
file|does not exist'):
+ _builder(table).execute_local()
+ else:
+ with pytest.raises(Exception):
+ _builder(table).execute_local()
+ assert indexes._ids(table) == list(range(8))
+
+
[email protected]('native', [False, True])
[email protected]('case', ['limit', 'query', 'missing-column',
'partition', 'dsl'])
+def test_invalid_search_inputs_fail(rest_catalog, native, case):
+ catalog, _ = rest_catalog
+ table = _table(catalog, {'read.native.enabled': str(native).lower()})
+ if case == 'limit':
+ builder = _builder(table, limit=0)
+ elif case == 'query':
+ builder = table.new_full_text_search_builder().with_limit(1)
+ elif case == 'missing-column':
+ builder = table.new_full_text_search_builder().with_query('missing',
match_query('paimon')).with_limit(1)
+ elif case == 'partition':
+ with pytest.raises(ValueError, match='Partition filter'):
+ _builder(table,
partition=table.new_read_builder().new_predicate_builder().equal('id', 1))
+ assert indexes._ids(table) == list(range(8))
+ return
+ else:
+ builder = _builder(table, query='paimon')
+ with pytest.raises(Exception):
+ builder.execute_local()
+ assert indexes._ids(table) == list(range(8))
+
+
[email protected]('mode', ['full', 'detail'])
[email protected]('options', [
+ {'full-text.stop-words': ['paimon']},
+ {'full-text.stop-words': ['lake'], 'full-text.lowercase': True}])
+def test_typed_analyzer_options_keep_raw_query_semantics(rest_catalog, mode,
options):
+ catalog, _ = rest_catalog
+ table = _table(catalog, {'full-text-index.search-mode': mode},
partial=True).copy(options)
+ scores = _compare(table)
+ expected = ({0, 1, 2, 4, 5, 6, 7} if options['full-text.stop-words'] ==
['paimon']
+ else {0, 1, 2, 4, 5, 6, 7, 8, 9, 10, 11})
+ assert set(scores) == expected
+ assert indexes._ids(table) == list(range(12))
+
+
[email protected]('mode', ['fast', 'full', 'detail'])
+def
test_composite_scalar_indexes_are_excluded_from_search_pre_filters(rest_catalog,
mode):
+ catalog, _ = rest_catalog
+ table = _table(catalog, {'scalar-index.search-mode': mode})
+ # Python intentionally supports single-column builds. Seed a composite
+ # definition through Rust to exercise Java-compatible search exclusion.
+ messages = (_native_table(table).new_global_index_build_builder()
+ .with_index_columns(['label',
'id']).with_index_type('btree').build())
+ indexes._commit(table, indexes.from_native_commit_messages(table,
messages))
+ pb = table.new_read_builder().new_predicate_builder()
+ predicate = pb.and_predicates([pb.equal('label', 'keep'),
pb.greater_than('id', 0)])
+ assert set(_compare(table, predicate=predicate)) == (set() if mode ==
'fast' else {1, 5, 7})
+ assert indexes._ids(table) == list(range(8))
+
+
[email protected]('mode', ['full', 'detail'])
+def
test_schema_only_rename_retains_read_schema_for_unindexed_rows(rest_catalog,
mode):
+ from pypaimon.schema.schema_change import SchemaChange
+ catalog, _ = rest_catalog
+ table = _table(catalog, {'full-text-index.search-mode': mode},
partial=True)
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ catalog.alter_table('default.indexes',
[SchemaChange.rename_column('content', 'body')], False)
+ table = catalog.get_table('default.indexes')
+ assert table.table_schema.id > snapshot.schema_id
+ assert set(_compare(table, column='body')) == {0, 1, 2, 4, 5, 6, 7, 8, 9,
10, 11}
+ native = (_native_table(table).new_full_text_search_builder()
+ .with_query('body', match_query('paimon')).with_limit(64))
+ plan = native.new_full_text_scan().scan()
+ assert set(native.new_full_text_read().read_plan(plan).row_ids()) == {0,
1, 2, 4, 5, 6, 7, 8, 9, 10, 11}
+ assert indexes._ids(table) == list(range(12))
+
+
[email protected]('mode', ['fast', 'full', 'detail'])
+def test_deleted_indexed_and_raw_rows_are_excluded_before_top_k(rest_catalog,
mode):
+ catalog, _ = rest_catalog
+ table = _table(catalog, {'full-text-index.search-mode': mode},
partial=True)
+ table = table.copy({'deletion-vectors.enabled': 'true'})
+ builder = table.new_batch_write_builder()
+ indexes._commit(table, builder.new_update().delete_by_row_id([0, 1, 8, 9]))
+ assert set(_compare(table)) == ({2, 4, 5, 6, 7} if mode == 'fast' else {2,
4, 5, 6, 7, 10, 11})
+ assert len(_compare(table, limit=3)) == 3
+ assert indexes._ids(table) == [2, 3, 4, 5, 6, 7, 10, 11]
+
+
[email protected]('native', [False, True])
[email protected]('limit', [-1, 0])
+def test_invalid_limit_has_same_value_error(rest_catalog, native, limit):
+ catalog, _ = rest_catalog
+ table = _table(catalog, {'read.native.enabled': str(native).lower()})
+ with pytest.raises(ValueError, match='Limit must be positive'):
+ _builder(table, limit=limit).execute_local()
+ assert indexes._ids(table) == list(range(8))
+
+
[email protected]('mode', ['fast', 'full', 'detail'])
+def test_updated_text_uses_invalidated_index_coverage(rest_catalog, mode):
+ catalog, _ = rest_catalog
+ table = _table(catalog, {'full-text-index.search-mode': mode,
+ 'global-index.column-update-action':
'DROP_PARTITION_INDEX'}, partial=True)
+ builder = table.new_batch_write_builder()
+ changed = pa.table({'_ROW_ID': pa.array([4], type=pa.int64()), 'content':
['rust']})
+ messages =
builder.new_update().with_update_type(['content']).update_by_arrow_with_row_id(changed)
+ indexes._commit(table, messages)
+ scores = _compare(table)
+ assert 4 not in scores
+ if mode != 'fast':
+ assert set(scores) == {0, 1, 2, 5, 6, 7, 8, 9, 10, 11}
+ assert indexes._ids(table) == list(range(12))
+
+
+def
test_unrepresentable_literal_qualifies_before_native_execution(rest_catalog):
+ catalog, _ = rest_catalog
+ table = _table(catalog)
+ predicate = table.new_read_builder().new_predicate_builder().equal('id',
1.0)
+ expected = _scores(_builder(table.copy({'read.native.enabled': 'false'}),
predicate=predicate).execute_local())
+ scan = FullTextSearchBuilderImpl.new_full_text_scan
+ with patch.object(FullTextSearchBuilderImpl, 'new_full_text_scan',
autospec=True, side_effect=scan) as selected:
+ assert _scores(_builder(table, predicate=predicate).execute_local())
== expected
+ selected.assert_called_once()
+ assert set(expected) == {1}
+ assert indexes._ids(table) == list(range(8))
+
+
[email protected]('native', [False, True])
+def test_query_auth_rejects_search_before_index_access(rest_catalog, native):
+ catalog, _ = rest_catalog
+ table = _table(catalog)
+ authorized = table.copy({'query-auth.enabled': 'true',
'read.native.enabled': str(native).lower()})
+ with pytest.raises(Exception, match='query-auth'):
+ _builder(authorized).execute_local()
+ assert indexes._ids(table) == list(range(8))
diff --git a/paimon-python/pypaimon/tests/vector_filter_exactness_test.py
b/paimon-python/pypaimon/tests/vector_filter_exactness_test.py
index 4a1d77861e..86e1e78c91 100644
--- a/paimon-python/pypaimon/tests/vector_filter_exactness_test.py
+++ b/paimon-python/pypaimon/tests/vector_filter_exactness_test.py
@@ -62,40 +62,67 @@ def query(table, predicate, batch=False):
@pytest.mark.parametrize("batch", [False, True])
@pytest.mark.parametrize("mode", ["full", "fast"])
@pytest.mark.parametrize("refine", [False, True])
-def test_btree_candidates_are_verified_before_top_k(table, caplog, pattern,
batch, mode, refine):
+def test_btree_string_filters_follow_java_exactness_before_top_k(table,
caplog, pattern, batch, mode, refine):
scalar_index(table)
table.raw_table = table.raw_table.copy({
"vector-index.search-mode": mode,
"global-index.filter.refine-from-data": str(refine).lower()})
result = query(table, "name LIKE '%s'" % pattern, batch).to_list()
- expected = [{"id": 1}] if refine else []
+ candidate_only = pattern == "%zeta%"
+ expected = [{"id": 1}] if refine or not candidate_only else []
assert result == ([expected, expected] if batch else expected)
- assert ("global-index.filter.refine-from-data=true" in caplog.text) ==
(not refine)
+ assert ("global-index.filter.refine-from-data=true" in caplog.text) ==
(candidate_only and not refine)
@pytest.mark.parametrize("kind, predicate", [
("btree", "name = 'beta zeta'"), ("btree", "name >= 'beta' AND name <
'gamma'"),
- ("btree", "name LIKE 'beta zeta'"), ("bitmap", "name LIKE '%zeta%'"),
+ ("btree", "name LIKE 'beta zeta'"), ("btree", "name LIKE 'beta%'"),
("bitmap", "name = 'beta zeta'"),
+ ("bitmap", "name LIKE 'beta%'"),
])
-def test_exact_indexes_do_not_read_filter_columns(table, kind, predicate):
[email protected]("refine", [False, True])
+def test_exact_indexes_do_not_read_filter_columns(table, kind, predicate,
refine):
scalar_index(table, kind)
- table.raw_table =
table.raw_table.copy({"global-index.filter.refine-from-data": "true"})
+ table.raw_table =
table.raw_table.copy({"global-index.filter.refine-from-data":
str(refine).lower()})
with patch.object(AbstractVectorSearchReadImpl, "_matching_candidate_rows",
side_effect=AssertionError("exact index recheck")):
assert query(table, predicate).to_list() == [{"id": 1}]
[email protected]("method, literal", [("contains", "zeta"),
("endswith", "zeta"), ("like", "%zeta%")])
@pytest.mark.parametrize("batch", [False, True])
@pytest.mark.parametrize("mode", ["full", "fast"])
@pytest.mark.parametrize("refine", [False, True])
-def test_mixed_btree_and_bitmap_preserve_exact_matches(table, batch, mode,
refine):
+def test_bitmap_string_filters_obey_search_refinement(table, method, literal,
batch, mode, refine):
+ scalar_index(table, "bitmap")
+ table.raw_table = table.raw_table.copy({
+ "vector-index.search-mode": mode,
"global-index.filter.refine-from-data": str(refine).lower()})
+ pb = table.raw_table.new_read_builder().new_predicate_builder()
+ predicate = getattr(pb, method)("name", literal)
+ if batch:
+ search =
(table.raw_table.new_batch_vector_search_builder().with_vector_column("embedding")
+ .with_query_vectors([[0., 1.], [0.,
1.]]).with_filter(predicate).with_limit(1))
+ result = [list(value.results()) for value in
search.execute_batch_local()]
+ else:
+ search =
(table.raw_table.new_vector_search_builder().with_vector_column("embedding")
+ .with_query_vector([0.,
1.]).with_filter(predicate).with_limit(1))
+ result = list(search.execute_local().results())
+ expected = [1] if refine else []
+ assert result == ([expected, expected] if batch else expected)
+
+
[email protected]("batch", [False, True])
[email protected]("mode", ["full", "fast"])
[email protected]("refine", [False, True])
+def test_mixed_btree_and_bitmap_obey_search_refinement(table, batch, mode,
refine):
scalar_index(table, "btree")
scalar_index(table, "bitmap")
table.raw_table = table.raw_table.copy({
"vector-index.search-mode": mode,
"global-index.filter.refine-from-data": str(refine).lower()})
+ verify = AbstractVectorSearchReadImpl._matching_candidate_rows
with patch.object(AbstractVectorSearchReadImpl, "_matching_candidate_rows",
- side_effect=AssertionError("exact index recheck")):
+ autospec=True, side_effect=verify) as read:
result = query(table, "name LIKE '%zeta%'", batch).to_list()
- expected = [{"id": 1}]
+ assert read.call_count == (1 if refine else 0)
+ expected = [{"id": 1}] if refine else []
assert result == ([expected, expected] if batch else expected)
