Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
LuciferYang wants to merge 1 commit into
base: master
Choose a base branch
from
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
86 changes: 80 additions & 6 deletions ...on-core/src/main/java/org/apache/paimon/table/source/AbstractDataEvolutionVectorRead.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -71,6 +70,7 @@
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;

import static org.apache.paimon.predicate.PredicateVisitor.collectFieldIds;
import static org.apache.paimon.utils.Preconditions.checkNotNull;

/** Base implementation for vector reads. */
Expand All @@ -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]));
Expand Down Expand Up @@ -132,6 +139,56 @@ private GlobalIndexer createGlobalIndexer(String indexType, GlobalIndexMeta meta
table.coreOptions().toConfiguration());
}

/**
* Moves index splits aside when the scalar index cannot EXACTLY evaluate {@link #filter} (it
* evaluated nothing, or only some of the filter's fields; see {@link #scalarMatchedRows}): the
* index path answers with all-or-nothing bitmaps and has no final-read filter, so an inexact
* bitmap can neither honor the filter (an empty bitmap silently drops every covered row, a
* superset lets non-matching rows into the top-K) nor be ignored. Instead, suppress the index
* splits and route the ranges the raw splits do not already cover through the raw search, where
* the exact final-read filter decides; already-covered ranges would duplicate that work. Splits
* routed this way reuse the vector index type so raw scoring keeps its metric. Must run after
* {@link #splitSearchSplits} and before the index splits are read, on the single thread
* performing the read; the computed pre-filter is cached for {@link #preFilters} to reuse.
*/
protected void demoteUncoverableIndexSplits(
List<IndexVectorSearchSplit> indexSplits, List<RawVectorSearchSplit> rawSplits) {
scalarPreFilter = null;
if (filter == null || indexSplits.isEmpty()) {
return;
}
RoaringNavigableMap64 matchedRows = scalarMatchedRows(indexSplits);
if (matchedRows != null) {
scalarPreFilter = matchedRows;
return;
}
String indexType = vectorIndexType(indexSplits);
List<Range> uncovered = new ArrayList<>();
List<IndexFileMeta> scalarIndexFiles = new ArrayList<>();
for (IndexVectorSearchSplit split : indexSplits) {
if (!coveredByRawSplits(split.rowRangeStart(), split.rowRangeEnd(), rawSplits)) {
uncovered.add(new Range(split.rowRangeStart(), split.rowRangeEnd()));
scalarIndexFiles.addAll(split.scalarIndexFiles());
}
}
if (!uncovered.isEmpty()) {
rawSplits.add(new RawVectorSearchSplit(uncovered, scalarIndexFiles, indexType));
}
indexSplits.clear();
}

private static boolean coveredByRawSplits(
long start, long end, List<RawVectorSearchSplit> rawSplits) {
for (RawVectorSearchSplit raw : rawSplits) {
for (Range range : raw.rowRanges()) {
if (range.from <= start && end <= range.to) {
return true;
}
}
}
return false;
}

protected List<RoaringNavigableMap64> preFilters(List<IndexVectorSearchSplit> splits) {
List<Range> indexedRowRanges = new ArrayList<>(splits.size());
for (IndexVectorSearchSplit split : splits) {
Expand All @@ -141,7 +198,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;
Expand All @@ -167,6 +225,14 @@ 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 it cannot
* evaluate the predicate exactly: no scalar index files, an unsupported function (e.g. {@code
* IS NOT NULL} on a multivalue-indexed array), or only some of the filter's fields contributed
* (an unsupported conjunct in an AND yields a superset). {@code null} means "cannot decide",
* never "no rows match": the index path trusts this bitmap as the exact include set with no
* final-read filter, so the caller must route these rows to the raw search instead.
*/
@Nullable
private RoaringNavigableMap64 scalarMatchedRows(List<IndexVectorSearchSplit> splits) {
if (filter == null) {
Expand All @@ -183,15 +249,23 @@ 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);
Optional<GlobalIndexEvaluator.Evaluation> result = scanner.scanWithCoverage(filter);
if (!result.isPresent()) {
return new RoaringNavigableMap64();
return null;
}
// Trust the bitmap as the exact match set only when every field the filter references
// actually contributed; a partially-evaluable predicate (an unsupported conjunct in an
// AND) yields a superset, which the index path cannot correct.
if (!result.get()
.contributingFieldIds()
.containsAll(collectFieldIds(table.rowType(), filter))) {
return null;
}
return result.get().results();
return result.get().result().results();
} catch (IOException e) {
throw new RuntimeException(e);
}
Expand Down
1 change: 1 addition & 0 deletions paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionBatchVectorRead.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,6 +71,7 @@ private List<GlobalIndexResult> readBatch(List<VectorSearchSplit> splits) {
List<IndexVectorSearchSplit> indexSplits = new ArrayList<>();
List<RawVectorSearchSplit> rawSplits = new ArrayList<>();
splitSearchSplits(splits, indexSplits, rawSplits);
demoteUncoverableIndexSplits(indexSplits, rawSplits);
if (indexSplits.isEmpty() && rawSplits.isEmpty()) {
List<GlobalIndexResult> empty = new ArrayList<>(n);
for (int i = 0; i < n; i++) {
Expand Down
1 change: 1 addition & 0 deletions paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionVectorRead.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,7 @@ protected GlobalIndexResult readSplits(List<? extends VectorSearchSplit> splits)
List<IndexVectorSearchSplit> indexSplits = new ArrayList<>();
List<RawVectorSearchSplit> rawSplits = new ArrayList<>();
splitSearchSplits(splits, indexSplits, rawSplits);
demoteUncoverableIndexSplits(indexSplits, rawSplits);
if (indexSplits.isEmpty() && rawSplits.isEmpty()) {
return GlobalIndexResult.createEmpty();
}
Expand Down
146 changes: 146 additions & 0 deletions paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -587,6 +589,150 @@ 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);
}

@Test
public void testVectorSearchFilterPartiallyEvaluableAnd() throws Exception {
catalog.createTable(
identifier("vector_search_partial_and_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_partial_and_scalar_predicate"));

// row 0 has NULL tags and is the closest to the query. The BTree index can evaluate
// `id >= 0` (all rows) but the multivalue index cannot evaluate `tags IS NOT NULL`, so
// the scalar index answers the AND with the `id`-only superset {0,1,2,3}. Trusting it
// would keep the index splits and pull the null-tags row 0 into the top-K; only demoting
// to the raw search, where the exact final-read filter drops row 0, 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);
buildAndCommitBTreeIndex(table, new int[] {0, 1, 2, 3}, new Range(0, 3));
buildAndCommitMultiValueIndex(table, vectors.length);

Predicate idFilter = new PredicateBuilder(table.rowType()).greaterOrEqual(0, 0);
Predicate tagsFilter =
new PredicateBuilder(table.rowType())
.isNotNull(table.rowType().getFieldIndex("tags"));
Predicate filter = PredicateBuilder.and(idFilter, tagsFilter);

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(
Expand Down
1 change: 1 addition & 0 deletions ...mmon/src/main/java/org/apache/paimon/flink/vectorsearch/FlinkDataEvolutionVectorRead.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,6 +92,7 @@ public GlobalIndexResult read(VectorScan.Plan plan) {
List<IndexVectorSearchSplit> indexSplits = new ArrayList<>();
List<RawVectorSearchSplit> rawSplits = new ArrayList<>();
splitSearchSplits(plan.splits(), indexSplits, rawSplits);
demoteUncoverableIndexSplits(indexSplits, rawSplits);
if (indexSplits.isEmpty() && rawSplits.isEmpty()) {
return GlobalIndexResult.createEmpty();
}
Expand Down
1 change: 1 addition & 0 deletions ...spark-common/src/main/java/org/apache/paimon/spark/read/SparkDataEvolutionVectorRead.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,6 +75,7 @@ public GlobalIndexResult read(VectorScan.Plan plan) {
List<IndexVectorSearchSplit> indexSplits = new ArrayList<>();
List<RawVectorSearchSplit> rawSplits = new ArrayList<>();
splitSearchSplits(plan.splits(), indexSplits, rawSplits);
demoteUncoverableIndexSplits(indexSplits, rawSplits);
if (indexSplits.isEmpty() && rawSplits.isEmpty()) {
return GlobalIndexResult.createEmpty();
}
Expand Down
Loading