[core] Fall back to raw vector search when scalar index cannot evaluate filter

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

@@ -88,6 +88,13 @@ public abstract class AbstractDataEvolutionVectorRead implements Serializable {
     /** Snapshot the plan was built against; pins filters and raw reads to it. */
     @Nullable protected Snapshot planSnapshot;
 
+    /**
+     * Scalar pre-filter computed while splitting index/raw splits for the current read; lets {@link
+     * #preFilters} reuse it instead of scanning the scalar index twice. Only written and read on
+     * the single thread performing the read.
+     */
+    @Nullable private transient RoaringNavigableMap64 scalarPreFilter;
+
     private static final Comparator<long[]> WEAKEST_SCORE_FIRST =
             Comparator.<long[]>comparingDouble(a -> Float.intBitsToFloat((int) a[1]))
                     .thenComparing((a, b) -> Long.compare(b[0], a[0]));
@@ -132,6 +139,42 @@ private GlobalIndexer createGlobalIndexer(String indexType, GlobalIndexMeta meta
                         table.coreOptions().toConfiguration());
     }
 
+    /**
+     * Splits the search splits into index and raw splits, then moves the index splits aside when
+     * the scalar index cannot evaluate {@link #filter}: the index path answers with all-or-nothing
+     * bitmaps, so with an unevaluable filter it can neither honor the filter (an empty bitmap
+     * silently drops every covered row) nor ignore it (the top-K would be polluted by non-matching
+     * rows). Instead, suppress the index splits and route their ranges through the raw search,
+     * where the exact final-read filter decides; the raw read merges ranges, so overlap with an
+     * existing raw split is deduplicated. Splits routed this way reuse the vector index type so raw
+     * scoring keeps its metric. The computed scalar pre-filter is cached for {@link #preFilters} to
+     * reuse.
+     */
+    protected void prepareSplits(
+            List<? extends VectorSearchSplit> splits,
+            List<IndexVectorSearchSplit> indexSplits,
+            List<RawVectorSearchSplit> rawSplits) {
+        splitSearchSplits(splits, indexSplits, rawSplits);
+        scalarPreFilter = null;
+        if (filter == null || indexSplits.isEmpty()) {
+            return;
+        }
+        RoaringNavigableMap64 matchedRows = scalarMatchedRows(indexSplits);
+        if (matchedRows != null) {
+            scalarPreFilter = matchedRows;
+            return;
+        }
+        List<Range> ranges = new ArrayList<>();
+        List<IndexFileMeta> scalarIndexFiles = new ArrayList<>();
+        for (IndexVectorSearchSplit split : indexSplits) {
+            ranges.add(new Range(split.rowRangeStart(), split.rowRangeEnd()));
+            scalarIndexFiles.addAll(split.scalarIndexFiles());
+        }
+        rawSplits.add(
+                new RawVectorSearchSplit(ranges, scalarIndexFiles, vectorIndexType(indexSplits)));
+        indexSplits.clear();
+    }
+
     protected List<RoaringNavigableMap64> preFilters(List<IndexVectorSearchSplit> splits) {
         List<Range> indexedRowRanges = new ArrayList<>(splits.size());
         for (IndexVectorSearchSplit split : splits) {
@@ -141,7 +184,8 @@ protected List<RoaringNavigableMap64> preFilters(List<IndexVectorSearchSplit> sp
         RoaringNavigableMap64 liveRows =
                 GlobalIndexLiveRowFilter.liveRows(
                         table, planSnapshot, partitionFilter, indexedRowRanges);
-        RoaringNavigableMap64 matchedRows = scalarMatchedRows(splits);
+        RoaringNavigableMap64 matchedRows =
+                scalarPreFilter != null ? scalarPreFilter : scalarMatchedRows(splits);
 
         List<RoaringNavigableMap64> includeRowIds = new ArrayList<>(splits.size());
         boolean hasFilter = false;
@@ -167,6 +211,11 @@ protected List<RoaringNavigableMap64> preFilters(List<IndexVectorSearchSplit> sp
         return hasFilter ? includeRowIds : Collections.emptyList();
     }
 
+    /**
+     * 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".
+     */
     @Nullable
     private RoaringNavigableMap64 scalarMatchedRows(List<IndexVectorSearchSplit> splits) {
         if (filter == null) {
@@ -183,13 +232,13 @@ private RoaringNavigableMap64 scalarMatchedRows(List<IndexVectorSearchSplit> spl
                 DataEvolutionGlobalIndexScanner.create(
                         table, planSnapshot, partitionFilter, scalarIndexFiles);
         if (!optionalScanner.isPresent()) {
-            return new RoaringNavigableMap64();
+            return null;
         }
 
         try (DataEvolutionGlobalIndexScanner scanner = optionalScanner.get()) {
             Optional<GlobalIndexResult> result = scanner.scan(filter);
             if (!result.isPresent()) {
-                return new RoaringNavigableMap64();
+                return null;
             }
             return result.get().results();
         } catch (IOException e) {
@@ -632,7 +681,7 @@ private FileStoreTable rawReadTable() {
         return table.copyWithoutTimeTravel(pinOptions);
     }
 
-    protected static void splitSearchSplits(
+    private static void splitSearchSplits(
             List<? extends VectorSearchSplit> splits,
             List<IndexVectorSearchSplit> indexSplits,
             List<RawVectorSearchSplit> rawSplits) {

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

@@ -70,7 +70,7 @@ private List<GlobalIndexResult> readBatch(List<VectorSearchSplit> splits) {
         int n = vectors.length;
         List<IndexVectorSearchSplit> indexSplits = new ArrayList<>();
         List<RawVectorSearchSplit> rawSplits = new ArrayList<>();
-        splitSearchSplits(splits, indexSplits, rawSplits);
+        prepareSplits(splits, indexSplits, rawSplits);
         if (indexSplits.isEmpty() && rawSplits.isEmpty()) {
             List<GlobalIndexResult> empty = new ArrayList<>(n);
             for (int i = 0; i < n; i++) {

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

@@ -68,7 +68,7 @@ public GlobalIndexResult read(VectorScan.Plan plan) {
     protected GlobalIndexResult readSplits(List<? extends VectorSearchSplit> splits) {
         List<IndexVectorSearchSplit> indexSplits = new ArrayList<>();
         List<RawVectorSearchSplit> rawSplits = new ArrayList<>();
-        splitSearchSplits(splits, indexSplits, rawSplits);
+        prepareSplits(splits, indexSplits, rawSplits);
         if (indexSplits.isEmpty() && rawSplits.isEmpty()) {
             return GlobalIndexResult.createEmpty();
         }

paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java

@@ -20,6 +20,7 @@
 
 import org.apache.paimon.CoreOptions;
 import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryString;
 import org.apache.paimon.data.GenericArray;
 import org.apache.paimon.data.GenericRow;
 import org.apache.paimon.data.InternalRow;
@@ -31,6 +32,7 @@
 import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
 import org.apache.paimon.globalindex.ResultEntry;
 import org.apache.paimon.globalindex.ScoredGlobalIndexResult;
+import org.apache.paimon.globalindex.bitmap.MultiValueGlobalIndexerFactory;
 import org.apache.paimon.globalindex.btree.BTreeGlobalIndexerFactory;
 import org.apache.paimon.globalindex.testvector.TestVectorGlobalIndexer;
 import org.apache.paimon.globalindex.testvector.TestVectorGlobalIndexerFactory;
@@ -587,6 +589,105 @@ public void testVectorSearchFullModeScansFilteredUnindexedData() throws Exceptio
         assertThat(result.results()).containsExactlyInAnyOrder(2L, 3L);
     }
 
+    @Test
+    public void testVectorSearchFilterIndexCannotEvaluate() throws Exception {
+        catalog.createTable(
+                identifier("vector_search_unsupported_scalar_predicate"),
+                withVectorSchemaOptions(
+                                Schema.newBuilder()
+                                        .column("id", DataTypes.INT())
+                                        .column("tags", new ArrayType(DataTypes.STRING()))
+                                        .column(
+                                                VECTOR_FIELD_NAME,
+                                                new ArrayType(DataTypes.FLOAT())))
+                        .build(),
+                false);
+        FileStoreTable table = getTable(identifier("vector_search_unsupported_scalar_predicate"));
+
+        // row 0 has NULL tags and is the closest to the query: dropping the index
+        // pre-filter alone would pollute the top-K with it, and keeping the old empty
+        // bitmap would drop every row — only routing the search through the exact
+        // final-read filter returns the closest *matching* rows
+        float[][] vectors = {{0.0f, 0.0f}, {1.0f, 0.0f}, {2.0f, 0.0f}, {3.0f, 0.0f}};
+        writeVectorsWithTags(table, vectors, 0);
+        buildAndCommitIndex(table, vectors);
+        buildAndCommitMultiValueIndex(table, vectors.length);
+
+        // IS NOT NULL on the multivalue-indexed array column is a function the reader
+        // cannot evaluate; the scalar pre-filter must step aside and let the exact
+        // final-read filter decide instead of answering "no rows match"
+        Predicate filter =
+                new PredicateBuilder(table.rowType())
+                        .isNotNull(table.rowType().getFieldIndex("tags"));
+        GlobalIndexResult result =
+                table.newVectorSearchBuilder()
+                        .withVector(new float[] {0.0f, 0.0f})
+                        .withLimit(2)
+                        .withVectorColumn(VECTOR_FIELD_NAME)
+                        .withFilter(filter)
+                        .executeLocal();
+
+        assertThat(result.results()).containsExactly(1L, 2L);
+        assertThat(readIds(table, result)).containsExactly(1, 2);
+    }
+
+    private void writeVectorsWithTags(FileStoreTable table, float[][] vectors, int nullTagsRow)
+            throws Exception {
+        BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
+        try (BatchTableWrite write = writeBuilder.newWrite();
+                BatchTableCommit commit = writeBuilder.newCommit()) {
+            for (int i = 0; i < vectors.length; i++) {
+                GenericArray tags =
+                        i == nullTagsRow
+                                ? null
+                                : new GenericArray(
+                                        new BinaryString[] {BinaryString.fromString("t" + i)});
+                write.write(GenericRow.of(i, tags, new GenericArray(vectors[i])));
+            }
+            commit.commit(write.prepareCommit());
+        }
+    }
+
+    private void buildAndCommitMultiValueIndex(FileStoreTable table, int rowCount)
+            throws Exception {
+        Options options = table.coreOptions().toConfiguration();
+        DataField tagsField = table.rowType().getField("tags");
+
+        GlobalIndexSingleColumnWriter writer =
+                (GlobalIndexSingleColumnWriter)
+                        GlobalIndexBuilderUtils.createIndexWriter(
+                                table,
+                                MultiValueGlobalIndexerFactory.IDENTIFIER,
+                                tagsField,
+                                options);
+        for (int row = 0; row < rowCount; row++) {
+            writer.write(BinaryString.fromString("t" + row), row);
+        }
+        List<ResultEntry> entries = writer.finish(rowCount);
+
+        List<IndexFileMeta> indexFiles =
+                GlobalIndexBuilderUtils.toIndexFileMetas(
+                        table.fileIO(),
+                        table.store().pathFactory().globalIndexFileFactory(),
+                        table.coreOptions(),
+                        new Range(0, rowCount - 1),
+                        tagsField.id(),
+                        MultiValueGlobalIndexerFactory.IDENTIFIER,
+                        entries);
+
+        DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles);
+        CommitMessage message =
+                new CommitMessageImpl(
+                        BinaryRow.EMPTY_ROW,
+                        0,
+                        null,
+                        dataIncrement,
+                        CompactIncrement.emptyIncrement());
+        try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) {
+            commit.commit(Collections.singletonList(message));
+        }
+    }
+
     @Test
     public void testVectorSearchRawSearchUsesScalarPreFilter() throws Exception {
         catalog.createTable(

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/vectorsearch/FlinkDataEvolutionVectorRead.java

@@ -91,7 +91,7 @@ public GlobalIndexResult read(VectorScan.Plan plan) {
         this.planSnapshot = plan.snapshot();
         List<IndexVectorSearchSplit> indexSplits = new ArrayList<>();
         List<RawVectorSearchSplit> rawSplits = new ArrayList<>();
-        splitSearchSplits(plan.splits(), indexSplits, rawSplits);
+        prepareSplits(plan.splits(), indexSplits, rawSplits);
         if (indexSplits.isEmpty() && rawSplits.isEmpty()) {
             return GlobalIndexResult.createEmpty();
         }

paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/read/SparkDataEvolutionVectorRead.java

@@ -74,7 +74,7 @@ public GlobalIndexResult read(VectorScan.Plan plan) {
         this.planSnapshot = plan.snapshot();
         List<IndexVectorSearchSplit> indexSplits = new ArrayList<>();
         List<RawVectorSearchSplit> rawSplits = new ArrayList<>();
-        splitSearchSplits(plan.splits(), indexSplits, rawSplits);
+        prepareSplits(plan.splits(), indexSplits, rawSplits);
         if (indexSplits.isEmpty() && rawSplits.isEmpty()) {
             return GlobalIndexResult.createEmpty();
         }