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 89dab0ef6c [vector] Support raw fallback for vector search (#8302)
89dab0ef6c is described below
commit 89dab0ef6c8ef7db22f4793920123f4f8cc6bd9c
Author: Jingsong Lee <[email protected]>
AuthorDate: Sun Jun 21 15:10:23 2026 +0800
[vector] Support raw fallback for vector search (#8302)
Support vector search over unindexed raw rows when
`global-index.search-mode` is configured as `full` or `detail`, while
keeping the default `fast` mode index-only. This improves vector search
freshness after new data is written before the vector global index is
rebuilt.
---
.../paimon/globalindex/VectorGlobalIndexer.java | 17 +-
.../testvector/TestVectorGlobalIndexer.java | 21 +-
.../paimon/globalindex/DataEvolutionBatchScan.java | 2 +-
.../paimon/globalindex/GlobalIndexCoverage.java | 146 ++++++++
.../paimon/globalindex/GlobalIndexScanner.java | 118 ++----
.../org/apache/paimon/index/IndexFileHandler.java | 6 +-
.../paimon/table/source/AbstractVectorRead.java | 411 +++++++++++++++++++-
.../paimon/table/source/BatchVectorRead.java | 7 +-
.../paimon/table/source/BatchVectorReadImpl.java | 56 ++-
.../table/source/BatchVectorSearchBuilderImpl.java | 5 +-
...earchSplit.java => IndexVectorSearchSplit.java} | 13 +-
...rSearchSplit.java => RawVectorSearchSplit.java} | 83 ++---
.../org/apache/paimon/table/source/VectorRead.java | 8 +-
.../apache/paimon/table/source/VectorReadImpl.java | 42 ++-
.../apache/paimon/table/source/VectorScanImpl.java | 153 +++++++-
.../table/source/VectorSearchBuilderImpl.java | 5 +-
.../paimon/table/source/VectorSearchSplit.java | 107 +-----
.../table/source/snapshot/TimeTravelUtil.java | 1 +
.../paimon/table/BtreeGlobalIndexTableTest.java | 25 ++
.../table/source/VectorSearchBuilderTest.java | 412 ++++++++++++++++++++-
.../lumina/index/LuminaVectorGlobalIndexer.java | 9 +-
.../paimon/spark/read/SparkVectorReadImpl.java | 272 +++++++++++---
.../spark/read/SparkVectorSearchBuilderImpl.java | 3 +-
.../paimon/spark/read/SparkVectorReadImplTest.java | 172 +++++++++
.../vector/index/NativeVectorGlobalIndexer.java | 11 +-
25 files changed, 1692 insertions(+), 413 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/VectorRead.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/VectorGlobalIndexer.java
similarity index 69%
copy from
paimon-core/src/main/java/org/apache/paimon/table/source/VectorRead.java
copy to
paimon-common/src/main/java/org/apache/paimon/globalindex/VectorGlobalIndexer.java
index 74e17e2845..63166ef7c1 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/VectorRead.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/VectorGlobalIndexer.java
@@ -16,18 +16,11 @@
* limitations under the License.
*/
-package org.apache.paimon.table.source;
+package org.apache.paimon.globalindex;
-import org.apache.paimon.globalindex.GlobalIndexResult;
+/** A {@link GlobalIndexer} that supports vector similarity search. */
+public interface VectorGlobalIndexer extends GlobalIndexer {
-import java.util.List;
-
-/** Vector read to read index files. */
-public interface VectorRead {
-
- default GlobalIndexResult read(VectorScan.Plan plan) {
- return read(plan.splits());
- }
-
- GlobalIndexResult read(List<VectorSearchSplit> splits);
+ /** Returns the metric name used to convert vector distances to comparable
scores. */
+ String metric();
}
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/testvector/TestVectorGlobalIndexer.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/testvector/TestVectorGlobalIndexer.java
index cd3a140681..f652719834 100644
---
a/paimon-common/src/test/java/org/apache/paimon/globalindex/testvector/TestVectorGlobalIndexer.java
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/testvector/TestVectorGlobalIndexer.java
@@ -21,7 +21,7 @@ package org.apache.paimon.globalindex.testvector;
import org.apache.paimon.globalindex.GlobalIndexIOMeta;
import org.apache.paimon.globalindex.GlobalIndexReader;
import org.apache.paimon.globalindex.GlobalIndexWriter;
-import org.apache.paimon.globalindex.GlobalIndexer;
+import org.apache.paimon.globalindex.VectorGlobalIndexer;
import org.apache.paimon.globalindex.io.GlobalIndexFileReader;
import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
import org.apache.paimon.options.Options;
@@ -32,12 +32,13 @@ import org.apache.paimon.types.FloatType;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.paimon.utils.Preconditions.checkArgument;
/**
- * A test-only {@link GlobalIndexer} for vector similarity search. Uses
brute-force linear scan for
- * ANN queries. No native library dependency required.
+ * A test-only {@link VectorGlobalIndexer} for vector similarity search. Uses
brute-force linear
+ * scan for ANN queries. No native library dependency required.
*
* <p>Supported distance metrics (configured via option {@code
test.vector.metric}):
*
@@ -47,7 +48,7 @@ import static
org.apache.paimon.utils.Preconditions.checkArgument;
* <li>{@code inner_product} - Inner product similarity (directly used as
score)
* </ul>
*/
-public class TestVectorGlobalIndexer implements GlobalIndexer {
+public class TestVectorGlobalIndexer implements VectorGlobalIndexer {
/** Option key for vector dimension. */
public static final String OPT_DIMENSION = "test.vector.dimension";
@@ -59,6 +60,8 @@ public class TestVectorGlobalIndexer implements GlobalIndexer
{
public static final String OPT_REQUIRED_OPTION_VALUE =
"test.vector.required-option.value";
+ private static final AtomicInteger METRIC_CALLS = new AtomicInteger();
+
private final DataType fieldType;
private final int dimension;
private final String metric;
@@ -96,7 +99,17 @@ public class TestVectorGlobalIndexer implements
GlobalIndexer {
return dimension;
}
+ @Override
public String metric() {
+ METRIC_CALLS.incrementAndGet();
return metric;
}
+
+ public static void resetMetricCalls() {
+ METRIC_CALLS.set(0);
+ }
+
+ public static int metricCalls() {
+ return METRIC_CALLS.get();
+ }
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java
b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java
index b03f27f3d3..76ae6195d5 100644
---
a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java
+++
b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java
@@ -282,7 +282,7 @@ public class DataEvolutionBatchScan implements
DataTableScan {
Optional<GlobalIndexResult> result = scanner.scan(filter);
if (result.isPresent()) {
LOG.info("Scan table '{}' with global index.", table.name());
- return result;
+ return
Optional.of(result.get().or(scanner.unindexedRows(filter)));
}
return Optional.empty();
} catch (IOException e) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexCoverage.java
b/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexCoverage.java
new file mode 100644
index 0000000000..6ef52eee2d
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexCoverage.java
@@ -0,0 +1,146 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.globalindex;
+
+import org.apache.paimon.CoreOptions.GlobalIndexSearchMode;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.ScanMode;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.snapshot.SnapshotReader;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Range;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.predicate.PredicateVisitor.collectFieldIds;
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+
+/** Row ranges covered and not covered by global index files. */
+public class GlobalIndexCoverage {
+
+ private final FileStoreTable table;
+ @Nullable private final Snapshot snapshot;
+ @Nullable private final PartitionPredicate partitionFilter;
+ private final Map<Integer, List<Range>> coverageByField;
+
+ public GlobalIndexCoverage(
+ FileStoreTable table,
+ @Nullable Snapshot snapshot,
+ @Nullable PartitionPredicate partitionFilter,
+ Collection<IndexFileMeta> indexFiles) {
+ this.table = table;
+ this.snapshot = snapshot;
+ this.partitionFilter = partitionFilter;
+ this.coverageByField = new HashMap<>();
+ for (IndexFileMeta indexFile : indexFiles) {
+ GlobalIndexMeta meta = checkNotNull(indexFile.globalIndexMeta());
+ Range range = new Range(meta.rowRangeStart(), meta.rowRangeEnd());
+ addCoverage(meta.indexFieldId(), range);
+ if (meta.extraFieldIds() != null) {
+ for (int extra : meta.extraFieldIds()) {
+ addCoverage(extra, range);
+ }
+ }
+ }
+ }
+
+ public List<Range> unindexedRanges(RowType rowType, @Nullable Predicate
predicate) {
+ return unindexedRanges(collectFieldIds(rowType, predicate));
+ }
+
+ public List<Range> unindexedRanges(int fieldId) {
+ return unindexedRanges(Collections.singleton(fieldId));
+ }
+
+ private void addCoverage(int fieldId, Range range) {
+ coverageByField.computeIfAbsent(fieldId, k -> new
ArrayList<>()).add(range);
+ }
+
+ private List<Range> indexedRanges(Collection<Integer> fieldIds) {
+ List<Range> ranges = null;
+ for (Integer fieldId : fieldIds) {
+ List<Range> fieldRanges = coverageByField.get(fieldId);
+ if (fieldRanges == null || fieldRanges.isEmpty()) {
+ return Collections.emptyList();
+ }
+ fieldRanges = Range.sortAndMergeOverlap(fieldRanges, true);
+ ranges = ranges == null ? fieldRanges : Range.and(ranges,
fieldRanges);
+ }
+ return ranges == null ? Collections.emptyList() :
Range.sortAndMergeOverlap(ranges, true);
+ }
+
+ private List<Range> unindexedRanges(Collection<Integer> fieldIds) {
+ GlobalIndexSearchMode searchMode =
table.coreOptions().globalIndexSearchMode();
+ if (searchMode == GlobalIndexSearchMode.FAST) {
+ return Collections.emptyList();
+ }
+ if (snapshot == null || snapshot.nextRowId() == null ||
snapshot.nextRowId() <= 0) {
+ return Collections.emptyList();
+ }
+
+ List<Range> dataRanges;
+ if (searchMode == GlobalIndexSearchMode.DETAIL) {
+ dataRanges = dataRangesByDataFiles();
+ } else {
+ dataRanges = Collections.singletonList(new Range(0,
snapshot.nextRowId() - 1));
+ }
+
+ List<Range> predicateIndexedRanges =
+ Range.sortAndMergeOverlap(indexedRanges(fieldIds), true);
+ List<Range> unindexedRanges = new ArrayList<>();
+ for (Range dataRange : Range.sortAndMergeOverlap(dataRanges, true)) {
+ unindexedRanges.addAll(dataRange.exclude(predicateIndexedRanges));
+ }
+ return Range.sortAndMergeOverlap(unindexedRanges, true);
+ }
+
+ private List<Range> dataRangesByDataFiles() {
+ SnapshotReader snapshotReader =
+ table.newSnapshotReader()
+ .withPartitionFilter(partitionFilter)
+ .withMode(ScanMode.ALL)
+ .withSnapshot(snapshot);
+ List<Range> dataRanges = new ArrayList<>();
+ for (Split split : snapshotReader.read().splits()) {
+ if (!(split instanceof DataSplit)) {
+ continue;
+ }
+ for (DataFileMeta file : ((DataSplit) split).dataFiles()) {
+ if (file.firstRowId() != null) {
+ dataRanges.add(file.nonNullRowIdRange());
+ }
+ }
+ }
+ return dataRanges;
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexScanner.java
b/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexScanner.java
index 93504d317c..7edb68365b 100644
---
a/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexScanner.java
+++
b/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexScanner.java
@@ -18,7 +18,6 @@
package org.apache.paimon.globalindex;
-import org.apache.paimon.CoreOptions.GlobalIndexSearchMode;
import org.apache.paimon.Snapshot;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
@@ -26,22 +25,19 @@ import
org.apache.paimon.globalindex.io.GlobalIndexFileReader;
import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.index.IndexPathFactory;
-import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.IndexManifestEntry;
import org.apache.paimon.options.Options;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.table.FileStoreTable;
-import org.apache.paimon.table.source.DataSplit;
-import org.apache.paimon.table.source.ScanMode;
-import org.apache.paimon.table.source.Split;
-import org.apache.paimon.table.source.snapshot.SnapshotReader;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.Filter;
import org.apache.paimon.utils.Range;
import org.apache.paimon.utils.RoaringNavigableMap64;
+import javax.annotation.Nullable;
+
import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
@@ -72,43 +68,33 @@ public class GlobalIndexScanner implements Closeable {
private final ExecutorService executor;
private final GlobalIndexEvaluator globalIndexEvaluator;
private final IndexPathFactory indexPathFactory;
- private final Map<Integer, List<Range>> coverageByField;
+ private final GlobalIndexCoverage coverage;
private final FileStoreTable table;
- private final Snapshot snapshot;
- private final PartitionPredicate partitionFilter;
private GlobalIndexScanner(
FileStoreTable table,
- Snapshot snapshot,
- PartitionPredicate partitionFilter,
+ @Nullable Snapshot snapshot,
+ @Nullable PartitionPredicate partitionFilter,
Options options,
RowType rowType,
FileIO fileIO,
IndexPathFactory indexPathFactory,
Collection<IndexFileMeta> indexFiles) {
this.table = table;
- this.snapshot = snapshot;
- this.partitionFilter = partitionFilter;
this.options = options;
this.rowType = rowType;
this.executor =
GlobalIndexReadThreadPool.getExecutorService(options.get(GLOBAL_INDEX_THREAD_NUM));
this.indexPathFactory = indexPathFactory;
+ this.coverage = new GlobalIndexCoverage(table, snapshot,
partitionFilter, indexFiles);
GlobalIndexFileReader indexFileReader = meta ->
fileIO.newInputStream(meta.filePath());
Map<Integer, IndexMetaFileGroup> indexMetas = new HashMap<>();
Map<Integer, List<IndexMetaFileGroup>> extraIndexMetas = new
HashMap<>();
- this.coverageByField = new HashMap<>();
for (IndexFileMeta indexFile : indexFiles) {
GlobalIndexMeta meta = checkNotNull(indexFile.globalIndexMeta());
String indexType = indexFile.indexType();
Range range = new Range(meta.rowRangeStart(), meta.rowRangeEnd());
int indexFieldId = meta.indexFieldId();
- coverageByField.computeIfAbsent(indexFieldId, k -> new
ArrayList<>()).add(range);
- if (meta.extraFieldIds() != null) {
- for (int extra : meta.extraFieldIds()) {
- coverageByField.computeIfAbsent(extra, k -> new
ArrayList<>()).add(range);
- }
- }
List<Integer> fieldIds = meta.getIndexedFieldIds();
IndexMetaFileGroup group = indexMetas.get(indexFieldId);
if (group == null) {
@@ -194,14 +180,21 @@ public class GlobalIndexScanner implements Closeable {
public static Optional<GlobalIndexScanner> create(
FileStoreTable table, Collection<IndexFileMeta> indexFiles) {
+ return create(table, null, indexFiles);
+ }
+
+ public static Optional<GlobalIndexScanner> create(
+ FileStoreTable table,
+ @Nullable PartitionPredicate partitionFilter,
+ Collection<IndexFileMeta> indexFiles) {
if (indexFiles.isEmpty()) {
return Optional.empty();
}
return Optional.of(
new GlobalIndexScanner(
- null,
- null,
- null,
+ table,
+ tryTravelOrLatest(table),
+ partitionFilter,
table.coreOptions().toConfiguration(),
table.rowType(),
table.fileIO(),
@@ -210,8 +203,10 @@ public class GlobalIndexScanner implements Closeable {
}
public static Optional<GlobalIndexScanner> create(
- FileStoreTable table, PartitionPredicate partitionFilter,
Predicate filter) {
- Snapshot snapshot = tryTravelOrLatest(table);
+ FileStoreTable table,
+ @Nullable PartitionPredicate partitionFilter,
+ @Nullable Predicate filter) {
+ @Nullable Snapshot snapshot = tryTravelOrLatest(table);
List<IndexFileMeta> indexFiles =
table.store().newIndexFileHandler()
.scan(snapshot, indexFileFilter(table,
partitionFilter, filter)).stream()
@@ -233,7 +228,9 @@ public class GlobalIndexScanner implements Closeable {
}
private static Filter<IndexManifestEntry> indexFileFilter(
- FileStoreTable table, PartitionPredicate partitionFilter,
Predicate filter) {
+ FileStoreTable table,
+ @Nullable PartitionPredicate partitionFilter,
+ @Nullable Predicate filter) {
if (filter == null) {
return entry -> false;
}
@@ -265,80 +262,17 @@ public class GlobalIndexScanner implements Closeable {
}
public Optional<GlobalIndexResult> scan(Predicate predicate) {
- Optional<GlobalIndexResult> result =
globalIndexEvaluator.evaluate(predicate);
- return result.map(indexedResultRows -> withUnindexedRows(predicate,
indexedResultRows));
+ return globalIndexEvaluator.evaluate(predicate);
}
- private GlobalIndexResult withUnindexedRows(
- Predicate predicate, GlobalIndexResult indexedResultRows) {
- if (indexedResultRows instanceof ScoredGlobalIndexResult
- || table == null
- || table.coreOptions().globalIndexSearchMode() ==
GlobalIndexSearchMode.FAST) {
- return indexedResultRows;
- }
-
+ public GlobalIndexResult unindexedRows(Predicate predicate) {
RoaringNavigableMap64 rows = new RoaringNavigableMap64();
- rows.or(indexedResultRows.results());
- for (Range range : unindexedRanges(predicate)) {
+ for (Range range : coverage.unindexedRanges(rowType, predicate)) {
rows.addRange(range);
}
return GlobalIndexResult.create(rows);
}
- private List<Range> indexedRanges(Predicate predicate) {
- List<Range> ranges = null;
- for (Integer fieldId : collectFieldIds(rowType, predicate)) {
- List<Range> fieldRanges = coverageByField.get(fieldId);
- if (fieldRanges == null || fieldRanges.isEmpty()) {
- return Collections.emptyList();
- }
- fieldRanges = Range.sortAndMergeOverlap(fieldRanges, true);
- ranges = ranges == null ? fieldRanges : Range.and(ranges,
fieldRanges);
- }
- return ranges == null ? Collections.emptyList() :
Range.sortAndMergeOverlap(ranges, true);
- }
-
- private List<Range> unindexedRanges(Predicate predicate) {
- if (snapshot == null || snapshot.nextRowId() == null ||
snapshot.nextRowId() <= 0) {
- return Collections.emptyList();
- }
-
- List<Range> dataRanges;
- if (table.coreOptions().globalIndexSearchMode() ==
GlobalIndexSearchMode.DETAIL) {
- dataRanges = dataRangesByDataFiles();
- } else {
- dataRanges = Collections.singletonList(new Range(0,
snapshot.nextRowId() - 1));
- }
-
- List<Range> predicateIndexedRanges =
- Range.sortAndMergeOverlap(indexedRanges(predicate), true);
- List<Range> unindexedRanges = new ArrayList<>();
- for (Range dataRange : Range.sortAndMergeOverlap(dataRanges, true)) {
- unindexedRanges.addAll(dataRange.exclude(predicateIndexedRanges));
- }
- return Range.sortAndMergeOverlap(unindexedRanges, true);
- }
-
- private List<Range> dataRangesByDataFiles() {
- SnapshotReader snapshotReader =
- table.newSnapshotReader()
- .withPartitionFilter(partitionFilter)
- .withMode(ScanMode.ALL)
- .withSnapshot(snapshot);
- List<Range> dataRanges = new ArrayList<>();
- for (Split split : snapshotReader.read().splits()) {
- if (!(split instanceof DataSplit)) {
- continue;
- }
- for (DataFileMeta file : ((DataSplit) split).dataFiles()) {
- if (file.firstRowId() != null) {
- dataRanges.add(file.nonNullRowIdRange());
- }
- }
- }
- return dataRanges;
- }
-
private Collection<GlobalIndexReader> createReaders(
GlobalIndexFileReader indexFileReadWrite,
IndexMetaFileGroup group,
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/IndexFileHandler.java
b/paimon-core/src/main/java/org/apache/paimon/index/IndexFileHandler.java
index cb9525cc5c..e848ed205e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/index/IndexFileHandler.java
+++ b/paimon-core/src/main/java/org/apache/paimon/index/IndexFileHandler.java
@@ -34,6 +34,8 @@ import org.apache.paimon.utils.IndexFilePathFactories;
import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.SnapshotManager;
+import javax.annotation.Nullable;
+
import java.io.IOException;
import java.io.UncheckedIOException;
import java.util.ArrayList;
@@ -96,7 +98,7 @@ public class IndexFileHandler {
return scan(snapshotManager.latestSnapshot(), indexType);
}
- public List<IndexManifestEntry> scan(Snapshot snapshot, String indexType) {
+ public List<IndexManifestEntry> scan(@Nullable Snapshot snapshot, String
indexType) {
if (snapshot == null) {
return Collections.emptyList();
}
@@ -115,7 +117,7 @@ public class IndexFileHandler {
}
public List<IndexManifestEntry> scan(
- Snapshot snapshot, Filter<IndexManifestEntry> readTFilter) {
+ @Nullable Snapshot snapshot, Filter<IndexManifestEntry>
readTFilter) {
if (snapshot == null) {
return Collections.emptyList();
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractVectorRead.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractVectorRead.java
index 12bec4fc2f..bb783415d9 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractVectorRead.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractVectorRead.java
@@ -18,6 +18,9 @@
package org.apache.paimon.table.source;
+import org.apache.paimon.data.InternalArray;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.InternalVector;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.globalindex.GlobalIndexIOMeta;
import org.apache.paimon.globalindex.GlobalIndexReader;
@@ -27,16 +30,23 @@ import org.apache.paimon.globalindex.GlobalIndexer;
import org.apache.paimon.globalindex.GlobalIndexerFactoryUtils;
import org.apache.paimon.globalindex.OffsetGlobalIndexReader;
import org.apache.paimon.globalindex.ScoredGlobalIndexResult;
+import org.apache.paimon.globalindex.VectorGlobalIndexer;
import org.apache.paimon.globalindex.io.GlobalIndexFileReader;
import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.index.IndexPathFactory;
+import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.predicate.BatchVectorSearch;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.VectorSearch;
+import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.SpecialFields;
import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypeRoot;
+import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.IOUtils;
+import org.apache.paimon.utils.Range;
import org.apache.paimon.utils.RoaringNavigableMap64;
import javax.annotation.Nullable;
@@ -63,18 +73,21 @@ public abstract class AbstractVectorRead implements
Serializable {
private static final long serialVersionUID = 1L;
protected final FileStoreTable table;
- private final Predicate filter;
+ @Nullable private final PartitionPredicate partitionFilter;
+ @Nullable private final Predicate filter;
protected final int limit;
protected final DataField vectorColumn;
protected final Map<String, String> options;
protected AbstractVectorRead(
FileStoreTable table,
- Predicate filter,
+ @Nullable PartitionPredicate partitionFilter,
+ @Nullable Predicate filter,
int limit,
DataField vectorColumn,
- Map<String, String> options) {
+ @Nullable Map<String, String> options) {
this.table = table;
+ this.partitionFilter = partitionFilter;
this.filter = filter;
this.limit = limit;
this.vectorColumn = vectorColumn;
@@ -84,38 +97,113 @@ public abstract class AbstractVectorRead implements
Serializable {
: Collections.unmodifiableMap(new HashMap<>(options));
}
- protected GlobalIndexer createGlobalIndexer(List<VectorSearchSplit>
splits) {
- IndexFileMeta firstFile = splits.get(0).vectorIndexFiles().get(0);
- String indexType = firstFile.indexType();
+ protected GlobalIndexer createGlobalIndexer(List<IndexVectorSearchSplit>
splits) {
+ IndexFileMeta firstFile = firstVectorIndexFile(splits);
GlobalIndexMeta firstMeta = checkNotNull(firstFile.globalIndexMeta());
- if (firstMeta.extraFieldIds() != null) {
- return GlobalIndexerFactoryUtils.load(indexType)
- .create(
- firstMeta.getIndexField(table.rowType()),
- firstMeta.getExtraFields(table.rowType()),
- table.coreOptions().toConfiguration());
- }
+ return createGlobalIndexer(firstFile.indexType(), firstMeta);
+ }
+
+ protected GlobalIndexer createGlobalIndexer(String indexType) {
return GlobalIndexerFactoryUtils.load(indexType)
.create(vectorColumn, table.coreOptions().toConfiguration());
}
- protected Optional<RoaringNavigableMap64>
preFilter(List<VectorSearchSplit> splits) {
+ private GlobalIndexer createGlobalIndexer(String indexType,
GlobalIndexMeta meta) {
+ if (meta.extraFieldIds() == null) {
+ return createGlobalIndexer(indexType);
+ }
+ return GlobalIndexerFactoryUtils.load(indexType)
+ .create(
+ meta.getIndexField(table.rowType()),
+ meta.getExtraFields(table.rowType()),
+ table.coreOptions().toConfiguration());
+ }
+
+ protected List<RoaringNavigableMap64>
preFilters(List<IndexVectorSearchSplit> splits) {
+ if (filter == null) {
+ return Collections.emptyList();
+ }
+
Set<IndexFileMeta> scalarIndexFiles =
new TreeSet<>(Comparator.comparing(IndexFileMeta::fileName));
- for (VectorSearchSplit split : splits) {
+ for (IndexVectorSearchSplit split : splits) {
scalarIndexFiles.addAll(split.scalarIndexFiles());
}
Optional<GlobalIndexScanner> optionalScanner =
- GlobalIndexScanner.create(table, scalarIndexFiles);
+ GlobalIndexScanner.create(table, partitionFilter,
scalarIndexFiles);
+ if (!optionalScanner.isPresent()) {
+ return emptyPreFilters(splits.size());
+ }
+
+ RoaringNavigableMap64 matchedRows;
+ try (GlobalIndexScanner scanner = optionalScanner.get()) {
+ Optional<GlobalIndexResult> result = scanner.scan(filter);
+ if (!result.isPresent()) {
+ return emptyPreFilters(splits.size());
+ }
+ matchedRows = result.get().results();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+
+ List<RoaringNavigableMap64> includeRowIds = new
ArrayList<>(splits.size());
+ for (IndexVectorSearchSplit split : splits) {
+ Range splitRange = new Range(split.rowRangeStart(),
split.rowRangeEnd());
+ RoaringNavigableMap64 splitRows = bitmapOf(splitRange);
+
+ RoaringNavigableMap64 include = new RoaringNavigableMap64();
+ include.or(matchedRows);
+ include.and(splitRows);
+ includeRowIds.add(include);
+ }
+ return includeRowIds;
+ }
+
+ private List<RoaringNavigableMap64> emptyPreFilters(int size) {
+ List<RoaringNavigableMap64> preFilters = new ArrayList<>(size);
+ for (int i = 0; i < size; i++) {
+ preFilters.add(new RoaringNavigableMap64());
+ }
+ return preFilters;
+ }
+
+ @Nullable
+ protected RoaringNavigableMap64 rawPreFilter(List<RawVectorSearchSplit>
splits) {
+ if (filter == null) {
+ return null;
+ }
+
+ RoaringNavigableMap64 rawRows = bitmapOf(rawRowRanges(splits));
+ if (rawRows.isEmpty()) {
+ return null;
+ }
+
+ Set<IndexFileMeta> scalarIndexFiles =
+ new TreeSet<>(Comparator.comparing(IndexFileMeta::fileName));
+ for (RawVectorSearchSplit split : splits) {
+ scalarIndexFiles.addAll(split.scalarIndexFiles());
+ }
+ Optional<GlobalIndexScanner> optionalScanner =
+ GlobalIndexScanner.create(table, partitionFilter,
scalarIndexFiles);
if (!optionalScanner.isPresent()) {
- return Optional.empty();
+ return null;
}
+
+ RoaringNavigableMap64 include = new RoaringNavigableMap64();
try (GlobalIndexScanner scanner = optionalScanner.get()) {
- return scanner.scan(filter).map(GlobalIndexResult::results);
+ Optional<GlobalIndexResult> result = scanner.scan(filter);
+ if (!result.isPresent()) {
+ return null;
+ }
+ include.or(result.get().results());
+ include.or(scanner.unindexedRows(filter).results());
} catch (IOException e) {
throw new RuntimeException(e);
}
+
+ include.and(rawRows);
+ return include;
}
protected CompletableFuture<Optional<ScoredGlobalIndexResult>> eval(
@@ -127,6 +215,10 @@ public abstract class AbstractVectorRead implements
Serializable {
float[] vector,
@Nullable RoaringNavigableMap64 includeRowIds,
ExecutorService executor) {
+ if (vectorIndexFiles.isEmpty()) {
+ return CompletableFuture.completedFuture(Optional.empty());
+ }
+
List<GlobalIndexIOMeta> indexIOMetaList =
buildIOMetaList(indexPathFactory, vectorIndexFiles);
@SuppressWarnings("resource")
@@ -151,6 +243,10 @@ public abstract class AbstractVectorRead implements
Serializable {
float[][] vectors,
@Nullable RoaringNavigableMap64 includeRowIds,
ExecutorService executor) {
+ if (vectorIndexFiles.isEmpty()) {
+ return
CompletableFuture.completedFuture(emptyOptionalResults(vectors.length));
+ }
+
List<GlobalIndexIOMeta> indexIOMetaList =
buildIOMetaList(indexPathFactory, vectorIndexFiles);
@SuppressWarnings("resource")
@@ -166,6 +262,34 @@ public abstract class AbstractVectorRead implements
Serializable {
.whenComplete((r, t) -> IOUtils.closeQuietly(reader));
}
+ protected ScoredGlobalIndexResult withRawSearch(
+ ScoredGlobalIndexResult result,
+ List<RawVectorSearchSplit> rawSplits,
+ @Nullable GlobalIndexer globalIndexer,
+ @Nullable RoaringNavigableMap64 preFilter,
+ float[] queryVector) {
+ List<Range> rawRowRanges = rawRowRanges(rawSplits);
+ if (rawRowRanges.isEmpty()) {
+ return result.topK(limit);
+ }
+
+ ScoredGlobalIndexResult rawResult =
+ readRawSearch(
+ rawRowRanges,
+ preFilter,
+ rawSearchIndexer(rawSplits, globalIndexer),
+ queryVector);
+ return result.or(rawResult).topK(limit);
+ }
+
+ protected ScoredGlobalIndexResult[] emptyScoredResults(int n) {
+ ScoredGlobalIndexResult[] results = new ScoredGlobalIndexResult[n];
+ for (int i = 0; i < n; i++) {
+ results[i] = ScoredGlobalIndexResult.createEmpty();
+ }
+ return results;
+ }
+
private List<GlobalIndexIOMeta> buildIOMetaList(
IndexPathFactory indexPathFactory, List<IndexFileMeta>
vectorIndexFiles) {
List<GlobalIndexIOMeta> indexIOMetaList = new ArrayList<>();
@@ -179,4 +303,255 @@ public abstract class AbstractVectorRead implements
Serializable {
}
return indexIOMetaList;
}
+
+ private static RoaringNavigableMap64 bitmapOf(Range range) {
+ RoaringNavigableMap64 bitmap = new RoaringNavigableMap64();
+ bitmap.addRange(range);
+ return bitmap;
+ }
+
+ private static RoaringNavigableMap64 bitmapOf(List<Range> ranges) {
+ RoaringNavigableMap64 bitmap = new RoaringNavigableMap64();
+ for (Range range : ranges) {
+ bitmap.addRange(range);
+ }
+ return bitmap;
+ }
+
+ protected ScoredGlobalIndexResult readRawSearch(
+ List<Range> rawRowRanges,
+ @Nullable RoaringNavigableMap64 preFilter,
+ @Nullable GlobalIndexer globalIndexer,
+ float[] queryVector) {
+ return readRawSearch(rawRowRanges, preFilter,
rawSearchMetric(globalIndexer), queryVector);
+ }
+
+ protected ScoredGlobalIndexResult readRawSearch(
+ List<Range> rawRowRanges,
+ @Nullable RoaringNavigableMap64 preFilter,
+ String metric,
+ float[] queryVector) {
+ RowType readType = SpecialFields.rowTypeWithRowId(table.rowType());
+ if (preFilter != null) {
+ rawRowRanges =
+ Range.and(
+ Range.sortAndMergeOverlap(rawRowRanges, true),
+ Range.sortAndMergeOverlap(preFilter.toRangeList(),
true));
+ }
+ if (rawRowRanges.isEmpty()) {
+ return ScoredGlobalIndexResult.createEmpty();
+ }
+
+ TableScan.Plan plan =
+ newRawReadBuilder(readType,
false).withRowRanges(rawRowRanges).newScan().plan();
+ ReadBuilder readBuilder = newRawReadBuilder(readType, true);
+ RoaringNavigableMap64 resultBitmap = new RoaringNavigableMap64();
+ Map<Long, Float> scoreMap = new HashMap<>();
+ int vectorIndex = readType.getFieldIndex(vectorColumn.name());
+ int rowIdIndex = readType.getFieldIndex(SpecialFields.ROW_ID.name());
+
+ try (RecordReader<InternalRow> reader =
+ readBuilder.newRead().executeFilter().createReader(plan)) {
+ reader.forEachRemaining(
+ row -> {
+ if (row.isNullAt(vectorIndex)) {
+ return;
+ }
+ float[] stored = getVector(row, vectorIndex);
+ if (stored.length != queryVector.length) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Query vector dimension mismatch:
expected %d, got %d",
+ stored.length,
queryVector.length));
+ }
+ long rowId = row.getLong(rowIdIndex);
+ resultBitmap.add(rowId);
+ scoreMap.put(rowId, computeScore(queryVector, stored,
metric));
+ });
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to read raw vectors for vector
search.", e);
+ }
+
+ return ScoredGlobalIndexResult.create(resultBitmap,
scoreMap::get).topK(limit);
+ }
+
+ private ReadBuilder newRawReadBuilder(RowType readType, boolean
includeFilter) {
+ ReadBuilder readBuilder =
table.newReadBuilder().withReadType(readType);
+ if (partitionFilter != null) {
+ readBuilder.withPartitionFilter(partitionFilter);
+ }
+ if (includeFilter && filter != null) {
+ readBuilder.withFilter(filter);
+ }
+ return readBuilder;
+ }
+
+ protected static void splitSearchSplits(
+ List<? extends VectorSearchSplit> splits,
+ List<IndexVectorSearchSplit> indexSplits,
+ List<RawVectorSearchSplit> rawSplits) {
+ for (VectorSearchSplit split : splits) {
+ if (split instanceof IndexVectorSearchSplit) {
+ indexSplits.add((IndexVectorSearchSplit) split);
+ } else if (split instanceof RawVectorSearchSplit) {
+ rawSplits.add((RawVectorSearchSplit) split);
+ }
+ }
+ }
+
+ protected static List<Range> rawRowRanges(List<RawVectorSearchSplit>
rawSplits) {
+ List<Range> rawRowRanges = new ArrayList<>();
+ for (RawVectorSearchSplit split : rawSplits) {
+ rawRowRanges.addAll(split.rowRanges());
+ }
+ return Range.sortAndMergeOverlap(rawRowRanges, true);
+ }
+
+ @Nullable
+ protected GlobalIndexer rawSearchIndexer(
+ List<RawVectorSearchSplit> rawSplits, @Nullable GlobalIndexer
globalIndexer) {
+ if (globalIndexer != null) {
+ return globalIndexer;
+ }
+ for (RawVectorSearchSplit split : rawSplits) {
+ String indexType = split.indexType();
+ if (indexType != null) {
+ return createGlobalIndexer(indexType);
+ }
+ }
+ return null;
+ }
+
+ private float[] getVector(InternalRow row, int vectorIndex) {
+ if (vectorColumn.type().getTypeRoot() == DataTypeRoot.VECTOR) {
+ InternalVector vector = row.getVector(vectorIndex);
+ return vector.toFloatArray();
+ } else if (vectorColumn.type().getTypeRoot() == DataTypeRoot.ARRAY) {
+ InternalArray array = row.getArray(vectorIndex);
+ return array.toFloatArray();
+ }
+ throw new IllegalArgumentException(
+ "Unsupported vector column type: " + vectorColumn.type());
+ }
+
+ protected String rawSearchMetric(@Nullable GlobalIndexer globalIndexer) {
+ String metric = null;
+ if (globalIndexer != null) {
+ if (!(globalIndexer instanceof VectorGlobalIndexer)) {
+ throw new IllegalArgumentException(
+ "Index type '"
+ + globalIndexer.getClass().getName()
+ + "' does not provide vector metric for raw
search.");
+ }
+ metric = ((VectorGlobalIndexer) globalIndexer).metric();
+ }
+ if (metric == null) {
+ metric = configuredRawSearchMetric();
+ }
+ return metric == null ? "l2" : normalizeMetric(metric);
+ }
+
+ @Nullable
+ private String configuredRawSearchMetric() {
+ String metric = configuredRawSearchMetric(options);
+ return metric == null ? configuredRawSearchMetric(table.options()) :
metric;
+ }
+
+ @Nullable
+ private String configuredRawSearchMetric(Map<String, String> options) {
+ String fieldPrefix = "fields." + vectorColumn.name() + ".";
+ String metric = option(options, fieldPrefix + "distance.metric");
+ if (metric == null) {
+ metric = option(options, fieldPrefix + "metric");
+ }
+ if (metric == null) {
+ metric = option(options, "test.vector.metric");
+ }
+ if (metric == null) {
+ metric = option(options, "lumina.distance.metric");
+ }
+ if (metric == null) {
+ metric = option(options, "distance.metric");
+ }
+ if (metric == null) {
+ metric = option(options, "metric");
+ }
+ if (metric != null) {
+ return metric;
+ }
+
+ for (Map.Entry<String, String> entry : options.entrySet()) {
+ String key = entry.getKey();
+ if (key.endsWith(".distance.metric") || key.endsWith(".metric")) {
+ String value = normalizeMetric(entry.getValue());
+ if (isRawSearchMetric(value)) {
+ if (metric != null && !metric.equals(value)) {
+ return null;
+ }
+ metric = value;
+ }
+ }
+ }
+ return metric;
+ }
+
+ @Nullable
+ private static String option(Map<String, String> options, String key) {
+ String value = options.get(key);
+ return value == null ? null : normalizeMetric(value);
+ }
+
+ private static boolean isRawSearchMetric(String metric) {
+ return "l2".equals(metric) || "cosine".equals(metric) ||
"inner_product".equals(metric);
+ }
+
+ private static String normalizeMetric(String metric) {
+ return metric.toLowerCase().replace('-', '_');
+ }
+
+ private static IndexFileMeta
firstVectorIndexFile(List<IndexVectorSearchSplit> splits) {
+ for (IndexVectorSearchSplit split : splits) {
+ if (!split.vectorIndexFiles().isEmpty()) {
+ return split.vectorIndexFiles().get(0);
+ }
+ }
+ throw new IllegalArgumentException("No vector index files found.");
+ }
+
+ private static float computeScore(float[] query, float[] stored, String
metric) {
+ if ("l2".equals(metric)) {
+ float sumSq = 0;
+ for (int i = 0; i < query.length; i++) {
+ float diff = query[i] - stored[i];
+ sumSq += diff * diff;
+ }
+ return 1.0f / (1.0f + sumSq);
+ } else if ("cosine".equals(metric)) {
+ float dot = 0;
+ float normA = 0;
+ float normB = 0;
+ for (int i = 0; i < query.length; i++) {
+ dot += query[i] * stored[i];
+ normA += query[i] * query[i];
+ normB += stored[i] * stored[i];
+ }
+ float denominator = (float) (Math.sqrt(normA) * Math.sqrt(normB));
+ return denominator == 0 ? 0 : dot / denominator;
+ } else if ("inner_product".equals(metric)) {
+ float dot = 0;
+ for (int i = 0; i < query.length; i++) {
+ dot += query[i] * stored[i];
+ }
+ return dot;
+ }
+ throw new IllegalArgumentException("Unknown vector search metric: " +
metric);
+ }
+
+ private static List<Optional<ScoredGlobalIndexResult>>
emptyOptionalResults(int n) {
+ List<Optional<ScoredGlobalIndexResult>> results = new ArrayList<>(n);
+ for (int i = 0; i < n; i++) {
+ results.add(Optional.empty());
+ }
+ return results;
+ }
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/BatchVectorRead.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/BatchVectorRead.java
index 7b2169779d..8a33fdc640 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/BatchVectorRead.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/BatchVectorRead.java
@@ -26,10 +26,5 @@ import java.util.List;
public interface BatchVectorRead {
/** Read batch results; result {@code i} corresponds to input vector
{@code i}. */
- default List<GlobalIndexResult> readBatch(VectorScan.Plan plan) {
- return readBatch(plan.splits());
- }
-
- /** Read batch results; result {@code i} corresponds to input vector
{@code i}. */
- List<GlobalIndexResult> readBatch(List<VectorSearchSplit> splits);
+ List<GlobalIndexResult> readBatch(VectorScan.Plan plan);
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/BatchVectorReadImpl.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/BatchVectorReadImpl.java
index 1da9ad554f..2a6fb07f7c 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/BatchVectorReadImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/BatchVectorReadImpl.java
@@ -23,11 +23,14 @@ import org.apache.paimon.globalindex.GlobalIndexResult;
import org.apache.paimon.globalindex.GlobalIndexer;
import org.apache.paimon.globalindex.ScoredGlobalIndexResult;
import org.apache.paimon.index.IndexPathFactory;
+import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.types.DataField;
import org.apache.paimon.utils.RoaringNavigableMap64;
+import javax.annotation.Nullable;
+
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -46,19 +49,27 @@ public class BatchVectorReadImpl extends AbstractVectorRead
implements BatchVect
public BatchVectorReadImpl(
FileStoreTable table,
- Predicate filter,
+ @Nullable PartitionPredicate partitionFilter,
+ @Nullable Predicate filter,
int limit,
DataField vectorColumn,
float[][] vectors,
- Map<String, String> options) {
- super(table, filter, limit, vectorColumn, options);
+ @Nullable Map<String, String> options) {
+ super(table, partitionFilter, filter, limit, vectorColumn, options);
this.vectors = vectors;
}
@Override
- public List<GlobalIndexResult> readBatch(List<VectorSearchSplit> splits) {
+ public List<GlobalIndexResult> readBatch(VectorScan.Plan plan) {
+ return readBatch(plan.splits());
+ }
+
+ private List<GlobalIndexResult> readBatch(List<VectorSearchSplit> splits) {
int n = vectors.length;
- if (splits.isEmpty()) {
+ List<IndexVectorSearchSplit> indexSplits = new ArrayList<>();
+ List<RawVectorSearchSplit> rawSplits = new ArrayList<>();
+ splitSearchSplits(splits, indexSplits, rawSplits);
+ if (indexSplits.isEmpty() && rawSplits.isEmpty()) {
List<GlobalIndexResult> empty = new ArrayList<>(n);
for (int i = 0; i < n; i++) {
empty.add(GlobalIndexResult.createEmpty());
@@ -66,9 +77,28 @@ public class BatchVectorReadImpl extends AbstractVectorRead
implements BatchVect
return empty;
}
- RoaringNavigableMap64 preFilter = preFilter(splits).orElse(null);
+ GlobalIndexer globalIndexer =
+ indexSplits.isEmpty() ? null :
createGlobalIndexer(indexSplits);
+ ScoredGlobalIndexResult[] indexedResults =
+ indexSplits.isEmpty()
+ ? emptyScoredResults(n)
+ : readIndexedBatch(indexSplits, globalIndexer);
+
+ List<GlobalIndexResult> results = new ArrayList<>(n);
+ RoaringNavigableMap64 rawPreFilter = rawPreFilter(rawSplits);
+ for (int i = 0; i < n; i++) {
+ results.add(
+ withRawSearch(
+ indexedResults[i], rawSplits, globalIndexer,
rawPreFilter, vectors[i]));
+ }
+ return results;
+ }
+
+ protected ScoredGlobalIndexResult[] readIndexedBatch(
+ List<IndexVectorSearchSplit> splits, GlobalIndexer globalIndexer) {
+ int n = vectors.length;
+ List<RoaringNavigableMap64> preFilters = preFilters(splits);
- GlobalIndexer globalIndexer = createGlobalIndexer(splits);
IndexPathFactory indexPathFactory =
table.store().pathFactory().globalIndexFileFactory();
int parallelism =
table.coreOptions().toConfiguration().get(GLOBAL_INDEX_THREAD_NUM);
@@ -76,7 +106,8 @@ public class BatchVectorReadImpl extends AbstractVectorRead
implements BatchVect
List<CompletableFuture<List<Optional<ScoredGlobalIndexResult>>>>
futures =
new ArrayList<>(splits.size());
- for (VectorSearchSplit split : splits) {
+ for (int i = 0; i < splits.size(); i++) {
+ IndexVectorSearchSplit split = splits.get(i);
futures.add(
evalBatch(
globalIndexer,
@@ -85,7 +116,7 @@ public class BatchVectorReadImpl extends AbstractVectorRead
implements BatchVect
split.rowRangeEnd(),
split.vectorIndexFiles(),
vectors,
- preFilter,
+ preFilters.isEmpty() ? null : preFilters.get(i),
executor));
}
@@ -104,11 +135,6 @@ public class BatchVectorReadImpl extends
AbstractVectorRead implements BatchVect
}
}
}
-
- List<GlobalIndexResult> results = new ArrayList<>(n);
- for (int i = 0; i < n; i++) {
- results.add(merged[i].topK(limit));
- }
- return results;
+ return merged;
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/BatchVectorSearchBuilderImpl.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/BatchVectorSearchBuilderImpl.java
index 3e8dd877df..8a34691b7a 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/BatchVectorSearchBuilderImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/BatchVectorSearchBuilderImpl.java
@@ -102,7 +102,7 @@ public class BatchVectorSearchBuilderImpl implements
BatchVectorSearchBuilder {
@Override
public VectorScan newVectorScan() {
- return new VectorScanImpl(table, partitionFilter, filter,
vectorColumn);
+ return new VectorScanImpl(table, partitionFilter, filter,
vectorColumn, options);
}
@Override
@@ -114,6 +114,7 @@ public class BatchVectorSearchBuilderImpl implements
BatchVectorSearchBuilder {
for (float[] vector : vectors) {
checkNotNull(vector, "Search vector element cannot be null");
}
- return new BatchVectorReadImpl(table, filter, limit, vectorColumn,
vectors, options);
+ return new BatchVectorReadImpl(
+ table, partitionFilter, filter, limit, vectorColumn, vectors,
options);
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/VectorSearchSplit.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/IndexVectorSearchSplit.java
similarity index 92%
copy from
paimon-core/src/main/java/org/apache/paimon/table/source/VectorSearchSplit.java
copy to
paimon-core/src/main/java/org/apache/paimon/table/source/IndexVectorSearchSplit.java
index 032c2be301..355b60d8b7 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/VectorSearchSplit.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/IndexVectorSearchSplit.java
@@ -26,12 +26,11 @@ import org.apache.paimon.io.DataOutputViewStreamWrapper;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
-import java.io.Serializable;
import java.util.List;
import java.util.Objects;
-/** Split of vector search. */
-public class VectorSearchSplit implements Serializable {
+/** Split to read vector index files. */
+public class IndexVectorSearchSplit extends VectorSearchSplit {
private static final long serialVersionUID = 1L;
@@ -45,7 +44,7 @@ public class VectorSearchSplit implements Serializable {
private transient List<IndexFileMeta> vectorIndexFiles;
private transient List<IndexFileMeta> scalarIndexFiles;
- public VectorSearchSplit(
+ public IndexVectorSearchSplit(
long rowRangeStart,
long rowRangeEnd,
List<IndexFileMeta> vectorIndexFiles,
@@ -87,7 +86,7 @@ public class VectorSearchSplit implements Serializable {
in.defaultReadObject();
int version = in.readInt();
if (version != VERSION) {
- throw new IOException("Unsupported VectorSearchSplit version: " +
version);
+ throw new IOException("Unsupported IndexVectorSearchSplit version:
" + version);
}
IndexFileMetaSerializer serializer = INDEX_SERIALIZER.get();
DataInputViewStreamWrapper view = new DataInputViewStreamWrapper(in);
@@ -102,7 +101,7 @@ public class VectorSearchSplit implements Serializable {
if (o == null || getClass() != o.getClass()) {
return false;
}
- VectorSearchSplit that = (VectorSearchSplit) o;
+ IndexVectorSearchSplit that = (IndexVectorSearchSplit) o;
return rowRangeStart == that.rowRangeStart
&& rowRangeEnd == that.rowRangeEnd
&& Objects.equals(vectorIndexFiles, that.vectorIndexFiles)
@@ -116,7 +115,7 @@ public class VectorSearchSplit implements Serializable {
@Override
public String toString() {
- return "VectorSearchSplit{"
+ return "IndexVectorSearchSplit{"
+ "rowRangeStart="
+ rowRangeStart
+ ", rowRangeEnd="
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/VectorSearchSplit.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/RawVectorSearchSplit.java
similarity index 59%
copy from
paimon-core/src/main/java/org/apache/paimon/table/source/VectorSearchSplit.java
copy to
paimon-core/src/main/java/org/apache/paimon/table/source/RawVectorSearchSplit.java
index 032c2be301..36221e8ae8 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/VectorSearchSplit.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/RawVectorSearchSplit.java
@@ -22,16 +22,20 @@ import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.index.IndexFileMetaSerializer;
import org.apache.paimon.io.DataInputViewStreamWrapper;
import org.apache.paimon.io.DataOutputViewStreamWrapper;
+import org.apache.paimon.utils.Range;
+
+import javax.annotation.Nullable;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
-import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.Objects;
-/** Split of vector search. */
-public class VectorSearchSplit implements Serializable {
+/** Split to scan raw vectors. */
+public class RawVectorSearchSplit extends VectorSearchSplit {
private static final long serialVersionUID = 1L;
@@ -40,46 +44,37 @@ public class VectorSearchSplit implements Serializable {
private static final ThreadLocal<IndexFileMetaSerializer> INDEX_SERIALIZER
=
ThreadLocal.withInitial(IndexFileMetaSerializer::new);
- private transient long rowRangeStart;
- private transient long rowRangeEnd;
- private transient List<IndexFileMeta> vectorIndexFiles;
+ private final List<Range> rowRanges;
+ @Nullable private final String indexType;
private transient List<IndexFileMeta> scalarIndexFiles;
- public VectorSearchSplit(
- long rowRangeStart,
- long rowRangeEnd,
- List<IndexFileMeta> vectorIndexFiles,
- List<IndexFileMeta> scalarIndexFiles) {
- this.rowRangeStart = rowRangeStart;
- this.rowRangeEnd = rowRangeEnd;
- this.vectorIndexFiles = vectorIndexFiles;
- this.scalarIndexFiles = scalarIndexFiles;
- }
-
- public long rowRangeStart() {
- return rowRangeStart;
+ public RawVectorSearchSplit(
+ List<Range> rowRanges,
+ List<IndexFileMeta> scalarIndexFiles,
+ @Nullable String indexType) {
+ this.rowRanges = Collections.unmodifiableList(new
ArrayList<>(rowRanges));
+ this.scalarIndexFiles = Collections.unmodifiableList(new
ArrayList<>(scalarIndexFiles));
+ this.indexType = indexType;
}
- public long rowRangeEnd() {
- return rowRangeEnd;
- }
-
- public List<IndexFileMeta> vectorIndexFiles() {
- return vectorIndexFiles;
+ public List<Range> rowRanges() {
+ return rowRanges;
}
public List<IndexFileMeta> scalarIndexFiles() {
return scalarIndexFiles;
}
+ @Nullable
+ public String indexType() {
+ return indexType;
+ }
+
private void writeObject(ObjectOutputStream out) throws IOException {
out.defaultWriteObject();
out.writeInt(VERSION);
IndexFileMetaSerializer serializer = INDEX_SERIALIZER.get();
DataOutputViewStreamWrapper view = new
DataOutputViewStreamWrapper(out);
- view.writeLong(rowRangeStart);
- view.writeLong(rowRangeEnd);
- serializer.serializeList(vectorIndexFiles, view);
serializer.serializeList(scalarIndexFiles, view);
}
@@ -87,14 +82,12 @@ public class VectorSearchSplit implements Serializable {
in.defaultReadObject();
int version = in.readInt();
if (version != VERSION) {
- throw new IOException("Unsupported VectorSearchSplit version: " +
version);
+ throw new IOException("Unsupported RawVectorSearchSplit version: "
+ version);
}
IndexFileMetaSerializer serializer = INDEX_SERIALIZER.get();
DataInputViewStreamWrapper view = new DataInputViewStreamWrapper(in);
- this.rowRangeStart = view.readLong();
- this.rowRangeEnd = view.readLong();
- this.vectorIndexFiles = serializer.deserializeList(view);
- this.scalarIndexFiles = serializer.deserializeList(view);
+ this.scalarIndexFiles =
+ Collections.unmodifiableList(new
ArrayList<>(serializer.deserializeList(view)));
}
@Override
@@ -102,29 +95,27 @@ public class VectorSearchSplit implements Serializable {
if (o == null || getClass() != o.getClass()) {
return false;
}
- VectorSearchSplit that = (VectorSearchSplit) o;
- return rowRangeStart == that.rowRangeStart
- && rowRangeEnd == that.rowRangeEnd
- && Objects.equals(vectorIndexFiles, that.vectorIndexFiles)
- && Objects.equals(scalarIndexFiles, that.scalarIndexFiles);
+ RawVectorSearchSplit that = (RawVectorSearchSplit) o;
+ return Objects.equals(rowRanges, that.rowRanges)
+ && Objects.equals(scalarIndexFiles, that.scalarIndexFiles)
+ && Objects.equals(indexType, that.indexType);
}
@Override
public int hashCode() {
- return Objects.hash(rowRangeStart, rowRangeEnd, vectorIndexFiles,
scalarIndexFiles);
+ return Objects.hash(rowRanges, scalarIndexFiles, indexType);
}
@Override
public String toString() {
- return "VectorSearchSplit{"
- + "rowRangeStart="
- + rowRangeStart
- + ", rowRangeEnd="
- + rowRangeEnd
- + ", vectorIndexFiles="
- + vectorIndexFiles
+ return "RawVectorSearchSplit{"
+ + "rowRanges="
+ + rowRanges
+ ", scalarIndexFiles="
+ scalarIndexFiles
+ + ", indexType='"
+ + indexType
+ + '\''
+ '}';
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/VectorRead.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/VectorRead.java
index 74e17e2845..4fc3c7aaf0 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/VectorRead.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/VectorRead.java
@@ -20,14 +20,8 @@ package org.apache.paimon.table.source;
import org.apache.paimon.globalindex.GlobalIndexResult;
-import java.util.List;
-
/** Vector read to read index files. */
public interface VectorRead {
- default GlobalIndexResult read(VectorScan.Plan plan) {
- return read(plan.splits());
- }
-
- GlobalIndexResult read(List<VectorSearchSplit> splits);
+ GlobalIndexResult read(VectorScan.Plan plan);
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/VectorReadImpl.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/VectorReadImpl.java
index eab7ab2737..f04a03c064 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/VectorReadImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/VectorReadImpl.java
@@ -23,11 +23,14 @@ import org.apache.paimon.globalindex.GlobalIndexResult;
import org.apache.paimon.globalindex.GlobalIndexer;
import org.apache.paimon.globalindex.ScoredGlobalIndexResult;
import org.apache.paimon.index.IndexPathFactory;
+import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.types.DataField;
import org.apache.paimon.utils.RoaringNavigableMap64;
+import javax.annotation.Nullable;
+
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -46,24 +49,42 @@ public class VectorReadImpl extends AbstractVectorRead
implements VectorRead {
public VectorReadImpl(
FileStoreTable table,
- Predicate filter,
+ @Nullable PartitionPredicate partitionFilter,
+ @Nullable Predicate filter,
int limit,
DataField vectorColumn,
float[] vector,
- Map<String, String> options) {
- super(table, filter, limit, vectorColumn, options);
+ @Nullable Map<String, String> options) {
+ super(table, partitionFilter, filter, limit, vectorColumn, options);
this.vector = vector;
}
@Override
- public GlobalIndexResult read(List<VectorSearchSplit> splits) {
- if (splits.isEmpty()) {
+ public GlobalIndexResult read(VectorScan.Plan plan) {
+ return readSplits(plan.splits());
+ }
+
+ protected GlobalIndexResult readSplits(List<? extends VectorSearchSplit>
splits) {
+ List<IndexVectorSearchSplit> indexSplits = new ArrayList<>();
+ List<RawVectorSearchSplit> rawSplits = new ArrayList<>();
+ splitSearchSplits(splits, indexSplits, rawSplits);
+ if (indexSplits.isEmpty() && rawSplits.isEmpty()) {
return GlobalIndexResult.createEmpty();
}
- RoaringNavigableMap64 preFilter = preFilter(splits).orElse(null);
+ GlobalIndexer globalIndexer =
+ indexSplits.isEmpty() ? null :
createGlobalIndexer(indexSplits);
+ ScoredGlobalIndexResult result =
+ indexSplits.isEmpty()
+ ? ScoredGlobalIndexResult.createEmpty()
+ : readIndexed(indexSplits, globalIndexer);
+ return withRawSearch(result, rawSplits, globalIndexer,
rawPreFilter(rawSplits), vector);
+ }
+
+ protected ScoredGlobalIndexResult readIndexed(
+ List<IndexVectorSearchSplit> splits, GlobalIndexer globalIndexer) {
+ List<RoaringNavigableMap64> preFilters = preFilters(splits);
- GlobalIndexer globalIndexer = createGlobalIndexer(splits);
IndexPathFactory indexPathFactory =
table.store().pathFactory().globalIndexFileFactory();
int parallelism =
table.coreOptions().toConfiguration().get(GLOBAL_INDEX_THREAD_NUM);
@@ -71,7 +92,8 @@ public class VectorReadImpl extends AbstractVectorRead
implements VectorRead {
List<CompletableFuture<Optional<ScoredGlobalIndexResult>>> futures =
new ArrayList<>(splits.size());
- for (VectorSearchSplit split : splits) {
+ for (int i = 0; i < splits.size(); i++) {
+ IndexVectorSearchSplit split = splits.get(i);
futures.add(
eval(
globalIndexer,
@@ -80,7 +102,7 @@ public class VectorReadImpl extends AbstractVectorRead
implements VectorRead {
split.rowRangeEnd(),
split.vectorIndexFiles(),
vector,
- preFilter,
+ preFilters.isEmpty() ? null : preFilters.get(i),
executor));
}
@@ -93,6 +115,6 @@ public class VectorReadImpl extends AbstractVectorRead
implements VectorRead {
merged = merged.or(splitResult.get());
}
}
- return merged.topK(limit);
+ return merged;
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/VectorScanImpl.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/VectorScanImpl.java
index 74ae2afe8d..63b2dadea3 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/VectorScanImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/VectorScanImpl.java
@@ -19,6 +19,7 @@
package org.apache.paimon.table.source;
import org.apache.paimon.Snapshot;
+import org.apache.paimon.globalindex.GlobalIndexCoverage;
import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileHandler;
import org.apache.paimon.index.IndexFileMeta;
@@ -31,7 +32,10 @@ import org.apache.paimon.types.DataField;
import org.apache.paimon.utils.Filter;
import org.apache.paimon.utils.Range;
+import javax.annotation.Nullable;
+
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -39,38 +43,40 @@ import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;
-import static org.apache.paimon.predicate.PredicateVisitor.collectFieldNames;
+import static org.apache.paimon.predicate.PredicateVisitor.collectFieldIds;
import static org.apache.paimon.utils.Preconditions.checkNotNull;
/** Implementation for {@link VectorScan}. */
public class VectorScanImpl implements VectorScan {
private final FileStoreTable table;
- private final PartitionPredicate partitionFilter;
- private final Predicate filter;
+ @Nullable private final PartitionPredicate partitionFilter;
+ @Nullable private final Predicate filter;
private final DataField vectorColumn;
+ private final Map<String, String> options;
public VectorScanImpl(
FileStoreTable table,
- PartitionPredicate partitionFilter,
- Predicate filter,
- DataField vectorColumn) {
+ @Nullable PartitionPredicate partitionFilter,
+ @Nullable Predicate filter,
+ DataField vectorColumn,
+ @Nullable Map<String, String> options) {
this.table = table;
this.partitionFilter = partitionFilter;
this.filter = filter;
this.vectorColumn = vectorColumn;
+ this.options =
+ options == null
+ ? Collections.emptyMap()
+ : Collections.unmodifiableMap(new HashMap<>(options));
}
@Override
public Plan scan() {
Objects.requireNonNull(vectorColumn, "Vector column must be set");
- Set<Integer> filterFieldIds =
- collectFieldNames(filter).stream()
- .filter(name -> table.rowType().containsField(name))
- .map(name -> table.rowType().getField(name).id())
- .collect(Collectors.toSet());
- Snapshot snapshot = TimeTravelUtil.tryTravelOrLatest(table);
+ Set<Integer> filterFieldIds = collectFieldIds(table.rowType(), filter);
+ @Nullable Snapshot snapshot = TimeTravelUtil.tryTravelOrLatest(table);
IndexFileHandler indexFileHandler =
table.store().newIndexFileHandler();
Filter<IndexManifestEntry> indexFileFilter =
entry -> {
@@ -96,14 +102,20 @@ public class VectorScanImpl implements VectorScan {
indexFileHandler.scan(snapshot, indexFileFilter).stream()
.map(IndexManifestEntry::indexFile)
.collect(Collectors.toList());
+ String vectorIndexType = vectorIndexType(allIndexFiles);
+ if (vectorIndexType == null) {
+ vectorIndexType = configuredVectorIndexType();
+ }
// Group vector index files by (rowRangeStart, rowRangeEnd)
Map<Range, List<IndexFileMeta>> vectorByRange = new HashMap<>();
+ List<IndexFileMeta> vectorIndexFiles = new ArrayList<>();
for (IndexFileMeta indexFile : allIndexFiles) {
GlobalIndexMeta meta = checkNotNull(indexFile.globalIndexMeta());
if (isPrimaryColumn(meta, vectorColumn.id())) {
Range range = new Range(meta.rowRangeStart(),
meta.rowRangeEnd());
vectorByRange.computeIfAbsent(range, k -> new
ArrayList<>()).add(indexFile);
+ vectorIndexFiles.add(indexFile);
}
}
@@ -124,16 +136,129 @@ public class VectorScanImpl implements VectorScan {
return
range.hasIntersection(globalIndex.rowRange());
})
.collect(Collectors.toList());
- splits.add(new VectorSearchSplit(range.from, range.to,
vectorFiles, scalarFiles));
+ splits.add(new IndexVectorSearchSplit(range.from, range.to,
vectorFiles, scalarFiles));
+ }
+
+ List<Range> rawRowRanges =
+ new GlobalIndexCoverage(table, snapshot, partitionFilter,
vectorIndexFiles)
+ .unindexedRanges(vectorColumn.id());
+ if (filter != null) {
+ rawRowRanges =
+ Range.sortAndMergeOverlap(
+ addAll(
+ rawRowRanges,
+ new GlobalIndexCoverage(
+ table,
+ snapshot,
+ partitionFilter,
+
scalarIndexFiles(allIndexFiles))
+ .unindexedRanges(table.rowType(),
filter)),
+ true);
+ }
+ if (!rawRowRanges.isEmpty()) {
+ splits.add(
+ new RawVectorSearchSplit(
+ rawRowRanges,
+ scalarIndexFiles(allIndexFiles, rawRowRanges),
+ vectorIndexType));
}
- return () -> splits;
+ return new Plan() {
+ @Override
+ public List<VectorSearchSplit> splits() {
+ return splits;
+ }
+ };
}
private static boolean isPrimaryColumn(GlobalIndexMeta meta, int fieldId) {
return meta.indexFieldId() == fieldId;
}
+ private List<IndexFileMeta> scalarIndexFiles(List<IndexFileMeta>
allIndexFiles) {
+ return allIndexFiles.stream()
+ .filter(
+ f -> {
+ GlobalIndexMeta globalIndex =
checkNotNull(f.globalIndexMeta());
+ return !isPrimaryColumn(globalIndex,
vectorColumn.id());
+ })
+ .collect(Collectors.toList());
+ }
+
+ private List<IndexFileMeta> scalarIndexFiles(
+ List<IndexFileMeta> allIndexFiles, List<Range> rowRanges) {
+ return allIndexFiles.stream()
+ .filter(
+ f -> {
+ GlobalIndexMeta globalIndex =
checkNotNull(f.globalIndexMeta());
+ if (isPrimaryColumn(globalIndex,
vectorColumn.id())) {
+ return false;
+ }
+ return hasIntersection(rowRanges,
globalIndex.rowRange());
+ })
+ .collect(Collectors.toList());
+ }
+
+ private static boolean hasIntersection(List<Range> ranges, Range range) {
+ for (Range r : ranges) {
+ if (r.hasIntersection(range)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static List<Range> addAll(List<Range> left, List<Range> right) {
+ List<Range> result = new ArrayList<>(left.size() + right.size());
+ result.addAll(left);
+ result.addAll(right);
+ return result;
+ }
+
+ @Nullable
+ private String vectorIndexType(List<IndexFileMeta> indexFiles) {
+ String indexType = null;
+ for (IndexFileMeta indexFile : indexFiles) {
+ GlobalIndexMeta meta = checkNotNull(indexFile.globalIndexMeta());
+ if (!isPrimaryColumn(meta, vectorColumn.id())) {
+ continue;
+ }
+ if (indexType == null) {
+ indexType = indexFile.indexType();
+ } else if (!indexType.equals(indexFile.indexType())) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Vector column '%s' has multiple index types:
%s and %s.",
+ vectorColumn.name(), indexType,
indexFile.indexType()));
+ }
+ }
+ return indexType;
+ }
+
+ @Nullable
+ private String configuredVectorIndexType() {
+ String indexType = option("index_type");
+ if (indexType == null) {
+ indexType = option("index-type");
+ }
+ if (indexType == null) {
+ indexType = option("vector.index-type");
+ }
+ if (indexType == null) {
+ indexType = option("fields." + vectorColumn.name() +
".index-type");
+ }
+ return indexType;
+ }
+
+ @Nullable
+ private String option(String key) {
+ String value = options.get(key);
+ if (value == null) {
+ value = table.options().get(key);
+ }
+ return value == null ? null : value.toLowerCase().trim();
+ }
+
private static boolean containsField(GlobalIndexMeta meta, int fieldId) {
if (meta.indexFieldId() == fieldId) {
return true;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/VectorSearchBuilderImpl.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/VectorSearchBuilderImpl.java
index 0ca9612436..996f8b4f19 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/VectorSearchBuilderImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/VectorSearchBuilderImpl.java
@@ -101,12 +101,13 @@ public class VectorSearchBuilderImpl implements
VectorSearchBuilder {
@Override
public VectorScan newVectorScan() {
- return new VectorScanImpl(table, partitionFilter, filter,
vectorColumn);
+ return new VectorScanImpl(table, partitionFilter, filter,
vectorColumn, options);
}
@Override
public VectorRead newVectorRead() {
checkNotNull(vector, "vector must be set via withVector()");
- return new VectorReadImpl(table, filter, limit, vectorColumn, vector,
options);
+ return new VectorReadImpl(
+ table, partitionFilter, filter, limit, vectorColumn, vector,
options);
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/VectorSearchSplit.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/VectorSearchSplit.java
index 032c2be301..82c813b9f2 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/VectorSearchSplit.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/VectorSearchSplit.java
@@ -18,113 +18,10 @@
package org.apache.paimon.table.source;
-import org.apache.paimon.index.IndexFileMeta;
-import org.apache.paimon.index.IndexFileMetaSerializer;
-import org.apache.paimon.io.DataInputViewStreamWrapper;
-import org.apache.paimon.io.DataOutputViewStreamWrapper;
-
-import java.io.IOException;
-import java.io.ObjectInputStream;
-import java.io.ObjectOutputStream;
import java.io.Serializable;
-import java.util.List;
-import java.util.Objects;
-/** Split of vector search. */
-public class VectorSearchSplit implements Serializable {
+/** Base split of vector search. */
+public abstract class VectorSearchSplit implements Serializable {
private static final long serialVersionUID = 1L;
-
- private static final int VERSION = 1;
-
- private static final ThreadLocal<IndexFileMetaSerializer> INDEX_SERIALIZER
=
- ThreadLocal.withInitial(IndexFileMetaSerializer::new);
-
- private transient long rowRangeStart;
- private transient long rowRangeEnd;
- private transient List<IndexFileMeta> vectorIndexFiles;
- private transient List<IndexFileMeta> scalarIndexFiles;
-
- public VectorSearchSplit(
- long rowRangeStart,
- long rowRangeEnd,
- List<IndexFileMeta> vectorIndexFiles,
- List<IndexFileMeta> scalarIndexFiles) {
- this.rowRangeStart = rowRangeStart;
- this.rowRangeEnd = rowRangeEnd;
- this.vectorIndexFiles = vectorIndexFiles;
- this.scalarIndexFiles = scalarIndexFiles;
- }
-
- public long rowRangeStart() {
- return rowRangeStart;
- }
-
- public long rowRangeEnd() {
- return rowRangeEnd;
- }
-
- public List<IndexFileMeta> vectorIndexFiles() {
- return vectorIndexFiles;
- }
-
- public List<IndexFileMeta> scalarIndexFiles() {
- return scalarIndexFiles;
- }
-
- private void writeObject(ObjectOutputStream out) throws IOException {
- out.defaultWriteObject();
- out.writeInt(VERSION);
- IndexFileMetaSerializer serializer = INDEX_SERIALIZER.get();
- DataOutputViewStreamWrapper view = new
DataOutputViewStreamWrapper(out);
- view.writeLong(rowRangeStart);
- view.writeLong(rowRangeEnd);
- serializer.serializeList(vectorIndexFiles, view);
- serializer.serializeList(scalarIndexFiles, view);
- }
-
- private void readObject(ObjectInputStream in) throws IOException,
ClassNotFoundException {
- in.defaultReadObject();
- int version = in.readInt();
- if (version != VERSION) {
- throw new IOException("Unsupported VectorSearchSplit version: " +
version);
- }
- IndexFileMetaSerializer serializer = INDEX_SERIALIZER.get();
- DataInputViewStreamWrapper view = new DataInputViewStreamWrapper(in);
- this.rowRangeStart = view.readLong();
- this.rowRangeEnd = view.readLong();
- this.vectorIndexFiles = serializer.deserializeList(view);
- this.scalarIndexFiles = serializer.deserializeList(view);
- }
-
- @Override
- public boolean equals(Object o) {
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- VectorSearchSplit that = (VectorSearchSplit) o;
- return rowRangeStart == that.rowRangeStart
- && rowRangeEnd == that.rowRangeEnd
- && Objects.equals(vectorIndexFiles, that.vectorIndexFiles)
- && Objects.equals(scalarIndexFiles, that.scalarIndexFiles);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(rowRangeStart, rowRangeEnd, vectorIndexFiles,
scalarIndexFiles);
- }
-
- @Override
- public String toString() {
- return "VectorSearchSplit{"
- + "rowRangeStart="
- + rowRangeStart
- + ", rowRangeEnd="
- + rowRangeEnd
- + ", vectorIndexFiles="
- + vectorIndexFiles
- + ", scalarIndexFiles="
- + scalarIndexFiles
- + '}';
- }
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java
index 28eceb69ce..786fecd0cb 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/TimeTravelUtil.java
@@ -65,6 +65,7 @@ public class TimeTravelUtil {
SCAN_TIMESTAMP_MILLIS.key()
};
+ @Nullable
public static Snapshot tryTravelOrLatest(FileStoreTable table) {
return tryTravelToSnapshot(table).orElseGet(() ->
table.latestSnapshot().orElse(null));
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java
index 8d963b11e9..0723ee5239 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java
@@ -168,6 +168,31 @@ public class BtreeGlobalIndexTableTest extends
DataEvolutionTestBase {
.containsExactly("a700");
}
+ @Test
+ public void testGlobalIndexScannerKeepsUnindexedRowsSeparate() throws
Exception {
+ write(500L);
+ createIndex("f1");
+ appendRows(500, 1000);
+
+ FileStoreTable table =
+ tableWithSearchMode((FileStoreTable)
catalog.getTable(identifier()), "full");
+ Predicate predicate =
+ new PredicateBuilder(table.rowType())
+ .in(
+ 1,
+ Arrays.asList(
+ BinaryString.fromString("a100"),
+ BinaryString.fromString("a700")));
+
+ try (GlobalIndexScanner scanner =
+ GlobalIndexScanner.create(table,
PartitionPredicate.ALWAYS_TRUE, predicate).get()) {
+ assertThat(scanner.scan(predicate).get().results().toRangeList())
+ .containsExactly(new Range(100L, 100L));
+
assertThat(scanner.unindexedRows(predicate).results().toRangeList())
+ .containsExactly(new Range(500L, 999L));
+ }
+ }
+
@Test
public void testBTreeGlobalIndexSearchModeUsesAllPredicateFieldCoverage()
throws Exception {
write(500L);
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 522d9fa732..e776dd8314 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
@@ -30,14 +30,21 @@ import
org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
import org.apache.paimon.globalindex.ResultEntry;
import org.apache.paimon.globalindex.ScoredGlobalIndexResult;
import org.apache.paimon.globalindex.btree.BTreeGlobalIndexerFactory;
+import org.apache.paimon.globalindex.testvector.TestVectorGlobalIndexer;
import org.apache.paimon.globalindex.testvector.TestVectorGlobalIndexerFactory;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.io.CompactIncrement;
import org.apache.paimon.io.DataIncrement;
import org.apache.paimon.options.Options;
import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.predicate.FieldRef;
+import org.apache.paimon.predicate.FieldTransform;
+import org.apache.paimon.predicate.GreaterOrEqual;
+import org.apache.paimon.predicate.LeafFunction;
+import org.apache.paimon.predicate.LeafPredicate;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.predicate.Transform;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.table.FileStoreTable;
@@ -240,6 +247,181 @@ public class VectorSearchBuilderTest extends
TableTestBase {
assertThat(result.results().isEmpty()).isTrue();
}
+ @Test
+ public void testFullModeRawOnlyUsesConfiguredMetric() throws Exception {
+ catalog.createTable(
+ identifier("full_search_raw_only_cosine_table"),
+ vectorSchemaBuilder(VECTOR_FIELD_NAME)
+ .option(CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key(),
"full")
+ .option("test.vector.metric", "cosine")
+ .build(),
+ false);
+ FileStoreTable table =
getTable(identifier("full_search_raw_only_cosine_table"));
+
+ float[][] vectors = {{100.0f, 0.0f}, {0.9f, 0.1f}};
+ writeVectors(table, vectors);
+
+ GlobalIndexResult result =
+ table.newVectorSearchBuilder()
+ .withVector(new float[] {1.0f, 0.0f})
+ .withLimit(1)
+ .withVectorColumn(VECTOR_FIELD_NAME)
+ .executeLocal();
+
+ assertThat(result.results()).containsExactly(0L);
+ }
+
+ @Test
+ public void testVectorSearchFullModeScansUnindexedData() throws Exception {
+ catalog.createTable(
+ identifier("full_search_cosine_table"),
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column(VECTOR_FIELD_NAME, new
ArrayType(DataTypes.FLOAT()))
+ .option(CoreOptions.BUCKET.key(), "-1")
+ .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(),
"true")
+ .option(CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key(),
"full")
+ .option("test.vector.dimension",
String.valueOf(DIMENSION))
+ .option("test.vector.metric", "cosine")
+ .build(),
+ false);
+ FileStoreTable table =
getTable(identifier("full_search_cosine_table"));
+
+ float[][] vectors = {
+ {0.0f, 1.0f},
+ {0.1f, 0.9f},
+ {0.5f, 0.0f},
+ {0.99f, 0.01f}
+ };
+ writeVectors(table, vectors);
+
+ buildAndCommitVectorIndex(table, new float[][] {vectors[0],
vectors[1]}, new Range(0, 1));
+
+ TestVectorGlobalIndexer.resetMetricCalls();
+ VectorScan.Plan plan =
+ table.newVectorSearchBuilder()
+ .withVector(new float[] {1.0f, 0.0f})
+ .withLimit(2)
+ .withVectorColumn(VECTOR_FIELD_NAME)
+ .newVectorScan()
+ .scan();
+ assertThat(indexVectorSearchSplits(plan.splits())).hasSize(1);
+ assertThat(rawVectorSearchSplits(plan.splits())).hasSize(1);
+ assertThat(rawVectorSearchSplits(plan.splits()).get(0).rowRanges())
+ .containsExactly(new Range(2, 3));
+
+ GlobalIndexResult result =
+ table.newVectorSearchBuilder()
+ .withVector(new float[] {1.0f, 0.0f})
+ .withLimit(2)
+ .withVectorColumn(VECTOR_FIELD_NAME)
+ .executeLocal();
+
+ assertThat(result.results()).containsExactlyInAnyOrder(2L, 3L);
+ assertThat(TestVectorGlobalIndexer.metricCalls()).isGreaterThan(0);
+ }
+
+ @Test
+ public void testVectorSearchFastModeSkipsUnindexedDataByDefault() throws
Exception {
+ catalog.createTable(
+ identifier("fast_search_table"),
+ vectorSchemaBuilder(VECTOR_FIELD_NAME).build(),
+ false);
+ FileStoreTable table = getTable(identifier("fast_search_table"));
+
+ float[][] vectors = {
+ {0.0f, 1.0f},
+ {0.1f, 0.9f},
+ {1.0f, 0.0f},
+ {0.95f, 0.05f}
+ };
+ writeVectors(table, vectors);
+
+ buildAndCommitVectorIndex(table, new float[][] {vectors[0],
vectors[1]}, new Range(0, 1));
+
+ GlobalIndexResult result =
+ table.newVectorSearchBuilder()
+ .withVector(new float[] {1.0f, 0.0f})
+ .withLimit(2)
+ .withVectorColumn(VECTOR_FIELD_NAME)
+ .executeLocal();
+
+ assertThat(result.results()).doesNotContain(2L, 3L);
+ }
+
+ @Test
+ public void testVectorSearchFullModeScansFilteredUnindexedData() throws
Exception {
+ catalog.createTable(
+ identifier("full_search_filtered_table"),
+ vectorSchemaBuilder(VECTOR_FIELD_NAME)
+ .option(CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key(),
"full")
+ .build(),
+ false);
+ FileStoreTable table =
getTable(identifier("full_search_filtered_table"));
+
+ float[][] vectors = {
+ {0.0f, 1.0f},
+ {0.1f, 0.9f},
+ {1.0f, 0.0f},
+ {0.95f, 0.05f}
+ };
+ writeVectors(table, vectors);
+
+ Range indexedRange = new Range(0, 1);
+ buildAndCommitVectorIndex(table, new float[][] {vectors[0],
vectors[1]}, indexedRange);
+ buildAndCommitBTreeIndex(table, new int[] {0, 1}, indexedRange);
+
+ Predicate filter = new
PredicateBuilder(table.rowType()).greaterOrEqual(0, 2);
+ GlobalIndexResult result =
+ table.newVectorSearchBuilder()
+ .withVector(new float[] {1.0f, 0.0f})
+ .withLimit(2)
+ .withVectorColumn(VECTOR_FIELD_NAME)
+ .withFilter(filter)
+ .executeLocal();
+
+ assertThat(result).isInstanceOf(ScoredGlobalIndexResult.class);
+ assertThat(result.results()).containsExactlyInAnyOrder(2L, 3L);
+ }
+
+ @Test
+ public void testVectorSearchRawSearchUsesScalarPreFilter() throws
Exception {
+ catalog.createTable(
+ identifier("raw_search_scalar_prefilter_table"),
+ vectorSchemaBuilder(VECTOR_FIELD_NAME)
+ .option(CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key(),
"full")
+ .build(),
+ false);
+ FileStoreTable table =
getTable(identifier("raw_search_scalar_prefilter_table"));
+
+ float[][] vectors = {
+ {100.0f, 0.0f},
+ {200.0f, 0.0f},
+ {1.0f, 0.0f},
+ {2.0f, 0.0f}
+ };
+ writeVectors(table, vectors);
+
+ buildAndCommitVectorIndex(table, new float[][] {vectors[0],
vectors[1]}, new Range(0, 1));
+ buildAndCommitBTreeIndex(table, new int[] {2, 3}, new Range(2, 3));
+
+ // Let scalar global index evaluate id >= 3, but let the final raw row
filter pass through.
+ // Without raw pre-filtering, raw topK would pick row 2 because it is
closer to the query.
+ Predicate filter = globalIndexOnlyIdGreaterOrEqual(table, 3);
+ VectorSearchBuilder builder =
+ table.newVectorSearchBuilder()
+ .withVector(new float[] {1.0f, 0.0f})
+ .withLimit(1)
+ .withVectorColumn(VECTOR_FIELD_NAME)
+ .withFilter(filter);
+ VectorScan.Plan plan = builder.newVectorScan().scan();
+
assertThat(rawVectorSearchSplits(plan.splits()).get(0).scalarIndexFiles()).isNotEmpty();
+
+ GlobalIndexResult result = builder.newVectorRead().read(plan);
+ assertThat(result.results()).containsExactly(3L);
+ }
+
@Test
public void testVectorSearchTopKLimit() throws Exception {
createTableDefault();
@@ -431,8 +613,13 @@ public class VectorSearchBuilderTest extends TableTestBase
{
@Test
public void testScanPartialRangeIntersection() throws Exception {
- createTableDefault();
- FileStoreTable table = getTableDefault();
+ catalog.createTable(
+ identifier("full_search_partial_scalar_table"),
+ vectorSchemaBuilder(VECTOR_FIELD_NAME)
+ .option(CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key(),
"full")
+ .build(),
+ false);
+ FileStoreTable table =
getTable(identifier("full_search_partial_scalar_table"));
// Write 10 rows
float[][] allVectors = new float[10][];
@@ -458,15 +645,19 @@ public class VectorSearchBuilderTest extends
TableTestBase {
.newVectorScan()
.scan();
- assertThat(plan.splits()).hasSize(1);
- VectorSearchSplit split = plan.splits().get(0);
+ assertThat(indexVectorSearchSplits(plan.splits())).hasSize(1);
+ assertThat(rawVectorSearchSplits(plan.splits())).hasSize(1);
+ IndexVectorSearchSplit split =
indexVectorSearchSplits(plan.splits()).get(0);
assertThat(split.rowRangeStart()).isEqualTo(0);
assertThat(split.rowRangeEnd()).isEqualTo(9);
assertThat(split.vectorIndexFiles()).isNotEmpty();
- // Scalar index [3,7] intersects vector range [0,9] → attached
+ // Scalar index [3,7] intersects vector range [0,9], so it is attached.
assertThat(split.scalarIndexFiles()).isNotEmpty();
+ assertThat(rawVectorSearchSplits(plan.splits()).get(0).rowRanges())
+ .containsExactly(new Range(0, 2), new Range(8, 9));
- // Read with pre-filter: id >= 5, btree covers [3,7] so rows 5,6,7
from btree
+ // Read with pre-filter: btree covers [3,7], while rows outside that
scalar index coverage
+ // are searched through raw splits and filtered by the raw table read.
GlobalIndexResult result =
table.newVectorSearchBuilder()
.withVector(new float[] {1.0f, 0.0f})
@@ -478,10 +669,9 @@ public class VectorSearchBuilderTest extends TableTestBase
{
assertThat(result).isInstanceOf(ScoredGlobalIndexResult.class);
assertThat(result.results().isEmpty()).isFalse();
- // Pre-filter restricts to rows matching id >= 5 from btree [3,7]
- for (long rowId : result.results()) {
- assertThat(rowId).isBetween(5L, 7L);
- }
+
+ assertThat(result.results()).containsAnyOf(5L, 6L, 7L);
+ assertThat(result.results()).doesNotContain(0L, 1L, 2L);
}
@Test
@@ -513,13 +703,119 @@ public class VectorSearchBuilderTest extends
TableTestBase {
VectorScan.Plan plan = searchBuilder.newVectorScan().scan();
assertThat(plan.splits()).hasSize(1);
// Scalar index is attached since field matches filter
- assertThat(plan.splits().get(0).scalarIndexFiles()).isNotEmpty();
+ assertThat(((IndexVectorSearchSplit)
plan.splits().get(0)).scalarIndexFiles()).isNotEmpty();
- // Read: preFilter returns empty bitmap → vector search returns no
results
+ // Read: preFilter returns empty bitmap, so vector search returns no
results.
GlobalIndexResult result = searchBuilder.newVectorRead().read(plan);
assertThat(result.results().isEmpty()).isTrue();
}
+ @Test
+ public void testPartialScalarPreFilterMustNotDropUnindexedScalarRows()
throws Exception {
+ catalog.createTable(
+ identifier("full_search_partial_scalar_unindexed_table"),
+ vectorSchemaBuilder(VECTOR_FIELD_NAME)
+ .option(CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key(),
"full")
+ .build(),
+ false);
+ FileStoreTable table =
getTable(identifier("full_search_partial_scalar_unindexed_table"));
+
+ float[][] vectors = new float[10][];
+ for (int i = 0; i < vectors.length; i++) {
+ vectors[i] = new float[] {Math.abs(i - 8), 0.0f};
+ }
+ writeVectors(table, vectors);
+
+ buildAndCommitVectorIndex(table, vectors, new Range(0, 9));
+ buildAndCommitBTreeIndex(table, new int[] {3, 4, 5, 6, 7}, new
Range(3, 7));
+
+ Predicate idFilter = new
PredicateBuilder(table.rowType()).greaterOrEqual(0, 8);
+ VectorSearchBuilder searchBuilder =
+ table.newVectorSearchBuilder()
+ .withVector(new float[] {0.0f, 0.0f})
+ .withLimit(1)
+ .withVectorColumn(VECTOR_FIELD_NAME)
+ .withFilter(idFilter);
+
+ VectorScan.Plan vectorPlan = searchBuilder.newVectorScan().scan();
+ GlobalIndexResult result =
searchBuilder.newVectorRead().read(vectorPlan);
+ assertThat(result.results()).contains(8L);
+
+ ReadBuilder readBuilder = table.newReadBuilder().withFilter(idFilter);
+ TableScan.Plan readPlan =
readBuilder.newScan().withGlobalIndexResult(result).plan();
+ List<Integer> ids = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
readBuilder.newRead().createReader(readPlan)) {
+ reader.forEachRemaining(row -> ids.add(row.getInt(0)));
+ }
+ assertThat(ids).containsExactly(8);
+ }
+
+ @Test
+ public void
testFullModeFilterWithoutScalarIndexMustNotLetVectorIndexPolluteTopK()
+ throws Exception {
+ catalog.createTable(
+ identifier("full_search_no_scalar_index_filter_table"),
+ vectorSchemaBuilder(VECTOR_FIELD_NAME)
+ .option(CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key(),
"full")
+ .build(),
+ false);
+ FileStoreTable table =
getTable(identifier("full_search_no_scalar_index_filter_table"));
+
+ float[][] vectors = new float[10][];
+ for (int i = 0; i < vectors.length; i++) {
+ vectors[i] = new float[] {Math.abs(i), 0.0f};
+ }
+ writeVectors(table, vectors);
+
+ buildAndCommitVectorIndex(table, vectors, new Range(0, 9));
+
+ Predicate idFilter = new
PredicateBuilder(table.rowType()).greaterOrEqual(0, 8);
+ VectorSearchBuilder searchBuilder =
+ table.newVectorSearchBuilder()
+ .withVector(new float[] {0.0f, 0.0f})
+ .withLimit(1)
+ .withVectorColumn(VECTOR_FIELD_NAME)
+ .withFilter(idFilter);
+
+ GlobalIndexResult result =
+
searchBuilder.newVectorRead().read(searchBuilder.newVectorScan().scan());
+
+ ReadBuilder readBuilder = table.newReadBuilder().withFilter(idFilter);
+ TableScan.Plan readPlan =
readBuilder.newScan().withGlobalIndexResult(result).plan();
+ List<Integer> ids = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
readBuilder.newRead().createReader(readPlan)) {
+ reader.forEachRemaining(row -> ids.add(row.getInt(0)));
+ }
+ assertThat(ids).containsExactly(8);
+ }
+
+ @Test
+ public void testFastModePartialScalarPreFilterOnlyUsesIndexedRows() throws
Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+
+ float[][] vectors = new float[10][];
+ for (int i = 0; i < vectors.length; i++) {
+ vectors[i] = new float[] {Math.abs(i - 8), 0.0f};
+ }
+ writeVectors(table, vectors);
+
+ buildAndCommitVectorIndex(table, vectors, new Range(0, 9));
+ buildAndCommitBTreeIndex(table, new int[] {3, 4, 5, 6, 7}, new
Range(3, 7));
+
+ Predicate idFilter = new
PredicateBuilder(table.rowType()).greaterOrEqual(0, 8);
+ VectorSearchBuilder searchBuilder =
+ table.newVectorSearchBuilder()
+ .withVector(new float[] {0.0f, 0.0f})
+ .withLimit(1)
+ .withVectorColumn(VECTOR_FIELD_NAME)
+ .withFilter(idFilter);
+
+ GlobalIndexResult result =
+
searchBuilder.newVectorRead().read(searchBuilder.newVectorScan().scan());
+ assertThat(result.results().isEmpty()).isTrue();
+ }
+
@Test
public void testVectorSearchRequiresVectorColumnAsPrimaryField() throws
Exception {
createTableDefault();
@@ -567,7 +863,7 @@ public class VectorSearchBuilderTest extends TableTestBase {
.scan();
assertThat(plan.splits()).hasSize(1);
- VectorSearchSplit original = plan.splits().get(0);
+ IndexVectorSearchSplit original = (IndexVectorSearchSplit)
plan.splits().get(0);
// Serialize
ByteArrayOutputStream bos = new ByteArrayOutputStream();
@@ -576,10 +872,10 @@ public class VectorSearchBuilderTest extends
TableTestBase {
}
// Deserialize
- VectorSearchSplit deserialized;
+ IndexVectorSearchSplit deserialized;
try (ObjectInputStream in =
new ObjectInputStream(new
ByteArrayInputStream(bos.toByteArray()))) {
- deserialized = (VectorSearchSplit) in.readObject();
+ deserialized = (IndexVectorSearchSplit) in.readObject();
}
// Verify all fields match
@@ -654,6 +950,37 @@ public class VectorSearchBuilderTest extends TableTestBase
{
assertThat(ids1).contains(4);
}
+ @Test
+ public void testBatchVectorSearchFullModeScansUnindexedData() throws
Exception {
+ catalog.createTable(
+ identifier("batch_full_search_table"),
+ vectorSchemaBuilder(VECTOR_FIELD_NAME)
+ .option(CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key(),
"full")
+ .build(),
+ false);
+ FileStoreTable table = getTable(identifier("batch_full_search_table"));
+
+ float[][] vectors = {
+ {0.0f, 1.0f},
+ {0.1f, 0.9f},
+ {1.0f, 0.0f},
+ {0.95f, 0.05f}
+ };
+ writeVectors(table, vectors);
+ buildAndCommitVectorIndex(table, new float[][] {vectors[0],
vectors[1]}, new Range(0, 1));
+
+ List<GlobalIndexResult> results =
+ table.newBatchVectorSearchBuilder()
+ .withVectors(new float[][] {{1.0f, 0.0f}, {0.0f,
1.0f}})
+ .withLimit(1)
+ .withVectorColumn(VECTOR_FIELD_NAME)
+ .executeBatchLocal();
+
+ assertThat(results).hasSize(2);
+ assertThat(results.get(0).results()).containsExactly(2L);
+ assertThat(results.get(1).results()).containsExactly(0L);
+ }
+
@Test
public void testBatchVectorSearchWithMultipleIndexFiles() throws Exception
{
createTableDefault();
@@ -950,12 +1277,14 @@ public class VectorSearchBuilderTest extends
TableTestBase {
VectorScan.Plan plan = searchBuilder.newVectorScan().scan();
assertThat(plan.splits()).isNotEmpty();
// Every split should have vector index files
- for (VectorSearchSplit split : plan.splits()) {
+ for (IndexVectorSearchSplit split :
indexVectorSearchSplits(plan.splits())) {
assertThat(split.vectorIndexFiles()).isNotEmpty();
}
// At least one split should have scalar (btree) index files
long scalarCount =
- plan.splits().stream().filter(s ->
!s.scalarIndexFiles().isEmpty()).count();
+ indexVectorSearchSplits(plan.splits()).stream()
+ .filter(s -> !s.scalarIndexFiles().isEmpty())
+ .count();
assertThat(scalarCount).isGreaterThan(0);
// --- Test VectorReadImpl: pre-filter should narrow results ---
@@ -1104,4 +1433,53 @@ public class VectorSearchBuilderTest extends
TableTestBase {
commit.commit(Collections.singletonList(message));
}
}
+
+ private List<IndexVectorSearchSplit>
indexVectorSearchSplits(List<VectorSearchSplit> splits) {
+ List<IndexVectorSearchSplit> indexSplits = new ArrayList<>();
+ for (VectorSearchSplit split : splits) {
+ if (split instanceof IndexVectorSearchSplit) {
+ indexSplits.add((IndexVectorSearchSplit) split);
+ }
+ }
+ return indexSplits;
+ }
+
+ private List<RawVectorSearchSplit>
rawVectorSearchSplits(List<VectorSearchSplit> splits) {
+ List<RawVectorSearchSplit> rawSplits = new ArrayList<>();
+ for (VectorSearchSplit split : splits) {
+ if (split instanceof RawVectorSearchSplit) {
+ rawSplits.add((RawVectorSearchSplit) split);
+ }
+ }
+ return rawSplits;
+ }
+
+ private Predicate globalIndexOnlyIdGreaterOrEqual(FileStoreTable table,
int literal) {
+ DataField idField = table.rowType().getField("id");
+ return new GlobalIndexOnlyLeafPredicate(
+ new FieldTransform(new FieldRef(0, idField.name(),
idField.type())),
+ GreaterOrEqual.INSTANCE,
+ Collections.singletonList(literal));
+ }
+
+ private static class GlobalIndexOnlyLeafPredicate extends LeafPredicate {
+
+ private static final long serialVersionUID = 1L;
+
+ private GlobalIndexOnlyLeafPredicate(
+ Transform transform, LeafFunction function, List<Object>
literals) {
+ super(transform, function, literals);
+ }
+
+ @Override
+ public boolean test(InternalRow row) {
+ return true;
+ }
+
+ @Override
+ public LeafPredicate copyWithNewInputs(List<Object> newInputs) {
+ return new GlobalIndexOnlyLeafPredicate(
+ transform().copyWithNewInputs(newInputs), function(),
literals());
+ }
+ }
}
diff --git
a/paimon-lumina/src/main/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexer.java
b/paimon-lumina/src/main/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexer.java
index 276cc5aa85..69b73a6bce 100644
---
a/paimon-lumina/src/main/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexer.java
+++
b/paimon-lumina/src/main/java/org/apache/paimon/lumina/index/LuminaVectorGlobalIndexer.java
@@ -21,7 +21,7 @@ package org.apache.paimon.lumina.index;
import org.apache.paimon.globalindex.GlobalIndexIOMeta;
import org.apache.paimon.globalindex.GlobalIndexReader;
import org.apache.paimon.globalindex.GlobalIndexWriter;
-import org.apache.paimon.globalindex.GlobalIndexer;
+import org.apache.paimon.globalindex.VectorGlobalIndexer;
import org.apache.paimon.globalindex.io.GlobalIndexFileReader;
import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
import org.apache.paimon.options.Options;
@@ -31,7 +31,7 @@ import java.util.List;
import java.util.concurrent.ExecutorService;
/** Lumina vector global indexer. */
-public class LuminaVectorGlobalIndexer implements GlobalIndexer {
+public class LuminaVectorGlobalIndexer implements VectorGlobalIndexer {
private final DataType fieldType;
private final LuminaVectorIndexOptions options;
@@ -53,4 +53,9 @@ public class LuminaVectorGlobalIndexer implements
GlobalIndexer {
ExecutorService executor) {
return new LuminaVectorGlobalIndexReader(fileReader, files, fieldType,
options, executor);
}
+
+ @Override
+ public String metric() {
+ return options.metric().getLuminaName();
+ }
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/read/SparkVectorReadImpl.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/read/SparkVectorReadImpl.java
index 1ddc9c9ac9..90b1eef331 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/read/SparkVectorReadImpl.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/read/SparkVectorReadImpl.java
@@ -25,16 +25,20 @@ import org.apache.paimon.globalindex.GlobalIndexer;
import org.apache.paimon.globalindex.GlobalIndexerFactoryUtils;
import org.apache.paimon.globalindex.ScoredGlobalIndexResult;
import org.apache.paimon.index.IndexPathFactory;
+import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.source.IndexVectorSearchSplit;
+import org.apache.paimon.table.source.RawVectorSearchSplit;
import org.apache.paimon.table.source.VectorReadImpl;
-import org.apache.paimon.table.source.VectorSearchSplit;
+import org.apache.paimon.table.source.VectorScan;
import org.apache.paimon.types.DataField;
import org.apache.paimon.utils.InstantiationUtil;
+import org.apache.paimon.utils.Range;
import org.apache.paimon.utils.RoaringNavigableMap64;
import org.apache.paimon.utils.SerializableFunction;
-import org.apache.spark.broadcast.Broadcast;
+import javax.annotation.Nullable;
import java.io.IOException;
import java.util.ArrayList;
@@ -56,67 +60,90 @@ public class SparkVectorReadImpl extends VectorReadImpl {
public SparkVectorReadImpl(
FileStoreTable table,
- Predicate filter,
+ @Nullable PartitionPredicate partitionFilter,
+ @Nullable Predicate filter,
int limit,
DataField vectorColumn,
float[] vector,
- Map<String, String> options) {
- super(table, filter, limit, vectorColumn, vector, options);
+ @Nullable Map<String, String> options) {
+ super(table, partitionFilter, filter, limit, vectorColumn, vector,
options);
}
@Override
- public GlobalIndexResult read(List<VectorSearchSplit> splits) {
- if (splits.isEmpty()) {
+ public GlobalIndexResult read(VectorScan.Plan plan) {
+ List<IndexVectorSearchSplit> indexSplits = new ArrayList<>();
+ List<RawVectorSearchSplit> rawSplits = new ArrayList<>();
+ splitSearchSplits(plan.splits(), indexSplits, rawSplits);
+ if (indexSplits.isEmpty() && rawSplits.isEmpty()) {
return GlobalIndexResult.createEmpty();
}
- int parallelism =
- Math.max(1,
table.coreOptions().toConfiguration().get(GLOBAL_INDEX_THREAD_NUM));
+ GlobalIndexer globalIndexer =
+ !indexSplits.isEmpty() && !rawSplits.isEmpty()
+ ? createGlobalIndexer(indexSplits)
+ : null;
+ ScoredGlobalIndexResult result =
+ indexSplits.isEmpty()
+ ? ScoredGlobalIndexResult.createEmpty()
+ : readIndexSplitsInSpark(indexSplits, globalIndexer);
+ return result.or(readRawSplitsInSpark(rawSplits, globalIndexer,
rawPreFilter(rawSplits)))
+ .topK(limit);
+ }
+
+ protected ScoredGlobalIndexResult readIndexSplitsInSpark(
+ List<IndexVectorSearchSplit> splits, @Nullable GlobalIndexer
globalIndexer) {
+ if (splits.isEmpty()) {
+ return ScoredGlobalIndexResult.createEmpty();
+ }
+
+ int parallelism = sparkParallelism();
if (splits.size() < parallelism * 2) {
- return super.read(splits);
+ return readIndexed(
+ splits, globalIndexer == null ?
createGlobalIndexer(splits) : globalIndexer);
}
- RoaringNavigableMap64 preFilter = preFilter(splits).orElse(null);
+ List<RoaringNavigableMap64> preFilters = preFilters(splits);
String indexType = splits.get(0).vectorIndexFiles().get(0).indexType();
- List<byte[]> splitBytes = new ArrayList<>(splits.size());
- for (VectorSearchSplit split : splits) {
+ List<SerializedSplit> serializedSplits = new
ArrayList<>(splits.size());
+ for (int i = 0; i < splits.size(); i++) {
try {
- splitBytes.add(InstantiationUtil.serializeObject(split));
+ IndexVectorSearchSplit split = splits.get(i);
+ RoaringNavigableMap64 preFilter = preFilters.isEmpty() ? null
: preFilters.get(i);
+ serializedSplits.add(
+ new SerializedSplit(
+ InstantiationUtil.serializeObject(split),
+ preFilter == null
+ ? null
+ :
InstantiationUtil.serializeObject(preFilter)));
} catch (IOException e) {
throw new RuntimeException("Failed to serialize
VectorSearchSplit", e);
}
}
- List<List<byte[]>> splitGroups = splitGroups(splitBytes, parallelism);
- SparkEngineContext engineContext = new SparkEngineContext();
- Broadcast<RoaringNavigableMap64> preFilterBroadcast =
- preFilter == null ? null : engineContext.broadcast(preFilter);
-
- SerializableFunction<List<byte[]>, byte[]> task =
+ List<List<SerializedSplit>> splitGroups =
splitGroups(serializedSplits, parallelism);
+ SerializableFunction<List<SerializedSplit>, byte[]> task =
group -> {
- GlobalIndexer globalIndexer =
+ GlobalIndexer taskGlobalIndexer =
GlobalIndexerFactoryUtils.load(indexType)
.create(vectorColumn,
table.coreOptions().toConfiguration());
IndexPathFactory indexPathFactory =
table.store().pathFactory().globalIndexFileFactory();
- RoaringNavigableMap64 includeRowIds =
- preFilterBroadcast == null ? null :
preFilterBroadcast.value();
ExecutorService executor =
GlobalIndexReadThreadPool.getExecutorService(
Math.min(parallelism, group.size()));
List<CompletableFuture<Optional<ScoredGlobalIndexResult>>>
futures =
new ArrayList<>(group.size());
- for (byte[] bytes : group) {
- VectorSearchSplit split = deserializeSplit(bytes);
+ for (SerializedSplit serializedSplit : group) {
+ IndexVectorSearchSplit split =
deserializeSplit(serializedSplit.split);
futures.add(
eval(
- globalIndexer,
+ taskGlobalIndexer,
indexPathFactory,
split.rowRangeStart(),
split.rowRangeEnd(),
split.vectorIndexFiles(),
vector,
- includeRowIds,
+
deserializePreFilter(serializedSplit.preFilter),
executor));
}
CompletableFuture.allOf(futures.toArray(new
CompletableFuture[0])).join();
@@ -139,15 +166,130 @@ public class SparkVectorReadImpl extends VectorReadImpl {
}
};
- List<byte[]> remoteResults;
- try {
- remoteResults = engineContext.map(splitGroups, task,
splitGroups.size());
- } finally {
- if (preFilterBroadcast != null) {
- preFilterBroadcast.unpersist(false);
+ List<byte[]> remoteResults = mapInSpark(splitGroups, task,
splitGroups.size());
+
+ return mergeRemoteResults(remoteResults);
+ }
+
+ protected ScoredGlobalIndexResult readRawSplitsInSpark(
+ List<RawVectorSearchSplit> splits,
+ @Nullable GlobalIndexer globalIndexer,
+ @Nullable RoaringNavigableMap64 preFilter) {
+ List<Range> rawRowRanges = rawRowRanges(splits);
+ if (rawRowRanges.isEmpty()) {
+ return ScoredGlobalIndexResult.createEmpty();
+ }
+
+ int parallelism = sparkParallelism();
+ if (rawRowCount(rawRowRanges) < parallelism * 2L) {
+ return readRawSearch(
+ rawRowRanges, preFilter, rawSearchIndexer(splits,
globalIndexer), vector);
+ }
+
+ String metric = rawSearchMetric(rawSearchIndexer(splits,
globalIndexer));
+ List<List<Range>> rangeGroups = rangeGroups(rawRowRanges, parallelism);
+ List<SerializedSplit> serializedSplits = new
ArrayList<>(rangeGroups.size());
+ for (List<Range> rangeGroup : rangeGroups) {
+ try {
+ serializedSplits.add(
+ new SerializedSplit(
+ InstantiationUtil.serializeObject(rangeGroup),
+ preFilter == null
+ ? null
+ :
InstantiationUtil.serializeObject(preFilter)));
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to serialize raw vector row
ranges", e);
}
}
+ List<List<SerializedSplit>> splitGroups =
splitGroups(serializedSplits, parallelism);
+ SerializableFunction<List<SerializedSplit>, byte[]> task =
+ group -> {
+ ScoredGlobalIndexResult result =
ScoredGlobalIndexResult.createEmpty();
+ for (SerializedSplit serializedSplit : group) {
+ List<Range> rowRanges =
deserializeRanges(serializedSplit.split);
+ ScoredGlobalIndexResult splitResult =
+ readRawSearch(
+ rowRanges,
+
deserializePreFilter(serializedSplit.preFilter),
+ metric,
+ vector);
+ result = result.or(splitResult);
+ }
+ result = result.topK(limit);
+ if (result.results().isEmpty()) {
+ return null;
+ }
+ try {
+ return new
GlobalIndexResultSerializer().serialize(result);
+ } catch (IOException e) {
+ throw new RuntimeException(
+ "Failed to serialize ScoredGlobalIndexResult",
e);
+ }
+ };
+
+ List<byte[]> remoteResults = mapInSpark(splitGroups, task,
splitGroups.size());
+ return mergeRemoteResults(remoteResults);
+ }
+
+ protected int sparkParallelism() {
+ return Math.max(1,
table.coreOptions().toConfiguration().get(GLOBAL_INDEX_THREAD_NUM));
+ }
+
+ protected SparkEngineContext createEngineContext() {
+ return new SparkEngineContext();
+ }
+
+ protected <I, O> List<O> mapInSpark(
+ List<I> data, SerializableFunction<I, O> func, int parallelism) {
+ return createEngineContext().map(data, func, parallelism);
+ }
+
+ private IndexVectorSearchSplit deserializeSplit(byte[] bytes) {
+ try {
+ return InstantiationUtil.deserializeObject(
+ bytes, Thread.currentThread().getContextClassLoader());
+ } catch (IOException | ClassNotFoundException e) {
+ throw new RuntimeException("Failed to deserialize
VectorSearchSplit", e);
+ }
+ }
+
+ private List<Range> deserializeRanges(byte[] bytes) {
+ try {
+ return InstantiationUtil.deserializeObject(
+ bytes, Thread.currentThread().getContextClassLoader());
+ } catch (IOException | ClassNotFoundException e) {
+ throw new RuntimeException("Failed to deserialize raw vector row
ranges", e);
+ }
+ }
+
+ @Nullable
+ private RoaringNavigableMap64 deserializePreFilter(@Nullable byte[] bytes)
{
+ if (bytes == null) {
+ return null;
+ }
+ try {
+ return InstantiationUtil.deserializeObject(
+ bytes, Thread.currentThread().getContextClassLoader());
+ } catch (IOException | ClassNotFoundException e) {
+ throw new RuntimeException("Failed to deserialize vector
pre-filter", e);
+ }
+ }
+
+ private List<List<SerializedSplit>> splitGroups(
+ List<SerializedSplit> serializedSplits, int parallelism) {
+ List<List<SerializedSplit>> groups = new ArrayList<>(parallelism);
+ int groupSize = (serializedSplits.size() + parallelism - 1) /
parallelism;
+ for (int start = 0; start < serializedSplits.size(); start +=
groupSize) {
+ groups.add(
+ new ArrayList<>(
+ serializedSplits.subList(
+ start, Math.min(start + groupSize,
serializedSplits.size()))));
+ }
+ return groups;
+ }
+
+ private ScoredGlobalIndexResult mergeRemoteResults(List<byte[]>
remoteResults) {
ScoredGlobalIndexResult result = ScoredGlobalIndexResult.createEmpty();
GlobalIndexResultSerializer serializer = new
GlobalIndexResultSerializer();
for (byte[] bytes : remoteResults) {
@@ -162,24 +304,58 @@ public class SparkVectorReadImpl extends VectorReadImpl {
return result.topK(limit);
}
- private VectorSearchSplit deserializeSplit(byte[] bytes) {
- try {
- return InstantiationUtil.deserializeObject(
- bytes, Thread.currentThread().getContextClassLoader());
- } catch (IOException | ClassNotFoundException e) {
- throw new RuntimeException("Failed to deserialize
VectorSearchSplit", e);
+ private List<List<Range>> rangeGroups(List<Range> ranges, int parallelism)
{
+ long rowCount = rawRowCount(ranges);
+ int groupCount = (int) Math.min(parallelism, rowCount);
+ long targetRowsPerGroup = (rowCount - 1) / groupCount + 1;
+
+ List<List<Range>> groups = new ArrayList<>(groupCount);
+ List<Range> currentGroup = new ArrayList<>();
+ long currentRows = 0;
+ for (Range range : ranges) {
+ long from = range.from;
+ while (from <= range.to) {
+ if (currentRows == targetRowsPerGroup) {
+ groups.add(currentGroup);
+ currentGroup = new ArrayList<>();
+ currentRows = 0;
+ }
+ long remainingGroupRows = targetRowsPerGroup - currentRows;
+ long to = Math.min(range.to, from + remainingGroupRows - 1);
+ Range next = new Range(from, to);
+ currentGroup.add(next);
+ currentRows += next.count();
+ from = to + 1;
+ }
+ }
+ if (!currentGroup.isEmpty()) {
+ groups.add(currentGroup);
}
+ return groups;
}
- private List<List<byte[]>> splitGroups(List<byte[]> splitBytes, int
parallelism) {
- List<List<byte[]>> groups = new ArrayList<>(parallelism);
- int groupSize = (splitBytes.size() + parallelism - 1) / parallelism;
- for (int start = 0; start < splitBytes.size(); start += groupSize) {
- groups.add(
- new ArrayList<>(
- splitBytes.subList(
- start, Math.min(start + groupSize,
splitBytes.size()))));
+ private long rawRowCount(List<Range> ranges) {
+ long rowCount = 0;
+ for (Range range : ranges) {
+ long count = range.count();
+ if (Long.MAX_VALUE - rowCount < count) {
+ return Long.MAX_VALUE;
+ }
+ rowCount += count;
+ }
+ return rowCount;
+ }
+
+ private static class SerializedSplit implements java.io.Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final byte[] split;
+ @Nullable private final byte[] preFilter;
+
+ private SerializedSplit(byte[] split, @Nullable byte[] preFilter) {
+ this.split = split;
+ this.preFilter = preFilter;
}
- return groups;
}
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/read/SparkVectorSearchBuilderImpl.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/read/SparkVectorSearchBuilderImpl.java
index be8d3d8cad..8704486258 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/read/SparkVectorSearchBuilderImpl.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/read/SparkVectorSearchBuilderImpl.java
@@ -38,6 +38,7 @@ public class SparkVectorSearchBuilderImpl extends
VectorSearchBuilderImpl {
@Override
public VectorRead newVectorRead() {
- return new SparkVectorReadImpl(table, filter, limit, vectorColumn,
vector, options);
+ return new SparkVectorReadImpl(
+ table, partitionFilter, filter, limit, vectorColumn, vector,
options);
}
}
diff --git
a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/read/SparkVectorReadImplTest.java
b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/read/SparkVectorReadImplTest.java
new file mode 100644
index 0000000000..c9a3e330a0
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/read/SparkVectorReadImplTest.java
@@ -0,0 +1,172 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.spark.read;
+
+import org.apache.paimon.globalindex.GlobalIndexResult;
+import org.apache.paimon.globalindex.GlobalIndexer;
+import org.apache.paimon.globalindex.ScoredGlobalIndexResult;
+import org.apache.paimon.table.source.IndexVectorSearchSplit;
+import org.apache.paimon.table.source.RawVectorSearchSplit;
+import org.apache.paimon.table.source.VectorScan;
+import org.apache.paimon.table.source.VectorSearchSplit;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.utils.Range;
+import org.apache.paimon.utils.RoaringNavigableMap64;
+
+import org.junit.jupiter.api.Test;
+
+import javax.annotation.Nullable;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link SparkVectorReadImpl}. */
+public class SparkVectorReadImplTest {
+
+ @Test
+ public void testRawSearchUsesSparkPath() {
+ TestingSparkVectorRead read = new TestingSparkVectorRead();
+ RawVectorSearchSplit rawSplit =
+ new RawVectorSearchSplit(
+ Collections.singletonList(new Range(42, 42)),
+ Collections.emptyList(),
+ null);
+ VectorScan.Plan plan = () -> Collections.singletonList(rawSplit);
+
+ GlobalIndexResult result = read.read(plan);
+
+ assertThat(read.rawSparkPathUsed).isTrue();
+ assertThat(result.results().contains(42L)).isTrue();
+ }
+
+ @Test
+ public void testRawSearchSplitsRangesAcrossSparkTasks() {
+ RecordingSparkVectorRead read = new RecordingSparkVectorRead();
+ RawVectorSearchSplit rawSplit =
+ new RawVectorSearchSplit(
+ Collections.singletonList(new Range(0, 63)),
Collections.emptyList(), null);
+
+ ScoredGlobalIndexResult result =
+ read.readRawSplitsInSpark(Collections.singletonList(rawSplit),
null, null);
+
+ assertThat(read.rawSearchRanges).containsExactly(new Range(0, 31), new
Range(32, 63));
+ assertThat(read.sparkParallelism).isEqualTo(2);
+ assertThat(result.results().getLongCardinality()).isEqualTo(64);
+ }
+
+ private static class TestingSparkVectorRead extends SparkVectorReadImpl {
+
+ private boolean rawSparkPathUsed;
+
+ private TestingSparkVectorRead() {
+ super(
+ null,
+ null,
+ null,
+ 10,
+ new DataField(0, "vec", new ArrayType(DataTypes.FLOAT())),
+ new float[] {1.0f},
+ null);
+ }
+
+ @Override
+ protected GlobalIndexResult readSplits(List<? extends
VectorSearchSplit> splits) {
+ throw new AssertionError("Raw search should not fall back to local
vector read.");
+ }
+
+ @Override
+ protected ScoredGlobalIndexResult readIndexSplitsInSpark(
+ List<IndexVectorSearchSplit> splits, @Nullable GlobalIndexer
globalIndexer) {
+ throw new AssertionError("Index search is not part of this test.");
+ }
+
+ @Override
+ protected ScoredGlobalIndexResult readRawSplitsInSpark(
+ List<RawVectorSearchSplit> splits,
+ @Nullable GlobalIndexer globalIndexer,
+ @Nullable RoaringNavigableMap64 preFilter) {
+ rawSparkPathUsed = true;
+ assertThat(splits).hasSize(1);
+ assertThat(globalIndexer).isNull();
+ assertThat(preFilter).isNull();
+
+ RoaringNavigableMap64 rows = new RoaringNavigableMap64();
+ rows.add(42L);
+ return ScoredGlobalIndexResult.create(rows, rowId -> 1.0f);
+ }
+ }
+
+ private static class RecordingSparkVectorRead extends SparkVectorReadImpl {
+
+ private final AtomicInteger nextTask = new AtomicInteger();
+ private final List<Range> rawSearchRanges =
+ Collections.synchronizedList(new java.util.ArrayList<>());
+ private int sparkParallelism;
+
+ private RecordingSparkVectorRead() {
+ super(
+ null,
+ null,
+ null,
+ 100,
+ new DataField(0, "vec", new ArrayType(DataTypes.FLOAT())),
+ new float[] {1.0f},
+ Collections.singletonMap("test.vector.metric", "l2"));
+ }
+
+ @Override
+ protected int sparkParallelism() {
+ return 2;
+ }
+
+ @Override
+ protected <I, O> List<O> mapInSpark(
+ List<I> data,
+ org.apache.paimon.utils.SerializableFunction<I, O> func,
+ int parallelism) {
+ sparkParallelism = parallelism;
+ return data.stream().map(func::apply).collect(Collectors.toList());
+ }
+
+ @Override
+ protected ScoredGlobalIndexResult readRawSearch(
+ List<Range> rawRowRanges,
+ @Nullable RoaringNavigableMap64 preFilter,
+ String metric,
+ float[] queryVector) {
+ assertThat(preFilter).isNull();
+ assertThat(metric).isEqualTo("l2");
+ assertThat(queryVector).containsExactly(1.0f);
+ rawSearchRanges.addAll(rawRowRanges);
+
+ RoaringNavigableMap64 rows = new RoaringNavigableMap64();
+ int scoreBase = nextTask.getAndIncrement();
+ for (Range range : rawRowRanges) {
+ rows.addRange(range);
+ }
+ return ScoredGlobalIndexResult.create(rows, rowId -> scoreBase +
(float) rowId);
+ }
+ }
+}
diff --git
a/paimon-vector/src/main/java/org/apache/paimon/vector/index/NativeVectorGlobalIndexer.java
b/paimon-vector/src/main/java/org/apache/paimon/vector/index/NativeVectorGlobalIndexer.java
index 7095b5bd2e..f45a97d34a 100644
---
a/paimon-vector/src/main/java/org/apache/paimon/vector/index/NativeVectorGlobalIndexer.java
+++
b/paimon-vector/src/main/java/org/apache/paimon/vector/index/NativeVectorGlobalIndexer.java
@@ -21,7 +21,7 @@ package org.apache.paimon.vector.index;
import org.apache.paimon.globalindex.GlobalIndexIOMeta;
import org.apache.paimon.globalindex.GlobalIndexReader;
import org.apache.paimon.globalindex.GlobalIndexWriter;
-import org.apache.paimon.globalindex.GlobalIndexer;
+import org.apache.paimon.globalindex.VectorGlobalIndexer;
import org.apache.paimon.globalindex.io.GlobalIndexFileReader;
import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
import org.apache.paimon.types.DataType;
@@ -32,7 +32,9 @@ import java.util.Objects;
import java.util.concurrent.ExecutorService;
/** Native vector global indexer backed by paimon-vector-index-java. */
-public class NativeVectorGlobalIndexer implements GlobalIndexer {
+public class NativeVectorGlobalIndexer implements VectorGlobalIndexer {
+
+ private static final String DEFAULT_METRIC = "inner_product";
private final DataType fieldType;
private final Map<String, String> options;
@@ -57,4 +59,9 @@ public class NativeVectorGlobalIndexer implements
GlobalIndexer {
ExecutorService executor) {
return new NativeVectorGlobalIndexReader(fileReader, files, fieldType,
executor);
}
+
+ @Override
+ public String metric() {
+ return options.getOrDefault("metric", DEFAULT_METRIC);
+ }
}
