You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/06/18 02:10:33 UTC
(paimon) branch master updated: [core][spark][python] Support hybrid search (#8271)
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 cabb0beb58 [core][spark][python] Support hybrid search (#8271)
cabb0beb58 is described below
commit cabb0beb58c65f27e5c5b71525e10f75adc57826
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Thu Jun 18 10:10:27 2026 +0800
[core][spark][python] Support hybrid search (#8271)
Replaces the unreleased multi-vector search API with a more general
hybrid search API that can combine multiple vector and full-text routes.
Spark exposes the new `hybrid_search(...)` table-valued function and
returns ranked results with `__paimon_search_score`.
---
docs/docs/multimodal-table/global-index.mdx | 429 +++++++++++++++-----
docs/docs/pypaimon/global-index.md | 142 -------
docs/redirects.js | 6 +-
docs/sidebars.js | 1 -
...orSearchRanker.java => HybridSearchRanker.java} | 48 +--
.../{MultiVectorSearch.java => HybridSearch.java} | 18 +-
...ctorSearchRoute.java => HybridSearchRoute.java} | 136 ++++++-
...RankerTest.java => HybridSearchRankerTest.java} | 9 +-
.../java/org/apache/paimon/table/FormatTable.java | 7 +-
...ctorSearchTable.java => HybridSearchTable.java} | 28 +-
.../java/org/apache/paimon/table/InnerTable.java | 8 +-
.../main/java/org/apache/paimon/table/Table.java | 8 +-
.../paimon/table/source/FullTextScanImpl.java | 11 +
.../paimon/table/source/FullTextSearchBuilder.java | 7 +
.../table/source/FullTextSearchBuilderImpl.java | 10 +-
.../paimon/table/source/HybridSearchBuilder.java | 177 ++++++++
.../table/source/HybridSearchBuilderImpl.java | 210 ++++++++++
.../table/source/MultiVectorSearchBuilder.java | 139 -------
.../table/source/MultiVectorSearchBuilderImpl.java | 160 --------
.../table/source/FullTextSearchBuilderTest.java | 45 +++
.../table/source/VectorSearchBuilderTest.java | 24 +-
paimon-python/pypaimon/table/file_store_table.py | 5 +
.../pypaimon/table/format/format_table.py | 3 +
.../pypaimon/table/iceberg/iceberg_table.py | 3 +
.../pypaimon/table/object/object_table.py | 5 +
.../pypaimon/table/source/full_text_scan.py | 18 +-
.../table/source/full_text_search_builder.py | 48 ++-
.../pypaimon/table/source/hybrid_search_builder.py | 450 +++++++++++++++++++++
.../pypaimon/table/system/system_table.py | 3 +
paimon-python/pypaimon/table/table.py | 5 +
.../pypaimon/tests/system/system_table_test.py | 1 +
.../pypaimon/tests/vector_search_filter_test.py | 176 ++++++++
.../scala/org/apache/paimon/spark/PaimonScan.scala | 4 +-
.../apache/paimon/spark/PaimonScanBuilder.scala | 8 +-
.../scala/org/apache/paimon/spark/PaimonScan.scala | 4 +-
...Impl.java => SparkHybridSearchBuilderImpl.java} | 12 +-
.../org/apache/paimon/spark/PaimonBaseScan.scala | 37 +-
.../paimon/spark/PaimonRecordReaderIterator.scala | 6 +-
.../scala/org/apache/paimon/spark/PaimonScan.scala | 4 +-
.../apache/paimon/spark/PaimonScanBuilder.scala | 8 +-
.../apache/paimon/spark/PaimonSparkTableBase.scala | 8 +-
.../plans/logical/PaimonTableValuedFunctions.scala | 182 ++++++---
.../org/apache/paimon/spark/read/BaseScan.scala | 6 +-
.../paimon/spark/schema/PaimonMetadataColumn.scala | 10 +-
.../plans/logical/VectorSearchQueryTest.scala | 58 ++-
.../apache/paimon/spark/SparkMultimodalITCase.java | 2 +-
.../apache/paimon/spark/sql/HybridSearchTest.scala | 250 ++++++++++++
.../paimon/spark/sql/MultiVectorSearchTest.scala | 85 ----
48 files changed, 2174 insertions(+), 850 deletions(-)
diff --git a/docs/docs/multimodal-table/global-index.mdx b/docs/docs/multimodal-table/global-index.mdx
index 7b9712a8f9..5a9788b8cb 100644
--- a/docs/docs/multimodal-table/global-index.mdx
+++ b/docs/docs/multimodal-table/global-index.mdx
@@ -81,10 +81,39 @@ CALL sys.create_global_index(
Once a BTree index is built, it is automatically used during scan when a filter predicate matches the indexed column.
+<Tabs groupId="btree-search">
+
+<TabItem value="sql" label="SQL">
+
```sql
SELECT * FROM my_table WHERE name IN ('a200', 'a300');
```
+</TabItem>
+
+<TabItem value="python-sdk" label="Python SDK">
+
+```python
+from pypaimon.common.predicate_builder import PredicateBuilder
+
+table = catalog.get_table("db.my_table")
+
+read_builder = table.new_read_builder()
+read_builder = read_builder.with_filter(
+ PredicateBuilder(table.fields)
+ .is_in("name", ["a200", "a300"])
+)
+
+scan = read_builder.new_scan()
+read = read_builder.new_read()
+pa_table = read.to_arrow(scan.plan().splits())
+print(pa_table)
+```
+
+</TabItem>
+
+</Tabs>
+
## Vector Index
Vector Index provides approximate nearest neighbor (ANN) search for vector similarity search scenarios such as
@@ -234,115 +263,50 @@ 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="python-sdk" label="Python SDK">
-<TabItem value="spark-sql" label="Spark SQL">
+```python
+table = catalog.get_table("db.my_table")
+
+# Step 1: Build vector search
+result = (
+ table.new_vector_search_builder()
+ .with_vector_column("embedding")
+ .with_query_vector([1.0, 2.0, 3.0])
+ .with_limit(5)
+ .with_option("ivf.nprobe", "16")
+ .execute_local()
+)
-```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');
+# Step 2: Read matching rows using the search result
+read_builder = table.new_read_builder()
+scan = read_builder.new_scan().with_global_index_result(result)
+plan = scan.plan()
+table_read = read_builder.new_read()
+pa_table = table_read.to_arrow(plan.splits())
+print(pa_table)
```
-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();
+You can also add a scalar filter to pre-filter rows before vector search:
-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));
- });
-}
+```python
+from pypaimon.common.predicate_builder import PredicateBuilder
+
+predicate = (
+ PredicateBuilder(table.fields)
+ .equal("category", "electronics")
+)
+
+result = (
+ table.new_vector_search_builder()
+ .with_vector_column("embedding")
+ .with_query_vector([1.0, 2.0, 3.0])
+ .with_limit(5)
+ .with_filter(predicate)
+ .execute_local()
+)
```
-For Java, use `Table.newMultiVectorSearchBuilder()` to configure routes, final limit, and ranker
-directly. `VectorSearch` and `MultiVectorSearch` are internal pushdown representations.
-
</TabItem>
</Tabs>
@@ -461,15 +425,17 @@ try (RecordReader<InternalRow> reader = readBuilder.newRead().createReader(plan)
<TabItem value="python-sdk" label="Python SDK">
```python
-table = catalog.get_table('db.my_table')
+table = catalog.get_table("db.my_table")
# Step 1: Build full-text search
-builder = table.new_full_text_search_builder()
-builder.with_text_column('content')
-builder.with_query_text('paimon lake format')
-builder.with_query_operator('and')
-builder.with_limit(10)
-result = builder.execute_local()
+result = (
+ table.new_full_text_search_builder()
+ .with_text_column("content")
+ .with_query_text("paimon lake format")
+ .with_query_operator("and")
+ .with_limit(10)
+ .execute_local()
+)
# Step 2: Read matching rows using the search result
read_builder = table.new_read_builder()
@@ -480,6 +446,251 @@ pa_table = table_read.to_arrow(plan.splits())
print(pa_table)
```
+PyPaimon reads the Tantivy tokenizer settings stored in the global index metadata. Indexes built
+with `tantivy.tokenizer=ngram` can be queried from Python when the installed `tantivy-py` package
+provides custom tokenizer support. Indexes built with `tantivy.tokenizer=jieba` can be queried
+when the `jieba` package is installed.
+
+</TabItem>
+
+</Tabs>
+
+## Hybrid Search
+
+Hybrid search queries multiple vector and full-text routes in one request and ranks the scored
+results before reading table rows. This is useful when one table stores several searchable
+representations for the same record, such as title embeddings, body embeddings, and content text.
+
+Before running hybrid search, create a global index for every vector or text 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'
+);
+
+CALL sys.create_global_index(
+ table => 'db.my_table',
+ index_column => 'content',
+ index_type => 'tantivy-fulltext'
+);
+```
+
+For Spark SQL, use the `hybrid_search(table_name, vector_routes, full_text_routes, limit[, ranker])`
+table-valued function. The fourth argument is the final number of ranked results to return.
+The optional fifth argument selects the ranker. Supported rankers are `rrf` and
+`weighted_score`; the default is `rrf`.
+
+The second argument is an array of vector route configs created by `named_struct`:
+
+| Field | Required | Default | Description |
+|---|---|---|---|
+| `field` or `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`. |
+
+The third argument is an array of full-text route configs created by `named_struct`:
+
+| Field | Required | Default | Description |
+|---|---|---|---|
+| `field` or `text_column` | Yes | N/A | Text column to search. |
+| `query_text` | Yes | N/A | Full-text query for this route. |
+| `query_operator` | No | `or` | Term operator. Supported values are `or` and `and`. |
+| `limit` | No | Final limit | Top K results to retrieve from this text column before ranking. |
+| `weight` | No | `1.0` | Weight for this route when ranking results. |
+| `options` | No | Empty map | Route-specific full-text search options. |
+
+Within each route array, every `named_struct` should use the same fields because
+Spark requires array elements to have the same struct type.
+
+<Tabs groupId="hybrid-search">
+
+<TabItem value="spark-sql" label="Spark SQL">
+
+Vector and full-text routes can be searched together:
+
+```sql
+SELECT id, __paimon_search_score
+FROM hybrid_search(
+ 'my_table',
+ array(
+ named_struct(
+ 'field', 'title_embedding',
+ 'query_vector', array(1.0f, 0.0f, 0.0f),
+ 'limit', 50,
+ 'weight', 2.0f,
+ 'options', map('ivf.nprobe', '32'))),
+ array(
+ named_struct(
+ 'field', 'content',
+ 'query_text', 'paimon search',
+ 'query_operator', 'or',
+ 'limit', 50,
+ 'weight', 1.0f,
+ 'options', map())),
+ 10,
+ 'rrf');
+```
+
+To search two vector indexes without a full-text route, pass `array()` as the third argument:
+
+```sql
+SELECT id, __paimon_search_score
+FROM hybrid_search(
+ 'my_table',
+ array(
+ named_struct(
+ 'field', 'title_embedding',
+ 'query_vector', array(1.0f, 0.0f, 0.0f),
+ 'limit', 50,
+ 'weight', 2.0f,
+ 'options', map('ivf.nprobe', '32')),
+ named_struct(
+ 'field', 'body_embedding',
+ 'query_vector', array(0.0f, 1.0f, 0.0f),
+ 'limit', 50,
+ 'weight', 1.0f,
+ 'options', map('ivf.nprobe', '32'))),
+ array(),
+ 10,
+ 'weighted_score');
+```
+
+Spark SQL adds `__paimon_search_score` to expose the ranked score.
+
+</TabItem>
+
+<TabItem value="java-api" label="Java API">
+
+```java
+Table table = catalog.getTable(identifier);
+
+HybridSearchBuilder searchBuilder =
+ table.newHybridSearchBuilder()
+ .addVectorRoute(
+ "title_embedding",
+ new float[] {1.0f, 0.0f, 0.0f},
+ 50,
+ 2.0f,
+ java.util.Collections.singletonMap("ivf.nprobe", "32"))
+ .addFullTextRoute(
+ "content",
+ "paimon search",
+ "or",
+ 50,
+ 1.0f,
+ java.util.Collections.emptyMap())
+ .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));
+ });
+}
+```
+
+To search two vector indexes without a full-text route, add two vector routes:
+
+```java
+GlobalIndexResult vectorOnlyResult =
+ table.newHybridSearchBuilder()
+ .addVectorRoute(
+ "title_embedding",
+ new float[] {1.0f, 0.0f, 0.0f},
+ 50,
+ 2.0f,
+ java.util.Collections.singletonMap("ivf.nprobe", "32"))
+ .addVectorRoute(
+ "body_embedding",
+ new float[] {0.0f, 1.0f, 0.0f},
+ 50,
+ 1.0f,
+ java.util.Collections.singletonMap("ivf.nprobe", "32"))
+ .withLimit(10)
+ .withWeightedScoreRanker()
+ .executeLocal();
+```
+
+For Java, use `Table.newHybridSearchBuilder()` to configure routes, final limit, and ranker
+directly. `VectorSearch` and `HybridSearch` are internal pushdown representations.
+
+</TabItem>
+
+<TabItem value="python-sdk" label="Python SDK">
+
+```python
+table = catalog.get_table("db.my_table")
+
+result = (
+ table.new_hybrid_search_builder()
+ .add_vector_route(
+ "title_embedding",
+ [1.0, 0.0, 0.0],
+ limit=50,
+ weight=2.0,
+ options={"ivf.nprobe": "32"},
+ )
+ .add_full_text_route(
+ "content",
+ "paimon search",
+ limit=50,
+ weight=1.0,
+ query_operator="or",
+ )
+ .with_limit(10)
+ .with_rrf_ranker()
+ .execute_local()
+)
+
+read_builder = table.new_read_builder()
+scan = read_builder.new_scan().with_global_index_result(result)
+plan = scan.plan()
+table_read = read_builder.new_read()
+pa_table = table_read.to_arrow(plan.splits())
+print(pa_table)
+```
+
+To search two vector indexes without a full-text route, add two vector routes:
+
+```python
+result = (
+ table.new_hybrid_search_builder()
+ .add_vector_route(
+ "title_embedding",
+ [1.0, 0.0, 0.0],
+ limit=50,
+ weight=2.0,
+ options={"ivf.nprobe": "32"},
+ )
+ .add_vector_route(
+ "body_embedding",
+ [0.0, 1.0, 0.0],
+ limit=50,
+ weight=1.0,
+ options={"ivf.nprobe": "32"},
+ )
+ .with_limit(10)
+ .with_weighted_score_ranker()
+ .execute_local()
+)
+```
+
</TabItem>
</Tabs>
diff --git a/docs/docs/pypaimon/global-index.md b/docs/docs/pypaimon/global-index.md
deleted file mode 100644
index 5795303a86..0000000000
--- a/docs/docs/pypaimon/global-index.md
+++ /dev/null
@@ -1,142 +0,0 @@
----
-title: "Global Index"
-sidebar_position: 6
----
-
-<!--
-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.
--->
-
-# Global Index
-
-PyPaimon supports querying global indexes built on Data Evolution (append) tables. Three index types are available:
-
-- **BTree Index**: B-tree based index for scalar column lookups. Supports equality, IN, range, and combined predicates.
-- **Vector Index (Lumina)**: Approximate nearest neighbor (ANN) index for vector similarity search.
-- **Full-Text Index (Tantivy)**: Full-text search index for text retrieval with relevance scoring.
-
-> Global indexes must be built beforehand (e.g., via Spark or Flink). See [Global Index](../multimodal-table/global-index) for how to create indexes.
-
-## BTree Index
-
-BTree index is automatically used during scan when a filter predicate matches the indexed column. No special API is needed — just set a filter on the read builder.
-
-```python
-import pypaimon
-
-catalog = pypaimon.create_catalog(...)
-table = catalog.get_table("db.my_table")
-
-# BTree index is used automatically when filtering on indexed columns
-read_builder = table.new_read_builder()
-read_builder = read_builder.with_filter(
- pypaimon.PredicateBuilder(table.fields)
- .in_("name", ["a200", "a300"])
-)
-
-scan = read_builder.new_scan()
-read = read_builder.new_read()
-splits = scan.plan().splits
-data = read.to_arrow(splits)
-```
-
-Supported predicates: `equal`, `not_equal`, `less_than`, `less_or_equal`, `greater_than`, `greater_or_equal`, `in_`, `not_in`, `between`, `is_null`, `is_not_null`.
-
-## Vector Index (Lumina)
-
-Use `VectorSearchBuilder` to perform approximate nearest neighbor search on a vector column, then read the matched rows.
-
-```python
-table = catalog.get_table("db.my_table")
-
-# Step 1: vector search to get matching row IDs
-builder = table.new_vector_search_builder()
-index_result = (
- builder
- .with_vector_column("embedding")
- .with_query_vector([1.0, 2.0, 3.0, ...])
- .with_limit(10)
- .execute_local()
-)
-
-# Step 2: read actual data for matched rows
-read_builder = table.new_read_builder()
-scan = read_builder.new_scan()
-scan.with_global_index_result(index_result)
-read = read_builder.new_read()
-data = read.to_arrow(scan.plan().splits)
-```
-
-You can also add a scalar filter to pre-filter rows before vector search:
-
-```python
-predicate = (
- pypaimon.PredicateBuilder(table.fields)
- .equal("category", "electronics")
-)
-
-index_result = (
- table.new_vector_search_builder()
- .with_vector_column("embedding")
- .with_query_vector([1.0, 2.0, 3.0, ...])
- .with_limit(10)
- .with_filter(predicate)
- .execute_local()
-)
-
-read_builder = table.new_read_builder()
-scan = read_builder.new_scan()
-scan.with_global_index_result(index_result)
-read = read_builder.new_read()
-data = read.to_arrow(scan.plan().splits)
-```
-
-## Full-Text Index (Tantivy)
-
-Use `FullTextSearchBuilder` to perform full-text search on a text column, then read the matched rows.
-PyPaimon reads the Tantivy tokenizer settings stored in the global index metadata. Indexes built
-with `tantivy.tokenizer=ngram` can be queried from Python when the installed `tantivy-py` package
-provides custom tokenizer support, including `Tokenizer.ngram`, `Tokenizer.simple`,
-`Tokenizer.whitespace`, `Tokenizer.raw`, `TextAnalyzerBuilder`, tokenizer filters, and
-`Index.register_tokenizer`. Indexes built with `tantivy.tokenizer=jieba` can be queried from Python
-when the `jieba` package is installed. Query terms use OR semantics by default; use
-`with_query_operator("and")` to require all terms.
-
-```python
-table = catalog.get_table("db.my_table")
-
-# Step 1: full-text search to get matching row IDs
-builder = table.new_full_text_search_builder()
-index_result = (
- builder
- .with_text_column("content")
- .with_query_text("search keywords")
- .with_query_operator("and")
- .with_limit(20)
- .execute_local()
-)
-
-# Step 2: read actual data for matched rows
-read_builder = table.new_read_builder()
-scan = read_builder.new_scan()
-scan.with_global_index_result(index_result)
-read = read_builder.new_read()
-data = read.to_arrow(scan.plan().splits)
-```
-
-For better performance when reading from remote storage, consider enabling the [Local Cache](../program-api/file-cache).
diff --git a/docs/redirects.js b/docs/redirects.js
index d3facf9b7b..c151681242 100644
--- a/docs/redirects.js
+++ b/docs/redirects.js
@@ -398,7 +398,11 @@ module.exports = [
},
{
"from": "/pypaimon/global-index.html",
- "to": "/pypaimon/global-index"
+ "to": "/multimodal-table/global-index"
+ },
+ {
+ "from": "/pypaimon/global-index",
+ "to": "/multimodal-table/global-index"
},
{
"from": "/pypaimon/manage-tags.html",
diff --git a/docs/sidebars.js b/docs/sidebars.js
index 6f6fa15afa..d28a03b65c 100644
--- a/docs/sidebars.js
+++ b/docs/sidebars.js
@@ -177,7 +177,6 @@ const sidebars = {
"pypaimon/daft",
"pypaimon/pytorch",
"pypaimon/data-evolution",
- "pypaimon/global-index",
"pypaimon/system-tables",
"pypaimon/fuse-support",
"pypaimon/pyjindosdk-support",
diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/MultiVectorSearchRanker.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/HybridSearchRanker.java
similarity index 80%
rename from paimon-common/src/main/java/org/apache/paimon/globalindex/MultiVectorSearchRanker.java
rename to paimon-common/src/main/java/org/apache/paimon/globalindex/HybridSearchRanker.java
index fc29adb1d3..1ab156c6cc 100644
--- a/paimon-common/src/main/java/org/apache/paimon/globalindex/MultiVectorSearchRanker.java
+++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/HybridSearchRanker.java
@@ -22,21 +22,19 @@ 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 {
+/** Ranker utilities for hybrid search results. */
+public class HybridSearchRanker {
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() {}
+ private HybridSearchRanker() {}
public static ScoredGlobalIndexResult rank(
String ranker, List<ScoredGlobalIndexResult> results, float[] weights, int limit) {
@@ -61,7 +59,7 @@ public class MultiVectorSearchRanker {
}
String normalized = ranker.trim().toLowerCase();
if (!RRF_RANKER.equals(normalized) && !WEIGHTED_SCORE_RANKER.equals(normalized)) {
- throw new IllegalArgumentException("Unsupported multi-vector ranker: " + ranker);
+ throw new IllegalArgumentException("Unsupported hybrid ranker: " + ranker);
}
return normalized;
}
@@ -84,8 +82,9 @@ public class MultiVectorSearchRanker {
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);
+ scores.compute(
+ rowId,
+ (k, oldScore) -> oldScore == null ? contribution : oldScore + contribution);
}
}
return topK(scores, limit);
@@ -108,8 +107,9 @@ public class MultiVectorSearchRanker {
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);
+ scores.compute(
+ rowId,
+ (k, oldScore) -> oldScore == null ? contribution : oldScore + contribution);
}
}
return topK(scores, limit);
@@ -121,14 +121,8 @@ public class MultiVectorSearchRanker {
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));
- }
- });
+ rowIds.sort(
+ (left, right) -> Float.compare(scoreGetter.score(right), scoreGetter.score(left)));
return rowIds;
}
@@ -137,17 +131,13 @@ public class MultiVectorSearchRanker {
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());
+ ranked.sort(
+ (left, 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());
@@ -168,7 +158,7 @@ public class MultiVectorSearchRanker {
return weights[index];
}
- /** Weighted result from one vector-search route. */
+ /** Weighted result from one search route. */
public static class WeightedResult implements Serializable {
private static final long serialVersionUID = 1L;
diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/MultiVectorSearch.java b/paimon-common/src/main/java/org/apache/paimon/predicate/HybridSearch.java
similarity index 77%
rename from paimon-common/src/main/java/org/apache/paimon/predicate/MultiVectorSearch.java
rename to paimon-common/src/main/java/org/apache/paimon/predicate/HybridSearch.java
index 0541edf473..0f6c55395c 100644
--- a/paimon-common/src/main/java/org/apache/paimon/predicate/MultiVectorSearch.java
+++ b/paimon-common/src/main/java/org/apache/paimon/predicate/HybridSearch.java
@@ -18,7 +18,7 @@
package org.apache.paimon.predicate;
-import org.apache.paimon.globalindex.MultiVectorSearchRanker;
+import org.apache.paimon.globalindex.HybridSearchRanker;
import java.io.Serializable;
import java.util.ArrayList;
@@ -26,20 +26,20 @@ import java.util.Collections;
import java.util.List;
/**
- * Multi-vector search over multiple vector columns.
+ * Hybrid search over vector and full-text routes.
*
- * <p>This is an internal pushdown representation. Use {@code Table.newMultiVectorSearchBuilder()}
- * to configure multi-vector search from Java.
+ * <p>This is an internal pushdown representation. Use {@code Table.newHybridSearchBuilder()} to
+ * configure hybrid search from Java.
*/
-public class MultiVectorSearch implements Serializable {
+public class HybridSearch implements Serializable {
private static final long serialVersionUID = 1L;
- private final List<MultiVectorSearchRoute> routes;
+ private final List<HybridSearchRoute> routes;
private final int limit;
private final String ranker;
- public MultiVectorSearch(List<MultiVectorSearchRoute> routes, int limit, String ranker) {
+ public HybridSearch(List<HybridSearchRoute> routes, int limit, String ranker) {
if (routes == null || routes.isEmpty()) {
throw new IllegalArgumentException("Routes cannot be null or empty");
}
@@ -48,10 +48,10 @@ public class MultiVectorSearch implements Serializable {
}
this.routes = Collections.unmodifiableList(new ArrayList<>(routes));
this.limit = limit;
- this.ranker = MultiVectorSearchRanker.normalizeRanker(ranker);
+ this.ranker = HybridSearchRanker.normalizeRanker(ranker);
}
- public List<MultiVectorSearchRoute> routes() {
+ public List<HybridSearchRoute> routes() {
return 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/HybridSearchRoute.java
similarity index 50%
rename from paimon-common/src/main/java/org/apache/paimon/predicate/MultiVectorSearchRoute.java
rename to paimon-common/src/main/java/org/apache/paimon/predicate/HybridSearchRoute.java
index 070828a963..4800479a73 100644
--- a/paimon-common/src/main/java/org/apache/paimon/predicate/MultiVectorSearchRoute.java
+++ b/paimon-common/src/main/java/org/apache/paimon/predicate/HybridSearchRoute.java
@@ -25,32 +25,87 @@ 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 {
+/** A single search route in a hybrid search. */
+public class HybridSearchRoute implements Serializable {
private static final long serialVersionUID = 1L;
+ /** Search route type. */
+ public enum RouteType {
+ VECTOR,
+ FULL_TEXT
+ }
+
+ private final RouteType routeType;
private final String fieldName;
private final float[] vector;
+ private final String queryText;
+ private final String queryOperator;
private final int limit;
private final float weight;
private final Map<String, String> options;
- public MultiVectorSearchRoute(String fieldName, float[] vector, int limit, float weight) {
+ public HybridSearchRoute(String fieldName, float[] vector, int limit, float weight) {
this(fieldName, vector, limit, weight, Collections.emptyMap());
}
- public MultiVectorSearchRoute(
+ public HybridSearchRoute(
+ String fieldName,
+ float[] vector,
+ int limit,
+ float weight,
+ Map<String, String> options) {
+ this(RouteType.VECTOR, fieldName, vector, null, "or", limit, weight, options);
+ }
+
+ public static HybridSearchRoute vector(
+ String fieldName,
+ float[] vector,
+ int limit,
+ float weight,
+ Map<String, String> options) {
+ return new HybridSearchRoute(fieldName, vector, limit, weight, options);
+ }
+
+ public static HybridSearchRoute fullText(
+ String fieldName,
+ String queryText,
+ String queryOperator,
+ int limit,
+ float weight,
+ Map<String, String> options) {
+ return new HybridSearchRoute(
+ RouteType.FULL_TEXT,
+ fieldName,
+ null,
+ queryText,
+ queryOperator,
+ limit,
+ weight,
+ options);
+ }
+
+ private HybridSearchRoute(
+ RouteType routeType,
String fieldName,
float[] vector,
+ String queryText,
+ String queryOperator,
int limit,
float weight,
Map<String, String> options) {
+ if (routeType == null) {
+ throw new IllegalArgumentException("Route type cannot be null");
+ }
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 (routeType == RouteType.VECTOR && vector == null) {
+ throw new IllegalArgumentException("Search vector cannot be null for vector route");
+ }
+ if (routeType == RouteType.FULL_TEXT && (queryText == null || queryText.isEmpty())) {
+ throw new IllegalArgumentException(
+ "Query text cannot be null or empty for full-text route");
}
if (limit <= 0) {
throw new IllegalArgumentException("Limit must be positive, got: " + limit);
@@ -58,8 +113,17 @@ public class MultiVectorSearchRoute implements Serializable {
if (weight <= 0) {
throw new IllegalArgumentException("Weight must be positive, got: " + weight);
}
+ String normalizedOperator =
+ queryOperator == null ? "or" : queryOperator.trim().toLowerCase();
+ if (!"or".equals(normalizedOperator) && !"and".equals(normalizedOperator)) {
+ throw new IllegalArgumentException(
+ "Query operator must be 'or' or 'and', got: " + queryOperator);
+ }
+ this.routeType = routeType;
this.fieldName = fieldName;
this.vector = vector;
+ this.queryText = queryText;
+ this.queryOperator = normalizedOperator;
this.limit = limit;
this.weight = weight;
this.options =
@@ -77,10 +141,30 @@ public class MultiVectorSearchRoute implements Serializable {
return fieldName;
}
+ public RouteType routeType() {
+ return routeType;
+ }
+
+ public boolean isVector() {
+ return routeType == RouteType.VECTOR;
+ }
+
+ public boolean isFullText() {
+ return routeType == RouteType.FULL_TEXT;
+ }
+
public float[] vector() {
return vector;
}
+ public String queryText() {
+ return queryText;
+ }
+
+ public String queryOperator() {
+ return queryOperator;
+ }
+
public int limit() {
return limit;
}
@@ -97,17 +181,31 @@ public class MultiVectorSearchRoute implements Serializable {
return new VectorSearch(vector, limit, fieldName, options);
}
+ public FullTextSearch toFullTextSearch() {
+ return new FullTextSearch(queryText, limit, fieldName, queryOperator);
+ }
+
@Override
public String toString() {
- return "FieldName(" + fieldName + "), Limit(" + limit + "), Weight(" + weight + ")";
+ return "Type("
+ + routeType
+ + "), FieldName("
+ + fieldName
+ + "), Limit("
+ + limit
+ + "), Weight("
+ + weight
+ + ")";
}
- /** Builder for {@link MultiVectorSearchRoute}. */
+ /** Builder for {@link HybridSearchRoute}. */
@Experimental
public static class Builder {
private String fieldName;
private float[] vector;
+ private String queryText;
+ private String queryOperator = "or";
private int limit;
private float weight = 1.0f;
private Map<String, String> options = new HashMap<>();
@@ -121,6 +219,11 @@ public class MultiVectorSearchRoute implements Serializable {
return vectorColumn(fieldName);
}
+ public Builder textColumn(String fieldName) {
+ this.fieldName = fieldName;
+ return this;
+ }
+
public Builder queryVector(float[] vector) {
this.vector = vector;
return this;
@@ -130,6 +233,16 @@ public class MultiVectorSearchRoute implements Serializable {
return queryVector(vector);
}
+ public Builder queryText(String queryText) {
+ this.queryText = queryText;
+ return this;
+ }
+
+ public Builder queryOperator(String queryOperator) {
+ this.queryOperator = queryOperator;
+ return this;
+ }
+
public Builder limit(int limit) {
this.limit = limit;
return this;
@@ -152,8 +265,11 @@ public class MultiVectorSearchRoute implements Serializable {
return this;
}
- public MultiVectorSearchRoute build() {
- return new MultiVectorSearchRoute(fieldName, vector, limit, weight, options);
+ public HybridSearchRoute build() {
+ if (queryText != null) {
+ return fullText(fieldName, queryText, queryOperator, limit, weight, options);
+ }
+ return HybridSearchRoute.vector(fieldName, vector, limit, weight, options);
}
}
}
diff --git a/paimon-common/src/test/java/org/apache/paimon/globalindex/MultiVectorSearchRankerTest.java b/paimon-common/src/test/java/org/apache/paimon/globalindex/HybridSearchRankerTest.java
similarity index 90%
rename from paimon-common/src/test/java/org/apache/paimon/globalindex/MultiVectorSearchRankerTest.java
rename to paimon-common/src/test/java/org/apache/paimon/globalindex/HybridSearchRankerTest.java
index 3aa5be4fd9..226be53dd2 100644
--- a/paimon-common/src/test/java/org/apache/paimon/globalindex/MultiVectorSearchRankerTest.java
+++ b/paimon-common/src/test/java/org/apache/paimon/globalindex/HybridSearchRankerTest.java
@@ -30,8 +30,8 @@ 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 {
+/** Tests for {@link HybridSearchRanker}. */
+public class HybridSearchRankerTest {
@Test
public void testRrfFavorsRowsReturnedByMultipleRoutes() {
@@ -39,8 +39,7 @@ public class MultiVectorSearchRankerTest {
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);
+ HybridSearchRanker.rrf(Arrays.asList(first, second), new float[] {1.0f, 1.0f}, 2);
assertThat(ranked.results().getIntCardinality()).isEqualTo(2);
assertThat(ranked.results()).contains(2L);
@@ -52,7 +51,7 @@ public class MultiVectorSearchRankerTest {
ScoredGlobalIndexResult result = result(new long[] {1, 2}, new float[] {0.3f, 0.2f});
ScoredGlobalIndexResult ranked =
- MultiVectorSearchRanker.weightedScore(
+ HybridSearchRanker.weightedScore(
Collections.singletonList(result), new float[] {3.0f}, 1);
assertThat(ranked.results().getIntCardinality()).isEqualTo(1);
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 ae23dcd8f1..86441c5dde 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,7 +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.HybridSearchBuilder;
import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.table.source.VectorSearchBuilder;
import org.apache.paimon.types.RowType;
@@ -282,9 +282,8 @@ public interface FormatTable extends Table {
}
@Override
- public MultiVectorSearchBuilder newMultiVectorSearchBuilder() {
- throw new UnsupportedOperationException(
- "FormatTable does not support multi-vector search.");
+ public HybridSearchBuilder newHybridSearchBuilder() {
+ throw new UnsupportedOperationException("FormatTable does not support hybrid search.");
}
@Override
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/MultiVectorSearchTable.java b/paimon-core/src/main/java/org/apache/paimon/table/HybridSearchTable.java
similarity index 68%
rename from paimon-core/src/main/java/org/apache/paimon/table/MultiVectorSearchTable.java
rename to paimon-core/src/main/java/org/apache/paimon/table/HybridSearchTable.java
index a1c0322ba8..d5a4a22810 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/MultiVectorSearchTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/HybridSearchTable.java
@@ -19,7 +19,7 @@
package org.apache.paimon.table;
import org.apache.paimon.fs.FileIO;
-import org.apache.paimon.predicate.MultiVectorSearch;
+import org.apache.paimon.predicate.HybridSearch;
import org.apache.paimon.table.source.InnerTableRead;
import org.apache.paimon.table.source.InnerTableScan;
import org.apache.paimon.types.RowType;
@@ -28,27 +28,26 @@ 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.
+ * A table wrapper to hold hybrid search information. This is used to pass hybrid 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 {
+public class HybridSearchTable implements ReadonlyTable {
private final InnerTable origin;
- private final MultiVectorSearch multiVectorSearch;
+ private final HybridSearch hybridSearch;
- private MultiVectorSearchTable(InnerTable origin, MultiVectorSearch multiVectorSearch) {
+ private HybridSearchTable(InnerTable origin, HybridSearch hybridSearch) {
this.origin = origin;
- this.multiVectorSearch = multiVectorSearch;
+ this.hybridSearch = hybridSearch;
}
- public static MultiVectorSearchTable create(
- InnerTable origin, MultiVectorSearch multiVectorSearch) {
- return new MultiVectorSearchTable(origin, multiVectorSearch);
+ public static HybridSearchTable create(InnerTable origin, HybridSearch hybridSearch) {
+ return new HybridSearchTable(origin, hybridSearch);
}
- public MultiVectorSearch multiVectorSearch() {
- return multiVectorSearch;
+ public HybridSearch hybridSearch() {
+ return hybridSearch;
}
public InnerTable origin() {
@@ -97,7 +96,6 @@ public class MultiVectorSearchTable implements ReadonlyTable {
@Override
public Table copy(Map<String, String> dynamicOptions) {
- return new MultiVectorSearchTable(
- (InnerTable) origin.copy(dynamicOptions), multiVectorSearch);
+ return new HybridSearchTable((InnerTable) origin.copy(dynamicOptions), hybridSearch);
}
}
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 fd863d643f..494aae1fe8 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
@@ -27,10 +27,10 @@ import org.apache.paimon.table.sink.StreamWriteBuilderImpl;
import org.apache.paimon.table.sink.WriteSelector;
import org.apache.paimon.table.source.FullTextSearchBuilder;
import org.apache.paimon.table.source.FullTextSearchBuilderImpl;
+import org.apache.paimon.table.source.HybridSearchBuilder;
+import org.apache.paimon.table.source.HybridSearchBuilderImpl;
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;
@@ -65,8 +65,8 @@ public interface InnerTable extends Table {
}
@Override
- default MultiVectorSearchBuilder newMultiVectorSearchBuilder() {
- return new MultiVectorSearchBuilderImpl(this);
+ default HybridSearchBuilder newHybridSearchBuilder() {
+ return new HybridSearchBuilderImpl(this);
}
@Override
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 aa05ba4660..3a7b1730b6 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,7 +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.HybridSearchBuilder;
import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.table.source.VectorSearchBuilder;
import org.apache.paimon.types.RowType;
@@ -227,10 +227,10 @@ public interface Table extends Serializable {
/** Returns a new vector search builder. */
VectorSearchBuilder newVectorSearchBuilder();
- /** Returns a new multi-vector search builder. */
- default MultiVectorSearchBuilder newMultiVectorSearchBuilder() {
+ /** Returns a new hybrid search builder. */
+ default HybridSearchBuilder newHybridSearchBuilder() {
throw new UnsupportedOperationException(
- getClass().getName() + " does not support multi-vector search.");
+ getClass().getName() + " does not support hybrid search.");
}
/** Returns a new full-text search builder. */
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextScanImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextScanImpl.java
index cc77d9121a..9b3c7fba2c 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextScanImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextScanImpl.java
@@ -23,6 +23,7 @@ import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileHandler;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.source.snapshot.TimeTravelUtil;
import org.apache.paimon.types.DataField;
@@ -42,10 +43,17 @@ import static org.apache.paimon.utils.Preconditions.checkNotNull;
public class FullTextScanImpl implements FullTextScan {
private final FileStoreTable table;
+ private final PartitionPredicate partitionFilter;
private final DataField textColumn;
public FullTextScanImpl(FileStoreTable table, DataField textColumn) {
+ this(table, null, textColumn);
+ }
+
+ public FullTextScanImpl(
+ FileStoreTable table, PartitionPredicate partitionFilter, DataField textColumn) {
this.table = table;
+ this.partitionFilter = partitionFilter;
this.textColumn = textColumn;
}
@@ -57,6 +65,9 @@ public class FullTextScanImpl implements FullTextScan {
IndexFileHandler indexFileHandler = table.store().newIndexFileHandler();
Filter<IndexManifestEntry> indexFileFilter =
entry -> {
+ if (partitionFilter != null && !partitionFilter.test(entry.partition())) {
+ return false;
+ }
GlobalIndexMeta globalIndex = entry.indexFile().globalIndexMeta();
if (globalIndex == null) {
return false;
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchBuilder.java b/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchBuilder.java
index 7ce4f11e1c..d7779ca0f5 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchBuilder.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchBuilder.java
@@ -19,12 +19,19 @@
package org.apache.paimon.table.source;
import org.apache.paimon.globalindex.GlobalIndexResult;
+import org.apache.paimon.partition.PartitionPredicate;
import java.io.Serializable;
/** Builder to build full-text search. */
public interface FullTextSearchBuilder extends Serializable {
+ /** Push partition filters. */
+ default FullTextSearchBuilder withPartitionFilter(PartitionPredicate partitionPredicate) {
+ throw new UnsupportedOperationException(
+ "This full-text search builder does not support partition filters.");
+ }
+
/** The top k results to return. */
FullTextSearchBuilder withLimit(int limit);
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchBuilderImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchBuilderImpl.java
index 33651e8045..dea7e8f5d1 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchBuilderImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchBuilderImpl.java
@@ -18,6 +18,7 @@
package org.apache.paimon.table.source;
+import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.InnerTable;
import org.apache.paimon.types.DataField;
@@ -36,11 +37,18 @@ public class FullTextSearchBuilderImpl implements FullTextSearchBuilder {
private DataField textColumn;
private String queryText;
private String queryOperator = "or";
+ private PartitionPredicate partitionFilter;
public FullTextSearchBuilderImpl(InnerTable table) {
this.table = (FileStoreTable) table;
}
+ @Override
+ public FullTextSearchBuilder withPartitionFilter(PartitionPredicate partitionFilter) {
+ this.partitionFilter = partitionFilter;
+ return this;
+ }
+
@Override
public FullTextSearchBuilder withLimit(int limit) {
this.limit = limit;
@@ -68,7 +76,7 @@ public class FullTextSearchBuilderImpl implements FullTextSearchBuilder {
@Override
public FullTextScan newFullTextScan() {
checkNotNull(textColumn, "Text column must be set via withTextColumn()");
- return new FullTextScanImpl(table, textColumn);
+ return new FullTextScanImpl(table, partitionFilter, textColumn);
}
@Override
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/HybridSearchBuilder.java b/paimon-core/src/main/java/org/apache/paimon/table/source/HybridSearchBuilder.java
new file mode 100644
index 0000000000..4fb6a644ca
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/HybridSearchBuilder.java
@@ -0,0 +1,177 @@
+/*
+ * 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.HybridSearchRoute;
+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 hybrid search. */
+public interface HybridSearchBuilder extends Serializable {
+
+ /** Push partition filters. */
+ HybridSearchBuilder withPartitionFilter(PartitionPredicate partitionPredicate);
+
+ /** Push pre-filter for route search. */
+ HybridSearchBuilder withFilter(Predicate predicate);
+
+ /** Add a hybrid-search route. */
+ HybridSearchBuilder addRoute(HybridSearchRoute route);
+
+ /** Add a vector-search route. */
+ default HybridSearchBuilder addVectorRoute(String vectorColumn, float[] vector, int limit) {
+ return addVectorRoute(vectorColumn, vector, limit, 1.0f);
+ }
+
+ /** Add a vector-search route. */
+ default HybridSearchBuilder addVectorRoute(
+ String vectorColumn, float[] vector, int limit, float weight) {
+ return addRoute(new HybridSearchRoute(vectorColumn, vector, limit, weight));
+ }
+
+ /** Add a vector-search route. */
+ default HybridSearchBuilder addVectorRoute(
+ String vectorColumn,
+ float[] vector,
+ int limit,
+ float weight,
+ Map<String, String> options) {
+ return addRoute(new HybridSearchRoute(vectorColumn, vector, limit, weight, options));
+ }
+
+ /** Add a full-text-search route. */
+ default HybridSearchBuilder addFullTextRoute(
+ String textColumn, String queryText, int limit, float weight) {
+ return addFullTextRoute(textColumn, queryText, "or", limit, weight, null);
+ }
+
+ /** Add a full-text-search route. */
+ default HybridSearchBuilder addFullTextRoute(
+ String textColumn,
+ String queryText,
+ String queryOperator,
+ int limit,
+ float weight,
+ Map<String, String> options) {
+ return addRoute(
+ HybridSearchRoute.fullText(
+ textColumn, queryText, queryOperator, limit, weight, options));
+ }
+
+ /** The final top k ranked results to return. */
+ HybridSearchBuilder withLimit(int limit);
+
+ /** Ranker for combining route results. */
+ HybridSearchBuilder withRanker(String ranker);
+
+ /** Use reciprocal rank fusion to combine route results. */
+ HybridSearchBuilder withRrfRanker();
+
+ /** Use weighted score to combine route results. */
+ HybridSearchBuilder withWeightedScoreRanker();
+
+ /** Create one search builder for each route so engines can dispatch route work. */
+ List<Route> routeBuilders();
+
+ /** Convert a search result into a weighted route result. */
+ RouteResult toRouteResult(Route route, GlobalIndexResult result);
+
+ /** Rank route results. */
+ ScoredGlobalIndexResult rank(List<RouteResult> routeResults);
+
+ /** Execute hybrid 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.executeLocal()));
+ }
+ return rank(routeResults);
+ }
+
+ /** A route and its configured search builder. */
+ class Route implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final HybridSearchRoute route;
+ private final VectorSearchBuilder vectorSearchBuilder;
+ private final FullTextSearchBuilder fullTextSearchBuilder;
+
+ public Route(HybridSearchRoute route, VectorSearchBuilder vectorSearchBuilder) {
+ this.route = route;
+ this.vectorSearchBuilder = vectorSearchBuilder;
+ this.fullTextSearchBuilder = null;
+ }
+
+ public Route(HybridSearchRoute route, FullTextSearchBuilder fullTextSearchBuilder) {
+ this.route = route;
+ this.vectorSearchBuilder = null;
+ this.fullTextSearchBuilder = fullTextSearchBuilder;
+ }
+
+ public HybridSearchRoute route() {
+ return route;
+ }
+
+ public VectorSearchBuilder vectorSearchBuilder() {
+ return vectorSearchBuilder;
+ }
+
+ public FullTextSearchBuilder fullTextSearchBuilder() {
+ return fullTextSearchBuilder;
+ }
+
+ public GlobalIndexResult executeLocal() {
+ if (route.isVector()) {
+ return vectorSearchBuilder.executeLocal();
+ }
+ return fullTextSearchBuilder.executeLocal();
+ }
+ }
+
+ /** A scored result produced by one route. */
+ class RouteResult implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final HybridSearchRoute route;
+ private final ScoredGlobalIndexResult result;
+
+ public RouteResult(HybridSearchRoute route, ScoredGlobalIndexResult result) {
+ this.route = route;
+ this.result = result;
+ }
+
+ public HybridSearchRoute route() {
+ return route;
+ }
+
+ public ScoredGlobalIndexResult result() {
+ return result;
+ }
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/HybridSearchBuilderImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/source/HybridSearchBuilderImpl.java
new file mode 100644
index 0000000000..1aa322815c
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/HybridSearchBuilderImpl.java
@@ -0,0 +1,210 @@
+/*
+ * 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.HybridSearchRanker;
+import org.apache.paimon.globalindex.ScoredGlobalIndexResult;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.predicate.HybridSearchRoute;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.table.InnerTable;
+import org.apache.paimon.utils.Pair;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
+/** Implementation for {@link HybridSearchBuilder}. */
+public class HybridSearchBuilderImpl implements HybridSearchBuilder {
+
+ private static final long serialVersionUID = 1L;
+
+ protected final InnerTable table;
+
+ protected final List<HybridSearchRoute> routes = new ArrayList<>();
+ protected int limit;
+ protected String ranker = HybridSearchRanker.RRF_RANKER;
+ protected PartitionPredicate partitionFilter;
+ protected Predicate filter;
+
+ public HybridSearchBuilderImpl(InnerTable table) {
+ this.table = table;
+ }
+
+ @Override
+ public HybridSearchBuilder withPartitionFilter(PartitionPredicate partitionFilter) {
+ addPartitionFilter(partitionFilter);
+ return this;
+ }
+
+ @Override
+ public HybridSearchBuilder withFilter(Predicate predicate) {
+ Pair<Optional<PartitionPredicate>, List<Predicate>> pair =
+ PartitionPredicate.splitPartitionPredicatesAndDataPredicates(
+ predicate, table.rowType(), table.partitionKeys());
+ if (pair.getLeft().isPresent()) {
+ addPartitionFilter(pair.getLeft().get());
+ }
+ if (!pair.getRight().isEmpty()) {
+ Predicate dataFilter = PredicateBuilder.and(pair.getRight());
+ if (this.filter == null) {
+ this.filter = dataFilter;
+ } else {
+ this.filter = PredicateBuilder.and(this.filter, dataFilter);
+ }
+ }
+ return this;
+ }
+
+ private void addPartitionFilter(PartitionPredicate partitionFilter) {
+ if (partitionFilter == null) {
+ return;
+ }
+ if (this.partitionFilter == null) {
+ this.partitionFilter = partitionFilter;
+ } else {
+ this.partitionFilter =
+ PartitionPredicate.and(Arrays.asList(this.partitionFilter, partitionFilter));
+ }
+ }
+
+ @Override
+ public HybridSearchBuilder addRoute(HybridSearchRoute route) {
+ this.routes.add(Objects.requireNonNull(route, "Route cannot be null"));
+ return this;
+ }
+
+ @Override
+ public HybridSearchBuilder withLimit(int limit) {
+ this.limit = limit;
+ return this;
+ }
+
+ @Override
+ public HybridSearchBuilder withRanker(String ranker) {
+ this.ranker = HybridSearchRanker.normalizeRanker(ranker);
+ return this;
+ }
+
+ @Override
+ public HybridSearchBuilder withRrfRanker() {
+ return withRanker(HybridSearchRanker.RRF_RANKER);
+ }
+
+ @Override
+ public HybridSearchBuilder withWeightedScoreRanker() {
+ return withRanker(HybridSearchRanker.WEIGHTED_SCORE_RANKER);
+ }
+
+ @Override
+ public List<Route> routeBuilders() {
+ validateSearch();
+
+ List<Route> routeBuilders = new ArrayList<>(routes.size());
+ for (HybridSearchRoute route : routes) {
+ if (route.isVector()) {
+ routeBuilders.add(new Route(route, newVectorSearchBuilder(route)));
+ } else {
+ routeBuilders.add(new Route(route, newFullTextSearchBuilder(route)));
+ }
+ }
+ 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);
+ }
+ if (filter != null) {
+ for (HybridSearchRoute route : routes) {
+ if (!route.isVector()) {
+ throw new UnsupportedOperationException(
+ "Hybrid search with full-text routes does not support non-partition "
+ + "filters because full-text indexes cannot apply row-id "
+ + "pre-filters before top-k ranking.");
+ }
+ }
+ }
+ }
+
+ @Override
+ public ScoredGlobalIndexResult rank(List<RouteResult> routeResults) {
+ validateSearch();
+
+ List<HybridSearchRanker.WeightedResult> weightedResults =
+ new ArrayList<>(routeResults.size());
+ for (RouteResult routeResult : routeResults) {
+ if (!routeResult.result().results().isEmpty()) {
+ weightedResults.add(
+ new HybridSearchRanker.WeightedResult(
+ routeResult.result(), routeResult.route().weight()));
+ }
+ }
+ return HybridSearchRanker.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(
+ "Hybrid search requires scored index results, but got: "
+ + result.getClass().getName());
+ }
+ }
+
+ protected VectorSearchBuilder newVectorSearchBuilder(HybridSearchRoute 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;
+ }
+
+ protected FullTextSearchBuilder newFullTextSearchBuilder(HybridSearchRoute route) {
+ FullTextSearchBuilder fullTextSearchBuilder =
+ table.newFullTextSearchBuilder()
+ .withQueryText(route.queryText())
+ .withQueryOperator(route.queryOperator())
+ .withTextColumn(route.fieldName())
+ .withLimit(route.limit());
+ if (partitionFilter != null) {
+ fullTextSearchBuilder.withPartitionFilter(partitionFilter);
+ }
+ return fullTextSearchBuilder;
+ }
+}
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
deleted file mode 100644
index b45d8fd2e0..0000000000
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/MultiVectorSearchBuilder.java
+++ /dev/null
@@ -1,139 +0,0 @@
-/*
- * 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
deleted file mode 100644
index 247301301a..0000000000
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/MultiVectorSearchBuilderImpl.java
+++ /dev/null
@@ -1,160 +0,0 @@
-/*
- * 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/FullTextSearchBuilderTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/FullTextSearchBuilderTest.java
index c52a2a46af..2d7b8c97fa 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/source/FullTextSearchBuilderTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/source/FullTextSearchBuilderTest.java
@@ -33,6 +33,8 @@ 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.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.table.FileStoreTable;
@@ -53,6 +55,7 @@ import java.util.Collections;
import java.util.List;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link FullTextSearchBuilder} using test-only brute-force full-text index. */
public class FullTextSearchBuilderTest extends TableTestBase {
@@ -112,6 +115,48 @@ public class FullTextSearchBuilderTest extends TableTestBase {
assertThat(ids).containsAnyOf(0, 1, 3);
}
+ @Test
+ public void testHybridSearchBuilderWithFullTextRoute() throws Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+
+ String[] documents = {
+ "Apache Paimon is a lake format",
+ "Paimon supports full-text search",
+ "Vector search is also supported"
+ };
+
+ writeDocuments(table, documents);
+ buildAndCommitIndex(table, documents);
+
+ ScoredGlobalIndexResult result =
+ table.newHybridSearchBuilder()
+ .addFullTextRoute(TEXT_FIELD_NAME, "Paimon", 3, 1.0f)
+ .withLimit(3)
+ .executeLocal();
+
+ assertThat(result.results().isEmpty()).isFalse();
+ assertThat(result.scoreGetter().score(result.results().iterator().next())).isGreaterThan(0);
+ }
+
+ @Test
+ public void testHybridSearchRejectsDataFilterWithFullTextRoute() throws Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+
+ Predicate idFilter = new PredicateBuilder(table.rowType()).equal(0, 1);
+
+ assertThatThrownBy(
+ () ->
+ table.newHybridSearchBuilder()
+ .addFullTextRoute(TEXT_FIELD_NAME, "Paimon", 3, 1.0f)
+ .withFilter(idFilter)
+ .withLimit(3)
+ .routeBuilders())
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining("does not support non-partition filters");
+ }
+
@Test
public void testFullTextSearchMultiTermQuery() throws Exception {
createTableDefault();
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 39379c66c1..75b6897e8e 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
@@ -84,7 +84,7 @@ public class VectorSearchBuilderTest extends TableTestBase {
.column(vectorFieldName, new ArrayType(DataTypes.FLOAT())));
}
- protected Schema.Builder multiVectorSchemaBuilder() {
+ protected Schema.Builder hybridVectorSchemaBuilder() {
return withVectorSchemaOptions(
Schema.newBuilder()
.column("id", DataTypes.INT())
@@ -101,12 +101,12 @@ public class VectorSearchBuilderTest extends TableTestBase {
}
@Test
- public void testMultiVectorSearchBuilderExposesRouteBuilders() throws Exception {
+ public void testHybridSearchBuilderExposesRouteBuilders() throws Exception {
catalog.createTable(
- identifier("multi_vector_builder_table"),
- multiVectorSchemaBuilder().build(),
+ identifier("hybrid_vector_builder_table"),
+ hybridVectorSchemaBuilder().build(),
false);
- FileStoreTable table = getTable(identifier("multi_vector_builder_table"));
+ FileStoreTable table = getTable(identifier("hybrid_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}};
@@ -114,18 +114,18 @@ public class VectorSearchBuilderTest extends TableTestBase {
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)
+ HybridSearchBuilder builder =
+ table.newHybridSearchBuilder()
+ .addVectorRoute("title_vec", new float[] {1.0f, 0.0f}, 2)
+ .addVectorRoute("body_vec", new float[] {0.0f, 1.0f}, 2, 2.0f)
.withLimit(2)
.withWeightedScoreRanker();
- List<MultiVectorSearchBuilder.Route> routes = builder.routeBuilders();
+ List<HybridSearchBuilder.Route> routes = builder.routeBuilders();
assertThat(routes).hasSize(2);
- List<MultiVectorSearchBuilder.RouteResult> routeResults = new ArrayList<>();
- for (MultiVectorSearchBuilder.Route route : routes) {
+ List<HybridSearchBuilder.RouteResult> routeResults = new ArrayList<>();
+ for (HybridSearchBuilder.Route route : routes) {
routeResults.add(
builder.toRouteResult(route, route.vectorSearchBuilder().executeLocal()));
}
diff --git a/paimon-python/pypaimon/table/file_store_table.py b/paimon-python/pypaimon/table/file_store_table.py
index 9fd61abed4..04f972d9e8 100644
--- a/paimon-python/pypaimon/table/file_store_table.py
+++ b/paimon-python/pypaimon/table/file_store_table.py
@@ -429,6 +429,11 @@ class FileStoreTable(Table):
VectorSearchBuilderImpl
return VectorSearchBuilderImpl(self)
+ def new_hybrid_search_builder(self) -> 'HybridSearchBuilder':
+ from pypaimon.table.source.hybrid_search_builder import \
+ HybridSearchBuilderImpl
+ return HybridSearchBuilderImpl(self)
+
def create_row_key_extractor(self) -> RowKeyExtractor:
bucket_mode = self.bucket_mode()
if bucket_mode == BucketMode.HASH_FIXED:
diff --git a/paimon-python/pypaimon/table/format/format_table.py b/paimon-python/pypaimon/table/format/format_table.py
index f0464329f4..795835f47a 100644
--- a/paimon-python/pypaimon/table/format/format_table.py
+++ b/paimon-python/pypaimon/table/format/format_table.py
@@ -109,3 +109,6 @@ class FormatTable(Table):
def new_vector_search_builder(self):
raise NotImplementedError("Format table does not support vector search.")
+
+ def new_hybrid_search_builder(self):
+ raise NotImplementedError("Format table does not support hybrid search.")
diff --git a/paimon-python/pypaimon/table/iceberg/iceberg_table.py b/paimon-python/pypaimon/table/iceberg/iceberg_table.py
index fa88009db3..51f099ee0f 100644
--- a/paimon-python/pypaimon/table/iceberg/iceberg_table.py
+++ b/paimon-python/pypaimon/table/iceberg/iceberg_table.py
@@ -114,3 +114,6 @@ class IcebergTable(Table):
def new_vector_search_builder(self):
raise NotImplementedError("IcebergTable does not support vector search.")
+
+ def new_hybrid_search_builder(self):
+ raise NotImplementedError("IcebergTable does not support hybrid search.")
diff --git a/paimon-python/pypaimon/table/object/object_table.py b/paimon-python/pypaimon/table/object/object_table.py
index 2b43c05d62..d12e00c616 100644
--- a/paimon-python/pypaimon/table/object/object_table.py
+++ b/paimon-python/pypaimon/table/object/object_table.py
@@ -112,3 +112,8 @@ class ObjectTable(Table):
raise NotImplementedError(
"ObjectTable is read-only and does not support vector search."
)
+
+ def new_hybrid_search_builder(self):
+ raise NotImplementedError(
+ "ObjectTable is read-only and does not support hybrid search."
+ )
diff --git a/paimon-python/pypaimon/table/source/full_text_scan.py b/paimon-python/pypaimon/table/source/full_text_scan.py
index 104d06a1b6..0da1607677 100644
--- a/paimon-python/pypaimon/table/source/full_text_scan.py
+++ b/paimon-python/pypaimon/table/source/full_text_scan.py
@@ -46,29 +46,37 @@ class FullTextScan(ABC):
class FullTextScanImpl(FullTextScan):
"""Implementation for FullTextScan."""
- def __init__(self, table: 'FileStoreTable', text_column: 'DataField'):
+ def __init__(
+ self,
+ table: 'FileStoreTable',
+ text_column: 'DataField',
+ partition_filter=None):
self._table = table
self._text_column = text_column
+ self._partition_filter = partition_filter
def scan(self) -> FullTextScanPlan:
from pypaimon.index.index_file_handler import IndexFileHandler
text_column = self._text_column
- snapshot = self._table.snapshot_manager().get_latest_snapshot()
from pypaimon.snapshot.time_travel_util import TimeTravelUtil
from pypaimon.common.options.options import Options
- travel_snapshot = TimeTravelUtil.try_travel_to_snapshot(
+ snapshot = TimeTravelUtil.try_travel_to_snapshot(
Options(self._table.table_schema.options),
self._table.tag_manager(),
self._table.snapshot_manager(),
)
- if travel_snapshot is not None:
- snapshot = travel_snapshot
+ if snapshot is None:
+ snapshot = self._table.snapshot_manager().get_latest_snapshot()
index_file_handler = IndexFileHandler(table=self._table)
+ partition_filter = self._partition_filter
def index_file_filter(entry):
+ if partition_filter is not None:
+ if not partition_filter.test(entry.partition):
+ return False
global_index_meta = entry.index_file.global_index_meta
if global_index_meta is None:
return False
diff --git a/paimon-python/pypaimon/table/source/full_text_search_builder.py b/paimon-python/pypaimon/table/source/full_text_search_builder.py
index b7d92c9934..44534a1337 100644
--- a/paimon-python/pypaimon/table/source/full_text_search_builder.py
+++ b/paimon-python/pypaimon/table/source/full_text_search_builder.py
@@ -20,6 +20,7 @@
from abc import ABC, abstractmethod
from typing import Optional
+from pypaimon.common.predicate_builder import PredicateBuilder
from pypaimon.globalindex.global_index_result import GlobalIndexResult
from pypaimon.table.source.full_text_read import FullTextRead, FullTextReadImpl
from pypaimon.table.source.full_text_scan import FullTextScan, FullTextScanImpl
@@ -48,6 +49,11 @@ class FullTextSearchBuilder(ABC):
"""The default operator for query terms. Supported values are 'or' and 'and'."""
pass
+ @abstractmethod
+ def with_partition_filter(self, partition_filter) -> 'FullTextSearchBuilder':
+ """Partition predicate used to prune index manifest entries."""
+ pass
+
@abstractmethod
def new_full_text_scan(self) -> FullTextScan:
"""Create full-text scan to scan index files."""
@@ -72,6 +78,7 @@ class FullTextSearchBuilderImpl(FullTextSearchBuilder):
self._text_column: Optional['DataField'] = None
self._query_text: Optional[str] = None
self._query_operator: str = "or"
+ self._partition_filter = None
def with_limit(self, limit: int) -> 'FullTextSearchBuilder':
self._limit = limit
@@ -92,10 +99,49 @@ class FullTextSearchBuilderImpl(FullTextSearchBuilder):
self._query_operator = query_operator
return self
+ def with_partition_filter(self, partition_filter) -> 'FullTextSearchBuilder':
+ if partition_filter is None:
+ self._partition_filter = None
+ return self
+ partition_keys = list(self._table.partition_keys or [])
+ if not partition_keys:
+ raise ValueError(
+ "with_partition_filter called on a non-partitioned table")
+ from pypaimon.read.push_down_utils import _get_all_fields
+ referenced = _get_all_fields(partition_filter)
+ extras = referenced - set(partition_keys)
+ if extras:
+ raise ValueError(
+ "Partition filter must reference only partition keys "
+ "(%s); got non-partition field(s): %s"
+ % (partition_keys, sorted(extras)))
+ rebuilt = self._rebuild_leaf_indices_by_name(
+ partition_filter,
+ {name: idx for idx, name in enumerate(partition_keys)},
+ )
+ if self._partition_filter is None:
+ self._partition_filter = rebuilt
+ else:
+ self._partition_filter = PredicateBuilder.and_predicates(
+ [self._partition_filter, rebuilt])
+ return self
+
+ @classmethod
+ def _rebuild_leaf_indices_by_name(cls, predicate, name_to_idx):
+ if predicate.method in ('and', 'or'):
+ return predicate.new_literals(
+ [cls._rebuild_leaf_indices_by_name(c, name_to_idx)
+ for c in (predicate.literals or [])])
+ return predicate.new_index(name_to_idx[predicate.field])
+
def new_full_text_scan(self) -> FullTextScan:
if self._text_column is None:
raise ValueError("Text column must be set via with_text_column()")
- return FullTextScanImpl(self._table, self._text_column)
+ return FullTextScanImpl(
+ self._table,
+ self._text_column,
+ partition_filter=self._partition_filter,
+ )
def new_full_text_read(self) -> FullTextRead:
if self._limit <= 0:
diff --git a/paimon-python/pypaimon/table/source/hybrid_search_builder.py b/paimon-python/pypaimon/table/source/hybrid_search_builder.py
new file mode 100644
index 0000000000..39a93a0167
--- /dev/null
+++ b/paimon-python/pypaimon/table/source/hybrid_search_builder.py
@@ -0,0 +1,450 @@
+# 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.
+
+"""Builder to build hybrid search."""
+
+import heapq
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+from typing import Dict, List, Optional
+
+from pypaimon.common.predicate_builder import PredicateBuilder
+from pypaimon.globalindex.global_index_result import GlobalIndexResult
+from pypaimon.globalindex.vector_search_result import (
+ DictBasedScoredIndexResult,
+ ScoredGlobalIndexResult,
+)
+
+RRF_RANKER = "rrf"
+WEIGHTED_SCORE_RANKER = "weighted_score"
+_RRF_K = 60.0
+
+
+def _normalize_ranker(ranker: Optional[str]) -> str:
+ if ranker is None or not ranker.strip():
+ return RRF_RANKER
+ normalized = ranker.strip().lower()
+ if normalized not in (RRF_RANKER, WEIGHTED_SCORE_RANKER):
+ raise ValueError("Unsupported hybrid ranker: %s" % ranker)
+ return normalized
+
+
+@dataclass
+class HybridSearchRoute:
+ """A single route in a hybrid search."""
+
+ route_type: str
+ field_name: str
+ limit: int
+ weight: float = 1.0
+ vector: Optional[List[float]] = None
+ query_text: Optional[str] = None
+ query_operator: str = "or"
+ options: Dict[str, str] = field(default_factory=dict)
+
+ VECTOR = "vector"
+ FULL_TEXT = "full_text"
+
+ def __post_init__(self):
+ if self.route_type not in (self.VECTOR, self.FULL_TEXT):
+ raise ValueError("Unsupported hybrid route type: %s" % self.route_type)
+ if not self.field_name:
+ raise ValueError("Field name cannot be None or empty")
+ if self.route_type == self.VECTOR and self.vector is None:
+ raise ValueError("Search vector cannot be None for vector route")
+ if self.route_type == self.FULL_TEXT and not self.query_text:
+ raise ValueError(
+ "Query text cannot be None or empty for full-text route")
+ if self.limit <= 0:
+ raise ValueError("Limit must be positive, got: %s" % self.limit)
+ if self.weight <= 0:
+ raise ValueError("Weight must be positive, got: %s" % self.weight)
+ operator = "or" if self.query_operator is None else (
+ self.query_operator.strip().lower())
+ if operator not in ("or", "and"):
+ raise ValueError(
+ "Query operator must be 'or' or 'and', got: %s"
+ % self.query_operator)
+ self.query_operator = operator
+ self.options = dict(self.options or {})
+
+ @classmethod
+ def vector_route(
+ cls,
+ field_name: str,
+ vector: List[float],
+ limit: int,
+ weight: float = 1.0,
+ options: Optional[Dict[str, str]] = None) -> 'HybridSearchRoute':
+ return cls(
+ route_type=cls.VECTOR,
+ field_name=field_name,
+ vector=vector,
+ limit=limit,
+ weight=weight,
+ options=dict(options or {}),
+ )
+
+ @classmethod
+ def full_text_route(
+ cls,
+ field_name: str,
+ query_text: str,
+ query_operator: str,
+ limit: int,
+ weight: float = 1.0,
+ options: Optional[Dict[str, str]] = None) -> 'HybridSearchRoute':
+ return cls(
+ route_type=cls.FULL_TEXT,
+ field_name=field_name,
+ query_text=query_text,
+ query_operator=query_operator,
+ limit=limit,
+ weight=weight,
+ options=dict(options or {}),
+ )
+
+ def is_vector(self) -> bool:
+ return self.route_type == self.VECTOR
+
+ def is_full_text(self) -> bool:
+ return self.route_type == self.FULL_TEXT
+
+
+@dataclass
+class HybridSearchRouteResult:
+ route: HybridSearchRoute
+ result: ScoredGlobalIndexResult
+
+
+class HybridSearchRouteBuilder:
+ """A route plus its configured search builder."""
+
+ def __init__(self, route, search_builder):
+ self.route = route
+ self.search_builder = search_builder
+
+ def execute_local(self):
+ return self.search_builder.execute_local()
+
+
+class HybridSearchBuilder(ABC):
+ """Builder to build hybrid search."""
+
+ @abstractmethod
+ def with_limit(self, limit: int) -> 'HybridSearchBuilder':
+ """The final top k ranked results to return."""
+ pass
+
+ @abstractmethod
+ def with_ranker(self, ranker: str) -> 'HybridSearchBuilder':
+ """Ranker for combining route results."""
+ pass
+
+ def with_rrf_ranker(self) -> 'HybridSearchBuilder':
+ """Use reciprocal rank fusion to combine route results."""
+ return self.with_ranker(RRF_RANKER)
+
+ def with_weighted_score_ranker(self) -> 'HybridSearchBuilder':
+ """Use weighted score to combine route results."""
+ return self.with_ranker(WEIGHTED_SCORE_RANKER)
+
+ @abstractmethod
+ def with_filter(self, predicate) -> 'HybridSearchBuilder':
+ """Scalar predicate used to pre-filter vector routes."""
+ pass
+
+ @abstractmethod
+ def with_partition_filter(self, partition_filter) -> 'HybridSearchBuilder':
+ """Partition predicate used to prune index manifest entries."""
+ pass
+
+ @abstractmethod
+ def add_route(self, route: HybridSearchRoute) -> 'HybridSearchBuilder':
+ """Add a hybrid-search route."""
+ pass
+
+ def add_vector_route(
+ self,
+ vector_column: str,
+ vector: List[float],
+ limit: int,
+ weight: float = 1.0,
+ options: Optional[Dict[str, str]] = None) -> 'HybridSearchBuilder':
+ """Add a vector-search route."""
+ return self.add_route(
+ HybridSearchRoute.vector_route(
+ vector_column, vector, limit, weight, options))
+
+ def add_full_text_route(
+ self,
+ text_column: str,
+ query_text: str,
+ limit: int,
+ weight: float = 1.0,
+ query_operator: str = "or",
+ options: Optional[Dict[str, str]] = None) -> 'HybridSearchBuilder':
+ """Add a full-text-search route."""
+ return self.add_route(
+ HybridSearchRoute.full_text_route(
+ text_column, query_text, query_operator, limit, weight, options))
+
+ @abstractmethod
+ def route_builders(self) -> List[HybridSearchRouteBuilder]:
+ """Create one search builder for each route."""
+ pass
+
+ @abstractmethod
+ def to_route_result(
+ self,
+ route_builder: HybridSearchRouteBuilder,
+ result: GlobalIndexResult) -> HybridSearchRouteResult:
+ """Convert a search result into a weighted route result."""
+ pass
+
+ @abstractmethod
+ def rank(
+ self,
+ route_results: List[HybridSearchRouteResult]) -> ScoredGlobalIndexResult:
+ """Rank route results."""
+ pass
+
+ def execute_local(self) -> ScoredGlobalIndexResult:
+ """Execute hybrid index search locally."""
+ route_builders = self.route_builders()
+ route_results = []
+ for route_builder in route_builders:
+ route_results.append(
+ self.to_route_result(
+ route_builder, route_builder.execute_local()))
+ return self.rank(route_results)
+
+
+class HybridSearchBuilderImpl(HybridSearchBuilder):
+ """Implementation for HybridSearchBuilder."""
+
+ def __init__(self, table):
+ self._table = table
+ self._routes = []
+ self._limit = 0
+ self._ranker = RRF_RANKER
+ self._filter = None
+ self._partition_filter = None
+
+ def with_limit(self, limit: int) -> 'HybridSearchBuilder':
+ self._limit = limit
+ return self
+
+ def with_ranker(self, ranker: str) -> 'HybridSearchBuilder':
+ self._ranker = _normalize_ranker(ranker)
+ return self
+
+ def with_filter(self, predicate) -> 'HybridSearchBuilder':
+ if predicate is None:
+ return self
+ partition_filter, data_filter = self._split_partition_filter(predicate)
+ if partition_filter is not None:
+ self._add_partition_filter(partition_filter)
+ if data_filter is not None:
+ if self._filter is None:
+ self._filter = data_filter
+ else:
+ self._filter = PredicateBuilder.and_predicates(
+ [self._filter, data_filter])
+ return self
+
+ def with_partition_filter(self, partition_filter) -> 'HybridSearchBuilder':
+ if partition_filter is not None:
+ self._add_partition_filter(
+ self._validate_and_rebuild_partition_filter(partition_filter))
+ return self
+
+ def add_route(self, route: HybridSearchRoute) -> 'HybridSearchBuilder':
+ if route is None:
+ raise ValueError("Route cannot be None")
+ self._routes.append(route)
+ return self
+
+ def route_builders(self) -> List[HybridSearchRouteBuilder]:
+ self._validate_search()
+ builders = []
+ for route in self._routes:
+ if route.is_vector():
+ builders.append(
+ HybridSearchRouteBuilder(
+ route, self._new_vector_search_builder(route)))
+ else:
+ builders.append(
+ HybridSearchRouteBuilder(
+ route, self._new_full_text_search_builder(route)))
+ return builders
+
+ def to_route_result(
+ self,
+ route_builder: HybridSearchRouteBuilder,
+ result: GlobalIndexResult) -> HybridSearchRouteResult:
+ if isinstance(result, ScoredGlobalIndexResult):
+ scored = result
+ elif result.results().is_empty():
+ scored = ScoredGlobalIndexResult.create_empty()
+ else:
+ raise ValueError(
+ "Hybrid search requires scored index results, but got: %s"
+ % result.__class__.__name__)
+ return HybridSearchRouteResult(route_builder.route, scored)
+
+ def rank(
+ self,
+ route_results: List[HybridSearchRouteResult]) -> ScoredGlobalIndexResult:
+ self._validate_search()
+ non_empty = [
+ route_result for route_result in route_results
+ if not route_result.result.results().is_empty()
+ ]
+ if self._ranker == WEIGHTED_SCORE_RANKER:
+ return self._weighted_score(non_empty)
+ return self._rrf(non_empty)
+
+ def _validate_search(self):
+ if not self._routes:
+ raise ValueError("Routes cannot be empty")
+ if self._limit <= 0:
+ raise ValueError("Limit must be positive, got: %s" % self._limit)
+ if self._filter is not None:
+ for route in self._routes:
+ if route.is_full_text():
+ raise ValueError(
+ "Hybrid search with full-text routes does not support "
+ "non-partition filters because full-text indexes cannot "
+ "apply row-id pre-filters before top-k ranking.")
+
+ def _new_vector_search_builder(self, route):
+ builder = (
+ self._table.new_vector_search_builder()
+ .with_vector_column(route.field_name)
+ .with_query_vector(route.vector)
+ .with_limit(route.limit)
+ .with_options(route.options)
+ )
+ if self._partition_filter is not None:
+ builder.with_partition_filter(self._partition_filter)
+ if self._filter is not None:
+ builder.with_filter(self._filter)
+ return builder
+
+ def _new_full_text_search_builder(self, route):
+ builder = (
+ self._table.new_full_text_search_builder()
+ .with_text_column(route.field_name)
+ .with_query_text(route.query_text)
+ .with_query_operator(route.query_operator)
+ .with_limit(route.limit)
+ )
+ if self._partition_filter is not None:
+ builder.with_partition_filter(self._partition_filter)
+ return builder
+
+ def _rrf(self, route_results):
+ scores = {}
+ for route_result in route_results:
+ result = route_result.result
+ score_getter = result.score_getter()
+ row_ids = sorted(
+ result.results(),
+ key=lambda row_id: (
+ -(score_getter(row_id) or 0.0), row_id))
+ for rank, row_id in enumerate(row_ids):
+ contribution = route_result.route.weight / (_RRF_K + rank + 1.0)
+ scores[row_id] = scores.get(row_id, 0.0) + contribution
+ return _top_k(scores, self._limit)
+
+ def _weighted_score(self, route_results):
+ scores = {}
+ for route_result in route_results:
+ result = route_result.result
+ score_getter = result.score_getter()
+ for row_id in result.results():
+ contribution = route_result.route.weight * (
+ score_getter(row_id) or 0.0)
+ scores[row_id] = scores.get(row_id, 0.0) + contribution
+ return _top_k(scores, self._limit)
+
+ def _split_partition_filter(self, predicate):
+ partition_keys = list(self._table.partition_keys or [])
+ if not partition_keys:
+ return None, predicate
+ from pypaimon.read.push_down_utils import _get_all_fields, _split_and
+ partition_key_set = set(partition_keys)
+ pk_to_idx = {name: idx for idx, name in enumerate(partition_keys)}
+ partition_parts = []
+ data_parts = []
+ for part in _split_and(predicate):
+ if _get_all_fields(part).issubset(partition_key_set):
+ partition_parts.append(
+ self._rebuild_leaf_indices_by_name(part, pk_to_idx))
+ else:
+ data_parts.append(part)
+ return (
+ PredicateBuilder.and_predicates(partition_parts),
+ PredicateBuilder.and_predicates(data_parts),
+ )
+
+ def _validate_and_rebuild_partition_filter(self, partition_filter):
+ partition_keys = list(self._table.partition_keys or [])
+ if not partition_keys:
+ raise ValueError(
+ "with_partition_filter called on a non-partitioned table")
+ from pypaimon.read.push_down_utils import _get_all_fields
+ referenced = _get_all_fields(partition_filter)
+ extras = referenced - set(partition_keys)
+ if extras:
+ raise ValueError(
+ "Partition filter must reference only partition keys "
+ "(%s); got non-partition field(s): %s"
+ % (partition_keys, sorted(extras)))
+ return self._rebuild_leaf_indices_by_name(
+ partition_filter,
+ {name: idx for idx, name in enumerate(partition_keys)})
+
+ def _add_partition_filter(self, partition_filter):
+ if self._partition_filter is None:
+ self._partition_filter = partition_filter
+ else:
+ self._partition_filter = PredicateBuilder.and_predicates(
+ [self._partition_filter, partition_filter])
+
+ @classmethod
+ def _rebuild_leaf_indices_by_name(cls, predicate, name_to_idx):
+ if predicate.method in ('and', 'or'):
+ return predicate.new_literals(
+ [cls._rebuild_leaf_indices_by_name(c, name_to_idx)
+ for c in (predicate.literals or [])])
+ return predicate.new_index(name_to_idx[predicate.field])
+
+
+def _top_k(scores, limit):
+ if not scores:
+ return ScoredGlobalIndexResult.create_empty()
+ heap = []
+ for row_id, score in scores.items():
+ item = (score, -row_id)
+ if len(heap) < limit:
+ heapq.heappush(heap, item)
+ elif item > heap[0]:
+ heapq.heapreplace(heap, item)
+ top_scores = {-neg_row_id: score for score, neg_row_id in heap}
+ return DictBasedScoredIndexResult(top_scores)
diff --git a/paimon-python/pypaimon/table/system/system_table.py b/paimon-python/pypaimon/table/system/system_table.py
index 69bc14d7dd..8432470076 100644
--- a/paimon-python/pypaimon/table/system/system_table.py
+++ b/paimon-python/pypaimon/table/system/system_table.py
@@ -102,6 +102,9 @@ class SystemTable(Table):
def new_vector_search_builder(self):
raise NotImplementedError(_READ_ONLY_MESSAGE)
+ def new_hybrid_search_builder(self):
+ raise NotImplementedError(_READ_ONLY_MESSAGE)
+
class SystemReadBuilder:
"""ReadBuilder-shaped facade exposing the system table's data.
diff --git a/paimon-python/pypaimon/table/table.py b/paimon-python/pypaimon/table/table.py
index 58a59f5b63..a89eab6e74 100644
--- a/paimon-python/pypaimon/table/table.py
+++ b/paimon-python/pypaimon/table/table.py
@@ -20,6 +20,7 @@ from abc import ABC, abstractmethod
from pypaimon.read.read_builder import ReadBuilder
from pypaimon.read.stream_read_builder import StreamReadBuilder
from pypaimon.table.source.full_text_search_builder import FullTextSearchBuilder
+from pypaimon.table.source.hybrid_search_builder import HybridSearchBuilder
from pypaimon.table.source.vector_search_builder import VectorSearchBuilder
from pypaimon.write.write_builder import BatchWriteBuilder, StreamWriteBuilder
@@ -50,3 +51,7 @@ class Table(ABC):
@abstractmethod
def new_vector_search_builder(self) -> VectorSearchBuilder:
"""Returns a new vector search builder."""
+
+ @abstractmethod
+ def new_hybrid_search_builder(self) -> HybridSearchBuilder:
+ """Returns a new hybrid search builder."""
diff --git a/paimon-python/pypaimon/tests/system/system_table_test.py b/paimon-python/pypaimon/tests/system/system_table_test.py
index fbb05cb22a..60a505679a 100644
--- a/paimon-python/pypaimon/tests/system/system_table_test.py
+++ b/paimon-python/pypaimon/tests/system/system_table_test.py
@@ -101,6 +101,7 @@ class SystemTableTest(unittest.TestCase):
"new_stream_write_builder",
"new_full_text_search_builder",
"new_vector_search_builder",
+ "new_hybrid_search_builder",
):
with self.assertRaises(NotImplementedError) as ctx:
getattr(sys_table, method_name)()
diff --git a/paimon-python/pypaimon/tests/vector_search_filter_test.py b/paimon-python/pypaimon/tests/vector_search_filter_test.py
index 1605d51a6c..90cb5bb0e4 100644
--- a/paimon-python/pypaimon/tests/vector_search_filter_test.py
+++ b/paimon-python/pypaimon/tests/vector_search_filter_test.py
@@ -81,6 +81,18 @@ class _StubTable:
return _G()
return _P()
+ def new_vector_search_builder(self):
+ from pypaimon.table.source.vector_search_builder import (
+ VectorSearchBuilderImpl,
+ )
+ return VectorSearchBuilderImpl(self)
+
+ def new_full_text_search_builder(self):
+ from pypaimon.table.source.full_text_search_builder import (
+ FullTextSearchBuilderImpl,
+ )
+ return FullTextSearchBuilderImpl(self)
+
def _field(fid, name, dtype="INT"):
return DataField(id=fid, name=name,
@@ -1253,6 +1265,170 @@ class VectorSearchPartitionedFilterTest(unittest.TestCase):
self.assertIn("non-partition", str(ctx.exception))
+class HybridSearchBuilderTest(unittest.TestCase):
+
+ def tearDown(self):
+ mock.patch.stopall()
+
+ def test_hybrid_search_ranks_multiple_routes(self):
+ from pypaimon.globalindex.vector_search_result import (
+ DictBasedScoredIndexResult,
+ )
+ from pypaimon.table.source.hybrid_search_builder import (
+ HybridSearchBuilderImpl,
+ )
+
+ id_field = _field(0, "id")
+ title_embedding = _field(1, "title_embedding", "FLOAT")
+ body_embedding = _field(2, "body_embedding", "FLOAT")
+ content = _field(3, "content", "STRING")
+ table = _StubTable(
+ fields=[id_field, title_embedding, body_embedding, content],
+ entries=[],
+ )
+
+ captured_builders = []
+
+ class _FakeRouteBuilder:
+ def __init__(self, result):
+ self._result = result
+ self.calls = []
+
+ def with_vector_column(self, name):
+ self.calls.append(("vector_column", name))
+ return self
+
+ def with_query_vector(self, vector):
+ self.calls.append(("query_vector", vector))
+ return self
+
+ def with_text_column(self, name):
+ self.calls.append(("text_column", name))
+ return self
+
+ def with_query_text(self, query_text):
+ self.calls.append(("query_text", query_text))
+ return self
+
+ def with_query_operator(self, query_operator):
+ self.calls.append(("query_operator", query_operator))
+ return self
+
+ def with_limit(self, limit):
+ self.calls.append(("limit", limit))
+ return self
+
+ def with_options(self, options):
+ self.calls.append(("options", options))
+ return self
+
+ def execute_local(self):
+ return self._result
+
+ vector_result = DictBasedScoredIndexResult({1: 0.9, 2: 0.8})
+ text_result = DictBasedScoredIndexResult({2: 0.7, 3: 0.6})
+
+ def new_vector_search_builder():
+ builder = _FakeRouteBuilder(vector_result)
+ captured_builders.append(builder)
+ return builder
+
+ def new_full_text_search_builder():
+ builder = _FakeRouteBuilder(text_result)
+ captured_builders.append(builder)
+ return builder
+
+ table.new_vector_search_builder = new_vector_search_builder
+ table.new_full_text_search_builder = new_full_text_search_builder
+
+ result = (
+ HybridSearchBuilderImpl(table)
+ .add_vector_route(
+ "title_embedding", [1.0, 0.0], 10, weight=1.0,
+ options={"ivf.nprobe": "32"})
+ .add_full_text_route(
+ "content", "paimon search", 10, weight=1.0,
+ query_operator="and")
+ .with_limit(2)
+ .with_rrf_ranker()
+ .execute_local()
+ )
+
+ self.assertEqual(2, result.results().cardinality())
+ self.assertIn(2, list(result.results()))
+ self.assertGreater(result.score_getter()(2), result.score_getter()(1))
+ self.assertIn(("options", {"ivf.nprobe": "32"}),
+ captured_builders[0].calls)
+ self.assertIn(("query_operator", "and"), captured_builders[1].calls)
+
+ def test_hybrid_search_rejects_data_filter_with_full_text_route(self):
+ from pypaimon.table.source.hybrid_search_builder import (
+ HybridSearchBuilderImpl,
+ )
+
+ id_field = _field(0, "id")
+ content = _field(1, "content", "STRING")
+ table = _StubTable(fields=[id_field, content], entries=[])
+ pb = PredicateBuilder(table.fields)
+
+ builder = (
+ HybridSearchBuilderImpl(table)
+ .add_full_text_route("content", "paimon search", 10)
+ .with_filter(pb.equal("id", 1))
+ .with_limit(5)
+ )
+
+ with self.assertRaises(ValueError) as ctx:
+ builder.route_builders()
+ self.assertIn("full-text routes", str(ctx.exception))
+
+ def test_hybrid_search_partition_filter_prunes_full_text_route(self):
+ from pypaimon.table.source.hybrid_search_builder import (
+ HybridSearchBuilderImpl,
+ )
+
+ pt_field = _field(0, "pt")
+ content = _field(1, "content", "STRING")
+ partition_pt1 = GenericRow([1], [pt_field])
+ partition_pt2 = GenericRow([2], [pt_field])
+ entries = [
+ _entry(partition_pt1, field_id=1, index_type="tantivy-fulltext",
+ file_name="ft-pt1.index", row_range_start=0,
+ row_range_end=4),
+ _entry(partition_pt2, field_id=1, index_type="tantivy-fulltext",
+ file_name="ft-pt2.index", row_range_start=5,
+ row_range_end=9),
+ ]
+ table = _StubTable(
+ fields=[pt_field, content],
+ entries=entries,
+ partition_fields=[pt_field],
+ )
+ _patch_snapshot(self, entries)
+
+ pb = PredicateBuilder(table.fields)
+ route_builders = (
+ HybridSearchBuilderImpl(table)
+ .add_full_text_route("content", "paimon search", 10)
+ .with_filter(pb.equal("pt", 2))
+ .with_limit(5)
+ .route_builders()
+ )
+ splits = (
+ route_builders[0]
+ .search_builder
+ .new_full_text_scan()
+ .scan()
+ .splits()
+ )
+
+ self.assertEqual(1, len(splits))
+ self.assertEqual(
+ ["ft-pt2.index"],
+ [f.file_name for f in splits[0].full_text_index_files],
+ )
+
+
class VectorSearchManySplitsTest(unittest.TestCase):
def test_vector_search_with_many_splits(self):
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 ec652fad3b..e1fd258186 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, MultiVectorSearch, Predicate, TopN, VectorSearch}
+import org.apache.paimon.predicate.{FullTextSearch, HybridSearch, Predicate, TopN, VectorSearch}
import org.apache.paimon.spark.read.VariantExtractionInfo
import org.apache.paimon.table.InnerTable
@@ -33,7 +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 pushedHybridSearch: Option[HybridSearch] = 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 daa1a56122..5253cc9894 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,11 +25,11 @@ import org.apache.spark.sql.connector.read.Scan
class PaimonScanBuilder(val table: InnerTable) extends PaimonBaseScanBuilder {
override def build(): Scan = {
- val (actualTable, vectorSearch, multiVectorSearch, fullTextSearch) = table match {
+ val (actualTable, vectorSearch, hybridSearch, fullTextSearch) = table match {
case vst: org.apache.paimon.table.VectorSearchTable =>
(vst.origin(), Option(vst.vectorSearch()), None, None)
- case mvst: org.apache.paimon.table.MultiVectorSearchTable =>
- (mvst.origin(), None, Option(mvst.multiVectorSearch()), None)
+ case hst: org.apache.paimon.table.HybridSearchTable =>
+ (hst.origin(), None, Option(hst.hybridSearch()), None)
case ftst: org.apache.paimon.table.FullTextSearchTable =>
(ftst.origin(), None, None, Option(ftst.fullTextSearch()))
case _ => (table, pushedVectorSearch, None, pushedFullTextSearch)
@@ -42,7 +42,7 @@ class PaimonScanBuilder(val table: InnerTable) extends PaimonBaseScanBuilder {
pushedLimit,
pushedTopN,
vectorSearch,
- multiVectorSearch,
+ hybridSearch,
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 c35589075b..6ecc3dd847 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, MultiVectorSearch, Predicate, TopN, VectorSearch}
+import org.apache.paimon.predicate.{FullTextSearch, HybridSearch, 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,7 +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 pushedHybridSearch: Option[HybridSearch] = 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/SparkHybridSearchBuilderImpl.java
similarity index 79%
rename from paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/read/SparkMultiVectorSearchBuilderImpl.java
rename to paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/read/SparkHybridSearchBuilderImpl.java
index cc42783094..7472e72bc9 100644
--- 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/SparkHybridSearchBuilderImpl.java
@@ -18,22 +18,22 @@
package org.apache.paimon.spark.read;
-import org.apache.paimon.predicate.MultiVectorSearchRoute;
+import org.apache.paimon.predicate.HybridSearchRoute;
import org.apache.paimon.table.InnerTable;
-import org.apache.paimon.table.source.MultiVectorSearchBuilderImpl;
+import org.apache.paimon.table.source.HybridSearchBuilderImpl;
import org.apache.paimon.table.source.VectorSearchBuilder;
-/** Spark-aware {@link MultiVectorSearchBuilderImpl}. */
-public class SparkMultiVectorSearchBuilderImpl extends MultiVectorSearchBuilderImpl {
+/** Spark-aware {@link HybridSearchBuilderImpl}. */
+public class SparkHybridSearchBuilderImpl extends HybridSearchBuilderImpl {
private static final long serialVersionUID = 1L;
- public SparkMultiVectorSearchBuilderImpl(InnerTable table) {
+ public SparkHybridSearchBuilderImpl(InnerTable table) {
super(table);
}
@Override
- protected VectorSearchBuilder newVectorSearchBuilder(MultiVectorSearchRoute route) {
+ protected VectorSearchBuilder newVectorSearchBuilder(HybridSearchRoute route) {
VectorSearchBuilder vectorSearchBuilder =
new SparkVectorSearchBuilderImpl(table)
.withVector(route.vector())
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 274c5b841c..b4453288cf 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, SparkMultiVectorSearchBuilderImpl, SparkVectorSearchBuilderImpl}
+import org.apache.paimon.spark.read.{BaseScan, BatchReadTagCleanupListener, PaimonSupportsRuntimeFiltering, SparkHybridSearchBuilderImpl, SparkVectorSearchBuilderImpl}
import org.apache.paimon.spark.sources.PaimonMicroBatchStream
import org.apache.paimon.spark.util.OptionUtils
import org.apache.paimon.table.{DataTable, FileStoreTable, InnerTable}
@@ -65,16 +65,16 @@ abstract class PaimonBaseScan(table: InnerTable)
private def evalGlobalIndexSearch(): GlobalIndexResult = {
val globalSearchCount =
- Seq(pushedVectorSearch, pushedMultiVectorSearch, pushedFullTextSearch).count(_.isDefined)
+ Seq(pushedVectorSearch, pushedHybridSearch, pushedFullTextSearch).count(_.isDefined)
if (globalSearchCount > 1) {
throw new UnsupportedOperationException(
- "Cannot push down vector search, multi-vector search and full-text search simultaneously.")
+ "Cannot push down vector search, hybrid search and full-text search simultaneously.")
}
if (pushedVectorSearch.isDefined) {
return evalVectorSearch()
}
- if (pushedMultiVectorSearch.isDefined) {
- return evalMultiVectorSearch()
+ if (pushedHybridSearch.isDefined) {
+ return evalHybridSearch()
}
if (pushedFullTextSearch.isDefined) {
return evalFullTextSearch()
@@ -104,18 +104,18 @@ abstract class PaimonBaseScan(table: InnerTable)
vectorBuilder.newVectorRead().read(vectorBuilder.newVectorScan().scan())
}
- private def evalMultiVectorSearch(): GlobalIndexResult = {
- val multiVectorSearch = pushedMultiVectorSearch.get
- val multiVectorSearchBuilder =
+ private def evalHybridSearch(): GlobalIndexResult = {
+ val hybridSearch = pushedHybridSearch.get
+ val hybridSearchBuilder =
if (CoreOptions.fromMap(table.options).vectorSearchDistributeEnabled()) {
- new SparkMultiVectorSearchBuilderImpl(table)
+ new SparkHybridSearchBuilderImpl(table)
} else {
- table.newMultiVectorSearchBuilder()
+ table.newHybridSearchBuilder()
}
- val builder = multiVectorSearchBuilder
- .withLimit(multiVectorSearch.limit())
- .withRanker(multiVectorSearch.ranker())
- multiVectorSearch.routes().asScala.foreach(route => builder.addRoute(route))
+ val builder = hybridSearchBuilder
+ .withLimit(hybridSearch.limit())
+ .withRanker(hybridSearch.ranker())
+ hybridSearch.routes().asScala.foreach(route => builder.addRoute(route))
if (pushedPartitionFilters.nonEmpty) {
builder.withPartitionFilter(PartitionPredicate.and(pushedPartitionFilters.asJava))
}
@@ -130,8 +130,17 @@ abstract class PaimonBaseScan(table: InnerTable)
val ftBuilder = table
.newFullTextSearchBuilder()
.withQueryText(fullTextSearch.queryText())
+ .withQueryOperator(fullTextSearch.queryOperator())
.withTextColumn(fullTextSearch.fieldName())
.withLimit(fullTextSearch.limit())
+ if (pushedPartitionFilters.nonEmpty) {
+ ftBuilder.withPartitionFilter(PartitionPredicate.and(pushedPartitionFilters.asJava))
+ }
+ if (pushedDataFilters.nonEmpty) {
+ throw new UnsupportedOperationException(
+ "Full-text search does not support non-partition filters because full-text indexes " +
+ "cannot apply row-id pre-filters before top-k ranking.")
+ }
ftBuilder.newFullTextRead().read(ftBuilder.newFullTextScan().scan())
}
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonRecordReaderIterator.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonRecordReaderIterator.scala
index 0d67001f0e..ac06698fe9 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonRecordReaderIterator.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonRecordReaderIterator.scala
@@ -22,7 +22,7 @@ import org.apache.paimon.data.{BinaryString, GenericRow, InternalRow => PaimonIn
import org.apache.paimon.fs.Path
import org.apache.paimon.reader.{FileRecordIterator, RecordReader, ScoreRecordIterator}
import org.apache.paimon.spark.schema.PaimonMetadataColumn
-import org.apache.paimon.spark.schema.PaimonMetadataColumn.{PARTITION_AND_BUCKET_META_COLUMNS, PATH_AND_INDEX_META_COLUMNS, VECTOR_SEARCH_SCORE_COLUMN}
+import org.apache.paimon.spark.schema.PaimonMetadataColumn.{PARTITION_AND_BUCKET_META_COLUMNS, PATH_AND_INDEX_META_COLUMNS, SEARCH_SCORE_COLUMN}
import org.apache.paimon.table.source.{DataSplit, Split}
import org.apache.paimon.utils.{CloseableIterator, Preconditions}
@@ -49,7 +49,7 @@ case class PaimonRecordReaderIterator(
private val needPathAndIndexMetadata =
metadataColumns.exists(c => PATH_AND_INDEX_META_COLUMNS.contains(c.name))
private val needScoreMetadata = {
- metadataColumns.exists(_.name == VECTOR_SEARCH_SCORE_COLUMN)
+ metadataColumns.exists(_.name == SEARCH_SCORE_COLUMN)
}
Preconditions.checkArgument(!needScoreMetadata || metadataColumns.size == 1)
@@ -179,7 +179,7 @@ case class PaimonRecordReaderIterator(
metadataColumns.zipWithIndex.foreach {
case (metadataColumn, index) =>
metadataColumn.name match {
- case PaimonMetadataColumn.VECTOR_SEARCH_SCORE_COLUMN =>
+ case PaimonMetadataColumn.SEARCH_SCORE_COLUMN =>
metadataRow.setField(index, fileRecordIterator.returnedScore())
}
}
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 09ece0bfea..412d92b7f4 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, MultiVectorSearch, Predicate, TopN, VectorSearch}
+import org.apache.paimon.predicate.{FullTextSearch, HybridSearch, 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,7 +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 pushedHybridSearch: Option[HybridSearch] = 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 9e2e853ee2..23f0924f1f 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,11 +129,11 @@ class PaimonScanBuilder(val table: InnerTable)
localScan match {
case Some(scan) => scan
case None =>
- val (actualTable, vectorSearch, multiVectorSearch, fullTextSearch) = table match {
+ val (actualTable, vectorSearch, hybridSearch, fullTextSearch) = table match {
case vst: org.apache.paimon.table.VectorSearchTable =>
(vst.origin(), Option(vst.vectorSearch()), None, None)
- case mvst: org.apache.paimon.table.MultiVectorSearchTable =>
- (mvst.origin(), None, Option(mvst.multiVectorSearch()), None)
+ case hst: org.apache.paimon.table.HybridSearchTable =>
+ (hst.origin(), None, Option(hst.hybridSearch()), None)
case ftst: org.apache.paimon.table.FullTextSearchTable =>
(ftst.origin(), None, None, Option(ftst.fullTextSearch()))
case _ => (table, pushedVectorSearch, None, pushedFullTextSearch)
@@ -147,7 +147,7 @@ class PaimonScanBuilder(val table: InnerTable)
pushedLimit,
pushedTopN,
vectorSearch,
- multiVectorSearch,
+ hybridSearch,
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 084174a823..bd69174be7 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,8 +118,12 @@ abstract class PaimonSparkTableBase(val table: Table)
_metadataColumns.append(PaimonMetadataColumn.ROW_ID)
_metadataColumns.append(PaimonMetadataColumn.SEQUENCE_NUMBER)
}
- if (table.isInstanceOf[VectorSearchTable] || table.isInstanceOf[MultiVectorSearchTable]) {
- _metadataColumns.append(PaimonMetadataColumn.VECTOR_SEARCH_SCORE)
+ if (
+ table.isInstanceOf[VectorSearchTable] ||
+ table.isInstanceOf[HybridSearchTable] ||
+ table.isInstanceOf[FullTextSearchTable]
+ ) {
+ _metadataColumns.append(PaimonMetadataColumn.SEARCH_SCORE)
}
_metadataColumns.appendAll(
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 7efd67960e..46af6bb38a 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,11 +19,11 @@
package org.apache.paimon.spark.catalyst.plans.logical
import org.apache.paimon.CoreOptions
-import org.apache.paimon.globalindex.MultiVectorSearchRanker
-import org.apache.paimon.predicate.{FullTextSearch, MultiVectorSearch, MultiVectorSearchRoute, VectorSearch}
+import org.apache.paimon.globalindex.HybridSearchRanker
+import org.apache.paimon.predicate.{FullTextSearch, HybridSearch, HybridSearchRoute, VectorSearch}
import org.apache.paimon.spark.SparkTable
import org.apache.paimon.spark.catalyst.plans.logical.PaimonTableValuedFunctions._
-import org.apache.paimon.table.{DataTable, FullTextSearchTable, InnerTable, MultiVectorSearchTable, VectorSearchTable}
+import org.apache.paimon.table.{DataTable, FullTextSearchTable, HybridSearchTable, InnerTable, VectorSearchTable}
import org.apache.paimon.table.source.snapshot.TimeTravelUtil.InconsistentTagBucketException
import org.apache.spark.sql.PaimonUtils.createDataset
@@ -47,7 +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 HYBRID_SEARCH = "hybrid_search"
val FULL_TEXT_SEARCH = "full_text_search"
val supportedFnNames: Seq[String] =
@@ -56,7 +56,7 @@ object PaimonTableValuedFunctions {
INCREMENTAL_BETWEEN_TIMESTAMP,
INCREMENTAL_TO_AUTO_TAG,
VECTOR_SEARCH,
- MULTI_VECTOR_SEARCH,
+ HYBRID_SEARCH,
FULL_TEXT_SEARCH)
def parsePositiveLimit(value: Any): Int = {
@@ -88,8 +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 HYBRID_SEARCH =>
+ FunctionRegistryBase.build[HybridSearchQuery](fnName, since = None)
case FULL_TEXT_SEARCH =>
FunctionRegistryBase.build[FullTextSearchQuery](fnName, since = None)
case _ =>
@@ -122,12 +122,12 @@ object PaimonTableValuedFunctions {
val ident: Identifier = Identifier.of(Array(dbName), tableName)
val sparkTable = sparkCatalog.loadTable(ident)
- // Handle vector_search and full_text_search specially
+ // Handle search table-valued functions specially.
tvf match {
case vsq: VectorSearchQuery =>
resolveVectorSearchQuery(sparkTable, sparkCatalog, ident, vsq, args.tail)
- case mvsq: MultiVectorSearchQuery =>
- resolveMultiVectorSearchQuery(sparkTable, sparkCatalog, ident, mvsq, args.tail)
+ case hsq: HybridSearchQuery =>
+ resolveHybridSearchQuery(sparkTable, sparkCatalog, ident, hsq, args.tail)
case ftsq: FullTextSearchQuery =>
resolveFullTextSearchQuery(sparkTable, sparkCatalog, ident, ftsq, args.tail)
case _ =>
@@ -167,24 +167,24 @@ object PaimonTableValuedFunctions {
}
}
- private def resolveMultiVectorSearchQuery(
+ private def resolveHybridSearchQuery(
sparkTable: Table,
sparkCatalog: TableCatalog,
ident: Identifier,
- mvsq: MultiVectorSearchQuery,
+ hsq: HybridSearchQuery,
argsWithoutTable: Seq[Expression]): LogicalPlan = {
sparkTable match {
case st @ SparkTable(innerTable: InnerTable) =>
- val multiVectorSearch = mvsq.createMultiVectorSearch(innerTable, argsWithoutTable)
- val multiVectorSearchTable = MultiVectorSearchTable.create(innerTable, multiVectorSearch)
+ val hybridSearch = hsq.createHybridSearch(innerTable, argsWithoutTable)
+ val hybridSearchTable = HybridSearchTable.create(innerTable, hybridSearch)
DataSourceV2Relation.create(
- st.copy(table = multiVectorSearchTable),
+ st.copy(table = hybridSearchTable),
Some(sparkCatalog),
Some(ident),
CaseInsensitiveStringMap.empty())
case _ =>
throw new RuntimeException(
- "multi_vector_search only supports Paimon SparkTable backed by InnerTable, " +
+ "hybrid_search only supports Paimon SparkTable backed by InnerTable, " +
s"but got table implementation: ${sparkTable.getClass.getName}")
}
}
@@ -450,73 +450,94 @@ case class VectorSearchQuery(override val args: Seq[Expression])
}
/**
- * Plan for the [[MULTI_VECTOR_SEARCH]] table-valued function.
+ * Plan for the [[HYBRID_SEARCH]] table-valued function.
*
- * Usage: multi_vector_search(table_name, routes, limit[, ranker])
+ * Usage: hybrid_search(table_name, vector_routes, full_text_routes, limit[, ranker])
* - table_name: the Paimon table to search
- * - routes: route config array with vector_column, query_vector, limit, weight, and options
- * fields
+ * - vector_routes: route config array with field, query_vector, limit, weight, and options fields
+ * - full_text_routes: route config array with field, query_text, query_operator, 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
+ * - ranker: optional ranker for combining results from multiple routes
*/
-case class MultiVectorSearchQuery(override val args: Seq[Expression])
- extends PaimonTableValueFunction(MULTI_VECTOR_SEARCH) {
+case class HybridSearchQuery(override val args: Seq[Expression])
+ extends PaimonTableValueFunction(HYBRID_SEARCH) {
override def parseArgs(args: Seq[Expression]): Map[String, String] = {
Map.empty
}
- def createMultiVectorSearch(
+ def createHybridSearch(
innerTable: InnerTable,
- argsWithoutTable: Seq[Expression]): MultiVectorSearch = {
- if (argsWithoutTable.size != 2 && argsWithoutTable.size != 3) {
+ argsWithoutTable: Seq[Expression]): HybridSearch = {
+ if (argsWithoutTable.size != 3 && argsWithoutTable.size != 4) {
throw new RuntimeException(
- s"$MULTI_VECTOR_SEARCH needs two or three parameters after table_name: " +
- s"routes, limit[, ranker]. " +
+ s"$HYBRID_SEARCH needs three or four parameters after table_name: " +
+ s"vector_routes, full_text_routes, limit[, ranker]. " +
s"Got ${argsWithoutTable.size} parameters after table_name.")
}
- val finalLimit = parsePositiveLimit(argsWithoutTable(1).eval())
+ val finalLimit = parsePositiveLimit(argsWithoutTable(2).eval())
val ranker =
- if (argsWithoutTable.size == 3) {
- VectorSearchQuery(Seq.empty).extractString(argsWithoutTable(2))
+ if (argsWithoutTable.size == 4) {
+ VectorSearchQuery(Seq.empty).extractString(argsWithoutTable(3))
} else {
- MultiVectorSearchRanker.RRF_RANKER
+ HybridSearchRanker.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
+ val routes =
+ (extractVectorRoutes(argsWithoutTable.head, finalLimit) ++
+ extractFullTextRoutes(argsWithoutTable(1), 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)
+ new HybridSearch(routes.asJava, finalLimit, ranker)
}
- private def extractRoutes(expr: Expression, defaultLimit: Int): Seq[MultiVectorSearchRoute] = {
+ private def extractVectorRoutes(expr: Expression, defaultLimit: Int): Seq[HybridSearchRoute] = {
expr match {
case CreateArray(elements, _) if elements != null =>
- elements.map(extractRoute(_, defaultLimit))
+ elements.map(extractVectorRoute(_, defaultLimit))
case _ =>
- throw new RuntimeException(s"Cannot extract multi-vector routes from expression: $expr")
+ throw new RuntimeException(s"Cannot extract vector routes from expression: $expr")
}
}
- private def extractRoute(expr: Expression, defaultLimit: Int): MultiVectorSearchRoute = {
+ private def extractFullTextRoutes(expr: Expression, defaultLimit: Int): Seq[HybridSearchRoute] = {
+ expr match {
+ case CreateArray(elements, _) if elements != null =>
+ elements.map(extractFullTextRoute(_, defaultLimit))
+ case _ =>
+ throw new RuntimeException(s"Cannot extract full-text routes from expression: $expr")
+ }
+ }
+
+ private def extractVectorRoute(expr: Expression, defaultLimit: Int): HybridSearchRoute = {
expr match {
case CreateNamedStruct(children) if children != null =>
- extractConfiguredRoute(children, defaultLimit)
+ extractConfiguredVectorRoute(children, defaultLimit)
case _ =>
- throw new RuntimeException(s"Cannot extract multi-vector route from expression: $expr")
+ throw new RuntimeException(s"Cannot extract vector route from expression: $expr")
}
}
- private def extractConfiguredRoute(
+ private def extractFullTextRoute(expr: Expression, defaultLimit: Int): HybridSearchRoute = {
+ expr match {
+ case CreateNamedStruct(children) if children != null =>
+ extractConfiguredFullTextRoute(children, defaultLimit)
+ case _ =>
+ throw new RuntimeException(s"Cannot extract full-text route from expression: $expr")
+ }
+ }
+
+ private def extractConfiguredVectorRoute(
children: Seq[Expression],
- defaultLimit: Int): MultiVectorSearchRoute = {
+ defaultLimit: Int): HybridSearchRoute = {
var columnName: Option[String] = None
var queryVector: Option[Array[Float]] = None
var limit: Option[Int] = None
@@ -526,7 +547,7 @@ case class MultiVectorSearchQuery(override val args: Seq[Expression])
children.grouped(2).foreach {
case Seq(keyExpr, valueExpr) =>
VectorSearchQuery(Seq.empty).extractString(keyExpr) match {
- case "vector_column" =>
+ case "field" | "vector_column" =>
columnName = Some(VectorSearchQuery(Seq.empty).extractString(valueExpr))
case "query_vector" =>
queryVector = Some(VectorSearchQuery(Seq.empty).extractQueryVector(valueExpr))
@@ -537,9 +558,8 @@ case class MultiVectorSearchQuery(override val args: Seq[Expression])
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.")
+ throw new IllegalArgumentException(s"Unsupported vector route field '$key'. " +
+ "Supported fields are field, vector_column, query_vector, limit, weight, and options.")
}
case other =>
throw new RuntimeException(s"Invalid route config entries: $other")
@@ -547,12 +567,60 @@ case class MultiVectorSearchQuery(override val args: Seq[Expression])
val routeColumn =
columnName.getOrElse(
- throw new IllegalArgumentException("Multi-vector route must define vector_column."))
- new MultiVectorSearchRoute(
+ throw new IllegalArgumentException("Vector route must define field or vector_column."))
+ HybridSearchRoute.vector(
routeColumn,
queryVector.getOrElse(
throw new IllegalArgumentException(
- s"Multi-vector route for column $routeColumn must define query_vector.")),
+ s"Vector route for column $routeColumn must define query_vector.")),
+ limit.getOrElse(defaultLimit),
+ weight.getOrElse(1.0f),
+ options.asJava
+ )
+ }
+
+ private def extractConfiguredFullTextRoute(
+ children: Seq[Expression],
+ defaultLimit: Int): HybridSearchRoute = {
+ var columnName: Option[String] = None
+ var queryText: Option[String] = None
+ var queryOperator: Option[String] = 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 "field" | "text_column" =>
+ columnName = Some(VectorSearchQuery(Seq.empty).extractString(valueExpr))
+ case "query_text" =>
+ queryText = Some(VectorSearchQuery(Seq.empty).extractString(valueExpr))
+ case "query_operator" =>
+ queryOperator = Some(VectorSearchQuery(Seq.empty).extractString(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 full-text route field '$key'. " +
+ "Supported fields are field, text_column, query_text, query_operator, limit, weight, and options.")
+ }
+ case other =>
+ throw new RuntimeException(s"Invalid route config entries: $other")
+ }
+
+ val routeColumn =
+ columnName.getOrElse(
+ throw new IllegalArgumentException("Full-text route must define field or text_column."))
+ HybridSearchRoute.fullText(
+ routeColumn,
+ queryText.getOrElse(
+ throw new IllegalArgumentException(
+ s"Full-text route for column $routeColumn must define query_text.")),
+ queryOperator.getOrElse("or"),
limit.getOrElse(defaultLimit),
weight.getOrElse(1.0f),
options.asJava
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 f06807ec44..b36a0f1f02 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, MultiVectorSearch, Predicate, TopN, VectorSearch}
+import org.apache.paimon.predicate.{FullTextSearch, HybridSearch, 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,7 +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 pushedHybridSearch: Option[HybridSearch] = None
def pushedFullTextSearch: Option[FullTextSearch] = None
def pushedVariantExtractions: Map[Seq[String], Seq[VariantExtractionInfo]] = Map.empty
@@ -205,7 +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("") +
+ pushedHybridSearch.map(hs => s", HybridSearch: [$hs]").getOrElse("") +
pushedFullTextSearch.map(fts => s", FullTextSearch: [$fts]").getOrElse("") +
pushedVariantsStr
}
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/schema/PaimonMetadataColumn.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/schema/PaimonMetadataColumn.scala
index 8438d5b175..d61dfb8c59 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/schema/PaimonMetadataColumn.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/schema/PaimonMetadataColumn.scala
@@ -56,7 +56,7 @@ object PaimonMetadataColumn {
val BUCKET_COLUMN = "__paimon_bucket"
val ROW_ID_COLUMN: String = SpecialFields.ROW_ID.name()
val SEQUENCE_NUMBER_COLUMN: String = SpecialFields.SEQUENCE_NUMBER.name()
- val VECTOR_SEARCH_SCORE_COLUMN: String = "__paimon_vector_search_score"
+ val SEARCH_SCORE_COLUMN: String = "__paimon_search_score"
val PATH_AND_INDEX_META_COLUMNS: Seq[String] = Seq(FILE_PATH_COLUMN, ROW_INDEX_COLUMN)
val PARTITION_AND_BUCKET_META_COLUMNS: Seq[String] = Seq(PARTITION_COLUMN, BUCKET_COLUMN)
@@ -69,7 +69,7 @@ object PaimonMetadataColumn {
BUCKET_COLUMN,
ROW_ID_COLUMN,
SEQUENCE_NUMBER_COLUMN,
- VECTOR_SEARCH_SCORE_COLUMN
+ SEARCH_SCORE_COLUMN
)
val ROW_INDEX: PaimonMetadataColumn =
@@ -89,8 +89,8 @@ object PaimonMetadataColumn {
SEQUENCE_NUMBER_COLUMN,
LongType,
preserveOnUpdate = false)
- val VECTOR_SEARCH_SCORE: PaimonMetadataColumn =
- PaimonMetadataColumn(Integer.MAX_VALUE - 106, VECTOR_SEARCH_SCORE_COLUMN, FloatType)
+ val SEARCH_SCORE: PaimonMetadataColumn =
+ PaimonMetadataColumn(Integer.MAX_VALUE - 106, SEARCH_SCORE_COLUMN, FloatType)
def dvMetaCols: Seq[PaimonMetadataColumn] = Seq(FILE_PATH, ROW_INDEX)
@@ -102,7 +102,7 @@ object PaimonMetadataColumn {
case BUCKET_COLUMN => BUCKET
case ROW_ID_COLUMN => ROW_ID
case SEQUENCE_NUMBER_COLUMN => SEQUENCE_NUMBER
- case VECTOR_SEARCH_SCORE_COLUMN => VECTOR_SEARCH_SCORE
+ case SEARCH_SCORE_COLUMN => SEARCH_SCORE
case _ =>
throw new IllegalArgumentException(s"$metadataColumn metadata column is not supported.")
}
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 12775ea540..b3b5811f2b 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
@@ -29,8 +29,8 @@ 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(
+ test("create hybrid search with vector route configs") {
+ val search = HybridSearchQuery(Seq.empty).createHybridSearch(
innerTable,
Seq(
CreateArray(
@@ -60,6 +60,7 @@ class VectorSearchQueryTest extends AnyFunSuite {
CreateMap(Seq(Literal("ivf.nprobe"), Literal("16")))
))
)),
+ CreateArray(Seq.empty),
Literal(3),
Literal("weighted_score")
)
@@ -75,8 +76,8 @@ class VectorSearchQueryTest extends AnyFunSuite {
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(
+ test("default hybrid vector route limit to final limit") {
+ val search = HybridSearchQuery(Seq.empty).createHybridSearch(
innerTable,
Seq(
CreateArray(
@@ -88,6 +89,7 @@ class VectorSearchQueryTest extends AnyFunSuite {
Literal("query_vector"),
CreateArray(Seq(Literal(1.0f), Literal(0.0f)))
)))),
+ CreateArray(Seq.empty),
Literal(7)
)
)
@@ -99,9 +101,45 @@ class VectorSearchQueryTest extends AnyFunSuite {
assert(search.routes().get(0).options().isEmpty)
}
- test("reject multi vector search query map") {
+ test("create hybrid search with full-text route configs") {
+ val search = HybridSearchQuery(Seq.empty).createHybridSearch(
+ innerTable,
+ Seq(
+ CreateArray(Seq.empty),
+ CreateArray(
+ Seq(
+ CreateNamedStruct(Seq(
+ Literal("field"),
+ Literal("content"),
+ Literal("query_text"),
+ Literal("paimon lake"),
+ Literal("query_operator"),
+ Literal("and"),
+ Literal("limit"),
+ Literal(20),
+ Literal("weight"),
+ Literal(1.5f),
+ Literal("options"),
+ CreateMap(Seq.empty)
+ ))
+ )),
+ Literal(5),
+ Literal("rrf")
+ )
+ )
+
+ assert(search.routes().size() == 1)
+ assert(search.routes().get(0).isFullText)
+ assert(search.routes().get(0).fieldName() == "content")
+ assert(search.routes().get(0).queryText() == "paimon lake")
+ assert(search.routes().get(0).queryOperator() == "and")
+ assert(search.routes().get(0).limit() == 20)
+ assert(search.routes().get(0).weight() == 1.5f)
+ }
+
+ test("reject hybrid search query map") {
val exception = intercept[RuntimeException] {
- MultiVectorSearchQuery(Seq.empty).createMultiVectorSearch(
+ HybridSearchQuery(Seq.empty).createHybridSearch(
innerTable,
Seq(
CreateMap(
@@ -110,12 +148,13 @@ class VectorSearchQueryTest extends AnyFunSuite {
CreateArray(Seq(Literal(1.0f), Literal(0.0f))),
Literal("body_vec"),
CreateArray(Seq(Literal(0.0f), Literal(1.0f))))),
+ CreateArray(Seq.empty),
Literal(3)
)
)
}
- assert(exception.getMessage.contains("Cannot extract multi-vector routes"))
+ assert(exception.getMessage.contains("Cannot extract vector routes"))
}
test("create vector search with string options") {
@@ -155,8 +194,9 @@ class VectorSearchQueryTest extends AnyFunSuite {
Array[DataType](
new ArrayType(DataTypes.FLOAT()),
new ArrayType(DataTypes.FLOAT()),
- new ArrayType(DataTypes.FLOAT())),
- Array[String]("v", "title_vec", "body_vec")
+ new ArrayType(DataTypes.FLOAT()),
+ DataTypes.STRING()),
+ Array[String]("v", "title_vec", "body_vec", "content")
)
override def invoke(proxy: Any, method: Method, args: Array[AnyRef]): AnyRef = {
diff --git a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkMultimodalITCase.java b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkMultimodalITCase.java
index 3a03ea182e..af3401ea2a 100644
--- a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkMultimodalITCase.java
+++ b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkMultimodalITCase.java
@@ -149,7 +149,7 @@ public class SparkMultimodalITCase {
.collectAsList();
assertThat(rows).hasSize(5);
String vectorSearchSql =
- "select gid, sid, embs, __paimon_vector_search_score "
+ "select gid, sid, embs, __paimon_search_score "
+ "from vector_search('my_db1.vector_test', 'embs', array(1.0f, 2.0f, 3.0f, 4.0f), 5) "
+ "where date = '20260420'";
Dataset<Row> df = spark.sql(vectorSearchSql);
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/HybridSearchTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/HybridSearchTest.scala
new file mode 100644
index 0000000000..7bfbfcfa18
--- /dev/null
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/HybridSearchTest.scala
@@ -0,0 +1,250 @@
+/*
+ * 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 hybrid search. */
+class HybridSearchTest extends PaimonSparkTestBase {
+
+ test("hybrid 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_search_score
+ |FROM hybrid_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'))),
+ | array(),
+ | 2,
+ | 'weighted_score')
+ |""".stripMargin)
+ .collect()
+
+ assert(result.length == 2)
+ assert(result.map(_.getInt(0)).contains(1))
+ assert(result.forall(row => !row.isNullAt(1)))
+ }
+ }
+
+ test("hybrid search ranks vector and full-text routes") {
+ withTable("T") {
+ spark.sql("""
+ |CREATE TABLE T (id INT, content STRING, 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')
+ |""".stripMargin)
+
+ spark.sql("""
+ |INSERT INTO T VALUES
+ | (0, 'lake format', array(1.0f, 0.0f)),
+ | (1, 'paimon hybrid search', array(0.9f, 0.1f)),
+ | (2, 'paimon full text', array(0.0f, 1.0f))
+ |""".stripMargin)
+
+ spark
+ .sql(s"CALL sys.create_global_index(table => 'test.T', index_column => 'vec', " +
+ s"index_type => '${TestVectorGlobalIndexerFactory.IDENTIFIER}')")
+ .collect()
+ spark
+ .sql("CALL sys.create_global_index(table => 'test.T', index_column => 'content', " +
+ "index_type => 'test-fulltext')")
+ .collect()
+
+ val result = spark
+ .sql("""
+ |SELECT id, __paimon_search_score
+ |FROM hybrid_search(
+ | 'T',
+ | array(
+ | named_struct(
+ | 'field', 'vec',
+ | 'query_vector', array(1.0f, 0.0f),
+ | 'limit', 2,
+ | 'weight', 1.0f,
+ | 'options', map())),
+ | array(
+ | named_struct(
+ | 'field', 'content',
+ | 'query_text', 'paimon',
+ | 'query_operator', 'or',
+ | 'limit', 2,
+ | 'weight', 1.0f,
+ | 'options', map())),
+ | 3,
+ | 'rrf')
+ |ORDER BY __paimon_search_score DESC
+ |""".stripMargin)
+ .collect()
+
+ assert(result.length == 3)
+ assert(result.head.getInt(0) == 1)
+ assert(result.forall(row => !row.isNullAt(1)))
+ }
+ }
+
+ test("hybrid full-text route uses partition filters before ranking") {
+ withTable("T") {
+ spark.sql("""
+ |CREATE TABLE T (id INT, content STRING, pt INT)
+ |PARTITIONED BY (pt)
+ |TBLPROPERTIES (
+ | 'bucket' = '-1',
+ | 'global-index.row-count-per-shard' = '10000',
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true')
+ |""".stripMargin)
+
+ spark.sql("""
+ |INSERT INTO T VALUES
+ | (0, 'paimon search', 1),
+ | (1, 'paimon', 2)
+ |""".stripMargin)
+
+ spark
+ .sql("CALL sys.create_global_index(table => 'test.T', index_column => 'content', " +
+ "index_type => 'test-fulltext')")
+ .collect()
+
+ val result = spark
+ .sql("""
+ |SELECT id
+ |FROM hybrid_search(
+ | 'T',
+ | array(),
+ | array(
+ | named_struct(
+ | 'field', 'content',
+ | 'query_text', 'paimon search',
+ | 'query_operator', 'or',
+ | 'limit', 1,
+ | 'weight', 1.0f,
+ | 'options', map())),
+ | 1)
+ |WHERE pt = 2
+ |""".stripMargin)
+ .collect()
+
+ assert(result.map(_.getInt(0)).toSeq == Seq(1))
+ }
+ }
+
+ test("hybrid full-text route rejects non-partition filters") {
+ withTable("T") {
+ spark.sql("""
+ |CREATE TABLE T (id INT, content STRING)
+ |TBLPROPERTIES (
+ | 'bucket' = '-1',
+ | 'global-index.row-count-per-shard' = '10000',
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true')
+ |""".stripMargin)
+
+ spark.sql("""
+ |INSERT INTO T VALUES
+ | (0, 'paimon search'),
+ | (1, 'paimon')
+ |""".stripMargin)
+
+ spark
+ .sql("CALL sys.create_global_index(table => 'test.T', index_column => 'content', " +
+ "index_type => 'test-fulltext')")
+ .collect()
+
+ val error = intercept[Exception] {
+ spark
+ .sql("""
+ |SELECT id
+ |FROM hybrid_search(
+ | 'T',
+ | array(),
+ | array(
+ | named_struct(
+ | 'field', 'content',
+ | 'query_text', 'paimon',
+ | 'query_operator', 'or',
+ | 'limit', 1,
+ | 'weight', 1.0f,
+ | 'options', map())),
+ | 1)
+ |WHERE id = 1
+ |""".stripMargin)
+ .collect()
+ }
+
+ assert(containsMessage(error, "does not support non-partition filters"))
+ }
+ }
+
+ private def containsMessage(error: Throwable, expected: String): Boolean = {
+ var current = error
+ while (current != null) {
+ if (current.getMessage != null && current.getMessage.contains(expected)) {
+ return true
+ }
+ current = current.getCause
+ }
+ false
+ }
+}
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
deleted file mode 100644
index e3a3bbe8b3..0000000000
--- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/MultiVectorSearchTest.scala
+++ /dev/null
@@ -1,85 +0,0 @@
-/*
- * 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)))
- }
- }
-}