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 c1adc84e17 [core] Fall back to raw vector search when scalar index
cannot evaluate filter (#9909)
c1adc84e17 is described below
commit c1adc84e17fd0c02ebefb75e51ee3011319370fc
Author: YangJie <[email protected]>
AuthorDate: Thu Sep 17 04:56:39 2026 -0400
[core] Fall back to raw vector search when scalar index cannot evaluate
filter (#9909)
---
.../source/AbstractDataEvolutionVectorRead.java | 57 +++++++++++-
.../table/source/DataEvolutionBatchVectorRead.java | 2 +-
.../table/source/DataEvolutionVectorRead.java | 2 +-
.../table/source/VectorSearchBuilderTest.java | 101 +++++++++++++++++++++
.../vectorsearch/FlinkDataEvolutionVectorRead.java | 2 +-
.../spark/read/SparkDataEvolutionVectorRead.java | 2 +-
6 files changed, 158 insertions(+), 8 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataEvolutionVectorRead.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataEvolutionVectorRead.java
index 83ea140270..0f0b43810d 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataEvolutionVectorRead.java
+++
b/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 @@ public abstract class AbstractDataEvolutionVectorRead
implements Serializable {
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 @@ public abstract class AbstractDataEvolutionVectorRead
implements Serializable {
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 @@ public abstract class AbstractDataEvolutionVectorRead
implements Serializable {
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 @@ public abstract class AbstractDataEvolutionVectorRead
implements Serializable {
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 @@ public abstract class AbstractDataEvolutionVectorRead
implements Serializable {
return table.copyWithoutTimeTravel(pinOptions);
}
- protected static void splitSearchSplits(
+ private static void splitSearchSplits(
List<? extends VectorSearchSplit> splits,
List<IndexVectorSearchSplit> indexSplits,
List<RawVectorSearchSplit> rawSplits) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionBatchVectorRead.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionBatchVectorRead.java
index 6c83a62b23..a94f0d6dba 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionBatchVectorRead.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionBatchVectorRead.java
@@ -70,7 +70,7 @@ public class DataEvolutionBatchVectorRead extends
AbstractDataEvolutionVectorRea
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++) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionVectorRead.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionVectorRead.java
index 9179dd7512..fbac5ef7b7 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionVectorRead.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionVectorRead.java
@@ -68,7 +68,7 @@ public class DataEvolutionVectorRead extends
AbstractDataEvolutionVectorRead imp
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();
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java
index efec98c1f3..e9c972d978 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java
@@ -20,6 +20,7 @@ package org.apache.paimon.table.source;
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.GlobalIndexResult;
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 class VectorSearchBuilderTest extends
TableTestBase {
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(
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/vectorsearch/FlinkDataEvolutionVectorRead.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/vectorsearch/FlinkDataEvolutionVectorRead.java
index 0e2b9a9df2..192bd54002 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/vectorsearch/FlinkDataEvolutionVectorRead.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/vectorsearch/FlinkDataEvolutionVectorRead.java
@@ -91,7 +91,7 @@ public class FlinkDataEvolutionVectorRead extends
DataEvolutionVectorRead {
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();
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/read/SparkDataEvolutionVectorRead.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/read/SparkDataEvolutionVectorRead.java
index a65f7dcb82..47e2a1114b 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/read/SparkDataEvolutionVectorRead.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/read/SparkDataEvolutionVectorRead.java
@@ -74,7 +74,7 @@ public class SparkDataEvolutionVectorRead extends
DataEvolutionVectorRead {
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();
}
