[python] Verify candidate-only scalar filters before vector top-k

docs/docs/pypaimon/multimodal-search.md

@@ -63,6 +63,15 @@ 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.
 
+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
+part of a conjunction is unsupported. Such index candidates are excluded with
+a warning by default, so the result can contain fewer than the requested rows.
+Set the table option `global-index.filter.refine-from-data=true` to verify those
+candidates before vector top-k selection. This reads the filter columns at the
+search snapshot and may scan every candidate row; exact index matches need no
+extra read. This applies to single and batch vector queries, locally and on Ray.
+
 Each execution of `search`, `search_vectors`, or `search_hybrid` reads one
 snapshot across candidate search, filtering, reranking, and result lookup.
 Concurrent commits become visible on the next execution, including when reusing

paimon-python/pypaimon/common/options/core_options.py

@@ -956,6 +956,15 @@ class CoreOptions:
         )
     )
 
+    GLOBAL_INDEX_FILTER_REFINE_FROM_DATA: ConfigOption[bool] = (
+        ConfigOptions.key("global-index.filter.refine-from-data")
+        .boolean_type()
+        .default_value(False)
+        .with_description(
+            "Whether vector search may read filter columns to verify candidate-only scalar index matches. "
+            "When false, inexact index candidates are excluded from the search.")
+    )
+
     GLOBAL_INDEX_THREAD_NUM: ConfigOption[int] = (
         ConfigOptions.key("global-index.thread-num")
         .int_type()
@@ -1742,6 +1751,9 @@ def global_index_external_path(self, default=None):
     def global_index_thread_num(self) -> Optional[int]:
         return self.options.get(CoreOptions.GLOBAL_INDEX_THREAD_NUM)
 
+    def global_index_filter_refine_from_data(self) -> bool:
+        return self.options.get(CoreOptions.GLOBAL_INDEX_FILTER_REFINE_FROM_DATA)
+
     def global_index_row_count_per_shard(self) -> int:
         return self.options.get(CoreOptions.GLOBAL_INDEX_ROW_COUNT_PER_SHARD)
 

paimon-python/pypaimon/globalindex/btree/btree_index_reader.py

@@ -239,16 +239,16 @@ def visit_not_in(self, literals: List[object]) -> Optional[GlobalIndexResult]:
         return GlobalIndexResult.create(result)
 
     def visit_starts_with(self, literal: object) -> Optional[GlobalIndexResult]:
-        return GlobalIndexResult.create(self._all_non_null_rows())
+        return GlobalIndexResult.create(self._all_non_null_rows(), is_exact=False)
 
     def visit_ends_with(self, literal: object) -> Optional[GlobalIndexResult]:
-        return GlobalIndexResult.create(self._all_non_null_rows())
+        return GlobalIndexResult.create(self._all_non_null_rows(), is_exact=False)
 
     def visit_contains(self, literal: object) -> Optional[GlobalIndexResult]:
-        return GlobalIndexResult.create(self._all_non_null_rows())
+        return GlobalIndexResult.create(self._all_non_null_rows(), is_exact=False)
 
     def visit_like(self, literal: object) -> Optional[GlobalIndexResult]:
-        return GlobalIndexResult.create(self._all_non_null_rows())
+        return GlobalIndexResult.create(self._all_non_null_rows(), is_exact=False)
 
     def visit_between(self, min_v: object, max_v: object) -> Optional[GlobalIndexResult]:
         return GlobalIndexResult.create(

paimon-python/pypaimon/globalindex/data_evolution_global_index_scanner.py

@@ -282,7 +282,8 @@ def add_file(self, index_type, range_key, io_meta):
 class _PaddingGlobalIndexReader(GlobalIndexReader):
     def __init__(self, wrapped, padding):
         self._wrapped = wrapped
-        self._padding = padding
+        # Padding rows have not been tested by this index.
+        self._padding = GlobalIndexResult.create(padding.results(), is_exact=False)
 
     def _pad(self, future):
         return _map_future(

paimon-python/pypaimon/globalindex/global_index_evaluator.py

@@ -130,7 +130,11 @@ def _combine_reader_results(
             if child_result is None:
                 continue
             if compound_result is not None:
-                compound_result = compound_result.and_(child_result)
+                # Readers answer the same predicate: an exact result intersected
+                # with a candidate superset remains exact.
+                is_exact = compound_result.is_exact() or child_result.is_exact()
+                compound_result = GlobalIndexResult.create(
+                    compound_result.and_(child_result).results(), is_exact=is_exact)
             else:
                 compound_result = child_result
             if compound_result.is_empty():
@@ -195,6 +199,11 @@ def _combine_results(
                     break
             if compound_result is None:
                 return None
+            if any(child is None for child in results):
+                # A dropped AND child can share a field with a supported child.
+                # Contributing field ids alone therefore cannot prove exactness.
+                compound_result = GlobalIndexResult.create(
+                    compound_result.results(), is_exact=False)
             return GlobalIndexEvaluation(compound_result,
                                          frozenset(contributing_field_ids))
 

paimon-python/pypaimon/globalindex/global_index_result.py

@@ -30,6 +30,10 @@ def results(self) -> RoaringBitmap64:
         """Returns the bitmap representing row ids."""
         pass
 
+    def is_exact(self) -> bool:
+        """Whether these row ids are matches rather than a candidate superset."""
+        return True
+
     def offset(self, start_offset: int) -> 'GlobalIndexResult':
         """Returns a new result with row IDs offset by the given amount."""
         if start_offset == 0:
@@ -38,18 +42,20 @@ def offset(self, start_offset: int) -> 'GlobalIndexResult':
         offset_bitmap = RoaringBitmap64()
         for row_id in bitmap:
             offset_bitmap.add(row_id + start_offset)
-        return SimpleGlobalIndexResult(offset_bitmap)
+        return SimpleGlobalIndexResult(offset_bitmap, self.is_exact())
 
     def and_(self, other: 'GlobalIndexResult') -> 'GlobalIndexResult':
         """Returns the intersection of this result and the other result."""
         return SimpleGlobalIndexResult(
-            RoaringBitmap64.and_(self.results(), other.results())
+            RoaringBitmap64.and_(self.results(), other.results()),
+            self.is_exact() and other.is_exact(),
         )
 
     def or_(self, other: 'GlobalIndexResult') -> 'GlobalIndexResult':
         """Returns the union of this result and the other result."""
         return SimpleGlobalIndexResult(
-            RoaringBitmap64.or_(self.results(), other.results())
+            RoaringBitmap64.or_(self.results(), other.results()),
+            self.is_exact() and other.is_exact(),
         )
 
     def is_empty(self) -> bool:
@@ -62,9 +68,9 @@ def create_empty() -> 'GlobalIndexResult':
         return SimpleGlobalIndexResult(RoaringBitmap64())
 
     @staticmethod
-    def create(bitmap: RoaringBitmap64) -> 'GlobalIndexResult':
+    def create(bitmap: RoaringBitmap64, is_exact: bool = True) -> 'GlobalIndexResult':
         """Returns a new GlobalIndexResult wrapping the given bitmap."""
-        return SimpleGlobalIndexResult(bitmap)
+        return SimpleGlobalIndexResult(bitmap, is_exact)
 
     @staticmethod
     def from_range(range_: Range) -> 'GlobalIndexResult':
@@ -84,8 +90,12 @@ def from_ranges(ranges: List[Range]) -> 'GlobalIndexResult':
 
 class SimpleGlobalIndexResult(GlobalIndexResult):
 
-    def __init__(self, result: RoaringBitmap64):
+    def __init__(self, result: RoaringBitmap64, is_exact: bool = True):
         self._result = result
+        self._is_exact = is_exact
+
+    def is_exact(self) -> bool:
+        return self._is_exact or self.is_empty()
 
     def results(self) -> RoaringBitmap64:
         return self._result

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

@@ -17,6 +17,7 @@
 
 """Vector search read to read index files."""
 
+import logging
 from abc import ABC, abstractmethod
 from collections import deque
 from concurrent.futures import ThreadPoolExecutor
@@ -181,11 +182,43 @@ def _scalar_matched_rows(self, splits, snapshot=None):
             return RoaringBitmap64()
         try:
             result = scanner.scan(self._filter)
-            if result is None:
-                return RoaringBitmap64()
-            return result.results()
         finally:
             scanner.close()
+        if result is not None and result.is_exact():
+            return result.results()
+        if not self._table.options.global_index_filter_refine_from_data():
+            logging.getLogger(__name__).warning(
+                "Scalar index candidates are excluded because the row filter %s cannot be "
+                "evaluated exactly. Set global-index.filter.refine-from-data=true to verify "
+                "candidates against the data; otherwise vector search may return fewer rows.",
+                self._filter)
+            return RoaringBitmap64()
+
+        candidates = RoaringBitmap64()
+        for split in splits:
+            candidates.add_range(split.row_range_start, split.row_range_end)
+        if result is not None:
+            candidates = RoaringBitmap64.and_(candidates, result.results())
+        return self._matching_candidate_rows(candidates, snapshot)
+
+    def _matching_candidate_rows(self, candidates, snapshot):
+        from pypaimon.read.table_read import _ClosableArrowBatchReader
+
+        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
 
     def _pre_filter(self, splits, snapshot=None):
         # Backwards-compatible helper used by older tests/callers.

paimon-python/pypaimon/tests/ray_vector_filter_exactness_test.py

@@ -0,0 +1,60 @@
+# 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 pytest
+
+pytest.importorskip("ray")
+
+from pypaimon.read.table_read import TableRead
+from pypaimon.table.source.vector_search_read import AbstractVectorSearchReadImpl
+from pypaimon.tests import ray_vector_search_test as ray_fixtures
+from pypaimon.tests import vector_filter_exactness_test as fixtures
+from pypaimon.tests.vector_filter_exactness_test import query, scalar_index
+
+ray_cluster = ray_fixtures.ray_cluster
+table = fixtures.table
+
+
+@pytest.mark.parametrize("batch", [False, True])
+@pytest.mark.parametrize("mode", ["full", "fast"])
+@pytest.mark.parametrize("refine", [False, True])
+def test_ray_applies_exact_row_filter_before_worker_top_k(table, ray_cluster, 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()})
+    original = AbstractVectorSearchReadImpl._matching_candidate_rows
+    arrow_read = TableRead._new_arrow_batch_reader
+    calls = []
+
+    def no_vectors(read, *args, **kwargs):
+        assert "embedding" not in [field.name for field in read.read_type]
+        return arrow_read(read, *args, **kwargs)
+
+    def verify(reader, candidates, snapshot):
+        calls.append(list(candidates))
+        assert snapshot is not None
+        with patch.object(TableRead, "_new_arrow_batch_reader", no_vectors):
+            return original(reader, candidates, snapshot)
+
+    with patch.object(AbstractVectorSearchReadImpl, "_matching_candidate_rows", verify):
+        result = query(table, "name LIKE '%zeta%'", batch).to_arrow(execution="ray", concurrency=2)
+    actual = [value.to_pylist() for value in result] if batch else result.to_pylist()
+    expected = [{"id": 1}] if refine else []
+    assert actual == ([expected, expected] if batch else expected)
+    assert calls == ([[0, 1, 2]] if refine else [])