[python] Support data prefilters for full-text and hybrid search

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()
 )
@@ -221,8 +231,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.

paimon-python/pypaimon/multimodal/query.py

@@ -599,7 +599,7 @@ def _execute_fts(self, query):
             .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()
 
 

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

@@ -39,6 +39,7 @@
 )
 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 @@ def __init__(
         text_column,
         query: str,
         partition_filter=None,
+        filter_=None,
     ):
         self._table = table
         self._limit = limit
@@ -72,6 +74,7 @@ def __init__(
                 % 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 @@ def read(self, splits: List[FullTextSearchSplit]) -> GlobalIndexResult:
             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 @@ def _eval(self, row_range_start, row_range_end, full_text_index_files,
         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 @@ def _read_raw_search(self, raw_row_ranges, index_type):
 
         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 @@ def _read_raw_search(self, raw_row_ranges, index_type):
             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,15 +329,19 @@ 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,
     )
 
     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

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

@@ -115,18 +115,17 @@ def index_file_filter(entry):
                     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)
 

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

@@ -48,6 +48,11 @@ def with_partition_filter(self, partition_filter) -> 'FullTextSearchBuilder':
         """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."""
@@ -72,6 +77,7 @@ def __init__(self, table: 'FileStoreTable'):
         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
@@ -82,6 +88,20 @@ def with_query(self, field_name: str, query: str) -> '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
@@ -132,6 +152,8 @@ def new_full_text_scan(self) -> FullTextScan:
     def new_full_text_read(self) -> FullTextRead:
         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
@@ -153,6 +175,7 @@ def new_full_text_read(self) -> FullTextRead:
             self._text_columns(),
             self._query,
             partition_filter=self._partition_filter,
+            filter_=self._filter,
         )
 
     def _text_columns(self):

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

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

@@ -361,13 +361,6 @@ def _validate_search(self):
             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 = (
@@ -391,6 +384,8 @@ def _new_full_text_search_builder(self, route):
         )
         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):

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

@@ -202,23 +202,9 @@ def _scalar_matched_rows(self, splits, snapshot=None):
         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.