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 a81d129702 [core][spark] Support multi-vector search with rankers
(#8258)
a81d129702 is described below
commit a81d12970293ea2386bb8c6c1b8a177062fe7467
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Jun 17 20:07:30 2026 +0800
[core][spark] Support multi-vector search with rankers (#8258)
Add first-phase multi-vector search support across Spark SQL and the
Java builder APIs. The implementation fans out to existing per-column
global vector indexes, ranks scored row ids with a configured ranker,
and then continues with the normal Paimon scan, so it does not require
any index format changes.
Spark SQL example:
```sql
SELECT id, __paimon_vector_search_score
FROM multi_vector_search(
'T',
array(
named_struct(
'vector_column', 'title_vec',
'query_vector', array(1.0f, 0.0f),
'limit', 20,
'weight', 2.0f,
'options', map('ivf.nprobe', '32')),
named_struct(
'vector_column', 'body_vec',
'query_vector', array(0.0f, 1.0f),
'limit', 20,
'weight', 1.0f,
'options', map('ivf.nprobe', '16'))),
10,
'rrf')
```
Java builder example:
```java
MultiVectorSearchBuilder searchBuilder =
table.newMultiVectorSearchBuilder()
.addRoute(
"title_vec",
new float[] {1.0f, 0.0f},
20,
2.0f,
java.util.Collections.singletonMap("ivf.nprobe",
"32"))
.addRoute(
"body_vec",
new float[] {0.0f, 1.0f},
20,
1.0f,
java.util.Collections.singletonMap("ivf.nprobe",
"16"))
.withLimit(10)
.withRrfRanker();
GlobalIndexResult result = searchBuilder.executeLocal();
ReadBuilder readBuilder = table.newReadBuilder();
TableScan.Plan plan =
readBuilder.newScan().withGlobalIndexResult(result).plan();
```
---
docs/docs/multimodal-table/global-index.mdx | 114 ++++++++++++
.../globalindex/MultiVectorSearchRanker.java | 192 +++++++++++++++++++++
.../apache/paimon/predicate/MultiVectorSearch.java | 70 ++++++++
.../paimon/predicate/MultiVectorSearchRoute.java | 159 +++++++++++++++++
.../org/apache/paimon/predicate/VectorSearch.java | 7 +-
.../globalindex/MultiVectorSearchRankerTest.java | 72 ++++++++
.../java/org/apache/paimon/table/FormatTable.java | 7 +
.../java/org/apache/paimon/table/InnerTable.java | 7 +
.../paimon/table/MultiVectorSearchTable.java | 103 +++++++++++
.../main/java/org/apache/paimon/table/Table.java | 7 +
.../table/source/MultiVectorSearchBuilder.java | 139 +++++++++++++++
.../table/source/MultiVectorSearchBuilderImpl.java | 160 +++++++++++++++++
.../table/source/VectorSearchBuilderTest.java | 118 +++++++++----
.../paimon/format/lance/LanceVectorSearchTest.java | 19 +-
.../scala/org/apache/paimon/spark/PaimonScan.scala | 3 +-
.../apache/paimon/spark/PaimonScanBuilder.scala | 17 +-
.../scala/org/apache/paimon/spark/PaimonScan.scala | 3 +-
.../read/SparkMultiVectorSearchBuilderImpl.java | 51 ++++++
.../org/apache/paimon/spark/PaimonBaseScan.scala | 32 +++-
.../scala/org/apache/paimon/spark/PaimonScan.scala | 3 +-
.../apache/paimon/spark/PaimonScanBuilder.scala | 17 +-
.../apache/paimon/spark/PaimonSparkTableBase.scala | 2 +-
.../plans/logical/PaimonTableValuedFunctions.scala | 169 +++++++++++++++++-
.../org/apache/paimon/spark/read/BaseScan.scala | 4 +-
.../plans/logical/VectorSearchQueryTest.scala | 99 ++++++++++-
.../paimon/spark/sql/MultiVectorSearchTest.scala | 85 +++++++++
26 files changed, 1578 insertions(+), 81 deletions(-)
diff --git a/docs/docs/multimodal-table/global-index.mdx
b/docs/docs/multimodal-table/global-index.mdx
index a7698f124a..7b9712a8f9 100644
--- a/docs/docs/multimodal-table/global-index.mdx
+++ b/docs/docs/multimodal-table/global-index.mdx
@@ -229,6 +229,120 @@ try (RecordReader<InternalRow> reader =
readBuilder.newRead().createReader(plan)
}
```
+For Java, use `Table.newVectorSearchBuilder()` to produce a global index
result, then pass
+the result to `TableScan.withGlobalIndexResult`.
+
+</TabItem>
+
+</Tabs>
+
+## Multi-Vector Search
+
+Multi-vector search queries multiple vector columns in one request and ranks
the scored results
+before reading table rows. This is useful when one table stores several
embeddings for the same
+record, such as title, body, image, or audio embeddings.
+
+Before running multi-vector search, create a vector global index for every
vector column used by
+the query:
+
+```sql
+CALL sys.create_global_index(
+ table => 'db.my_table',
+ index_column => 'title_embedding',
+ index_type => 'ivf-pq',
+ options => 'ivf-pq.distance.metric=cosine,ivf-pq.nlist=256,ivf-pq.pq.m=16'
+);
+
+CALL sys.create_global_index(
+ table => 'db.my_table',
+ index_column => 'body_embedding',
+ index_type => 'ivf-pq',
+ options => 'ivf-pq.distance.metric=cosine,ivf-pq.nlist=256,ivf-pq.pq.m=16'
+);
+```
+
+For Spark SQL, use the `multi_vector_search(table_name, routes, limit[,
ranker])`
+table-valued function. The third argument is the final number of ranked
results to return.
+The optional fourth argument selects the ranker. Supported rankers are `rrf`
and
+`weighted_score`; the default is `rrf`.
+
+The second argument is an array of route configs created by `named_struct`:
+
+| Field | Required | Default | Description |
+|---|---|---|---|
+| `vector_column` | Yes | N/A | Vector column to search. |
+| `query_vector` | Yes | N/A | Query vector for this route. |
+| `limit` | No | Final limit | Top K results to retrieve from this vector
column before ranking. |
+| `weight` | No | `1.0` | Weight for this route when ranking results. |
+| `options` | No | Empty map | Route-specific vector search options such as
`ivf.nprobe` and `hnsw.ef_search`. |
+
+When using an array of route configs, every `named_struct` should use the same
fields because
+Spark requires array elements to have the same struct type.
+
+<Tabs groupId="multi-vector-search">
+
+<TabItem value="spark-sql" label="Spark SQL">
+
+```sql
+SELECT id, __paimon_vector_search_score
+FROM multi_vector_search(
+ 'my_table',
+ array(
+ named_struct(
+ 'vector_column', 'title_embedding',
+ 'query_vector', array(1.0f, 0.0f, 0.0f),
+ 'limit', 50,
+ 'weight', 2.0f,
+ 'options', map('ivf.nprobe', '32')),
+ named_struct(
+ 'vector_column', 'body_embedding',
+ 'query_vector', array(0.0f, 1.0f, 0.0f),
+ 'limit', 50,
+ 'weight', 1.0f,
+ 'options', map('ivf.nprobe', '16'))),
+ 10,
+ 'rrf');
+```
+
+Spark SQL adds `__paimon_vector_search_score` to expose the ranked score.
+
+</TabItem>
+
+<TabItem value="java-api" label="Java API">
+
+```java
+Table table = catalog.getTable(identifier);
+
+MultiVectorSearchBuilder searchBuilder =
+ table.newMultiVectorSearchBuilder()
+ .addRoute(
+ "title_embedding",
+ new float[] {1.0f, 0.0f, 0.0f},
+ 50,
+ 2.0f,
+ java.util.Collections.singletonMap("ivf.nprobe", "32"))
+ .addRoute(
+ "body_embedding",
+ new float[] {0.0f, 1.0f, 0.0f},
+ 50,
+ 1.0f,
+ java.util.Collections.singletonMap("ivf.nprobe", "16"))
+ .withLimit(10)
+ .withRrfRanker();
+GlobalIndexResult result = searchBuilder.executeLocal();
+
+ReadBuilder readBuilder = table.newReadBuilder();
+TableScan.Plan plan =
readBuilder.newScan().withGlobalIndexResult(result).plan();
+try (RecordReader<InternalRow> reader =
readBuilder.newRead().createReader(plan)) {
+ reader.forEachRemaining(row -> {
+ System.out.println("id=" + row.getInt(0));
+ });
+}
+```
+
+For Java, use `Table.newMultiVectorSearchBuilder()` to configure routes, final
limit, and ranker
+directly. `VectorSearch` and `MultiVectorSearch` are internal pushdown
representations.
+
</TabItem>
</Tabs>
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/MultiVectorSearchRanker.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/MultiVectorSearchRanker.java
new file mode 100644
index 0000000000..fc29adb1d3
--- /dev/null
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/MultiVectorSearchRanker.java
@@ -0,0 +1,192 @@
+/*
+ * 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.utils.RoaringNavigableMap64;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/** Ranker utilities for multi-vector search results. */
+public class MultiVectorSearchRanker {
+
+ public static final String RRF_RANKER = "rrf";
+ public static final String WEIGHTED_SCORE_RANKER = "weighted_score";
+
+ private static final float RRF_K = 60.0f;
+
+ private MultiVectorSearchRanker() {}
+
+ public static ScoredGlobalIndexResult rank(
+ String ranker, List<ScoredGlobalIndexResult> results, float[]
weights, int limit) {
+ List<WeightedResult> weightedResults = new ArrayList<>(results.size());
+ for (int i = 0; i < results.size(); i++) {
+ weightedResults.add(new WeightedResult(results.get(i),
weightAt(weights, i)));
+ }
+ return rank(ranker, weightedResults, limit);
+ }
+
+ public static ScoredGlobalIndexResult rank(
+ String ranker, List<WeightedResult> results, int limit) {
+ if (WEIGHTED_SCORE_RANKER.equals(normalizeRanker(ranker))) {
+ return weightedScore(results, limit);
+ }
+ return rrf(results, limit);
+ }
+
+ public static String normalizeRanker(String ranker) {
+ if (ranker == null || ranker.trim().isEmpty()) {
+ return RRF_RANKER;
+ }
+ String normalized = ranker.trim().toLowerCase();
+ if (!RRF_RANKER.equals(normalized) &&
!WEIGHTED_SCORE_RANKER.equals(normalized)) {
+ throw new IllegalArgumentException("Unsupported multi-vector
ranker: " + ranker);
+ }
+ return normalized;
+ }
+
+ public static ScoredGlobalIndexResult rrf(
+ List<ScoredGlobalIndexResult> results, float[] weights, int limit)
{
+ List<WeightedResult> weightedResults = new ArrayList<>(results.size());
+ for (int i = 0; i < results.size(); i++) {
+ weightedResults.add(new WeightedResult(results.get(i),
weightAt(weights, i)));
+ }
+ return rrf(weightedResults, limit);
+ }
+
+ public static ScoredGlobalIndexResult rrf(List<WeightedResult> results,
int limit) {
+ Map<Long, Float> scores = new HashMap<>();
+ for (WeightedResult weightedResult : results) {
+ ScoredGlobalIndexResult result = weightedResult.result();
+ float weight = weightedResult.weight();
+ List<Long> ranked = rankedRowIds(result);
+ for (int rank = 0; rank < ranked.size(); rank++) {
+ Long rowId = ranked.get(rank);
+ float contribution = weight / (RRF_K + rank + 1.0f);
+ Float oldScore = scores.get(rowId);
+ scores.put(rowId, oldScore == null ? contribution : oldScore +
contribution);
+ }
+ }
+ return topK(scores, limit);
+ }
+
+ public static ScoredGlobalIndexResult weightedScore(
+ List<ScoredGlobalIndexResult> results, float[] weights, int limit)
{
+ List<WeightedResult> weightedResults = new ArrayList<>(results.size());
+ for (int i = 0; i < results.size(); i++) {
+ weightedResults.add(new WeightedResult(results.get(i),
weightAt(weights, i)));
+ }
+ return weightedScore(weightedResults, limit);
+ }
+
+ public static ScoredGlobalIndexResult weightedScore(List<WeightedResult>
results, int limit) {
+ Map<Long, Float> scores = new HashMap<>();
+ for (WeightedResult weightedResult : results) {
+ ScoredGlobalIndexResult result = weightedResult.result();
+ float weight = weightedResult.weight();
+ ScoreGetter scoreGetter = result.scoreGetter();
+ for (long rowId : result.results()) {
+ float contribution = weight * scoreGetter.score(rowId);
+ Float oldScore = scores.get(rowId);
+ scores.put(rowId, oldScore == null ? contribution : oldScore +
contribution);
+ }
+ }
+ return topK(scores, limit);
+ }
+
+ private static List<Long> rankedRowIds(ScoredGlobalIndexResult result) {
+ List<Long> rowIds = new ArrayList<>();
+ for (long rowId : result.results()) {
+ rowIds.add(rowId);
+ }
+ final ScoreGetter scoreGetter = result.scoreGetter();
+ Collections.sort(
+ rowIds,
+ new Comparator<Long>() {
+ @Override
+ public int compare(Long left, Long right) {
+ return Float.compare(scoreGetter.score(right),
scoreGetter.score(left));
+ }
+ });
+ return rowIds;
+ }
+
+ private static ScoredGlobalIndexResult topK(Map<Long, Float> scores, int
limit) {
+ if (scores.isEmpty()) {
+ return ScoredGlobalIndexResult.createEmpty();
+ }
+ List<Map.Entry<Long, Float>> ranked = new
ArrayList<>(scores.entrySet());
+ Collections.sort(
+ ranked,
+ new Comparator<Map.Entry<Long, Float>>() {
+ @Override
+ public int compare(Map.Entry<Long, Float> left,
Map.Entry<Long, Float> right) {
+ int scoreCompare = Float.compare(right.getValue(),
left.getValue());
+ if (scoreCompare != 0) {
+ return scoreCompare;
+ }
+ return Long.compare(left.getKey(), right.getKey());
+ }
+ });
+
+ int size = Math.min(limit, ranked.size());
+ RoaringNavigableMap64 bitmap = new RoaringNavigableMap64();
+ Map<Long, Float> topScores = new HashMap<>();
+ for (int i = 0; i < size; i++) {
+ Map.Entry<Long, Float> entry = ranked.get(i);
+ bitmap.add(entry.getKey());
+ topScores.put(entry.getKey(), entry.getValue());
+ }
+ return ScoredGlobalIndexResult.create(bitmap, topScores::get);
+ }
+
+ private static float weightAt(float[] weights, int index) {
+ if (weights == null || index >= weights.length) {
+ return 1.0f;
+ }
+ return weights[index];
+ }
+
+ /** Weighted result from one vector-search route. */
+ public static class WeightedResult implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final ScoredGlobalIndexResult result;
+ private final float weight;
+
+ public WeightedResult(ScoredGlobalIndexResult result, float weight) {
+ this.result = result;
+ this.weight = weight;
+ }
+
+ public ScoredGlobalIndexResult result() {
+ return result;
+ }
+
+ public float weight() {
+ return weight;
+ }
+ }
+}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/predicate/MultiVectorSearch.java
b/paimon-common/src/main/java/org/apache/paimon/predicate/MultiVectorSearch.java
new file mode 100644
index 0000000000..0541edf473
--- /dev/null
+++
b/paimon-common/src/main/java/org/apache/paimon/predicate/MultiVectorSearch.java
@@ -0,0 +1,70 @@
+/*
+ * 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.predicate;
+
+import org.apache.paimon.globalindex.MultiVectorSearchRanker;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Multi-vector search over multiple vector columns.
+ *
+ * <p>This is an internal pushdown representation. Use {@code
Table.newMultiVectorSearchBuilder()}
+ * to configure multi-vector search from Java.
+ */
+public class MultiVectorSearch implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final List<MultiVectorSearchRoute> routes;
+ private final int limit;
+ private final String ranker;
+
+ public MultiVectorSearch(List<MultiVectorSearchRoute> routes, int limit,
String ranker) {
+ if (routes == null || routes.isEmpty()) {
+ throw new IllegalArgumentException("Routes cannot be null or
empty");
+ }
+ if (limit <= 0) {
+ throw new IllegalArgumentException("Limit must be positive, got: "
+ limit);
+ }
+ this.routes = Collections.unmodifiableList(new ArrayList<>(routes));
+ this.limit = limit;
+ this.ranker = MultiVectorSearchRanker.normalizeRanker(ranker);
+ }
+
+ public List<MultiVectorSearchRoute> routes() {
+ return routes;
+ }
+
+ public int limit() {
+ return limit;
+ }
+
+ public String ranker() {
+ return ranker;
+ }
+
+ @Override
+ public String toString() {
+ return "Ranker(" + ranker + "), Limit(" + limit + "), Routes(" +
routes + ")";
+ }
+}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/predicate/MultiVectorSearchRoute.java
b/paimon-common/src/main/java/org/apache/paimon/predicate/MultiVectorSearchRoute.java
new file mode 100644
index 0000000000..070828a963
--- /dev/null
+++
b/paimon-common/src/main/java/org/apache/paimon/predicate/MultiVectorSearchRoute.java
@@ -0,0 +1,159 @@
+/*
+ * 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.predicate;
+
+import org.apache.paimon.annotation.Experimental;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+/** A single vector-search route in a multi-vector search. */
+public class MultiVectorSearchRoute implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String fieldName;
+ private final float[] vector;
+ private final int limit;
+ private final float weight;
+ private final Map<String, String> options;
+
+ public MultiVectorSearchRoute(String fieldName, float[] vector, int limit,
float weight) {
+ this(fieldName, vector, limit, weight, Collections.emptyMap());
+ }
+
+ public MultiVectorSearchRoute(
+ String fieldName,
+ float[] vector,
+ int limit,
+ float weight,
+ Map<String, String> options) {
+ if (fieldName == null || fieldName.isEmpty()) {
+ throw new IllegalArgumentException("Field name cannot be null or
empty");
+ }
+ if (vector == null) {
+ throw new IllegalArgumentException("Search vector cannot be null");
+ }
+ if (limit <= 0) {
+ throw new IllegalArgumentException("Limit must be positive, got: "
+ limit);
+ }
+ if (weight <= 0) {
+ throw new IllegalArgumentException("Weight must be positive, got:
" + weight);
+ }
+ this.fieldName = fieldName;
+ this.vector = vector;
+ this.limit = limit;
+ this.weight = weight;
+ this.options =
+ options == null
+ ? Collections.emptyMap()
+ : Collections.unmodifiableMap(new HashMap<>(options));
+ }
+
+ @Experimental
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ public String fieldName() {
+ return fieldName;
+ }
+
+ public float[] vector() {
+ return vector;
+ }
+
+ public int limit() {
+ return limit;
+ }
+
+ public float weight() {
+ return weight;
+ }
+
+ public Map<String, String> options() {
+ return options;
+ }
+
+ public VectorSearch toVectorSearch() {
+ return new VectorSearch(vector, limit, fieldName, options);
+ }
+
+ @Override
+ public String toString() {
+ return "FieldName(" + fieldName + "), Limit(" + limit + "), Weight(" +
weight + ")";
+ }
+
+ /** Builder for {@link MultiVectorSearchRoute}. */
+ @Experimental
+ public static class Builder {
+
+ private String fieldName;
+ private float[] vector;
+ private int limit;
+ private float weight = 1.0f;
+ private Map<String, String> options = new HashMap<>();
+
+ public Builder vectorColumn(String fieldName) {
+ this.fieldName = fieldName;
+ return this;
+ }
+
+ public Builder field(String fieldName) {
+ return vectorColumn(fieldName);
+ }
+
+ public Builder queryVector(float[] vector) {
+ this.vector = vector;
+ return this;
+ }
+
+ public Builder vector(float[] vector) {
+ return queryVector(vector);
+ }
+
+ public Builder limit(int limit) {
+ this.limit = limit;
+ return this;
+ }
+
+ public Builder weight(float weight) {
+ this.weight = weight;
+ return this;
+ }
+
+ public Builder option(String key, String value) {
+ this.options.put(key, value);
+ return this;
+ }
+
+ public Builder options(Map<String, String> options) {
+ if (options != null) {
+ this.options.putAll(options);
+ }
+ return this;
+ }
+
+ public MultiVectorSearchRoute build() {
+ return new MultiVectorSearchRoute(fieldName, vector, limit,
weight, options);
+ }
+ }
+}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/predicate/VectorSearch.java
b/paimon-common/src/main/java/org/apache/paimon/predicate/VectorSearch.java
index c2b608346b..5f046c6307 100644
--- a/paimon-common/src/main/java/org/apache/paimon/predicate/VectorSearch.java
+++ b/paimon-common/src/main/java/org/apache/paimon/predicate/VectorSearch.java
@@ -28,7 +28,12 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
-/** VectorSearch to perform vector similarity search. * */
+/**
+ * VectorSearch to perform vector similarity search.
+ *
+ * <p>This is an internal pushdown representation. Use {@code
Table.newVectorSearchBuilder()} to
+ * configure vector search from Java.
+ */
public class VectorSearch implements Serializable {
private static final long serialVersionUID = 1L;
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/MultiVectorSearchRankerTest.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/MultiVectorSearchRankerTest.java
new file mode 100644
index 0000000000..3aa5be4fd9
--- /dev/null
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/MultiVectorSearchRankerTest.java
@@ -0,0 +1,72 @@
+/*
+ * 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.utils.RoaringNavigableMap64;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.within;
+
+/** Tests for {@link MultiVectorSearchRanker}. */
+public class MultiVectorSearchRankerTest {
+
+ @Test
+ public void testRrfFavorsRowsReturnedByMultipleRoutes() {
+ ScoredGlobalIndexResult first = result(new long[] {1, 2}, new float[]
{0.9f, 0.8f});
+ ScoredGlobalIndexResult second = result(new long[] {2, 3}, new float[]
{0.7f, 0.6f});
+
+ ScoredGlobalIndexResult ranked =
+ MultiVectorSearchRanker.rrf(
+ Arrays.asList(first, second), new float[] {1.0f,
1.0f}, 2);
+
+ assertThat(ranked.results().getIntCardinality()).isEqualTo(2);
+ assertThat(ranked.results()).contains(2L);
+
assertThat(ranked.scoreGetter().score(2L)).isGreaterThan(ranked.scoreGetter().score(1L));
+ }
+
+ @Test
+ public void testWeightedScoreUsesAlignedWeightsAfterEmptyRouteIsSkipped() {
+ ScoredGlobalIndexResult result = result(new long[] {1, 2}, new float[]
{0.3f, 0.2f});
+
+ ScoredGlobalIndexResult ranked =
+ MultiVectorSearchRanker.weightedScore(
+ Collections.singletonList(result), new float[] {3.0f},
1);
+
+ assertThat(ranked.results().getIntCardinality()).isEqualTo(1);
+ assertThat(ranked.results()).contains(1L);
+ assertThat(ranked.scoreGetter().score(1L)).isCloseTo(0.9f,
within(0.000001f));
+ }
+
+ private ScoredGlobalIndexResult result(long[] rowIds, float[] scores) {
+ RoaringNavigableMap64 bitmap = new RoaringNavigableMap64();
+ Map<Long, Float> scoreMap = new HashMap<>();
+ for (int i = 0; i < rowIds.length; i++) {
+ bitmap.add(rowIds[i]);
+ scoreMap.put(rowIds[i], scores[i]);
+ }
+ return ScoredGlobalIndexResult.create(bitmap, scoreMap::get);
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/FormatTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/FormatTable.java
index 4149683d7a..ae23dcd8f1 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/FormatTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/FormatTable.java
@@ -32,6 +32,7 @@ import org.apache.paimon.table.format.FormatReadBuilder;
import org.apache.paimon.table.sink.BatchWriteBuilder;
import org.apache.paimon.table.sink.StreamWriteBuilder;
import org.apache.paimon.table.source.FullTextSearchBuilder;
+import org.apache.paimon.table.source.MultiVectorSearchBuilder;
import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.table.source.VectorSearchBuilder;
import org.apache.paimon.types.RowType;
@@ -280,6 +281,12 @@ public interface FormatTable extends Table {
throw new UnsupportedOperationException("FormatTable does not
support vector search.");
}
+ @Override
+ public MultiVectorSearchBuilder newMultiVectorSearchBuilder() {
+ throw new UnsupportedOperationException(
+ "FormatTable does not support multi-vector search.");
+ }
+
@Override
public FullTextSearchBuilder newFullTextSearchBuilder() {
throw new UnsupportedOperationException(
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/InnerTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/InnerTable.java
index d360597744..fd863d643f 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/InnerTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/InnerTable.java
@@ -29,6 +29,8 @@ import org.apache.paimon.table.source.FullTextSearchBuilder;
import org.apache.paimon.table.source.FullTextSearchBuilderImpl;
import org.apache.paimon.table.source.InnerTableRead;
import org.apache.paimon.table.source.InnerTableScan;
+import org.apache.paimon.table.source.MultiVectorSearchBuilder;
+import org.apache.paimon.table.source.MultiVectorSearchBuilderImpl;
import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.table.source.ReadBuilderImpl;
import org.apache.paimon.table.source.StreamDataTableScan;
@@ -62,6 +64,11 @@ public interface InnerTable extends Table {
return new VectorSearchBuilderImpl(this);
}
+ @Override
+ default MultiVectorSearchBuilder newMultiVectorSearchBuilder() {
+ return new MultiVectorSearchBuilderImpl(this);
+ }
+
@Override
default FullTextSearchBuilder newFullTextSearchBuilder() {
return new FullTextSearchBuilderImpl(this);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/MultiVectorSearchTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/MultiVectorSearchTable.java
new file mode 100644
index 0000000000..a1c0322ba8
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/table/MultiVectorSearchTable.java
@@ -0,0 +1,103 @@
+/*
+ * 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.table;
+
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.predicate.MultiVectorSearch;
+import org.apache.paimon.table.source.InnerTableRead;
+import org.apache.paimon.table.source.InnerTableScan;
+import org.apache.paimon.types.RowType;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * A table wrapper to hold multi-vector search information. This is used to
pass multi-vector search
+ * pushdown information from logical plan optimization to physical plan
execution. For now, it is
+ * only used by internal for Spark engine.
+ */
+public class MultiVectorSearchTable implements ReadonlyTable {
+
+ private final InnerTable origin;
+ private final MultiVectorSearch multiVectorSearch;
+
+ private MultiVectorSearchTable(InnerTable origin, MultiVectorSearch
multiVectorSearch) {
+ this.origin = origin;
+ this.multiVectorSearch = multiVectorSearch;
+ }
+
+ public static MultiVectorSearchTable create(
+ InnerTable origin, MultiVectorSearch multiVectorSearch) {
+ return new MultiVectorSearchTable(origin, multiVectorSearch);
+ }
+
+ public MultiVectorSearch multiVectorSearch() {
+ return multiVectorSearch;
+ }
+
+ public InnerTable origin() {
+ return origin;
+ }
+
+ @Override
+ public String name() {
+ return origin.name();
+ }
+
+ @Override
+ public RowType rowType() {
+ return origin.rowType();
+ }
+
+ @Override
+ public List<String> primaryKeys() {
+ return origin.primaryKeys();
+ }
+
+ @Override
+ public List<String> partitionKeys() {
+ return origin.partitionKeys();
+ }
+
+ @Override
+ public Map<String, String> options() {
+ return origin.options();
+ }
+
+ @Override
+ public FileIO fileIO() {
+ return origin.fileIO();
+ }
+
+ @Override
+ public InnerTableRead newRead() {
+ return origin.newRead();
+ }
+
+ @Override
+ public InnerTableScan newScan() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Table copy(Map<String, String> dynamicOptions) {
+ return new MultiVectorSearchTable(
+ (InnerTable) origin.copy(dynamicOptions), multiVectorSearch);
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/Table.java
b/paimon-core/src/main/java/org/apache/paimon/table/Table.java
index f58259b1fc..aa05ba4660 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/Table.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/Table.java
@@ -29,6 +29,7 @@ import org.apache.paimon.stats.Statistics;
import org.apache.paimon.table.sink.BatchWriteBuilder;
import org.apache.paimon.table.sink.StreamWriteBuilder;
import org.apache.paimon.table.source.FullTextSearchBuilder;
+import org.apache.paimon.table.source.MultiVectorSearchBuilder;
import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.table.source.VectorSearchBuilder;
import org.apache.paimon.types.RowType;
@@ -226,6 +227,12 @@ public interface Table extends Serializable {
/** Returns a new vector search builder. */
VectorSearchBuilder newVectorSearchBuilder();
+ /** Returns a new multi-vector search builder. */
+ default MultiVectorSearchBuilder newMultiVectorSearchBuilder() {
+ throw new UnsupportedOperationException(
+ getClass().getName() + " does not support multi-vector
search.");
+ }
+
/** Returns a new full-text search builder. */
FullTextSearchBuilder newFullTextSearchBuilder();
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/MultiVectorSearchBuilder.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/MultiVectorSearchBuilder.java
new file mode 100644
index 0000000000..b45d8fd2e0
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/MultiVectorSearchBuilder.java
@@ -0,0 +1,139 @@
+/*
+ * 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.table.source;
+
+import org.apache.paimon.globalindex.GlobalIndexResult;
+import org.apache.paimon.globalindex.ScoredGlobalIndexResult;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.predicate.MultiVectorSearchRoute;
+import org.apache.paimon.predicate.Predicate;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/** Builder to build multi-vector search. */
+public interface MultiVectorSearchBuilder extends Serializable {
+
+ /** Push partition filters. */
+ MultiVectorSearchBuilder withPartitionFilter(PartitionPredicate
partitionPredicate);
+
+ /** Push pre-filter for vector search. */
+ MultiVectorSearchBuilder withFilter(Predicate predicate);
+
+ /** Add a vector-search route. */
+ MultiVectorSearchBuilder addRoute(MultiVectorSearchRoute route);
+
+ /** Add a vector-search route. */
+ default MultiVectorSearchBuilder addRoute(String vectorColumn, float[]
vector, int limit) {
+ return addRoute(vectorColumn, vector, limit, 1.0f);
+ }
+
+ /** Add a vector-search route. */
+ default MultiVectorSearchBuilder addRoute(
+ String vectorColumn, float[] vector, int limit, float weight) {
+ return addRoute(new MultiVectorSearchRoute(vectorColumn, vector,
limit, weight));
+ }
+
+ /** Add a vector-search route. */
+ default MultiVectorSearchBuilder addRoute(
+ String vectorColumn,
+ float[] vector,
+ int limit,
+ float weight,
+ Map<String, String> options) {
+ return addRoute(new MultiVectorSearchRoute(vectorColumn, vector,
limit, weight, options));
+ }
+
+ /** The final top k ranked results to return. */
+ MultiVectorSearchBuilder withLimit(int limit);
+
+ /** Ranker for combining route results. */
+ MultiVectorSearchBuilder withRanker(String ranker);
+
+ /** Use reciprocal rank fusion to combine route results. */
+ MultiVectorSearchBuilder withRrfRanker();
+
+ /** Use weighted score to combine route results. */
+ MultiVectorSearchBuilder withWeightedScoreRanker();
+
+ /** Create one vector-search builder for each route so engines can
dispatch route work. */
+ List<Route> routeBuilders();
+
+ /** Convert a vector-search result into a weighted route result. */
+ RouteResult toRouteResult(Route route, GlobalIndexResult result);
+
+ /** Rank route results. */
+ ScoredGlobalIndexResult rank(List<RouteResult> routeResults);
+
+ /** Execute multi-vector index search in local. */
+ default ScoredGlobalIndexResult executeLocal() {
+ List<Route> routes = routeBuilders();
+ List<RouteResult> routeResults = new ArrayList<>(routes.size());
+ for (Route route : routes) {
+ routeResults.add(toRouteResult(route,
route.vectorSearchBuilder().executeLocal()));
+ }
+ return rank(routeResults);
+ }
+
+ /** A route and its configured vector-search builder. */
+ class Route implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final MultiVectorSearchRoute route;
+ private final VectorSearchBuilder vectorSearchBuilder;
+
+ public Route(MultiVectorSearchRoute route, VectorSearchBuilder
vectorSearchBuilder) {
+ this.route = route;
+ this.vectorSearchBuilder = vectorSearchBuilder;
+ }
+
+ public MultiVectorSearchRoute route() {
+ return route;
+ }
+
+ public VectorSearchBuilder vectorSearchBuilder() {
+ return vectorSearchBuilder;
+ }
+ }
+
+ /** A scored result produced by one route. */
+ class RouteResult implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final MultiVectorSearchRoute route;
+ private final ScoredGlobalIndexResult result;
+
+ public RouteResult(MultiVectorSearchRoute route,
ScoredGlobalIndexResult result) {
+ this.route = route;
+ this.result = result;
+ }
+
+ public MultiVectorSearchRoute route() {
+ return route;
+ }
+
+ public ScoredGlobalIndexResult result() {
+ return result;
+ }
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/MultiVectorSearchBuilderImpl.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/MultiVectorSearchBuilderImpl.java
new file mode 100644
index 0000000000..247301301a
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/MultiVectorSearchBuilderImpl.java
@@ -0,0 +1,160 @@
+/*
+ * 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.table.source;
+
+import org.apache.paimon.globalindex.GlobalIndexResult;
+import org.apache.paimon.globalindex.MultiVectorSearchRanker;
+import org.apache.paimon.globalindex.ScoredGlobalIndexResult;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.predicate.MultiVectorSearchRoute;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.table.InnerTable;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+
+/** Implementation for {@link MultiVectorSearchBuilder}. */
+public class MultiVectorSearchBuilderImpl implements MultiVectorSearchBuilder {
+
+ private static final long serialVersionUID = 1L;
+
+ protected final InnerTable table;
+
+ protected final List<MultiVectorSearchRoute> routes = new ArrayList<>();
+ protected int limit;
+ protected String ranker = MultiVectorSearchRanker.RRF_RANKER;
+ protected PartitionPredicate partitionFilter;
+ protected Predicate filter;
+
+ public MultiVectorSearchBuilderImpl(InnerTable table) {
+ this.table = table;
+ }
+
+ @Override
+ public MultiVectorSearchBuilder withPartitionFilter(PartitionPredicate
partitionFilter) {
+ this.partitionFilter = partitionFilter;
+ return this;
+ }
+
+ @Override
+ public MultiVectorSearchBuilder withFilter(Predicate predicate) {
+ if (this.filter == null) {
+ this.filter = predicate;
+ } else {
+ this.filter = PredicateBuilder.and(this.filter, predicate);
+ }
+ return this;
+ }
+
+ @Override
+ public MultiVectorSearchBuilder addRoute(MultiVectorSearchRoute route) {
+ this.routes.add(Objects.requireNonNull(route, "Route cannot be null"));
+ return this;
+ }
+
+ @Override
+ public MultiVectorSearchBuilder withLimit(int limit) {
+ this.limit = limit;
+ return this;
+ }
+
+ @Override
+ public MultiVectorSearchBuilder withRanker(String ranker) {
+ this.ranker = MultiVectorSearchRanker.normalizeRanker(ranker);
+ return this;
+ }
+
+ @Override
+ public MultiVectorSearchBuilder withRrfRanker() {
+ return withRanker(MultiVectorSearchRanker.RRF_RANKER);
+ }
+
+ @Override
+ public MultiVectorSearchBuilder withWeightedScoreRanker() {
+ return withRanker(MultiVectorSearchRanker.WEIGHTED_SCORE_RANKER);
+ }
+
+ @Override
+ public List<Route> routeBuilders() {
+ validateSearch();
+
+ List<Route> routeBuilders = new ArrayList<>(routes.size());
+ for (MultiVectorSearchRoute route : routes) {
+ VectorSearchBuilder vectorSearchBuilder =
newVectorSearchBuilder(route);
+ routeBuilders.add(new Route(route, vectorSearchBuilder));
+ }
+ return routeBuilders;
+ }
+
+ private void validateSearch() {
+ if (routes.isEmpty()) {
+ throw new IllegalArgumentException("Routes cannot be empty");
+ }
+ if (limit <= 0) {
+ throw new IllegalArgumentException("Limit must be positive, got: "
+ limit);
+ }
+ }
+
+ @Override
+ public ScoredGlobalIndexResult rank(List<RouteResult> routeResults) {
+ validateSearch();
+
+ List<MultiVectorSearchRanker.WeightedResult> weightedResults =
+ new ArrayList<>(routeResults.size());
+ for (RouteResult routeResult : routeResults) {
+ if (!routeResult.result().results().isEmpty()) {
+ weightedResults.add(
+ new MultiVectorSearchRanker.WeightedResult(
+ routeResult.result(),
routeResult.route().weight()));
+ }
+ }
+ return MultiVectorSearchRanker.rank(ranker, weightedResults, limit);
+ }
+
+ @Override
+ public RouteResult toRouteResult(Route route, GlobalIndexResult result) {
+ if (result instanceof ScoredGlobalIndexResult) {
+ return new RouteResult(route.route(), (ScoredGlobalIndexResult)
result);
+ } else if (result.results().isEmpty()) {
+ return new RouteResult(route.route(),
ScoredGlobalIndexResult.createEmpty());
+ } else {
+ throw new UnsupportedOperationException(
+ "Multi-vector search requires scored vector index results,
but got: "
+ + result.getClass().getName());
+ }
+ }
+
+ protected VectorSearchBuilder
newVectorSearchBuilder(MultiVectorSearchRoute route) {
+ VectorSearchBuilder vectorSearchBuilder =
+ table.newVectorSearchBuilder()
+ .withVector(route.vector())
+ .withVectorColumn(route.fieldName())
+ .withLimit(route.limit())
+ .withOptions(route.options());
+ if (partitionFilter != null) {
+ vectorSearchBuilder.withPartitionFilter(partitionFilter);
+ }
+ if (filter != null) {
+ vectorSearchBuilder.withFilter(filter);
+ }
+ return vectorSearchBuilder;
+ }
+}
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 17ff21bb61..39379c66c1 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
@@ -74,15 +74,65 @@ public class VectorSearchBuilderTest extends TableTestBase {
@Override
protected Schema schemaDefault() {
- return Schema.newBuilder()
- .column("id", DataTypes.INT())
- .column(VECTOR_FIELD_NAME, new ArrayType(DataTypes.FLOAT()))
- .option(CoreOptions.BUCKET.key(), "-1")
+ return vectorSchemaBuilder(VECTOR_FIELD_NAME).build();
+ }
+
+ protected Schema.Builder vectorSchemaBuilder(String vectorFieldName) {
+ return withVectorSchemaOptions(
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column(vectorFieldName, new
ArrayType(DataTypes.FLOAT())));
+ }
+
+ protected Schema.Builder multiVectorSchemaBuilder() {
+ return withVectorSchemaOptions(
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("title_vec", new ArrayType(DataTypes.FLOAT()))
+ .column("body_vec", new ArrayType(DataTypes.FLOAT())));
+ }
+
+ protected Schema.Builder withVectorSchemaOptions(Schema.Builder builder) {
+ return builder.option(CoreOptions.BUCKET.key(), "-1")
.option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
.option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true")
.option("test.vector.dimension", String.valueOf(DIMENSION))
- .option("test.vector.metric", "l2")
- .build();
+ .option("test.vector.metric", "l2");
+ }
+
+ @Test
+ public void testMultiVectorSearchBuilderExposesRouteBuilders() throws
Exception {
+ catalog.createTable(
+ identifier("multi_vector_builder_table"),
+ multiVectorSchemaBuilder().build(),
+ false);
+ FileStoreTable table =
getTable(identifier("multi_vector_builder_table"));
+
+ float[][] titleVectors = {{1.0f, 0.0f}, {0.9f, 0.1f}, {0.0f, 1.0f}};
+ float[][] bodyVectors = {{0.0f, 1.0f}, {0.1f, 0.9f}, {1.0f, 0.0f}};
+ writeTwoVectorColumns(table, titleVectors, bodyVectors);
+ buildAndCommitIndex(table, "title_vec", titleVectors);
+ buildAndCommitIndex(table, "body_vec", bodyVectors);
+
+ MultiVectorSearchBuilder builder =
+ table.newMultiVectorSearchBuilder()
+ .addRoute("title_vec", new float[] {1.0f, 0.0f}, 2)
+ .addRoute("body_vec", new float[] {0.0f, 1.0f}, 2,
2.0f)
+ .withLimit(2)
+ .withWeightedScoreRanker();
+ List<MultiVectorSearchBuilder.Route> routes = builder.routeBuilders();
+
+ assertThat(routes).hasSize(2);
+
+ List<MultiVectorSearchBuilder.RouteResult> routeResults = new
ArrayList<>();
+ for (MultiVectorSearchBuilder.Route route : routes) {
+ routeResults.add(
+ builder.toRouteResult(route,
route.vectorSearchBuilder().executeLocal()));
+ }
+ ScoredGlobalIndexResult ranked = builder.rank(routeResults);
+
+ assertThat(ranked.results().getIntCardinality()).isEqualTo(2);
+ assertThat(ranked.results()).contains(1L);
}
@Test
@@ -133,13 +183,7 @@ public class VectorSearchBuilderTest extends TableTestBase
{
// Create a table with cosine metric
catalog.createTable(
identifier("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("test.vector.dimension",
String.valueOf(DIMENSION))
+ vectorSchemaBuilder(VECTOR_FIELD_NAME)
.option("test.vector.metric", "cosine")
.build(),
false);
@@ -231,14 +275,7 @@ public class VectorSearchBuilderTest extends TableTestBase
{
public void testVectorSearchThreadsOptions() throws Exception {
catalog.createTable(
identifier("options_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("test.vector.dimension",
String.valueOf(DIMENSION))
- .option("test.vector.metric", "l2")
+ vectorSchemaBuilder(VECTOR_FIELD_NAME)
.option("test.vector.required-option.key",
"ivf.nprobe")
.option("test.vector.required-option.value", "16")
.build(),
@@ -311,16 +348,12 @@ public class VectorSearchBuilderTest extends
TableTestBase {
// Create a partitioned table
catalog.createTable(
identifier("partitioned_table"),
- Schema.newBuilder()
- .column("pt", DataTypes.INT())
- .column("id", DataTypes.INT())
- .column(VECTOR_FIELD_NAME, new
ArrayType(DataTypes.FLOAT()))
- .partitionKeys("pt")
- .option(CoreOptions.BUCKET.key(), "-1")
- .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
- .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(),
"true")
- .option("test.vector.dimension",
String.valueOf(DIMENSION))
- .option("test.vector.metric", "l2")
+ withVectorSchemaOptions(
+ Schema.newBuilder()
+ .column("pt", DataTypes.INT())
+ .column("id", DataTypes.INT())
+ .column(VECTOR_FIELD_NAME, new
ArrayType(DataTypes.FLOAT()))
+ .partitionKeys("pt"))
.build(),
false);
FileStoreTable table = getTable(identifier("partitioned_table"));
@@ -553,9 +586,30 @@ public class VectorSearchBuilderTest extends TableTestBase
{
}
}
+ private void writeTwoVectorColumns(
+ FileStoreTable table, float[][] titleVectors, float[][]
bodyVectors) throws Exception {
+ BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = writeBuilder.newWrite();
+ BatchTableCommit commit = writeBuilder.newCommit()) {
+ for (int i = 0; i < titleVectors.length; i++) {
+ write.write(
+ GenericRow.of(
+ i,
+ new GenericArray(titleVectors[i]),
+ new GenericArray(bodyVectors[i])));
+ }
+ commit.commit(write.prepareCommit());
+ }
+ }
+
private void buildAndCommitIndex(FileStoreTable table, float[][] vectors)
throws Exception {
+ buildAndCommitIndex(table, VECTOR_FIELD_NAME, vectors);
+ }
+
+ private void buildAndCommitIndex(FileStoreTable table, String fieldName,
float[][] vectors)
+ throws Exception {
Options options = table.coreOptions().toConfiguration();
- DataField vectorField = table.rowType().getField(VECTOR_FIELD_NAME);
+ DataField vectorField = table.rowType().getField(fieldName);
GlobalIndexSingletonWriter writer =
(GlobalIndexSingletonWriter)
diff --git
a/paimon-lance/src/test/java/org/apache/paimon/format/lance/LanceVectorSearchTest.java
b/paimon-lance/src/test/java/org/apache/paimon/format/lance/LanceVectorSearchTest.java
index a8107977bb..d02f55db8e 100644
---
a/paimon-lance/src/test/java/org/apache/paimon/format/lance/LanceVectorSearchTest.java
+++
b/paimon-lance/src/test/java/org/apache/paimon/format/lance/LanceVectorSearchTest.java
@@ -26,8 +26,6 @@ import org.apache.paimon.fs.Path;
import org.apache.paimon.options.Options;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.table.source.VectorSearchBuilderTest;
-import org.apache.paimon.types.ArrayType;
-import org.apache.paimon.types.DataTypes;
import org.apache.paimon.utils.TraceableFileIO;
import org.junit.jupiter.api.BeforeEach;
@@ -52,16 +50,13 @@ public class LanceVectorSearchTest extends
VectorSearchBuilderTest {
@Override
protected Schema schemaDefault() {
- return Schema.newBuilder()
- .column("id", DataTypes.INT())
- .column("vec", 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.FILE_FORMAT.key(), "lance")
- .option("test.vector.dimension", "2")
- .option("test.vector.metric", "l2")
- .build();
+ return vectorSchemaBuilder("vec").build();
+ }
+
+ @Override
+ protected Schema.Builder withVectorSchemaOptions(Schema.Builder builder) {
+ return super.withVectorSchemaOptions(builder)
+ .option(CoreOptions.FILE_FORMAT.key(), "lance");
}
@Disabled("Cosine metric uses Tantivy index which requires Hadoop
dependencies")
diff --git
a/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
b/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
index 9af6c8f8b5..ec652fad3b 100644
---
a/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
+++
b/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
@@ -19,7 +19,7 @@
package org.apache.paimon.spark
import org.apache.paimon.partition.PartitionPredicate
-import org.apache.paimon.predicate.{FullTextSearch, Predicate, TopN,
VectorSearch}
+import org.apache.paimon.predicate.{FullTextSearch, MultiVectorSearch,
Predicate, TopN, VectorSearch}
import org.apache.paimon.spark.read.VariantExtractionInfo
import org.apache.paimon.table.InnerTable
@@ -33,6 +33,7 @@ case class PaimonScan(
override val pushedLimit: Option[Int] = None,
override val pushedTopN: Option[TopN] = None,
override val pushedVectorSearch: Option[VectorSearch] = None,
+ override val pushedMultiVectorSearch: Option[MultiVectorSearch] = None,
override val pushedFullTextSearch: Option[FullTextSearch] = None,
override val pushedVariantExtractions: Map[Seq[String],
Seq[VariantExtractionInfo]] = Map.empty,
bucketedScanDisabled: Boolean = true)
diff --git
a/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/PaimonScanBuilder.scala
b/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/PaimonScanBuilder.scala
index ef1f68c09f..daa1a56122 100644
---
a/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/PaimonScanBuilder.scala
+++
b/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/PaimonScanBuilder.scala
@@ -25,18 +25,14 @@ import org.apache.spark.sql.connector.read.Scan
class PaimonScanBuilder(val table: InnerTable) extends PaimonBaseScanBuilder {
override def build(): Scan = {
- val (actualTable, vectorSearch, fullTextSearch) = table match {
+ val (actualTable, vectorSearch, multiVectorSearch, fullTextSearch) = table
match {
case vst: org.apache.paimon.table.VectorSearchTable =>
- val tableVectorSearch = Option(vst.vectorSearch())
- val vs = (tableVectorSearch, pushedVectorSearch) match {
- case (Some(_), _) => tableVectorSearch
- case (None, Some(_)) => pushedVectorSearch
- case (None, None) => None
- }
- (vst.origin(), vs, None)
+ (vst.origin(), Option(vst.vectorSearch()), None, None)
+ case mvst: org.apache.paimon.table.MultiVectorSearchTable =>
+ (mvst.origin(), None, Option(mvst.multiVectorSearch()), None)
case ftst: org.apache.paimon.table.FullTextSearchTable =>
- (ftst.origin(), None, Option(ftst.fullTextSearch()))
- case _ => (table, pushedVectorSearch, pushedFullTextSearch)
+ (ftst.origin(), None, None, Option(ftst.fullTextSearch()))
+ case _ => (table, pushedVectorSearch, None, pushedFullTextSearch)
}
PaimonScan(
actualTable,
@@ -46,6 +42,7 @@ class PaimonScanBuilder(val table: InnerTable) extends
PaimonBaseScanBuilder {
pushedLimit,
pushedTopN,
vectorSearch,
+ multiVectorSearch,
fullTextSearch)
}
}
diff --git
a/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
b/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
index 9bedc9bdf1..c35589075b 100644
---
a/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
+++
b/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
@@ -19,7 +19,7 @@
package org.apache.paimon.spark
import org.apache.paimon.partition.PartitionPredicate
-import org.apache.paimon.predicate.{FullTextSearch, Predicate, TopN,
VectorSearch}
+import org.apache.paimon.predicate.{FullTextSearch, MultiVectorSearch,
Predicate, TopN, VectorSearch}
import org.apache.paimon.spark.read.VariantExtractionInfo
import org.apache.paimon.table.{BucketMode, FileStoreTable, InnerTable}
import org.apache.paimon.table.source.{DataSplit, Split}
@@ -37,6 +37,7 @@ case class PaimonScan(
override val pushedLimit: Option[Int],
override val pushedTopN: Option[TopN],
override val pushedVectorSearch: Option[VectorSearch],
+ override val pushedMultiVectorSearch: Option[MultiVectorSearch] = None,
override val pushedFullTextSearch: Option[FullTextSearch] = None,
override val pushedVariantExtractions: Map[Seq[String],
Seq[VariantExtractionInfo]] = Map.empty,
bucketedScanDisabled: Boolean = false)
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/read/SparkMultiVectorSearchBuilderImpl.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/read/SparkMultiVectorSearchBuilderImpl.java
new file mode 100644
index 0000000000..cc42783094
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/read/SparkMultiVectorSearchBuilderImpl.java
@@ -0,0 +1,51 @@
+/*
+ * 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.predicate.MultiVectorSearchRoute;
+import org.apache.paimon.table.InnerTable;
+import org.apache.paimon.table.source.MultiVectorSearchBuilderImpl;
+import org.apache.paimon.table.source.VectorSearchBuilder;
+
+/** Spark-aware {@link MultiVectorSearchBuilderImpl}. */
+public class SparkMultiVectorSearchBuilderImpl extends
MultiVectorSearchBuilderImpl {
+
+ private static final long serialVersionUID = 1L;
+
+ public SparkMultiVectorSearchBuilderImpl(InnerTable table) {
+ super(table);
+ }
+
+ @Override
+ protected VectorSearchBuilder
newVectorSearchBuilder(MultiVectorSearchRoute route) {
+ VectorSearchBuilder vectorSearchBuilder =
+ new SparkVectorSearchBuilderImpl(table)
+ .withVector(route.vector())
+ .withVectorColumn(route.fieldName())
+ .withLimit(route.limit())
+ .withOptions(route.options());
+ if (partitionFilter != null) {
+ vectorSearchBuilder.withPartitionFilter(partitionFilter);
+ }
+ if (filter != null) {
+ vectorSearchBuilder.withFilter(filter);
+ }
+ return vectorSearchBuilder;
+ }
+}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScan.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScan.scala
index d7f6cd23d3..274c5b841c 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScan.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScan.scala
@@ -23,7 +23,7 @@ import org.apache.paimon.globalindex.GlobalIndexResult
import org.apache.paimon.partition.PartitionPredicate
import org.apache.paimon.predicate.PredicateBuilder
import org.apache.paimon.spark.metric.SparkMetricRegistry
-import org.apache.paimon.spark.read.{BaseScan, BatchReadTagCleanupListener,
PaimonSupportsRuntimeFiltering, SparkVectorSearchBuilderImpl}
+import org.apache.paimon.spark.read.{BaseScan, BatchReadTagCleanupListener,
PaimonSupportsRuntimeFiltering, SparkMultiVectorSearchBuilderImpl,
SparkVectorSearchBuilderImpl}
import org.apache.paimon.spark.sources.PaimonMicroBatchStream
import org.apache.paimon.spark.util.OptionUtils
import org.apache.paimon.table.{DataTable, FileStoreTable, InnerTable}
@@ -64,13 +64,18 @@ abstract class PaimonBaseScan(table: InnerTable)
}
private def evalGlobalIndexSearch(): GlobalIndexResult = {
- if (pushedVectorSearch.isDefined && pushedFullTextSearch.isDefined) {
+ val globalSearchCount =
+ Seq(pushedVectorSearch, pushedMultiVectorSearch,
pushedFullTextSearch).count(_.isDefined)
+ if (globalSearchCount > 1) {
throw new UnsupportedOperationException(
- "Cannot push down both vector search and full-text search
simultaneously.")
+ "Cannot push down vector search, multi-vector search and full-text
search simultaneously.")
}
if (pushedVectorSearch.isDefined) {
return evalVectorSearch()
}
+ if (pushedMultiVectorSearch.isDefined) {
+ return evalMultiVectorSearch()
+ }
if (pushedFullTextSearch.isDefined) {
return evalFullTextSearch()
}
@@ -99,6 +104,27 @@ abstract class PaimonBaseScan(table: InnerTable)
vectorBuilder.newVectorRead().read(vectorBuilder.newVectorScan().scan())
}
+ private def evalMultiVectorSearch(): GlobalIndexResult = {
+ val multiVectorSearch = pushedMultiVectorSearch.get
+ val multiVectorSearchBuilder =
+ if (CoreOptions.fromMap(table.options).vectorSearchDistributeEnabled()) {
+ new SparkMultiVectorSearchBuilderImpl(table)
+ } else {
+ table.newMultiVectorSearchBuilder()
+ }
+ val builder = multiVectorSearchBuilder
+ .withLimit(multiVectorSearch.limit())
+ .withRanker(multiVectorSearch.ranker())
+ multiVectorSearch.routes().asScala.foreach(route =>
builder.addRoute(route))
+ if (pushedPartitionFilters.nonEmpty) {
+
builder.withPartitionFilter(PartitionPredicate.and(pushedPartitionFilters.asJava))
+ }
+ if (pushedDataFilters.nonEmpty) {
+ builder.withFilter(PredicateBuilder.and(pushedDataFilters.asJava))
+ }
+ builder.executeLocal()
+ }
+
private def evalFullTextSearch(): GlobalIndexResult = {
val fullTextSearch = pushedFullTextSearch.get
val ftBuilder = table
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
index a2b04c3ac4..09ece0bfea 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScan.scala
@@ -20,7 +20,7 @@ package org.apache.paimon.spark
import org.apache.paimon.CoreOptions.BucketFunctionType
import org.apache.paimon.partition.PartitionPredicate
-import org.apache.paimon.predicate.{FullTextSearch, Predicate, TopN,
VectorSearch}
+import org.apache.paimon.predicate.{FullTextSearch, MultiVectorSearch,
Predicate, TopN, VectorSearch}
import org.apache.paimon.spark.commands.BucketExpression.quote
import org.apache.paimon.spark.read.VariantExtractionInfo
import org.apache.paimon.table.{BucketMode, FileStoreTable, InnerTable}
@@ -43,6 +43,7 @@ case class PaimonScan(
override val pushedLimit: Option[Int],
override val pushedTopN: Option[TopN],
override val pushedVectorSearch: Option[VectorSearch],
+ override val pushedMultiVectorSearch: Option[MultiVectorSearch] = None,
override val pushedFullTextSearch: Option[FullTextSearch] = None,
override val pushedVariantExtractions: Map[Seq[String],
Seq[VariantExtractionInfo]] = Map.empty,
bucketedScanDisabled: Boolean = false)
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScanBuilder.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScanBuilder.scala
index e6f98363a0..9e2e853ee2 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScanBuilder.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonScanBuilder.scala
@@ -129,18 +129,14 @@ class PaimonScanBuilder(val table: InnerTable)
localScan match {
case Some(scan) => scan
case None =>
- val (actualTable, vectorSearch, fullTextSearch) = table match {
+ val (actualTable, vectorSearch, multiVectorSearch, fullTextSearch) =
table match {
case vst: org.apache.paimon.table.VectorSearchTable =>
- val tableVectorSearch = Option(vst.vectorSearch())
- val vs = (tableVectorSearch, pushedVectorSearch) match {
- case (Some(_), _) => tableVectorSearch
- case (None, Some(_)) => pushedVectorSearch
- case (None, None) => None
- }
- (vst.origin(), vs, None)
+ (vst.origin(), Option(vst.vectorSearch()), None, None)
+ case mvst: org.apache.paimon.table.MultiVectorSearchTable =>
+ (mvst.origin(), None, Option(mvst.multiVectorSearch()), None)
case ftst: org.apache.paimon.table.FullTextSearchTable =>
- (ftst.origin(), None, Option(ftst.fullTextSearch()))
- case _ => (table, pushedVectorSearch, pushedFullTextSearch)
+ (ftst.origin(), None, None, Option(ftst.fullTextSearch()))
+ case _ => (table, pushedVectorSearch, None, pushedFullTextSearch)
}
PaimonScan(
@@ -151,6 +147,7 @@ class PaimonScanBuilder(val table: InnerTable)
pushedLimit,
pushedTopN,
vectorSearch,
+ multiVectorSearch,
fullTextSearch,
acceptedVariantExtractions
)
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSparkTableBase.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSparkTableBase.scala
index 94b9128444..084174a823 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSparkTableBase.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonSparkTableBase.scala
@@ -118,7 +118,7 @@ abstract class PaimonSparkTableBase(val table: Table)
_metadataColumns.append(PaimonMetadataColumn.ROW_ID)
_metadataColumns.append(PaimonMetadataColumn.SEQUENCE_NUMBER)
}
- if (table.isInstanceOf[VectorSearchTable]) {
+ if (table.isInstanceOf[VectorSearchTable] ||
table.isInstanceOf[MultiVectorSearchTable]) {
_metadataColumns.append(PaimonMetadataColumn.VECTOR_SEARCH_SCORE)
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonTableValuedFunctions.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonTableValuedFunctions.scala
index 6dc0c77ced..7efd67960e 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonTableValuedFunctions.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonTableValuedFunctions.scala
@@ -19,10 +19,11 @@
package org.apache.paimon.spark.catalyst.plans.logical
import org.apache.paimon.CoreOptions
-import org.apache.paimon.predicate.{FullTextSearch, VectorSearch}
+import org.apache.paimon.globalindex.MultiVectorSearchRanker
+import org.apache.paimon.predicate.{FullTextSearch, MultiVectorSearch,
MultiVectorSearchRoute, VectorSearch}
import org.apache.paimon.spark.SparkTable
import
org.apache.paimon.spark.catalyst.plans.logical.PaimonTableValuedFunctions._
-import org.apache.paimon.table.{DataTable, FullTextSearchTable, InnerTable,
VectorSearchTable}
+import org.apache.paimon.table.{DataTable, FullTextSearchTable, InnerTable,
MultiVectorSearchTable, VectorSearchTable}
import
org.apache.paimon.table.source.snapshot.TimeTravelUtil.InconsistentTagBucketException
import org.apache.spark.sql.PaimonUtils.createDataset
@@ -30,7 +31,7 @@ import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.FunctionIdentifier
import org.apache.spark.sql.catalyst.analysis.FunctionRegistryBase
import
org.apache.spark.sql.catalyst.analysis.TableFunctionRegistry.TableFunctionBuilder
-import org.apache.spark.sql.catalyst.expressions.{Attribute, CreateArray,
CreateMap, Expression, ExpressionInfo, Literal}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, CreateArray,
CreateMap, CreateNamedStruct, Expression, ExpressionInfo, Literal}
import org.apache.spark.sql.catalyst.plans.logical.{LeafNode, LogicalPlan}
import org.apache.spark.sql.catalyst.util.MapData
import org.apache.spark.sql.connector.catalog.{Identifier, Table, TableCatalog}
@@ -46,6 +47,7 @@ object PaimonTableValuedFunctions {
val INCREMENTAL_BETWEEN_TIMESTAMP = "paimon_incremental_between_timestamp"
val INCREMENTAL_TO_AUTO_TAG = "paimon_incremental_to_auto_tag"
val VECTOR_SEARCH = "vector_search"
+ val MULTI_VECTOR_SEARCH = "multi_vector_search"
val FULL_TEXT_SEARCH = "full_text_search"
val supportedFnNames: Seq[String] =
@@ -54,6 +56,7 @@ object PaimonTableValuedFunctions {
INCREMENTAL_BETWEEN_TIMESTAMP,
INCREMENTAL_TO_AUTO_TAG,
VECTOR_SEARCH,
+ MULTI_VECTOR_SEARCH,
FULL_TEXT_SEARCH)
def parsePositiveLimit(value: Any): Int = {
@@ -85,6 +88,8 @@ object PaimonTableValuedFunctions {
FunctionRegistryBase.build[IncrementalToAutoTag](fnName, since = None)
case VECTOR_SEARCH =>
FunctionRegistryBase.build[VectorSearchQuery](fnName, since = None)
+ case MULTI_VECTOR_SEARCH =>
+ FunctionRegistryBase.build[MultiVectorSearchQuery](fnName, since =
None)
case FULL_TEXT_SEARCH =>
FunctionRegistryBase.build[FullTextSearchQuery](fnName, since = None)
case _ =>
@@ -121,6 +126,8 @@ object PaimonTableValuedFunctions {
tvf match {
case vsq: VectorSearchQuery =>
resolveVectorSearchQuery(sparkTable, sparkCatalog, ident, vsq,
args.tail)
+ case mvsq: MultiVectorSearchQuery =>
+ resolveMultiVectorSearchQuery(sparkTable, sparkCatalog, ident, mvsq,
args.tail)
case ftsq: FullTextSearchQuery =>
resolveFullTextSearchQuery(sparkTable, sparkCatalog, ident, ftsq,
args.tail)
case _ =>
@@ -160,6 +167,28 @@ object PaimonTableValuedFunctions {
}
}
+ private def resolveMultiVectorSearchQuery(
+ sparkTable: Table,
+ sparkCatalog: TableCatalog,
+ ident: Identifier,
+ mvsq: MultiVectorSearchQuery,
+ argsWithoutTable: Seq[Expression]): LogicalPlan = {
+ sparkTable match {
+ case st @ SparkTable(innerTable: InnerTable) =>
+ val multiVectorSearch = mvsq.createMultiVectorSearch(innerTable,
argsWithoutTable)
+ val multiVectorSearchTable = MultiVectorSearchTable.create(innerTable,
multiVectorSearch)
+ DataSourceV2Relation.create(
+ st.copy(table = multiVectorSearchTable),
+ Some(sparkCatalog),
+ Some(ident),
+ CaseInsensitiveStringMap.empty())
+ case _ =>
+ throw new RuntimeException(
+ "multi_vector_search only supports Paimon SparkTable backed by
InnerTable, " +
+ s"but got table implementation: ${sparkTable.getClass.getName}")
+ }
+ }
+
private def resolveFullTextSearchQuery(
sparkTable: Table,
sparkCatalog: TableCatalog,
@@ -338,7 +367,7 @@ case class VectorSearchQuery(override val args:
Seq[Expression])
new VectorSearch(queryVector, limit, columnName, options.asJava)
}
- private def extractQueryVector(expr: Expression): Array[Float] = {
+ def extractQueryVector(expr: Expression): Array[Float] = {
expr match {
case Literal(arrayData, _) if arrayData != null =>
val arr =
arrayData.asInstanceOf[org.apache.spark.sql.catalyst.util.ArrayData]
@@ -356,7 +385,7 @@ case class VectorSearchQuery(override val args:
Seq[Expression])
}
}
- private def extractOptions(expr: Expression): Map[String, String] = {
+ def extractOptions(expr: Expression): Map[String, String] = {
expr match {
case CreateMap(children, _) if children != null =>
children
@@ -410,7 +439,7 @@ case class VectorSearchQuery(override val args:
Seq[Expression])
}
}
- private def extractString(expr: Expression): String =
stringValue(expr.eval())
+ def extractString(expr: Expression): String = stringValue(expr.eval())
private def stringValue(value: Any): String = {
if (value == null) {
@@ -420,6 +449,134 @@ case class VectorSearchQuery(override val args:
Seq[Expression])
}
}
+/**
+ * Plan for the [[MULTI_VECTOR_SEARCH]] table-valued function.
+ *
+ * Usage: multi_vector_search(table_name, routes, limit[, ranker])
+ * - table_name: the Paimon table to search
+ * - routes: route config array with vector_column, query_vector, limit,
weight, and options
+ * fields
+ * - limit: the final number of ranked top results to return
+ * - ranker: optional ranker for combining results from multiple vector
columns
+ */
+case class MultiVectorSearchQuery(override val args: Seq[Expression])
+ extends PaimonTableValueFunction(MULTI_VECTOR_SEARCH) {
+
+ override def parseArgs(args: Seq[Expression]): Map[String, String] = {
+ Map.empty
+ }
+
+ def createMultiVectorSearch(
+ innerTable: InnerTable,
+ argsWithoutTable: Seq[Expression]): MultiVectorSearch = {
+ if (argsWithoutTable.size != 2 && argsWithoutTable.size != 3) {
+ throw new RuntimeException(
+ s"$MULTI_VECTOR_SEARCH needs two or three parameters after table_name:
" +
+ s"routes, limit[, ranker]. " +
+ s"Got ${argsWithoutTable.size} parameters after table_name.")
+ }
+ val finalLimit = parsePositiveLimit(argsWithoutTable(1).eval())
+ val ranker =
+ if (argsWithoutTable.size == 3) {
+ VectorSearchQuery(Seq.empty).extractString(argsWithoutTable(2))
+ } else {
+ MultiVectorSearchRanker.RRF_RANKER
+ }
+
+ val routes = extractRoutes(argsWithoutTable.head, finalLimit).map {
+ route =>
+ val columnName = route.fieldName()
+ if (!innerTable.rowType().containsField(columnName)) {
+ throw new RuntimeException(
+ s"Column $columnName does not exist in table ${innerTable.name()}")
+ }
+ route
+ }.toList
+
+ new MultiVectorSearch(routes.asJava, finalLimit, ranker)
+ }
+
+ private def extractRoutes(expr: Expression, defaultLimit: Int):
Seq[MultiVectorSearchRoute] = {
+ expr match {
+ case CreateArray(elements, _) if elements != null =>
+ elements.map(extractRoute(_, defaultLimit))
+ case _ =>
+ throw new RuntimeException(s"Cannot extract multi-vector routes from
expression: $expr")
+ }
+ }
+
+ private def extractRoute(expr: Expression, defaultLimit: Int):
MultiVectorSearchRoute = {
+ expr match {
+ case CreateNamedStruct(children) if children != null =>
+ extractConfiguredRoute(children, defaultLimit)
+ case _ =>
+ throw new RuntimeException(s"Cannot extract multi-vector route from
expression: $expr")
+ }
+ }
+
+ private def extractConfiguredRoute(
+ children: Seq[Expression],
+ defaultLimit: Int): MultiVectorSearchRoute = {
+ var columnName: Option[String] = None
+ var queryVector: Option[Array[Float]] = None
+ var limit: Option[Int] = None
+ var weight: Option[Float] = None
+ var options = Map.empty[String, String]
+
+ children.grouped(2).foreach {
+ case Seq(keyExpr, valueExpr) =>
+ VectorSearchQuery(Seq.empty).extractString(keyExpr) match {
+ case "vector_column" =>
+ columnName =
Some(VectorSearchQuery(Seq.empty).extractString(valueExpr))
+ case "query_vector" =>
+ queryVector =
Some(VectorSearchQuery(Seq.empty).extractQueryVector(valueExpr))
+ case "limit" =>
+ limit = Some(parsePositiveLimit(valueExpr.eval()))
+ case "weight" =>
+ weight = Some(parsePositiveFloat(valueExpr.eval(), "weight"))
+ case "options" =>
+ options = VectorSearchQuery(Seq.empty).extractOptions(valueExpr)
+ case key =>
+ throw new IllegalArgumentException(
+ s"Unsupported multi-vector route field '$key'. " +
+ "Supported fields are vector_column, query_vector, limit,
weight, and options.")
+ }
+ case other =>
+ throw new RuntimeException(s"Invalid route config entries: $other")
+ }
+
+ val routeColumn =
+ columnName.getOrElse(
+ throw new IllegalArgumentException("Multi-vector route must define
vector_column."))
+ new MultiVectorSearchRoute(
+ routeColumn,
+ queryVector.getOrElse(
+ throw new IllegalArgumentException(
+ s"Multi-vector route for column $routeColumn must define
query_vector.")),
+ limit.getOrElse(defaultLimit),
+ weight.getOrElse(1.0f),
+ options.asJava
+ )
+ }
+
+ private def parsePositiveFloat(value: Any, name: String): Float = {
+ val parsed = value match {
+ case f: Float => f
+ case d: Double => d.toFloat
+ case i: Int => i.toFloat
+ case l: Long => l.toFloat
+ case s: String => s.toFloat
+ case u: UTF8String => u.toString.toFloat
+ case other => throw new RuntimeException(s"Invalid $name type:
${other.getClass.getName}")
+ }
+ if (parsed <= 0) {
+ throw new IllegalArgumentException(s"$name must be positive, but got:
$parsed")
+ }
+ parsed
+ }
+
+}
+
/**
* Plan for the [[FULL_TEXT_SEARCH]] table-valued function.
*
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/BaseScan.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/BaseScan.scala
index 33b5d4e176..f06807ec44 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/BaseScan.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/BaseScan.scala
@@ -20,7 +20,7 @@ package org.apache.paimon.spark.read
import org.apache.paimon.CoreOptions
import org.apache.paimon.partition.PartitionPredicate
-import org.apache.paimon.predicate.{FullTextSearch, Predicate, TopN,
VectorSearch}
+import org.apache.paimon.predicate.{FullTextSearch, MultiVectorSearch,
Predicate, TopN, VectorSearch}
import org.apache.paimon.spark.{PaimonBatch, PaimonInputPartition,
PaimonNumSplitMetric, PaimonPartitionSizeMetric, PaimonReadBatchTimeMetric,
PaimonResultedTableFilesMetric, PaimonResultedTableFilesTaskMetric,
SparkTypeUtils}
import org.apache.paimon.spark.schema.PaimonMetadataColumn
import org.apache.paimon.spark.schema.PaimonMetadataColumn._
@@ -51,6 +51,7 @@ trait BaseScan extends Scan with SupportsReportStatistics
with Logging {
def pushedLimit: Option[Int] = None
def pushedTopN: Option[TopN] = None
def pushedVectorSearch: Option[VectorSearch] = None
+ def pushedMultiVectorSearch: Option[MultiVectorSearch] = None
def pushedFullTextSearch: Option[FullTextSearch] = None
def pushedVariantExtractions: Map[Seq[String], Seq[VariantExtractionInfo]] =
Map.empty
@@ -204,6 +205,7 @@ trait BaseScan extends Scan with SupportsReportStatistics
with Logging {
pushedTopN.map(topN => s", TopN: [$topN]").getOrElse("") +
pushedLimit.map(limit => s", Limit: [$limit]").getOrElse("") +
pushedVectorSearch.map(vs => s", VectorSearch: [$vs]").getOrElse("") +
+ pushedMultiVectorSearch.map(mvs => s", MultiVectorSearch:
[$mvs]").getOrElse("") +
pushedFullTextSearch.map(fts => s", FullTextSearch:
[$fts]").getOrElse("") +
pushedVariantsStr
}
diff --git
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/catalyst/plans/logical/VectorSearchQueryTest.scala
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/catalyst/plans/logical/VectorSearchQueryTest.scala
index d3f1b8d265..12775ea540 100644
---
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/catalyst/plans/logical/VectorSearchQueryTest.scala
+++
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/catalyst/plans/logical/VectorSearchQueryTest.scala
@@ -21,7 +21,7 @@ package org.apache.paimon.spark.catalyst.plans.logical
import org.apache.paimon.table.InnerTable
import org.apache.paimon.types.{ArrayType, DataType, DataTypes, RowType}
-import org.apache.spark.sql.catalyst.expressions.{CreateArray, CreateMap,
Expression, Literal}
+import org.apache.spark.sql.catalyst.expressions.{CreateArray, CreateMap,
CreateNamedStruct, Expression, Literal}
import org.scalatest.funsuite.AnyFunSuite
import java.lang.reflect.{InvocationHandler, Method, Proxy}
@@ -29,6 +29,95 @@ import java.lang.reflect.{InvocationHandler, Method, Proxy}
/** Tests for [[VectorSearchQuery]]. */
class VectorSearchQueryTest extends AnyFunSuite {
+ test("create multi vector search with route configs") {
+ val search = MultiVectorSearchQuery(Seq.empty).createMultiVectorSearch(
+ innerTable,
+ Seq(
+ CreateArray(
+ Seq(
+ CreateNamedStruct(Seq(
+ Literal("vector_column"),
+ Literal("title_vec"),
+ Literal("query_vector"),
+ CreateArray(Seq(Literal(1.0f), Literal(0.0f))),
+ Literal("limit"),
+ Literal(20),
+ Literal("weight"),
+ Literal(2.0f),
+ Literal("options"),
+ CreateMap(Seq(Literal("ivf.nprobe"), Literal("32")))
+ )),
+ CreateNamedStruct(Seq(
+ Literal("vector_column"),
+ Literal("body_vec"),
+ Literal("query_vector"),
+ CreateArray(Seq(Literal(0.0f), Literal(1.0f))),
+ Literal("limit"),
+ Literal(10),
+ Literal("weight"),
+ Literal(1.0f),
+ Literal("options"),
+ CreateMap(Seq(Literal("ivf.nprobe"), Literal("16")))
+ ))
+ )),
+ Literal(3),
+ Literal("weighted_score")
+ )
+ )
+
+ assert(search.ranker() == "weighted_score")
+ assert(search.routes().size() == 2)
+ assert(search.routes().get(0).limit() == 20)
+ assert(search.routes().get(0).weight() == 2.0f)
+ assert(search.routes().get(0).options().get("ivf.nprobe") == "32")
+ assert(search.routes().get(1).limit() == 10)
+ assert(search.routes().get(1).weight() == 1.0f)
+ assert(search.routes().get(1).options().get("ivf.nprobe") == "16")
+ }
+
+ test("default multi vector route limit to final limit") {
+ val search = MultiVectorSearchQuery(Seq.empty).createMultiVectorSearch(
+ innerTable,
+ Seq(
+ CreateArray(
+ Seq(
+ CreateNamedStruct(
+ Seq(
+ Literal("vector_column"),
+ Literal("title_vec"),
+ Literal("query_vector"),
+ CreateArray(Seq(Literal(1.0f), Literal(0.0f)))
+ )))),
+ Literal(7)
+ )
+ )
+
+ assert(search.limit() == 7)
+ assert(search.ranker() == "rrf")
+ assert(search.routes().get(0).limit() == 7)
+ assert(search.routes().get(0).weight() == 1.0f)
+ assert(search.routes().get(0).options().isEmpty)
+ }
+
+ test("reject multi vector search query map") {
+ val exception = intercept[RuntimeException] {
+ MultiVectorSearchQuery(Seq.empty).createMultiVectorSearch(
+ innerTable,
+ Seq(
+ CreateMap(
+ Seq(
+ Literal("title_vec"),
+ CreateArray(Seq(Literal(1.0f), Literal(0.0f))),
+ Literal("body_vec"),
+ CreateArray(Seq(Literal(0.0f), Literal(1.0f))))),
+ Literal(3)
+ )
+ )
+ }
+
+ assert(exception.getMessage.contains("Cannot extract multi-vector routes"))
+ }
+
test("create vector search with string options") {
val vectorSearch = createVectorSearch(
Literal("v"),
@@ -62,7 +151,13 @@ class VectorSearchQueryTest extends AnyFunSuite {
Array(classOf[InnerTable]),
new InvocationHandler {
private val rowType =
- RowType.of(Array[DataType](new ArrayType(DataTypes.FLOAT())),
Array[String]("v"))
+ RowType.of(
+ Array[DataType](
+ new ArrayType(DataTypes.FLOAT()),
+ new ArrayType(DataTypes.FLOAT()),
+ new ArrayType(DataTypes.FLOAT())),
+ Array[String]("v", "title_vec", "body_vec")
+ )
override def invoke(proxy: Any, method: Method, args:
Array[AnyRef]): AnyRef = {
method.getName match {
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/MultiVectorSearchTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/MultiVectorSearchTest.scala
new file mode 100644
index 0000000000..e3a3bbe8b3
--- /dev/null
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/MultiVectorSearchTest.scala
@@ -0,0 +1,85 @@
+/*
+ * 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.sql
+
+import org.apache.paimon.globalindex.testvector.TestVectorGlobalIndexerFactory
+import org.apache.paimon.spark.PaimonSparkTestBase
+
+/** Tests for multi-vector search. */
+class MultiVectorSearchTest extends PaimonSparkTestBase {
+
+ test("multi vector search ranks results from multiple vector columns") {
+ withTable("T") {
+ spark.sql("""
+ |CREATE TABLE T (id INT, title_vec ARRAY<FLOAT>, body_vec
ARRAY<FLOAT>)
+ |TBLPROPERTIES (
+ | 'bucket' = '-1',
+ | 'global-index.row-count-per-shard' = '10000',
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'test.vector.dimension' = '2',
+ | 'test.vector.required-option.key' = 'ivf.nprobe',
+ | 'test.vector.required-option.value' = '16')
+ |""".stripMargin)
+
+ spark.sql("""
+ |INSERT INTO T VALUES
+ | (0, array(1.0f, 0.0f), array(0.0f, 1.0f)),
+ | (1, array(0.9f, 0.1f), array(0.1f, 0.9f)),
+ | (2, array(0.0f, 1.0f), array(1.0f, 0.0f))
+ |""".stripMargin)
+
+ spark
+ .sql(s"CALL sys.create_global_index(table => 'test.T', index_column =>
'title_vec', " +
+ s"index_type => '${TestVectorGlobalIndexerFactory.IDENTIFIER}')")
+ .collect()
+ spark
+ .sql(s"CALL sys.create_global_index(table => 'test.T', index_column =>
'body_vec', " +
+ s"index_type => '${TestVectorGlobalIndexerFactory.IDENTIFIER}')")
+ .collect()
+
+ val result = spark
+ .sql("""
+ |SELECT id, __paimon_vector_search_score
+ |FROM multi_vector_search(
+ | 'T',
+ | array(
+ | named_struct(
+ | 'vector_column', 'title_vec',
+ | 'query_vector', array(1.0f, 0.0f),
+ | 'limit', 2,
+ | 'weight', 2.0f,
+ | 'options', map('ivf.nprobe', '16')),
+ | named_struct(
+ | 'vector_column', 'body_vec',
+ | 'query_vector', array(0.0f, 1.0f),
+ | 'limit', 2,
+ | 'weight', 1.0f,
+ | 'options', map('ivf.nprobe', '16'))),
+ | 2,
+ | 'weighted_score')
+ |""".stripMargin)
+ .collect()
+
+ assert(result.length == 2)
+ assert(result.map(_.getInt(0)).contains(1))
+ assert(result.forall(row => !row.isNullAt(1)))
+ }
+ }
+}
