[core] Refine candidate-only scalar index answers before vector top-k

docs/docs/multimodal-table/global-index/full-text.mdx

@@ -300,8 +300,13 @@ BM25 statistics are always those of the full corpus:
   Bitmap, Multivalue, FM) are decided by the index, the same way
   [vector search pre-filters](./vector#vector-search) rows. When the index can
   only produce candidates (a conjunct no index could evaluate, or `contains`,
-  ends-with and `LIKE` on a BTree index), the candidates are verified by reading
-  their filter columns.
+  ends-with and `LIKE` on a BTree index), those candidates are never ranked as
+  if they matched: with `global-index.filter.refine-from-data=true` they are
+  verified by reading their filter columns, otherwise (the default) they are
+  excluded and a warning is logged, so the result may hold fewer than `limit`
+  rows. The read runs on the caller and can cover every candidate row, which is
+  why it is opt-in; prefer an index that answers the predicate exactly, such as
+  Bitmap or FM for `contains`.
 - Rows whose filter columns are not covered by a scalar index follow
   `scalar-index.search-mode`:
 

docs/docs/multimodal-table/global-index/vector.mdx

@@ -285,6 +285,13 @@ The scalar filter is evaluated with matching scalar global indexes before vector
 BTree index for frequently used metadata filters, such as `category`, `tenant_id`, or `event_time`,
 so vector search can restrict the candidate row ids before running ANN search.
 
+An index answer that is only a candidate set — `contains`, ends-with or `LIKE` on a BTree index,
+or a conjunction with a member no index can evaluate — is never ranked as if it matched, because
+a non-matching but closer row would take a top-k slot from a matching one. By default such
+candidates are excluded and a warning is logged, so the result may hold fewer than `limit` rows;
+set `global-index.filter.refine-from-data=true` to verify them by reading their filter columns
+instead. That read runs on the caller and can cover every candidate row.
+
 </TabItem>
 
 </Tabs>

docs/generated/core_configuration.html

@@ -866,6 +866,12 @@
             <td>String</td>
             <td>Global index root directory, if not set, the global index files will be stored under the &lt;table-root-directory&gt;/index.</td>
         </tr>
+        <tr>
+            <td><h5>global-index.filter.refine-from-data</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>Whether a vector, hybrid or full-text search may read the filter columns of candidate rows to verify a row filter that the scalar global index can only answer with a superset, such as contains, ends-with or like on a BTree index or a conjunction with a member no index can evaluate. When false, such candidates are excluded from the search, which never returns a non-matching row but may return fewer than the requested top-k. When true, the read runs on the caller and may cover every candidate row.</td>
+        </tr>
         <tr>
             <td><h5>global-index.row-count-per-shard</h5></td>
             <td style="word-wrap: break-word;">100000</td>

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -3000,6 +3000,21 @@ public String toString() {
                     .defaultValue(GlobalIndexSearchMode.FAST)
                     .withDescription("Search mode for full-text index queries.");
 
+    public static final ConfigOption<Boolean> GLOBAL_INDEX_FILTER_REFINE_FROM_DATA =
+            key("global-index.filter.refine-from-data")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Whether a vector, hybrid or full-text search may read the filter "
+                                    + "columns of candidate rows to verify a row filter that the "
+                                    + "scalar global index can only answer with a superset, such "
+                                    + "as contains, ends-with or like on a BTree index or a "
+                                    + "conjunction with a member no index can evaluate. When "
+                                    + "false, such candidates are excluded from the search, which "
+                                    + "never returns a non-matching row but may return fewer than "
+                                    + "the requested top-k. When true, the read runs on the caller "
+                                    + "and may cover every candidate row.");
+
     public static final ConfigOption<Integer> GLOBAL_INDEX_THREAD_NUM =
             key("global-index.thread-num")
                     .intType()
@@ -4781,6 +4796,10 @@ public GlobalIndexSearchMode scalarIndexSearchMode() {
         return indexSearchMode(SCALAR_INDEX_SEARCH_MODE);
     }
 
+    public boolean globalIndexFilterRefineFromData() {
+        return options.get(GLOBAL_INDEX_FILTER_REFINE_FROM_DATA);
+    }
+
     public GlobalIndexSearchMode vectorIndexSearchMode() {
         return indexSearchMode(VECTOR_INDEX_SEARCH_MODE);
     }

paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataEvolutionVectorRead.java

@@ -28,7 +28,6 @@
 import org.apache.paimon.globalindex.GlobalIndexEvaluator;
 import org.apache.paimon.globalindex.GlobalIndexIOMeta;
 import org.apache.paimon.globalindex.GlobalIndexReader;
-import org.apache.paimon.globalindex.GlobalIndexResult;
 import org.apache.paimon.globalindex.GlobalIndexer;
 import org.apache.paimon.globalindex.GlobalIndexerFactoryUtils;
 import org.apache.paimon.globalindex.OffsetGlobalIndexReader;
@@ -54,6 +53,9 @@
 import org.apache.paimon.utils.Range;
 import org.apache.paimon.utils.RoaringNavigableMap64;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import javax.annotation.Nullable;
 
 import java.io.IOException;
@@ -76,6 +78,9 @@
 /** Base implementation for vector reads. */
 public abstract class AbstractDataEvolutionVectorRead implements Serializable {
 
+    private static final Logger LOG =
+            LoggerFactory.getLogger(AbstractDataEvolutionVectorRead.class);
+
     private static final long serialVersionUID = 1L;
 
     protected final FileStoreTable table;
@@ -212,9 +217,14 @@ protected List<RoaringNavigableMap64> preFilters(List<IndexVectorSearchSplit> sp
     }
 
     /**
-     * Row ids the scalar index reports as matching {@link #filter}, or {@code null} when the index
-     * cannot evaluate the predicate (no scalar index files, or a function the reader does not
-     * support). {@code null} means "cannot decide", never "no rows match".
+     * Rows of the indexed splits that satisfy {@link #filter} according to the scalar global
+     * indexes, or {@code null} when no index can evaluate the predicate (no scalar index files, or
+     * a function the reader does not support); {@code null} means "cannot decide", never "no rows
+     * match". The set is exact: an index answer that may be a superset (see {@link
+     * FilteredRowIdReader#isExact}) is refined from the data when {@code
+     * global-index.filter.refine-from-data} allows it and excluded otherwise, because a superset
+     * ranked by the ANN would push matching rows out of the top-k where the engine-side filter
+     * cannot bring them back.
      */
     @Nullable
     private RoaringNavigableMap64 scalarMatchedRows(List<IndexVectorSearchSplit> splits) {
@@ -224,8 +234,10 @@ private RoaringNavigableMap64 scalarMatchedRows(List<IndexVectorSearchSplit> spl
 
         Set<IndexFileMeta> scalarIndexFiles =
                 new TreeSet<>(Comparator.comparing(IndexFileMeta::fileName));
+        RoaringNavigableMap64 splitRows = new RoaringNavigableMap64();
         for (IndexVectorSearchSplit split : splits) {
             scalarIndexFiles.addAll(split.scalarIndexFiles());
+            splitRows.addRange(new Range(split.rowRangeStart(), split.rowRangeEnd()));
         }
 
         Optional<DataEvolutionGlobalIndexScanner> optionalScanner =
@@ -236,11 +248,21 @@ private RoaringNavigableMap64 scalarMatchedRows(List<IndexVectorSearchSplit> spl
         }
 
         try (DataEvolutionGlobalIndexScanner scanner = optionalScanner.get()) {
-            Optional<GlobalIndexResult> result = scanner.scan(filter);
-            if (!result.isPresent()) {
+            Optional<GlobalIndexEvaluator.Evaluation> evaluation = scanner.scanWithCoverage(filter);
+            if (!evaluation.isPresent()) {
                 return null;
             }
-            return result.get().results();
+            RoaringNavigableMap64 matched = evaluation.get().result().results();
+            if (FilteredRowIdReader.isExact(table.rowType(), filter, evaluation.get())) {
+                return matched;
+            }
+            if (!table.coreOptions().globalIndexFilterRefineFromData()) {
+                FilteredRowIdReader.warnCandidatesExcluded(LOG, table, filter);
+                return new RoaringNavigableMap64();
+            }
+            RoaringNavigableMap64 candidates = RoaringNavigableMap64.and(matched, splitRows);
+            return new FilteredRowIdReader(table, planSnapshot, partitionFilter, filter)
+                    .matchingRowIds(candidates);
         } catch (IOException e) {
             throw new RuntimeException(e);
         }

paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextRead.java

@@ -36,13 +36,7 @@
 import org.apache.paimon.index.IndexFileMeta;
 import org.apache.paimon.index.IndexPathFactory;
 import org.apache.paimon.partition.PartitionPredicate;
-import org.apache.paimon.predicate.CompoundPredicate;
-import org.apache.paimon.predicate.Contains;
-import org.apache.paimon.predicate.EndsWith;
 import org.apache.paimon.predicate.FullTextSearch;
-import org.apache.paimon.predicate.LeafFunction;
-import org.apache.paimon.predicate.LeafPredicate;
-import org.apache.paimon.predicate.Like;
 import org.apache.paimon.predicate.Predicate;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.types.DataField;
@@ -68,7 +62,6 @@
 import java.util.concurrent.ExecutorService;
 
 import static org.apache.paimon.CoreOptions.GLOBAL_INDEX_THREAD_NUM;
-import static org.apache.paimon.predicate.PredicateVisitor.collectFieldIds;
 import static org.apache.paimon.utils.Preconditions.checkNotNull;
 
 /** Implementation for {@link FullTextRead}. */
@@ -183,7 +176,8 @@ private GlobalIndexResult read(
      *       index. When the index answer may be a superset (a conjunct it could not evaluate was
      *       dropped, or a {@code contains} / {@code endsWith} / {@code like} leaf, which BTree
      *       answers with every non-null row), the candidates are refined by reading their filter
-     *       columns.
+     *       columns if {@code global-index.filter.refine-from-data} allows it, and excluded
+     *       otherwise.
      *   <li>Rows whose filter columns are not covered follow {@code scalar-index.search-mode}:
      *       excluded in {@code fast}, otherwise decided by reading their filter columns.
      * </ul>
@@ -237,10 +231,17 @@ private RoaringNavigableMap64 matchedRows(
                 RoaringNavigableMap64 fromIndex =
                         RoaringNavigableMap64.and(
                                 evaluation.get().result().results(), decidedByIndex);
-                if (!isExact(evaluation.get()) && !fromIndex.isEmpty()) {
-                    fromIndex =
-                            new FilteredRowIdReader(table, planSnapshot, partitionFilter, filter)
-                                    .matchingRowIds(fromIndex);
+                if (!FilteredRowIdReader.isExact(table.rowType(), filter, evaluation.get())
+                        && !fromIndex.isEmpty()) {
+                    if (table.coreOptions().globalIndexFilterRefineFromData()) {
+                        fromIndex =
+                                new FilteredRowIdReader(
+                                                table, planSnapshot, partitionFilter, filter)
+                                        .matchingRowIds(fromIndex);
+                    } else {
+                        FilteredRowIdReader.warnCandidatesExcluded(LOG, table, filter);
+                        fromIndex = new RoaringNavigableMap64();
+                    }
                 }
                 matched.or(fromIndex);
             } else {
@@ -270,37 +271,6 @@ private Optional<GlobalIndexEvaluator.Evaluation> evaluateWithIndexes(
         }
     }
 
-    /**
-     * Whether an index evaluation is an exact match set. The global index contract only promises
-     * candidates: a conjunct no index could evaluate is dropped, and BTree answers substring
-     * predicates with every non-null row. Both cases are refined from the data.
-     */
-    private boolean isExact(GlobalIndexEvaluator.Evaluation evaluation) {
-        Set<Integer> filterFieldIds = collectFieldIds(table.rowType(), filter);
-        if (!evaluation.contributingFieldIds().containsAll(filterFieldIds)) {
-            return false;
-        }
-        return !hasCandidateOnlyLeaf(filter);
-    }
-
-    private static boolean hasCandidateOnlyLeaf(Predicate predicate) {
-        if (predicate instanceof CompoundPredicate) {
-            for (Predicate child : ((CompoundPredicate) predicate).children()) {
-                if (hasCandidateOnlyLeaf(child)) {
-                    return true;
-                }
-            }
-            return false;
-        }
-        if (predicate instanceof LeafPredicate) {
-            LeafFunction function = ((LeafPredicate) predicate).function();
-            return function instanceof Contains
-                    || function instanceof EndsWith
-                    || function instanceof Like;
-        }
-        return false;
-    }
-
     private void warnUnindexedFilter() {
         if (table.coreOptions().scalarIndexSearchMode() == GlobalIndexSearchMode.FAST) {
             LOG.warn(

paimon-core/src/main/java/org/apache/paimon/table/source/FilteredRowIdReader.java

@@ -21,7 +21,14 @@
 import org.apache.paimon.CoreOptions;
 import org.apache.paimon.Snapshot;
 import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.globalindex.GlobalIndexEvaluator;
 import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.predicate.CompoundPredicate;
+import org.apache.paimon.predicate.Contains;
+import org.apache.paimon.predicate.EndsWith;
+import org.apache.paimon.predicate.LeafFunction;
+import org.apache.paimon.predicate.LeafPredicate;
+import org.apache.paimon.predicate.Like;
 import org.apache.paimon.predicate.Predicate;
 import org.apache.paimon.predicate.PredicateVisitor;
 import org.apache.paimon.reader.RecordReader;
@@ -31,6 +38,8 @@
 import org.apache.paimon.utils.Range;
 import org.apache.paimon.utils.RoaringNavigableMap64;
 
+import org.slf4j.Logger;
+
 import javax.annotation.Nullable;
 
 import java.io.IOException;
@@ -40,6 +49,8 @@
 import java.util.Map;
 import java.util.Set;
 
+import static org.apache.paimon.predicate.PredicateVisitor.collectFieldIds;
+
 /**
  * Evaluates a row predicate on the data for a given set of row ids and returns the ids that satisfy
  * it. Only the filter columns and the row id are read, so this is the exact counterpart of a scalar
@@ -64,6 +75,53 @@ class FilteredRowIdReader {
         this.filter = filter;
     }
 
+    /**
+     * Whether a scalar global index evaluation of {@code filter} is an exact match set. The global
+     * index contract only promises candidates: {@link GlobalIndexEvaluator} drops a conjunct no
+     * index can evaluate, and BTree answers {@code contains} / {@code endsWith} / {@code like} with
+     * every non-null row. A search that ranks rows before the engine filters them must refine a
+     * non-exact answer through {@link #matchingRowIds} first.
+     */
+    static boolean isExact(
+            RowType rowType, Predicate filter, GlobalIndexEvaluator.Evaluation evaluation) {
+        Set<Integer> filterFieldIds = collectFieldIds(rowType, filter);
+        if (!evaluation.contributingFieldIds().containsAll(filterFieldIds)) {
+            return false;
+        }
+        return !hasCandidateOnlyLeaf(filter);
+    }
+
+    private static boolean hasCandidateOnlyLeaf(Predicate predicate) {
+        if (predicate instanceof CompoundPredicate) {
+            for (Predicate child : ((CompoundPredicate) predicate).children()) {
+                if (hasCandidateOnlyLeaf(child)) {
+                    return true;
+                }
+            }
+            return false;
+        }
+        if (predicate instanceof LeafPredicate) {
+            LeafFunction function = ((LeafPredicate) predicate).function();
+            return function instanceof Contains
+                    || function instanceof EndsWith
+                    || function instanceof Like;
+        }
+        return false;
+    }
+
+    /** Logs that a candidate-only index answer was dropped because refinement is disabled. */
+    static void warnCandidatesExcluded(Logger log, FileStoreTable table, Predicate filter) {
+        log.warn(
+                "The scalar global index can only answer the row filter {} on table {} with "
+                        + "candidates, and {} is false, so those rows are excluded from the "
+                        + "search; the result may hold fewer rows than requested. Set the option "
+                        + "to true to verify the candidates against the data, or build an index "
+                        + "that answers the predicate exactly.",
+                filter,
+                table.name(),
+                CoreOptions.GLOBAL_INDEX_FILTER_REFINE_FROM_DATA.key());
+    }
+
     /** The subset of {@code rows} whose data satisfies the filter. */
     RoaringNavigableMap64 matchingRowIds(RoaringNavigableMap64 rows) {
         RoaringNavigableMap64 matching = new RoaringNavigableMap64();