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 ae8f9d5ea0 [python] Support data prefilters for full-text and hybrid
search (#10006)
ae8f9d5ea0 is described below
commit ae8f9d5ea0c51aa56be3d4acd98e95f815720294
Author: chaoyang <[email protected]>
AuthorDate: Mon Sep 21 11:34:21 2026 +0800
[python] Support data prefilters for full-text and hybrid search (#10006)
---
docs/docs/pypaimon/multimodal-search.md | 23 ++-
paimon-python/pypaimon/multimodal/query.py | 2 +-
.../pypaimon/table/source/full_text_read.py | 43 ++++-
.../pypaimon/table/source/full_text_scan.py | 23 ++-
.../table/source/full_text_search_builder.py | 23 +++
.../table/source/global_index_row_filter.py | 42 +++++
.../pypaimon/table/source/hybrid_search_builder.py | 9 +-
.../pypaimon/table/source/vector_search_read.py | 18 +--
.../pypaimon/tests/full_text_scalar_filter_test.py | 179 +++++++++++++++++++++
.../pypaimon/tests/vector_search_filter_test.py | 7 +-
10 files changed, 319 insertions(+), 50 deletions(-)
diff --git a/docs/docs/pypaimon/multimodal-search.md
b/docs/docs/pypaimon/multimodal-search.md
index 05d38be456..1c56a5e04b 100644
--- a/docs/docs/pypaimon/multimodal-search.md
+++ b/docs/docs/pypaimon/multimodal-search.md
@@ -60,8 +60,18 @@ has exactly one text column. To target a specific text
column, pass `column`.
Use `pre_filter` to prune search candidates before ranking. Use `where()` to
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.
+accept SQL-like predicate strings. On data-evolution tables, full-text
+`pre_filter` supports ordinary data columns as well as partition columns.
+For primary-key full-text search, only partition predicates are supported.
+
+Full-text data predicates are evaluated before Top-K selection. PyPaimon reads
+the predicate's columns and row IDs at the search snapshot, then passes the
+matching row IDs to the full-text index. Scalar indexes can prune this read,
+but partial scalar-index coverage does not exclude matching rows covered by
+the full-text search plan. This filter read can scan all candidate rows;
+partition-only predicates keep the existing partition-pruning path. For
+unindexed data in `full-text-index.search-mode=full`, filtering preserves the
+unfiltered corpus used to calculate BM25 statistics.
For data-evolution vector search, a scalar index may return candidates rather
than exact
matches, for example for BTree string-prefix or substring predicates, or when
@@ -92,7 +102,7 @@ neighbors = (
)
matches = (
- docs.search("paimon vector", column="content")
+ docs.search("paimon vector", column="content", pre_filter="category =
'lake'")
.limit(10)
.to_pandas()
)
@@ -272,8 +282,11 @@ text column when the table has exactly one text column. To
target a specific
text column, pass `column` to `pm.text_route`.
`pre_filter` is applied before ranking. It accepts a SQL-like predicate string.
-When a hybrid query has a full-text route, `pre_filter` must only reference
-partition columns.
+On data-evolution tables, ordinary data predicates are applied to both vector
+and full-text routes before each route selects its candidates. The vector
+route retains the `global-index.filter.refine-from-data` behavior described
+above, while the full-text route verifies data predicates through a
filter-column
+read. Partition-only predicates prune both routes without that extra read.
```python
# This example assumes the table is partitioned by dt.
diff --git a/paimon-python/pypaimon/multimodal/query.py
b/paimon-python/pypaimon/multimodal/query.py
index a7c8a569a2..cb218fdc1e 100644
--- a/paimon-python/pypaimon/multimodal/query.py
+++ b/paimon-python/pypaimon/multimodal/query.py
@@ -685,7 +685,7 @@ class TextQuery(_PreFilterQuery):
.with_limit(limit)
)
if query._pre_filter is not None:
- builder = builder.with_partition_filter(query._pre_filter)
+ builder = builder.with_filter(query._pre_filter)
return builder.execute_local()
diff --git a/paimon-python/pypaimon/table/source/full_text_read.py
b/paimon-python/pypaimon/table/source/full_text_read.py
index cccd1a90e1..c4236b6646 100644
--- a/paimon-python/pypaimon/table/source/full_text_read.py
+++ b/paimon-python/pypaimon/table/source/full_text_read.py
@@ -39,6 +39,7 @@ from pypaimon.table.source.full_text_search_split import (
)
from pypaimon.table.source.full_text_scan import FullTextScanPlan
from pypaimon.utils.range import Range
+from pypaimon.utils.roaring_bitmap import RoaringBitmap64
class FullTextRead(ABC):
@@ -62,6 +63,7 @@ class DataEvolutionFullTextRead(FullTextRead):
text_column,
query: str,
partition_filter=None,
+ filter_=None,
):
self._table = table
self._limit = limit
@@ -72,6 +74,7 @@ class DataEvolutionFullTextRead(FullTextRead):
% self._text_columns)
self._query = query
self._partition_filter = partition_filter
+ self._filter = filter_
def read_plan(self, plan: FullTextScanPlan) -> GlobalIndexResult:
reader = copy(self)
@@ -88,9 +91,25 @@ 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)
- raw_result = self._read_raw_search(
- _raw_row_ranges(raw_splits), _index_type(index_splits))
+ 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)
return indexed_result.or_(raw_result).top_k(self._limit)
def _eval_column_query(
@@ -170,7 +189,7 @@ class DataEvolutionFullTextRead(FullTextRead):
future.add_done_callback(lambda _: reader.close())
return future
- def _read_raw_search(self, raw_row_ranges, index_type):
+ def _read_raw_search(self, raw_row_ranges, index_type,
include_row_ids=None):
raw_row_ranges = Range.sort_and_merge_overlap(raw_row_ranges, True)
if not raw_row_ranges:
return DictBasedScoredIndexResult({})
@@ -179,6 +198,15 @@ class DataEvolutionFullTextRead(FullTextRead):
row_range_start = raw_row_ranges[0].from_
row_range_end = raw_row_ranges[-1].to
+ search_kwargs = {}
+ if include_row_ids is not None:
+ raw_rows = GlobalIndexResult.from_ranges(raw_row_ranges).results()
+ include_row_ids = RoaringBitmap64.and_(include_row_ids, raw_rows)
+ if include_row_ids.is_empty():
+ return DictBasedScoredIndexResult({})
+ search_kwargs["include_row_ids"] = include_row_ids
+ # Build the same corpus as an unfiltered query so BM25 statistics stay
+ # unchanged. Filter row IDs in the native search, before selecting
hits.
table = self._read_raw_rows(raw_row_ranges)
if table is None or table.num_rows == 0:
return DictBasedScoredIndexResult({})
@@ -196,6 +224,7 @@ class DataEvolutionFullTextRead(FullTextRead):
row_range_start,
self._query,
_candidate_limit(row_range_start, row_range_end),
+ **search_kwargs,
).top_k(self._limit)
def _read_raw_rows(self, raw_row_ranges):
@@ -300,7 +329,7 @@ def _index_type(index_splits):
return None
-def _search_raw_full_text(index_bytes, row_range_start, query, limit):
+def _search_raw_full_text(index_bytes, row_range_start, query, limit,
include_row_ids=None):
from paimon_ftindex import FullTextIndexReader
from pypaimon.globalindex.full_text.native_full_text_global_index_reader
import (
PaimonFullTextInput,
@@ -308,7 +337,11 @@ def _search_raw_full_text(index_bytes, row_range_start,
query, limit):
reader = FullTextIndexReader(PaimonFullTextInput(BytesIO(index_bytes)))
try:
- row_ids, scores = reader.search(query, limit=limit)
+ search_kwargs = {}
+ if include_row_ids is not None:
+ relative_ids =
GlobalIndexResult.create(include_row_ids).offset(-row_range_start).results()
+ search_kwargs["filter_bytes"] = relative_ids.serialize()
+ row_ids, scores = reader.search(query, limit=limit, **search_kwargs)
return DictBasedScoredIndexResult(
{
row_range_start + row_id: score
diff --git a/paimon-python/pypaimon/table/source/full_text_scan.py
b/paimon-python/pypaimon/table/source/full_text_scan.py
index 29b44ee5b0..ae5d089d0b 100644
--- a/paimon-python/pypaimon/table/source/full_text_scan.py
+++ b/paimon-python/pypaimon/table/source/full_text_scan.py
@@ -115,18 +115,17 @@ class DataEvolutionFullTextScan(FullTextScan):
IndexFullTextSearchSplit(
column_name, range_key.from_, range_key.to, files))
- if all_index_files:
- raw_row_ranges = DataEvolutionGlobalIndexCoverage(
- self._table,
- snapshot,
- partition_filter,
- all_index_files,
- ).unindexed_ranges(
- list(text_column_ids),
- search_mode=self._table.options.full_text_index_search_mode(),
- )
- if raw_row_ranges:
- splits.append(RawFullTextSearchSplit(raw_row_ranges))
+ raw_row_ranges = DataEvolutionGlobalIndexCoverage(
+ self._table,
+ snapshot,
+ partition_filter,
+ all_index_files,
+ ).unindexed_ranges(
+ list(text_column_ids),
+ search_mode=self._table.options.full_text_index_search_mode(),
+ )
+ if raw_row_ranges:
+ splits.append(RawFullTextSearchSplit(raw_row_ranges))
return FullTextScanPlan(splits, snapshot)
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 420f918ac5..72b34a92dd 100644
--- a/paimon-python/pypaimon/table/source/full_text_search_builder.py
+++ b/paimon-python/pypaimon/table/source/full_text_search_builder.py
@@ -49,6 +49,11 @@ class FullTextSearchBuilder(ABC):
"""Partition predicate used to prune index manifest entries."""
pass
+ @abstractmethod
+ def with_filter(self, predicate) -> 'FullTextSearchBuilder':
+ """Filter data-evolution rows before full-text ranking."""
+ pass
+
@abstractmethod
def new_full_text_scan(self) -> FullTextScan:
"""Create full-text scan to scan index files."""
@@ -73,6 +78,7 @@ class FullTextSearchBuilderImpl(FullTextSearchBuilder):
self._field_name: Optional[str] = None
self._query: Optional[str] = None
self._partition_filter = None
+ self._filter = None
def with_limit(self, limit: int) -> 'FullTextSearchBuilder':
self._limit = limit
@@ -83,6 +89,20 @@ class FullTextSearchBuilderImpl(FullTextSearchBuilder):
self._query = query
return self
+ def with_filter(self, predicate) -> 'FullTextSearchBuilder':
+ if predicate is None:
+ return self
+ from pypaimon.read.push_down_utils import _split_and, _get_all_fields
+
+ partition_keys = set(self._table.partition_keys or [])
+ for part in _split_and(predicate):
+ if partition_keys and
_get_all_fields(part).issubset(partition_keys):
+ self.with_partition_filter(part)
+ else:
+ self._filter = (part if self._filter is None else
+ PredicateBuilder.and_predicates([self._filter,
part]))
+ return self
+
def with_partition_filter(self, partition_filter) ->
'FullTextSearchBuilder':
if partition_filter is None:
self._partition_filter = None
@@ -135,6 +155,8 @@ class FullTextSearchBuilderImpl(FullTextSearchBuilder):
reject_search_under_query_auth(self._table)
if self._limit <= 0:
raise ValueError("Limit must be positive, set via with_limit()")
+ if self._filter is not None and not
self._table.options.data_evolution_enabled():
+ raise NotImplementedError("Full-text row filters require a
data-evolution table.")
definition = self._primary_key_full_text_definition()
if definition is not None:
from pypaimon.common.options.core_options import
GlobalIndexSearchMode
@@ -156,6 +178,7 @@ class FullTextSearchBuilderImpl(FullTextSearchBuilder):
self._text_columns(),
self._query,
partition_filter=self._partition_filter,
+ filter_=self._filter,
)
def _text_columns(self):
diff --git a/paimon-python/pypaimon/table/source/global_index_row_filter.py
b/paimon-python/pypaimon/table/source/global_index_row_filter.py
new file mode 100644
index 0000000000..0dc087f56a
--- /dev/null
+++ b/paimon-python/pypaimon/table/source/global_index_row_filter.py
@@ -0,0 +1,42 @@
+# 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.
+
+"""Exact row filtering for data-evolution search candidates."""
+
+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 matching_rows(table, predicate, candidates, partition_filter=None,
snapshot=None):
+ """Read only filter dependencies and row IDs, closing both reader and
source."""
+ matched = RoaringBitmap64()
+ if candidates.is_empty():
+ return matched
+ table = table_at_snapshot(table, snapshot)
+ builder = (table.new_read_builder().with_filter(predicate)
+ .with_projection([SpecialFields.ROW_ID.name]))
+ if partition_filter is not None:
+ builder = builder.with_partition_filter(partition_filter)
+ splits =
builder.new_scan().with_row_ranges(candidates.to_range_list()).plan().splits()
+ reader, batches = builder.new_read()._new_arrow_batch_reader(splits)
+ with _ClosableArrowBatchReader(reader, batches) as batch_reader:
+ for batch in batch_reader:
+ for row_id in batch.column(SpecialFields.ROW_ID.name).to_pylist():
+ matched.add(row_id)
+ return matched
diff --git a/paimon-python/pypaimon/table/source/hybrid_search_builder.py
b/paimon-python/pypaimon/table/source/hybrid_search_builder.py
index 01736816db..42914c2f27 100644
--- a/paimon-python/pypaimon/table/source/hybrid_search_builder.py
+++ b/paimon-python/pypaimon/table/source/hybrid_search_builder.py
@@ -363,13 +363,6 @@ class HybridSearchBuilderImpl(HybridSearchBuilder):
raise ValueError("Routes cannot be empty")
if self._limit <= 0:
raise ValueError("Limit must be positive, got: %s" % self._limit)
- if self._filter is not None:
- for route in self._routes:
- if route.is_full_text():
- raise ValueError(
- "Hybrid search with full-text routes does not support "
- "non-partition filters because full-text indexes
cannot "
- "apply row-id pre-filters before top-k ranking.")
def _new_vector_search_builder(self, route):
builder = (
@@ -393,6 +386,8 @@ class HybridSearchBuilderImpl(HybridSearchBuilder):
)
if self._partition_filter is not None:
builder.with_partition_filter(self._partition_filter)
+ if self._filter is not None:
+ builder.with_filter(self._filter)
return builder
def _rrf(self, route_results):
diff --git a/paimon-python/pypaimon/table/source/vector_search_read.py
b/paimon-python/pypaimon/table/source/vector_search_read.py
index 59909966d0..0df1fec983 100644
--- a/paimon-python/pypaimon/table/source/vector_search_read.py
+++ b/paimon-python/pypaimon/table/source/vector_search_read.py
@@ -202,23 +202,9 @@ class AbstractVectorSearchReadImpl:
return self._matching_candidate_rows(candidates, snapshot)
def _matching_candidate_rows(self, candidates, snapshot):
- from pypaimon.read.table_read import _ClosableArrowBatchReader
+ from pypaimon.table.source.global_index_row_filter import matching_rows
- matched = RoaringBitmap64()
- if candidates.is_empty():
- return matched
- table = global_index_live_row_filter.table_at_snapshot(self._table,
snapshot)
- builder = (table.new_read_builder().with_filter(self._filter)
- .with_projection([SpecialFields.ROW_ID.name]))
- if self._partition_filter is not None:
- builder = builder.with_partition_filter(self._partition_filter)
- splits =
builder.new_scan().with_row_ranges(candidates.to_range_list()).plan().splits()
- reader, batches = builder.new_read()._new_arrow_batch_reader(splits)
- with _ClosableArrowBatchReader(reader, batches) as batch_reader:
- for batch in batch_reader:
- for row_id in
batch.column(SpecialFields.ROW_ID.name).to_pylist():
- matched.add(row_id)
- return matched
+ return matching_rows(self._table, self._filter, candidates,
self._partition_filter, snapshot)
def _pre_filter(self, splits, snapshot=None):
# Backwards-compatible helper used by older tests/callers.
diff --git a/paimon-python/pypaimon/tests/full_text_scalar_filter_test.py
b/paimon-python/pypaimon/tests/full_text_scalar_filter_test.py
new file mode 100644
index 0000000000..523f573dea
--- /dev/null
+++ b/paimon-python/pypaimon/tests/full_text_scalar_filter_test.py
@@ -0,0 +1,179 @@
+# 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.
+
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+import pypaimon.multimodal as pm
+from pypaimon.table.source.full_text_read import DataEvolutionFullTextRead
+from pypaimon.table.source.global_index_row_filter import matching_rows
+
+
[email protected]
+def docs(tmp_path):
+ pytest.importorskip("paimon_ftindex")
+ schema = pa.schema([("id", pa.int64()), ("text", pa.string()), ("label",
pa.string()),
+ ("pt", pa.string()), ("embedding",
pa.list_(pa.float32(), 2))])
+ table = pm.connect(options={"warehouse": str(tmp_path)}).create_table(
+ "docs", schema=schema, partitioned=["pt"], options={
+ "file.format": "parquet", "vector.file.format": "parquet",
"read.batch-size": "1",
+ "full-text-index.search-mode": "full", "vector-index.search-mode":
"full"})
+ table.add(pa.table({"id": [0, 1, 2], "text": ["paimon", "paimon paimon",
"paimon long document"],
+ "label": ["other", "other", "target"], "pt": ["a"] * 3,
+ "embedding": [[0., 1.], [1., 1.], [2., 1.]]},
schema=schema))
+ table.raw_table.copy({"deletion-vectors.enabled":
"false"}).create_global_index("text", "full-text")
+ 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 builder(docs, predicate=None, limit=10):
+ result = docs.raw_table.new_full_text_search_builder().with_query(
+ "text", '{"match":{"query":"paimon"}}').with_limit(limit)
+ if predicate is not None:
+ from pypaimon.common.where_parser import parse_where_clause
+ result.with_filter(parse_where_clause(predicate,
docs.raw_table.fields))
+ return result
+
+
+def scores(result):
+ return {row_id: result.score_getter()(row_id) for row_id in
result.results()}
+
+
[email protected]("kind", [None, "btree", "bitmap"])
[email protected]("raw", [False, True])
+def test_full_text_data_filters_precede_top_k_and_preserve_scores(docs, kind,
raw):
+ 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]}
+ public = docs.search("paimon", column="text", pre_filter="label LIKE
'%target%'").select(["id"]).limit(1)
+ assert public.to_list() == [{"id": best}]
+
+
+def test_partial_scalar_coverage_does_not_drop_full_text_matches(docs):
+ 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}]
+
+
[email protected]("predicate, expected", [
+ ("pt = 'b' AND label = 'target'", [4]),
+ ("pt = 'a' OR id = 4", [0, 1, 2, 4]),
+ ("id = 2", [2]),
+])
+def test_partition_and_data_predicates_keep_boolean_semantics(docs, predicate,
expected):
+ append_rows(docs)
+ actual = docs.search("paimon", column="text",
pre_filter=predicate).select(["id"]).limit(10).to_list()
+ assert sorted(row["id"] for row in actual) == expected
+
+
+def test_partition_only_filter_keeps_existing_path(docs):
+ with patch("pypaimon.table.source.global_index_row_filter.matching_rows",
+ side_effect=AssertionError("partition-only data read")):
+ 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):
+ 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() == expected
+
+
+def test_non_partitioned_table_filters_before_top_k(tmp_path):
+ pytest.importorskip("paimon_ftindex")
+ schema = pa.schema([("id", pa.int64()), ("text", pa.string())])
+ docs = pm.connect(options={"warehouse": str(tmp_path)}).create_table(
+ "docs", schema=schema, options={"file.format": "parquet"})
+ docs.add(pa.table({"id": [0, 1], "text": ["paimon", "paimon long
document"]}, schema=schema))
+ docs.raw_table.copy({"deletion-vectors.enabled":
"false"}).create_global_index("text", "full-text")
+ assert docs.search("paimon", column="text").where("id =
1").limit(1).to_list() == []
+ assert docs.search("paimon", column="text", pre_filter="id =
1").select(["id"]).limit(1).to_list() == [
+ {"id": 1}]
+
+
+def test_empty_matches_skip_native_search(docs):
+ with patch.object(DataEvolutionFullTextRead, "_eval",
side_effect=AssertionError("empty search")):
+ assert docs.search("paimon", column="text", pre_filter="id <
0").limit(1).to_list() == []
+
+
[email protected]("selector", ["snapshot", "tag"])
+def test_filters_use_historical_snapshot_and_deletions(docs, selector):
+ saved = docs.raw_table.snapshot_manager().get_latest_snapshot()
+ docs.raw_table.create_tag("saved", snapshot_id=saved.id)
+ options = {"snapshot_id": saved.id} if selector == "snapshot" else
{"tag_name": "saved"}
+ docs.delete("id = 2")
+ search = docs.search("paimon", column="text", pre_filter="label =
'target'", **options)
+ assert search.select(["id"]).limit(1).to_list() == [{"id": 2}]
+ assert docs.search("paimon", column="text", pre_filter="label =
'target'").limit(1).to_list() == []
+
+
+def test_commit_during_filter_read_keeps_execution_snapshot(docs):
+ search = docs.search("paimon", column="text", pre_filter="label =
'target'").select(["id"]).limit(1)
+
+ def verify(table, predicate, candidates, partition_filter=None,
snapshot=None):
+ docs.update("id = 2", {"label": "changed"})
+ return matching_rows(table, predicate, candidates, partition_filter,
snapshot)
+
+ with patch("pypaimon.table.source.global_index_row_filter.matching_rows",
verify):
+ assert search.to_list() == [{"id": 2}]
+ assert search.to_list() == []
+
+
+def test_hybrid_applies_data_filter_to_both_routes(docs):
+ search = docs.search_hybrid([
+ pm.vector_route("embedding", [0., 1.], limit=1),
+ pm.text_route("paimon", column="text", limit=1),
+ ], pre_filter="label = 'target'").select(["id"]).limit(1)
+ assert search.to_list() == [{"id": 2}]
+
+
+def test_data_filter_does_not_project_vectors_or_text(docs):
+ from pypaimon.read.table_read import TableRead
+ original = TableRead._new_arrow_batch_reader
+ projections = []
+
+ def read(reader, *args, **kwargs):
+ projections.append([field.name for field in reader.read_type])
+ return original(reader, *args, **kwargs)
+
+ with patch.object(TableRead, "_new_arrow_batch_reader", read):
+ assert docs.search("paimon", column="text", pre_filter="label =
'target'").select(["id"]).limit(1).to_list()
+ assert projections == [["_ROW_ID"]]
+ assert all("embedding" not in fields and "text" not in fields for fields
in projections)
+
+
+def test_primary_key_row_filter_is_explicitly_unsupported(docs):
+ docs.raw_table = docs.raw_table.copy({"data-evolution.enabled": "false"})
+ with pytest.raises(NotImplementedError, match="data-evolution"):
+ builder(docs, "id = 2").new_full_text_read()
diff --git a/paimon-python/pypaimon/tests/vector_search_filter_test.py
b/paimon-python/pypaimon/tests/vector_search_filter_test.py
index 1d2731666f..837e792ea7 100644
--- a/paimon-python/pypaimon/tests/vector_search_filter_test.py
+++ b/paimon-python/pypaimon/tests/vector_search_filter_test.py
@@ -2683,7 +2683,7 @@ class HybridSearchBuilderTest(unittest.TestCase):
self.assertEqual(("query", "content", match_query("paimon search",
"And")),
captured_builders[1].calls[0])
- def test_hybrid_search_rejects_data_filter_with_full_text_route(self):
+ def test_hybrid_search_forwards_data_filter_to_full_text_route(self):
from pypaimon.table.source.hybrid_search_builder import (
HybridSearchBuilderImpl,
)
@@ -2701,9 +2701,8 @@ class HybridSearchBuilderTest(unittest.TestCase):
.with_limit(5)
)
- with self.assertRaises(ValueError) as ctx:
- builder.route_builders()
- self.assertIn("full-text routes", str(ctx.exception))
+ route = builder.route_builders()[0]
+ self.assertEqual(pb.equal("id", 1), route.search_builder._filter)
def test_hybrid_search_rejects_full_text_route_options(self):
from pypaimon.table.source.hybrid_search_builder import (
