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/22 04:00:11 UTC
(paimon) branch master updated: [core] Align full-text search with JSON DSL (#8308)
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 84d5acbea9 [core] Align full-text search with JSON DSL (#8308)
84d5acbea9 is described below
commit 84d5acbea9716ba9fa3e0edd87c7175f650b2be0
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Mon Jun 22 12:00:06 2026 +0800
[core] Align full-text search with JSON DSL (#8308)
Align Paimon full-text search with the LanceDB-style FTS query DSL and
make JSON DSL the public full-text search API surface. This covers Java,
PyPaimon, Spark SQL table-valued functions, Tantivy JNI parsing, hybrid
full-text routes, docs, and index scan correctness.
---
.../multimodal-table/global-index/full-text.mdx | 249 +++++-
.../global-index/hybrid-search.mdx | 16 +-
.../paimon/globalindex/GlobalIndexerFactory.java | 4 +
.../org/apache/paimon/predicate/FullTextQuery.java | 865 +++++++++++++++++++++
.../apache/paimon/predicate/FullTextSearch.java | 52 +-
.../apache/paimon/predicate/HybridSearchRoute.java | 85 +-
.../TestFullTextGlobalIndexReader.java | 68 +-
.../TestFullTextGlobalIndexerFactory.java | 5 +
.../apache/paimon/predicate/FullTextQueryTest.java | 175 +++++
.../paimon/table/source/FullTextReadImpl.java | 265 ++++++-
.../paimon/table/source/FullTextScanImpl.java | 62 +-
.../paimon/table/source/FullTextSearchBuilder.java | 11 +-
.../table/source/FullTextSearchBuilderImpl.java | 42 +-
.../paimon/table/source/FullTextSearchSplit.java | 27 +-
.../paimon/table/source/HybridSearchBuilder.java | 16 +-
.../table/source/HybridSearchBuilderImpl.java | 4 +-
.../table/source/FullTextSearchBuilderTest.java | 343 +++++++-
paimon-python/pypaimon/cli/cli_table.py | 15 +-
paimon-python/pypaimon/globalindex/__init__.py | 20 +
.../pypaimon/globalindex/full_text_query.py | 512 ++++++++++++
.../pypaimon/globalindex/full_text_search.py | 44 +-
.../tantivy_full_text_global_index_reader.py | 79 +-
.../pypaimon/table/source/full_text_read.py | 152 +++-
.../pypaimon/table/source/full_text_scan.py | 32 +-
.../table/source/full_text_search_builder.py | 56 +-
.../table/source/full_text_search_split.py | 11 +-
.../pypaimon/table/source/hybrid_search_builder.py | 40 +-
.../pypaimon/tests/e2e/java_py_read_write_test.py | 33 +-
.../pypaimon/tests/vector_search_filter_test.py | 340 +++++++-
.../org/apache/paimon/spark/PaimonBaseScan.scala | 4 +-
.../plans/logical/PaimonTableValuedFunctions.scala | 80 +-
.../plans/logical/VectorSearchQueryTest.scala | 52 +-
.../paimon/spark/sql/FullTextSearchTest.scala | 59 +-
.../apache/paimon/spark/sql/HybridSearchTest.scala | 12 +-
paimon-tantivy/paimon-tantivy-index/README.md | 9 +-
.../index/TantivyFullTextGlobalIndexReader.java | 6 +-
.../index/TantivyFullTextGlobalIndexerFactory.java | 5 +
.../index/TantivyFullTextGlobalIndexTest.java | 49 +-
paimon-tantivy/paimon-tantivy-jni/rust/src/lib.rs | 608 +++++++++++++--
.../org/apache/paimon/tantivy/TantivySearcher.java | 7 +
40 files changed, 3958 insertions(+), 556 deletions(-)
diff --git a/docs/docs/multimodal-table/global-index/full-text.mdx b/docs/docs/multimodal-table/global-index/full-text.mdx
index cce27419c8..d51129d7c8 100644
--- a/docs/docs/multimodal-table/global-index/full-text.mdx
+++ b/docs/docs/multimodal-table/global-index/full-text.mdx
@@ -113,23 +113,49 @@ Set `tantivy.with-position=false` to reduce index size when phrase queries are n
## Full-Text Search
-The query operator controls how multiple query terms are matched:
-
-| Operator | Meaning |
-|---|---|
-| `or` | Return rows matching any query term. |
-| `and` | Return rows matching all query terms. |
+Full-text search accepts a JSON query DSL. The root JSON object should contain one query type.
+Paimon supports `match`, `match_phrase` / `phrase`, `boost`, `multi_match`, and `boolean`.
+The same JSON DSL is used by Spark SQL `full_text_search(...)`, Java `FullTextQuery.fromJson(...)`,
+Python `FullTextQuery.from_json(...)`, and full-text routes in Hybrid Search.
<Tabs groupId="fulltext-search">
<TabItem value="spark-sql" label="Spark SQL">
```sql
--- Search for top-10 documents matching any query term. The default query operator is 'or'.
-SELECT * FROM full_text_search('my_table', 'content', 'paimon lake format', 10);
+-- Search for top-10 documents matching any query term. The default query operator is 'Or'.
+SELECT * FROM full_text_search(
+ 'my_table',
+ '{"match":{"column":"content","terms":"paimon lake format"}}',
+ 10
+);
-- Search for top-10 documents matching all query terms.
-SELECT * FROM full_text_search('my_table', 'content', 'paimon lake format', 10, 'and');
+SELECT * FROM full_text_search(
+ 'my_table',
+ '{"match":{"column":"content","terms":"paimon lake format","operator":"And"}}',
+ 10
+);
+
+-- Structured query DSL. The JSON shape follows LanceDB full-text queries:
+-- match, match_phrase, boost, multi_match, and boolean.
+SELECT * FROM full_text_search(
+ 'my_table',
+ '{"match_phrase":{"column":"content","terms":"paimon lake","slop":1}}',
+ 10
+);
+
+SELECT * FROM full_text_search(
+ 'my_table',
+ '{"boolean":{"must":[{"match":{"column":"content","terms":"paimon"}},{"match":{"column":"content","terms":"format"}}],"must_not":[{"match":{"column":"content","terms":"vector"}}]}}',
+ 10
+);
+
+SELECT * FROM full_text_search(
+ 'my_table',
+ '{"multi_match":{"query":"paimon","columns":["title","content"],"boost":[2.0,1.0],"operator":"Or"}}',
+ 10
+);
```
</TabItem>
@@ -141,10 +167,8 @@ Table table = catalog.getTable(identifier);
// Step 1: Build full-text search
GlobalIndexResult result = table.newFullTextSearchBuilder()
- .withQueryText("paimon lake format")
- .withQueryOperator("and")
+ .withQuery(FullTextQuery.phrase("paimon lake", "content", 1))
.withLimit(10)
- .withTextColumn("content")
.executeLocal();
// Step 2: Read matching rows using the search result
@@ -162,14 +186,14 @@ try (RecordReader<InternalRow> reader = readBuilder.newRead().createReader(plan)
<TabItem value="python-sdk" label="Python SDK">
```python
+from pypaimon.globalindex.full_text_query import MatchQuery
+
table = catalog.get_table("db.my_table")
# Step 1: Build full-text search
result = (
table.new_full_text_search_builder()
- .with_text_column("content")
- .with_query_text("paimon lake format")
- .with_query_operator("and")
+ .with_query(MatchQuery("paimon lake format", "content", operator="And"))
.with_limit(10)
.execute_local()
)
@@ -191,3 +215,198 @@ when the `jieba` package is installed.
</TabItem>
</Tabs>
+
+## Query DSL Reference
+
+### Match
+
+Use `match` to search terms in one text column. By default, a row matches if any query term
+matches.
+
+```json
+{
+ "match": {
+ "column": "content",
+ "terms": "paimon lake format",
+ "operator": "or",
+ "boost": 2.0,
+ "fuzziness": 1,
+ "max_expansions": 50,
+ "prefix_length": 0
+ }
+}
+```
+
+| Field | Required | Default | Description |
+|---|---|---|---|
+| `column` | Yes | | Text column to search. |
+| `terms` | Yes | | Query text. `query` is also accepted as an alias. |
+| `operator` | No | `or` | How query terms are combined. Supported values are `or` and `and`. |
+| `boost` | No | `1.0` | Positive score multiplier for this query. |
+| `fuzziness` | No | `0` | Edit distance for fuzzy matching. Supported numeric values are `0`, `1`, and `2`. Use `null` or `auto` to leave fuzziness unset. |
+| `max_expansions` | No | `50` | Maximum fuzzy term expansions. `maxExpansions` is also accepted. |
+| `prefix_length` | No | `0` | Number of leading characters that must match exactly for fuzzy matching. `prefixLength` is also accepted. |
+
+### Phrase
+
+Use `match_phrase` to match terms in order. Phrase queries require the index to store positions,
+so keep `tantivy.with-position=true` when building indexes that need phrase search.
+
+```json
+{
+ "match_phrase": {
+ "column": "content",
+ "terms": "paimon lake",
+ "slop": 1
+ }
+}
+```
+
+`phrase` is accepted as an alias for `match_phrase`, and `query` is accepted as an alias for
+`terms`. Paimon serializes phrase queries as `match_phrase`.
+
+| Field | Required | Default | Description |
+|---|---|---|---|
+| `column` | Yes | | Text column to search. |
+| `terms` | Yes | | Phrase text. `query` is also accepted as an alias. |
+| `slop` | No | `0` | Maximum number of term-position moves allowed in the phrase. |
+
+### Multi Match
+
+Use `multi_match` to search the same query text across multiple columns in one full-text query.
+Column boosts are applied inside the full-text score before the final top-k is selected.
+
+```json
+{
+ "multi_match": {
+ "query": "paimon lake",
+ "columns": ["title", "content"],
+ "boost": [2.0, 1.0],
+ "operator": "or"
+ }
+}
+```
+
+| Field | Required | Default | Description |
+|---|---|---|---|
+| `query` | Yes | | Query text searched in every listed column. |
+| `columns` | Yes | | Non-empty array of text columns. |
+| `boost` | No | all `1.0` | Per-column score multipliers. The array length must match `columns`. `boosts` is also accepted. |
+| `operator` | No | `or` | How query terms are combined within each column. Supported values are `or` and `and`. |
+
+### Boost
+
+Use `boost` to reduce the score of documents that also match a negative query.
+
+```json
+{
+ "boost": {
+ "positive": {
+ "match": {
+ "column": "content",
+ "terms": "paimon"
+ }
+ },
+ "negative": {
+ "match": {
+ "column": "content",
+ "terms": "vector"
+ }
+ },
+ "negative_boost": 0.3
+ }
+}
+```
+
+| Field | Required | Default | Description |
+|---|---|---|---|
+| `positive` | Yes | | Query that contributes the main score. |
+| `negative` | Yes | | Query used to down-rank matching rows. |
+| `negative_boost` | No | `0.5` | Positive multiplier applied to rows that also match `negative`. `negativeBoost` is also accepted. |
+
+### Boolean
+
+Use `boolean` to combine nested queries. `must` clauses restrict matches, `should` clauses add
+matches and score, and `must_not` clauses exclude matches.
+
+```json
+{
+ "boolean": {
+ "must": [
+ {
+ "match": {
+ "column": "content",
+ "terms": "paimon"
+ }
+ }
+ ],
+ "should": [
+ {
+ "match_phrase": {
+ "column": "content",
+ "terms": "lake format"
+ }
+ }
+ ],
+ "must_not": [
+ {
+ "match": {
+ "column": "content",
+ "terms": "vector"
+ }
+ }
+ ]
+ }
+}
+```
+
+| Field | Required | Default | Description |
+|---|---|---|---|
+| `must` | No | empty | Array of nested queries that every result must match. |
+| `should` | No | empty | Array of nested queries that can match and contribute score. |
+| `must_not` | No | empty | Array of nested queries that matching rows must not satisfy. |
+
+`boolean` also accepts a `queries` array for explicit occurrence labels:
+
+```json
+{
+ "boolean": {
+ "queries": [
+ {
+ "occur": "must",
+ "query": {
+ "match": {
+ "column": "content",
+ "terms": "paimon"
+ }
+ }
+ },
+ [
+ "must_not",
+ {
+ "match": {
+ "column": "content",
+ "terms": "vector"
+ }
+ }
+ ]
+ ]
+ }
+}
+```
+
+Supported `occur` values are `should`, `must`, and `must_not`.
+
+### Validation Notes
+
+- Query text and column names cannot be empty.
+- Boost values and `negative_boost` must be positive.
+- `max_expansions` must be positive.
+- `prefix_length` and `slop` must be non-negative.
+- `fuzziness` must be `0`, `1`, `2`, `null`, or `auto`.
+- `max_expansions` and `prefix_length` are part of the DSL for LanceDB compatibility. The current
+ Tantivy backend accepts the default values only: `max_expansions=50` and `prefix_length=0`.
+- PyPaimon can parse the same DSL, but its local Tantivy reader does not support every advanced
+ scoring option yet. Unsupported options fail fast instead of changing query semantics silently.
+- Multi-column queries such as `multi_match` require full-text indexes for every referenced
+ column.
diff --git a/docs/docs/multimodal-table/global-index/hybrid-search.mdx b/docs/docs/multimodal-table/global-index/hybrid-search.mdx
index 7c3f8395e3..52068d2fed 100644
--- a/docs/docs/multimodal-table/global-index/hybrid-search.mdx
+++ b/docs/docs/multimodal-table/global-index/hybrid-search.mdx
@@ -84,9 +84,7 @@ The third argument is an array of full-text route configs created by `named_stru
| 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`. |
+| `query` | Yes | N/A | LanceDB-style full-text query JSON for this route. |
| `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. |
@@ -117,9 +115,7 @@ FROM hybrid_search(
'options', map('ivf.nprobe', '32'))),
array(
named_struct(
- 'field', 'content',
- 'query_text', 'paimon search',
- 'query_operator', 'or',
+ 'query', '{"match":{"column":"content","terms":"paimon search"}}',
'limit', 50,
'weight', 1.0f,
'options', map())),
@@ -169,9 +165,7 @@ HybridSearchBuilder searchBuilder =
2.0f,
java.util.Collections.singletonMap("ivf.nprobe", "32"))
.addFullTextRoute(
- "content",
- "paimon search",
- "or",
+ "{\"match\":{\"column\":\"content\",\"terms\":\"paimon search\"}}",
50,
1.0f,
java.util.Collections.emptyMap())
@@ -230,11 +224,9 @@ result = (
options={"ivf.nprobe": "32"},
)
.add_full_text_route(
- "content",
- "paimon search",
+ '{"match":{"column":"content","terms":"paimon search"}}',
limit=50,
weight=1.0,
- query_operator="or",
)
.with_limit(10)
.with_rrf_ranker()
diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexerFactory.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexerFactory.java
index 741f91f42a..995d197607 100644
--- a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexerFactory.java
+++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexerFactory.java
@@ -29,6 +29,10 @@ public interface GlobalIndexerFactory {
String identifier();
+ default boolean supportsFullTextSearch() {
+ return false;
+ }
+
GlobalIndexer create(DataField indexField, Options options);
/**
diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/FullTextQuery.java b/paimon-common/src/main/java/org/apache/paimon/predicate/FullTextQuery.java
new file mode 100644
index 0000000000..0b627876ed
--- /dev/null
+++ b/paimon-common/src/main/java/org/apache/paimon/predicate/FullTextQuery.java
@@ -0,0 +1,865 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.predicate;
+
+import org.apache.paimon.utils.JsonSerdeUtil;
+
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+
+/** Structured full-text query DSL aligned with LanceDB FTS query JSON. */
+public abstract class FullTextQuery implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ /** Operator for combining terms in a match query. */
+ public enum Operator {
+ OR,
+ AND;
+
+ public static Operator fromString(String value) {
+ if (value == null) {
+ return OR;
+ }
+ String normalized = value.trim().toUpperCase(Locale.ROOT);
+ if ("OR".equals(normalized)) {
+ return OR;
+ }
+ if ("AND".equals(normalized)) {
+ return AND;
+ }
+ throw new IllegalArgumentException(
+ "Full-text query operator must be 'or' or 'and', got: " + value);
+ }
+
+ public String jsonValue() {
+ return name().substring(0, 1) + name().substring(1).toLowerCase(Locale.ROOT);
+ }
+ }
+
+ /** Occurrence type for boolean full-text query clauses. */
+ public enum Occur {
+ SHOULD,
+ MUST,
+ MUST_NOT;
+
+ static Occur fromString(String value) {
+ checkNotNull(value, "Boolean query occur cannot be null");
+ String normalized = value.trim().toUpperCase(Locale.ROOT);
+ if ("SHOULD".equals(normalized)) {
+ return SHOULD;
+ }
+ if ("MUST".equals(normalized)) {
+ return MUST;
+ }
+ if ("MUST_NOT".equals(normalized) || "MUSTNOT".equals(normalized)) {
+ return MUST_NOT;
+ }
+ throw new IllegalArgumentException("Unknown boolean query occur: " + value);
+ }
+ }
+
+ public static Match match(String query, String column) {
+ return new Match(query, column, 1.0f, 0, 50, Operator.OR, 0);
+ }
+
+ public static Match match(String query, String column, String operator) {
+ return match(query, column, Operator.fromString(operator));
+ }
+
+ public static Match match(String query, String column, Operator operator) {
+ return new Match(query, column, 1.0f, 0, 50, operator, 0);
+ }
+
+ public static Phrase phrase(String query, String column) {
+ return new Phrase(query, column, 0);
+ }
+
+ public static Phrase phrase(String query, String column, int slop) {
+ return new Phrase(query, column, slop);
+ }
+
+ public static Boost boost(FullTextQuery positive, FullTextQuery negative) {
+ return new Boost(positive, negative, 0.5f);
+ }
+
+ public static Boost boost(FullTextQuery positive, FullTextQuery negative, float negativeBoost) {
+ return new Boost(positive, negative, negativeBoost);
+ }
+
+ public static MultiMatch multiMatch(String query, List<String> columns) {
+ return new MultiMatch(query, columns, null, Operator.OR);
+ }
+
+ public static MultiMatch multiMatch(String query, List<String> columns, List<Float> boosts) {
+ return new MultiMatch(query, columns, boosts, Operator.OR);
+ }
+
+ public static MultiMatch multiMatch(String query, List<String> columns, String operator) {
+ return multiMatch(query, columns, null, operator);
+ }
+
+ public static MultiMatch multiMatch(
+ String query, List<String> columns, List<Float> boosts, String operator) {
+ return new MultiMatch(query, columns, boosts, Operator.fromString(operator));
+ }
+
+ public static MultiMatch multiMatch(
+ String query, List<String> columns, List<Float> boosts, Operator operator) {
+ return new MultiMatch(query, columns, boosts, operator);
+ }
+
+ public static BooleanQuery bool(List<Clause> queries) {
+ return new BooleanQuery(queries);
+ }
+
+ public static BooleanQuery bool() {
+ return new BooleanQuery(Collections.emptyList(), false);
+ }
+
+ public static Clause occurShould(FullTextQuery query) {
+ return new Clause(Occur.SHOULD, query);
+ }
+
+ public static Clause occurMust(FullTextQuery query) {
+ return new Clause(Occur.MUST, query);
+ }
+
+ public static Clause occurMustNot(FullTextQuery query) {
+ return new Clause(Occur.MUST_NOT, query);
+ }
+
+ public static FullTextQuery fromJson(String json) {
+ JsonNode node;
+ try {
+ node = JsonSerdeUtil.OBJECT_MAPPER_INSTANCE.readTree(json);
+ } catch (Exception e) {
+ throw new IllegalArgumentException("Failed to parse full-text query JSON: " + json, e);
+ }
+ return fromJsonNode(node);
+ }
+
+ private static FullTextQuery fromJsonNode(JsonNode node) {
+ checkArgument(node != null && node.isObject(), "Full-text query JSON must be an object.");
+ if (node.has("match")) {
+ JsonNode match = node.get("match");
+ return new Match(
+ textValueAny(match, null, "terms", "query"),
+ textValue(match, "column", null),
+ floatValue(match, "boost", 1.0f),
+ fuzzinessValue(match),
+ intValueAny(match, 50, "max_expansions", "maxExpansions"),
+ Operator.fromString(textValue(match, "operator", "or")),
+ intValueAny(match, 0, "prefix_length", "prefixLength"));
+ }
+ if (node.has("phrase") || node.has("match_phrase")) {
+ JsonNode phrase = node.has("phrase") ? node.get("phrase") : node.get("match_phrase");
+ return new Phrase(
+ textValueAny(phrase, null, "terms", "query"),
+ textValue(phrase, "column", null),
+ intValue(phrase, "slop", 0));
+ }
+ if (node.has("boost")) {
+ JsonNode boost = node.get("boost");
+ return new Boost(
+ fromJsonNode(boost.get("positive")),
+ fromJsonNode(boost.get("negative")),
+ floatValueAny(boost, 0.5f, "negative_boost", "negativeBoost"));
+ }
+ if (node.has("multi_match")) {
+ JsonNode multiMatch = node.get("multi_match");
+ return new MultiMatch(
+ textValue(multiMatch, "query", null),
+ stringList(multiMatch.get("columns")),
+ nullableFloatList(fieldValue(multiMatch, "boost", "boosts")),
+ Operator.fromString(textValue(multiMatch, "operator", "or")));
+ }
+ if (node.has("boolean")) {
+ JsonNode bool = node.get("boolean");
+ List<Clause> clauses = new ArrayList<>();
+ addClauses(clauses, Occur.SHOULD, bool.get("should"));
+ addClauses(clauses, Occur.MUST, bool.get("must"));
+ addClauses(clauses, Occur.MUST_NOT, bool.get("must_not"));
+ if (bool.has("queries")) {
+ for (JsonNode clause : bool.get("queries")) {
+ Occur occur;
+ JsonNode queryNode;
+ if (clause.isArray() && clause.size() == 2) {
+ occur = Occur.fromString(clause.get(0).asText());
+ queryNode = clause.get(1);
+ } else {
+ occur = Occur.fromString(textValue(clause, "occur", null));
+ queryNode = clause.get("query");
+ }
+ clauses.add(new Clause(occur, fromJsonNode(queryNode)));
+ }
+ }
+ return new BooleanQuery(clauses);
+ }
+ throw new IllegalArgumentException("Unknown full-text query JSON: " + node);
+ }
+
+ private static void addClauses(List<Clause> clauses, Occur occur, @Nullable JsonNode node) {
+ if (node == null || node.isNull()) {
+ return;
+ }
+ checkArgument(node.isArray(), "Boolean query clauses must be arrays.");
+ for (JsonNode child : node) {
+ clauses.add(new Clause(occur, fromJsonNode(child)));
+ }
+ }
+
+ private static String textValue(JsonNode node, String field, @Nullable String defaultValue) {
+ JsonNode value = node == null ? null : node.get(field);
+ if (value == null || value.isNull()) {
+ return defaultValue;
+ }
+ return value.asText();
+ }
+
+ private static String textValueAny(
+ JsonNode node, @Nullable String defaultValue, String... fields) {
+ JsonNode value = fieldValue(node, fields);
+ if (value == null || value.isNull()) {
+ return defaultValue;
+ }
+ return value.asText();
+ }
+
+ @Nullable
+ private static JsonNode fieldValue(JsonNode node, String... fields) {
+ if (node == null) {
+ return null;
+ }
+ for (String field : fields) {
+ JsonNode value = node.get(field);
+ if (value != null) {
+ return value;
+ }
+ }
+ return null;
+ }
+
+ private static Integer nullableIntValue(
+ JsonNode node, String field, @Nullable Integer defaultValue) {
+ JsonNode value = node == null ? null : node.get(field);
+ if (value == null || value.isNull()) {
+ return defaultValue;
+ }
+ return value.asInt();
+ }
+
+ private static int intValue(JsonNode node, String field, int defaultValue) {
+ Integer value = nullableIntValue(node, field, defaultValue);
+ return value == null ? defaultValue : value;
+ }
+
+ private static int intValueAny(JsonNode node, int defaultValue, String... fields) {
+ JsonNode value = fieldValue(node, fields);
+ if (value == null || value.isNull()) {
+ return defaultValue;
+ }
+ return value.asInt();
+ }
+
+ @Nullable
+ private static Integer fuzzinessValue(JsonNode node) {
+ JsonNode value = fieldValue(node, "fuzziness");
+ if (value == null) {
+ return 0;
+ }
+ if (value.isNull()) {
+ return null;
+ }
+ if (value.isTextual() && "auto".equalsIgnoreCase(value.asText())) {
+ return null;
+ }
+ return value.asInt();
+ }
+
+ private static float floatValue(JsonNode node, String field, float defaultValue) {
+ JsonNode value = node == null ? null : node.get(field);
+ if (value == null || value.isNull()) {
+ return defaultValue;
+ }
+ return (float) value.asDouble();
+ }
+
+ private static float floatValueAny(JsonNode node, float defaultValue, String... fields) {
+ JsonNode value = fieldValue(node, fields);
+ if (value == null || value.isNull()) {
+ return defaultValue;
+ }
+ return (float) value.asDouble();
+ }
+
+ private static List<String> stringList(JsonNode node) {
+ checkArgument(node != null && node.isArray(), "columns must be an array.");
+ List<String> values = new ArrayList<>();
+ for (JsonNode value : node) {
+ values.add(value.asText());
+ }
+ return values;
+ }
+
+ @Nullable
+ private static List<Float> nullableFloatList(@Nullable JsonNode node) {
+ if (node == null || node.isNull()) {
+ return null;
+ }
+ checkArgument(node.isArray(), "boost must be an array.");
+ List<Float> values = new ArrayList<>();
+ for (JsonNode value : node) {
+ values.add((float) value.asDouble());
+ }
+ return values;
+ }
+
+ public final String toJson() {
+ return JsonSerdeUtil.toFlatJson(toRootMap());
+ }
+
+ public abstract String queryText();
+
+ /** Returns all columns referenced by this query. */
+ public abstract List<String> columns();
+
+ public final String singleColumn() {
+ List<String> columns = columns();
+ checkArgument(
+ columns.size() == 1,
+ "Full-text query must reference exactly one column, but got: %s",
+ columns);
+ return columns.get(0);
+ }
+
+ abstract Map<String, Object> toRootMap();
+
+ private static void checkColumn(String column) {
+ checkArgument(column != null && !column.isEmpty(), "Column cannot be null or empty.");
+ }
+
+ private static void checkTerms(String terms) {
+ checkArgument(terms != null && !terms.isEmpty(), "Query terms cannot be null or empty.");
+ }
+
+ private static void checkBoost(float boost, String name) {
+ checkArgument(boost > 0, "%s must be positive, got: %s", name, boost);
+ }
+
+ private static List<String> distinctColumns(Iterable<String> columns) {
+ List<String> result = new ArrayList<>();
+ Set<String> seen = new HashSet<>();
+ for (String column : columns) {
+ if (seen.add(column)) {
+ result.add(column);
+ }
+ }
+ return Collections.unmodifiableList(result);
+ }
+
+ /** Match query. */
+ public static final class Match extends FullTextQuery {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String terms;
+ private final String column;
+ private final float boost;
+ @Nullable private final Integer fuzziness;
+ private final int maxExpansions;
+ private final Operator operator;
+ private final int prefixLength;
+
+ public Match(
+ String terms,
+ String column,
+ float boost,
+ @Nullable Integer fuzziness,
+ int maxExpansions,
+ Operator operator,
+ int prefixLength) {
+ checkColumn(column);
+ checkTerms(terms);
+ checkBoost(boost, "Boost factor");
+ checkArgument(
+ maxExpansions > 0, "maxExpansions must be positive, got: %s", maxExpansions);
+ checkArgument(
+ prefixLength >= 0, "prefixLength must be non-negative, got: %s", prefixLength);
+ if (fuzziness != null) {
+ checkArgument(fuzziness >= 0, "fuzziness must be non-negative, got: %s", fuzziness);
+ checkArgument(fuzziness <= 2, "fuzziness must be <= 2, got: %s", fuzziness);
+ }
+ this.column = column;
+ this.terms = terms;
+ this.boost = boost;
+ this.fuzziness = fuzziness;
+ this.maxExpansions = maxExpansions;
+ this.operator = operator == null ? Operator.OR : operator;
+ this.prefixLength = prefixLength;
+ }
+
+ public Match withBoost(float boost) {
+ return new Match(
+ terms, column, boost, fuzziness, maxExpansions, operator, prefixLength);
+ }
+
+ public Match withFuzziness(@Nullable Integer fuzziness) {
+ return new Match(
+ terms, column, boost, fuzziness, maxExpansions, operator, prefixLength);
+ }
+
+ public Match withMaxExpansions(int maxExpansions) {
+ return new Match(
+ terms, column, boost, fuzziness, maxExpansions, operator, prefixLength);
+ }
+
+ public Match withOperator(Operator operator) {
+ return new Match(
+ terms, column, boost, fuzziness, maxExpansions, operator, prefixLength);
+ }
+
+ public Match withOperator(String operator) {
+ return withOperator(Operator.fromString(operator));
+ }
+
+ public Match withPrefixLength(int prefixLength) {
+ return new Match(
+ terms, column, boost, fuzziness, maxExpansions, operator, prefixLength);
+ }
+
+ public String column() {
+ return column;
+ }
+
+ public String query() {
+ return terms;
+ }
+
+ public String terms() {
+ return terms;
+ }
+
+ public float boost() {
+ return boost;
+ }
+
+ @Nullable
+ public Integer fuzziness() {
+ return fuzziness;
+ }
+
+ public int maxExpansions() {
+ return maxExpansions;
+ }
+
+ public Operator operator() {
+ return operator;
+ }
+
+ public int prefixLength() {
+ return prefixLength;
+ }
+
+ @Override
+ public String queryText() {
+ return terms;
+ }
+
+ @Override
+ public List<String> columns() {
+ return Collections.singletonList(column);
+ }
+
+ @Override
+ Map<String, Object> toRootMap() {
+ Map<String, Object> body = new LinkedHashMap<>();
+ body.put("column", column);
+ body.put("terms", terms);
+ body.put("boost", boost);
+ body.put("fuzziness", fuzziness);
+ body.put("max_expansions", maxExpansions);
+ body.put("operator", operator.jsonValue());
+ body.put("prefix_length", prefixLength);
+ Map<String, Object> root = new LinkedHashMap<>();
+ root.put("match", body);
+ return root;
+ }
+ }
+
+ /** Phrase query. */
+ public static final class Phrase extends FullTextQuery {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String terms;
+ private final String column;
+ private final int slop;
+
+ public Phrase(String terms, String column, int slop) {
+ checkColumn(column);
+ checkTerms(terms);
+ checkArgument(slop >= 0, "slop must be non-negative, got: %s", slop);
+ this.column = column;
+ this.terms = terms;
+ this.slop = slop;
+ }
+
+ public String column() {
+ return column;
+ }
+
+ public String query() {
+ return terms;
+ }
+
+ public String terms() {
+ return terms;
+ }
+
+ public int slop() {
+ return slop;
+ }
+
+ @Override
+ public String queryText() {
+ return terms;
+ }
+
+ @Override
+ public List<String> columns() {
+ return Collections.singletonList(column);
+ }
+
+ @Override
+ Map<String, Object> toRootMap() {
+ Map<String, Object> body = new LinkedHashMap<>();
+ body.put("column", column);
+ body.put("terms", terms);
+ body.put("slop", slop);
+ Map<String, Object> root = new LinkedHashMap<>();
+ root.put("match_phrase", body);
+ return root;
+ }
+ }
+
+ /** Boost query. */
+ public static final class Boost extends FullTextQuery {
+
+ private static final long serialVersionUID = 1L;
+
+ private final FullTextQuery positive;
+ private final FullTextQuery negative;
+ private final float negativeBoost;
+
+ public Boost(FullTextQuery positive, FullTextQuery negative, float negativeBoost) {
+ this.positive = checkNotNull(positive, "Positive boost query cannot be null.");
+ this.negative = checkNotNull(negative, "Negative boost query cannot be null.");
+ checkBoost(negativeBoost, "Negative boost");
+ this.negativeBoost = negativeBoost;
+ }
+
+ public FullTextQuery positive() {
+ return positive;
+ }
+
+ public FullTextQuery negative() {
+ return negative;
+ }
+
+ public float negativeBoost() {
+ return negativeBoost;
+ }
+
+ @Override
+ public String queryText() {
+ return positive.queryText();
+ }
+
+ @Override
+ public List<String> columns() {
+ List<String> columns = new ArrayList<>();
+ columns.addAll(positive.columns());
+ columns.addAll(negative.columns());
+ return distinctColumns(columns);
+ }
+
+ @Override
+ Map<String, Object> toRootMap() {
+ Map<String, Object> body = new LinkedHashMap<>();
+ body.put("positive", positive.toRootMap());
+ body.put("negative", negative.toRootMap());
+ body.put("negative_boost", negativeBoost);
+ Map<String, Object> root = new LinkedHashMap<>();
+ root.put("boost", body);
+ return root;
+ }
+ }
+
+ /** Multi-match query. */
+ public static final class MultiMatch extends FullTextQuery {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String query;
+ private final List<String> columns;
+ private final List<Float> boosts;
+ private final Operator operator;
+
+ public MultiMatch(
+ String query,
+ List<String> columns,
+ @Nullable List<Float> boosts,
+ Operator operator) {
+ checkTerms(query);
+ checkArgument(
+ columns != null && !columns.isEmpty(), "columns cannot be null or empty.");
+ List<String> checkedColumns = new ArrayList<>();
+ for (String column : columns) {
+ checkColumn(column);
+ checkedColumns.add(column);
+ }
+ List<Float> checkedBoosts;
+ if (boosts == null) {
+ checkedBoosts = new ArrayList<>(checkedColumns.size());
+ for (int i = 0; i < checkedColumns.size(); i++) {
+ checkedBoosts.add(1.0f);
+ }
+ } else {
+ checkArgument(
+ boosts.size() == checkedColumns.size(),
+ "The number of boosts must match the number of columns.");
+ checkedBoosts = new ArrayList<>(boosts.size());
+ for (Float boost : boosts) {
+ checkArgument(boost != null, "Boost cannot be null.");
+ checkBoost(boost, "Boost");
+ checkedBoosts.add(boost);
+ }
+ }
+ this.query = query;
+ this.columns = Collections.unmodifiableList(checkedColumns);
+ this.boosts = Collections.unmodifiableList(checkedBoosts);
+ this.operator = operator == null ? Operator.OR : operator;
+ }
+
+ public String query() {
+ return query;
+ }
+
+ @Override
+ public String queryText() {
+ return query;
+ }
+
+ @Override
+ public List<String> columns() {
+ return columns;
+ }
+
+ public List<Float> boosts() {
+ return boosts;
+ }
+
+ public Operator operator() {
+ return operator;
+ }
+
+ @Override
+ Map<String, Object> toRootMap() {
+ Map<String, Object> body = new LinkedHashMap<>();
+ body.put("query", query);
+ body.put("columns", columns);
+ body.put("boost", boosts);
+ body.put("operator", operator.jsonValue());
+ Map<String, Object> root = new LinkedHashMap<>();
+ root.put("multi_match", body);
+ return root;
+ }
+ }
+
+ /** Boolean query clause. */
+ public static final class Clause implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final Occur occur;
+ private final FullTextQuery query;
+
+ public Clause(Occur occur, FullTextQuery query) {
+ this.occur = checkNotNull(occur, "Boolean query occur cannot be null.");
+ this.query = checkNotNull(query, "Boolean query cannot be null.");
+ }
+
+ public Occur occur() {
+ return occur;
+ }
+
+ public FullTextQuery query() {
+ return query;
+ }
+ }
+
+ /** Boolean query. */
+ public static final class BooleanQuery extends FullTextQuery {
+
+ private static final long serialVersionUID = 1L;
+
+ private final List<Clause> queries;
+ private final List<FullTextQuery> should;
+ private final List<FullTextQuery> must;
+ private final List<FullTextQuery> mustNot;
+
+ public BooleanQuery(List<Clause> queries) {
+ this(queries, true);
+ }
+
+ private BooleanQuery(List<Clause> queries, boolean checkNonEmpty) {
+ List<Clause> checkedQueries = new ArrayList<>();
+ if (queries != null) {
+ for (Clause query : queries) {
+ checkedQueries.add(checkNotNull(query, "Boolean query clause cannot be null."));
+ }
+ }
+ if (checkNonEmpty) {
+ checkArgument(
+ !checkedQueries.isEmpty(),
+ "Boolean query must contain at least one clause.");
+ }
+ this.queries = Collections.unmodifiableList(checkedQueries);
+ this.should = filterQueries(checkedQueries, Occur.SHOULD);
+ this.must = filterQueries(checkedQueries, Occur.MUST);
+ this.mustNot = filterQueries(checkedQueries, Occur.MUST_NOT);
+ }
+
+ public BooleanQuery(
+ List<FullTextQuery> should, List<FullTextQuery> must, List<FullTextQuery> mustNot) {
+ this(toClauses(should, must, mustNot));
+ }
+
+ private static List<Clause> toClauses(
+ List<FullTextQuery> should, List<FullTextQuery> must, List<FullTextQuery> mustNot) {
+ List<Clause> clauses = new ArrayList<>();
+ addClauses(clauses, Occur.SHOULD, should);
+ addClauses(clauses, Occur.MUST, must);
+ addClauses(clauses, Occur.MUST_NOT, mustNot);
+ return clauses;
+ }
+
+ private static void addClauses(
+ List<Clause> clauses, Occur occur, @Nullable List<FullTextQuery> queries) {
+ if (queries == null) {
+ return;
+ }
+ for (FullTextQuery query : queries) {
+ clauses.add(new Clause(occur, query));
+ }
+ }
+
+ private static List<FullTextQuery> filterQueries(List<Clause> queries, Occur occur) {
+ List<FullTextQuery> result = new ArrayList<>();
+ for (Clause query : queries) {
+ if (query.occur() == occur) {
+ result.add(query.query());
+ }
+ }
+ return Collections.unmodifiableList(result);
+ }
+
+ public BooleanQuery should(FullTextQuery query) {
+ return with(Occur.SHOULD, query);
+ }
+
+ public BooleanQuery must(FullTextQuery query) {
+ return with(Occur.MUST, query);
+ }
+
+ public BooleanQuery mustNot(FullTextQuery query) {
+ return with(Occur.MUST_NOT, query);
+ }
+
+ private BooleanQuery with(Occur occur, FullTextQuery query) {
+ List<Clause> next = new ArrayList<>(queries);
+ next.add(new Clause(occur, query));
+ return new BooleanQuery(next);
+ }
+
+ public List<Clause> queries() {
+ return queries;
+ }
+
+ public List<FullTextQuery> should() {
+ return should;
+ }
+
+ public List<FullTextQuery> must() {
+ return must;
+ }
+
+ public List<FullTextQuery> mustNot() {
+ return mustNot;
+ }
+
+ @Override
+ public String queryText() {
+ return "";
+ }
+
+ @Override
+ public List<String> columns() {
+ List<String> columns = new ArrayList<>();
+ for (Clause query : queries) {
+ columns.addAll(query.query().columns());
+ }
+ return distinctColumns(columns);
+ }
+
+ @Override
+ Map<String, Object> toRootMap() {
+ Map<String, Object> body = new LinkedHashMap<>();
+ body.put("should", queryMaps(should));
+ body.put("must", queryMaps(must));
+ body.put("must_not", queryMaps(mustNot));
+ Map<String, Object> root = new LinkedHashMap<>();
+ root.put("boolean", body);
+ return root;
+ }
+
+ private static List<Map<String, Object>> queryMaps(List<FullTextQuery> queries) {
+ List<Map<String, Object>> result = new ArrayList<>();
+ for (FullTextQuery query : queries) {
+ result.add(query.toRootMap());
+ }
+ return result;
+ }
+ }
+}
diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/FullTextSearch.java b/paimon-common/src/main/java/org/apache/paimon/predicate/FullTextSearch.java
index d25225ec69..b9731434c2 100644
--- a/paimon-common/src/main/java/org/apache/paimon/predicate/FullTextSearch.java
+++ b/paimon-common/src/main/java/org/apache/paimon/predicate/FullTextSearch.java
@@ -19,63 +19,51 @@
package org.apache.paimon.predicate;
import java.io.Serializable;
+import java.util.List;
-/** FullTextSearch to perform full-text search on a text column. */
+/** FullTextSearch to perform full-text search with a structured query. */
public class FullTextSearch implements Serializable {
private static final long serialVersionUID = 1L;
- private final String queryText;
- private final String fieldName;
+ private final FullTextQuery query;
private final int limit;
- private final String queryOperator;
- public FullTextSearch(String queryText, int limit, String fieldName) {
- this(queryText, limit, fieldName, "or");
- }
-
- public FullTextSearch(String queryText, int limit, String fieldName, String queryOperator) {
- if (queryText == null || queryText.isEmpty()) {
- throw new IllegalArgumentException("Query text cannot be null or empty");
+ public FullTextSearch(FullTextQuery query, int limit) {
+ if (query == null) {
+ throw new IllegalArgumentException("Query cannot be null");
}
if (limit <= 0) {
throw new IllegalArgumentException("Limit must be positive, got: " + limit);
}
- if (fieldName == null || fieldName.isEmpty()) {
- throw new IllegalArgumentException("Field name cannot be null or empty");
- }
- this.queryText = queryText;
+ this.query = query;
this.limit = limit;
- this.fieldName = fieldName;
- 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.queryOperator = normalizedOperator;
- }
-
- public String queryText() {
- return queryText;
}
public int limit() {
return limit;
}
+ public List<String> columns() {
+ return query.columns();
+ }
+
public String fieldName() {
- return fieldName;
+ return query.singleColumn();
+ }
+
+ public FullTextQuery query() {
+ return query;
}
- public String queryOperator() {
- return queryOperator;
+ public String queryJson() {
+ return query.toJson();
}
@Override
public String toString() {
return String.format(
- "FullTextSearch{field=%s, query='%s', limit=%d, operator=%s}",
- fieldName, queryText, limit, queryOperator);
+ "FullTextSearch{columns=%s, limit=%d, queryJson=%s}",
+ columns(), limit, queryJson());
}
}
diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/HybridSearchRoute.java b/paimon-common/src/main/java/org/apache/paimon/predicate/HybridSearchRoute.java
index 4800479a73..beddab745a 100644
--- a/paimon-common/src/main/java/org/apache/paimon/predicate/HybridSearchRoute.java
+++ b/paimon-common/src/main/java/org/apache/paimon/predicate/HybridSearchRoute.java
@@ -39,8 +39,7 @@ public class HybridSearchRoute implements Serializable {
private final RouteType routeType;
private final String fieldName;
private final float[] vector;
- private final String queryText;
- private final String queryOperator;
+ private final FullTextQuery fullTextQuery;
private final int limit;
private final float weight;
private final Map<String, String> options;
@@ -55,7 +54,7 @@ public class HybridSearchRoute implements Serializable {
int limit,
float weight,
Map<String, String> options) {
- this(RouteType.VECTOR, fieldName, vector, null, "or", limit, weight, options);
+ this(RouteType.VECTOR, fieldName, vector, null, limit, weight, options);
}
public static HybridSearchRoute vector(
@@ -68,29 +67,17 @@ public class HybridSearchRoute implements Serializable {
}
public static HybridSearchRoute fullText(
- String fieldName,
- String queryText,
- String queryOperator,
- int limit,
- float weight,
- Map<String, String> options) {
+ String queryJson, int limit, float weight, Map<String, String> options) {
+ FullTextQuery query = FullTextQuery.fromJson(queryJson);
return new HybridSearchRoute(
- RouteType.FULL_TEXT,
- fieldName,
- null,
- queryText,
- queryOperator,
- limit,
- weight,
- options);
+ RouteType.FULL_TEXT, query.columns().get(0), null, query, limit, weight, options);
}
private HybridSearchRoute(
RouteType routeType,
String fieldName,
float[] vector,
- String queryText,
- String queryOperator,
+ FullTextQuery fullTextQuery,
int limit,
float weight,
Map<String, String> options) {
@@ -103,9 +90,8 @@ public class HybridSearchRoute implements Serializable {
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 (routeType == RouteType.FULL_TEXT && fullTextQuery == null) {
+ throw new IllegalArgumentException("Query cannot be null for full-text route");
}
if (limit <= 0) {
throw new IllegalArgumentException("Limit must be positive, got: " + limit);
@@ -113,17 +99,10 @@ public class HybridSearchRoute 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.fullTextQuery = fullTextQuery;
this.limit = limit;
this.weight = weight;
this.options =
@@ -157,12 +136,8 @@ public class HybridSearchRoute implements Serializable {
return vector;
}
- public String queryText() {
- return queryText;
- }
-
- public String queryOperator() {
- return queryOperator;
+ public FullTextQuery fullTextQuery() {
+ return fullTextQuery;
}
public int limit() {
@@ -182,7 +157,14 @@ public class HybridSearchRoute implements Serializable {
}
public FullTextSearch toFullTextSearch() {
- return new FullTextSearch(queryText, limit, fieldName, queryOperator);
+ return new FullTextSearch(fullTextQuery, limit);
+ }
+
+ public Iterable<String> columns() {
+ if (isFullText()) {
+ return fullTextQuery.columns();
+ }
+ return Collections.singletonList(fieldName);
}
@Override
@@ -204,8 +186,7 @@ public class HybridSearchRoute implements Serializable {
private String fieldName;
private float[] vector;
- private String queryText;
- private String queryOperator = "or";
+ private FullTextQuery fullTextQuery;
private int limit;
private float weight = 1.0f;
private Map<String, String> options = new HashMap<>();
@@ -219,11 +200,6 @@ public class HybridSearchRoute 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;
@@ -233,13 +209,9 @@ public class HybridSearchRoute implements Serializable {
return queryVector(vector);
}
- public Builder queryText(String queryText) {
- this.queryText = queryText;
- return this;
- }
-
- public Builder queryOperator(String queryOperator) {
- this.queryOperator = queryOperator;
+ public Builder query(String queryJson) {
+ this.fullTextQuery = FullTextQuery.fromJson(queryJson);
+ this.fieldName = fullTextQuery.columns().get(0);
return this;
}
@@ -266,8 +238,15 @@ public class HybridSearchRoute implements Serializable {
}
public HybridSearchRoute build() {
- if (queryText != null) {
- return fullText(fieldName, queryText, queryOperator, limit, weight, options);
+ if (fullTextQuery != null) {
+ return new HybridSearchRoute(
+ RouteType.FULL_TEXT,
+ fullTextQuery.columns().get(0),
+ null,
+ fullTextQuery,
+ limit,
+ weight,
+ options);
}
return HybridSearchRoute.vector(fieldName, vector, limit, weight, options);
}
diff --git a/paimon-common/src/test/java/org/apache/paimon/globalindex/testfulltext/TestFullTextGlobalIndexReader.java b/paimon-common/src/test/java/org/apache/paimon/globalindex/testfulltext/TestFullTextGlobalIndexReader.java
index d4ff47cca4..1f6240f529 100644
--- a/paimon-common/src/test/java/org/apache/paimon/globalindex/testfulltext/TestFullTextGlobalIndexReader.java
+++ b/paimon-common/src/test/java/org/apache/paimon/globalindex/testfulltext/TestFullTextGlobalIndexReader.java
@@ -25,6 +25,7 @@ import org.apache.paimon.globalindex.GlobalIndexResult;
import org.apache.paimon.globalindex.ScoredGlobalIndexResult;
import org.apache.paimon.globalindex.io.GlobalIndexFileReader;
import org.apache.paimon.predicate.FieldRef;
+import org.apache.paimon.predicate.FullTextQuery;
import org.apache.paimon.predicate.FullTextSearch;
import org.apache.paimon.utils.RoaringNavigableMap64;
@@ -72,22 +73,18 @@ public class TestFullTextGlobalIndexReader implements GlobalIndexReader {
throw new RuntimeException("Failed to load test full-text index", e);
}
- String queryText = fullTextSearch.queryText();
int limit = fullTextSearch.limit();
int effectiveK = Math.min(limit, count);
if (effectiveK <= 0) {
return CompletableFuture.completedFuture(Optional.empty());
}
- String[] queryTerms = queryText.toLowerCase(Locale.ROOT).split("\\s+");
- boolean requireAllTerms = "and".equals(fullTextSearch.queryOperator());
-
// Min-heap: smallest score at head, so we evict the weakest candidate.
PriorityQueue<ScoredRow> topK =
new PriorityQueue<>(effectiveK + 1, Comparator.comparingDouble(s -> s.score));
for (int i = 0; i < count; i++) {
- float score = computeScore(documents[i], queryTerms, requireAllTerms);
+ float score = computeScore(documents[i], fullTextSearch.query());
if (score <= 0) {
continue;
}
@@ -114,8 +111,41 @@ public class TestFullTextGlobalIndexReader implements GlobalIndexReader {
Optional.of(ScoredGlobalIndexResult.create(resultBitmap, scoreMap::get)));
}
- private static float computeScore(
- String document, String[] queryTerms, boolean requireAllTerms) {
+ private static float computeScore(String document, FullTextQuery query) {
+ if (query instanceof FullTextQuery.Match) {
+ FullTextQuery.Match match = (FullTextQuery.Match) query;
+ return computeMatchScore(
+ document, match.terms(), match.operator() == FullTextQuery.Operator.AND)
+ * match.boost();
+ }
+ if (query instanceof FullTextQuery.Phrase) {
+ FullTextQuery.Phrase phrase = (FullTextQuery.Phrase) query;
+ return document.toLowerCase(Locale.ROOT)
+ .contains(phrase.terms().toLowerCase(Locale.ROOT))
+ ? 1.0f
+ : 0.0f;
+ }
+ if (query instanceof FullTextQuery.Boost) {
+ FullTextQuery.Boost boost = (FullTextQuery.Boost) query;
+ float score = computeScore(document, boost.positive());
+ if (computeScore(document, boost.negative()) > 0) {
+ score *= boost.negativeBoost();
+ }
+ return score;
+ }
+ if (query instanceof FullTextQuery.MultiMatch) {
+ throw new IllegalArgumentException(
+ "multi_match is not supported by single-column full-text indexes");
+ }
+ if (query instanceof FullTextQuery.BooleanQuery) {
+ return computeBooleanScore(document, (FullTextQuery.BooleanQuery) query);
+ }
+ throw new IllegalArgumentException("Unsupported full-text query: " + query);
+ }
+
+ private static float computeMatchScore(
+ String document, String queryText, boolean requireAllTerms) {
+ String[] queryTerms = queryText.toLowerCase(Locale.ROOT).split("\\s+");
String lowerDoc = document.toLowerCase(Locale.ROOT);
float score = 0;
for (String term : queryTerms) {
@@ -128,6 +158,30 @@ public class TestFullTextGlobalIndexReader implements GlobalIndexReader {
return score;
}
+ private static float computeBooleanScore(String document, FullTextQuery.BooleanQuery query) {
+ float score = 0;
+ for (FullTextQuery child : query.must()) {
+ float childScore = computeScore(document, child);
+ if (childScore <= 0) {
+ return 0;
+ }
+ score += childScore;
+ }
+ for (FullTextQuery child : query.mustNot()) {
+ if (computeScore(document, child) > 0) {
+ return 0;
+ }
+ }
+ float shouldScore = 0;
+ for (FullTextQuery child : query.should()) {
+ shouldScore += computeScore(document, child);
+ }
+ if (query.must().isEmpty() && !query.should().isEmpty() && shouldScore <= 0) {
+ return 0;
+ }
+ return score + shouldScore;
+ }
+
private void ensureLoaded() throws IOException {
if (documents != null) {
return;
diff --git a/paimon-common/src/test/java/org/apache/paimon/globalindex/testfulltext/TestFullTextGlobalIndexerFactory.java b/paimon-common/src/test/java/org/apache/paimon/globalindex/testfulltext/TestFullTextGlobalIndexerFactory.java
index b501c2df10..e381fef383 100644
--- a/paimon-common/src/test/java/org/apache/paimon/globalindex/testfulltext/TestFullTextGlobalIndexerFactory.java
+++ b/paimon-common/src/test/java/org/apache/paimon/globalindex/testfulltext/TestFullTextGlobalIndexerFactory.java
@@ -36,6 +36,11 @@ public class TestFullTextGlobalIndexerFactory implements GlobalIndexerFactory {
return IDENTIFIER;
}
+ @Override
+ public boolean supportsFullTextSearch() {
+ return true;
+ }
+
@Override
public GlobalIndexer create(DataField field, Options options) {
return new TestFullTextGlobalIndexer(field.type(), options);
diff --git a/paimon-common/src/test/java/org/apache/paimon/predicate/FullTextQueryTest.java b/paimon-common/src/test/java/org/apache/paimon/predicate/FullTextQueryTest.java
new file mode 100644
index 0000000000..2cd4e2a491
--- /dev/null
+++ b/paimon-common/src/test/java/org/apache/paimon/predicate/FullTextQueryTest.java
@@ -0,0 +1,175 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.predicate;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link FullTextQuery}. */
+public class FullTextQueryTest {
+
+ @Test
+ public void testMatchQueryJson() {
+ FullTextQuery query =
+ FullTextQuery.fromJson(
+ "{\"match\":{\"column\":\"content\",\"terms\":\"paimon search\","
+ + "\"operator\":\"And\",\"boost\":2.0,\"fuzziness\":1}}");
+
+ assertThat(query).isInstanceOf(FullTextQuery.Match.class);
+ FullTextQuery.Match match = (FullTextQuery.Match) query;
+ assertThat(match.column()).isEqualTo("content");
+ assertThat(match.terms()).isEqualTo("paimon search");
+ assertThat(match.operator()).isEqualTo(FullTextQuery.Operator.AND);
+ assertThat(match.boost()).isEqualTo(2.0f);
+ assertThat(match.fuzziness()).isEqualTo(1);
+ assertThat(query.toJson())
+ .isEqualTo(
+ "{\"match\":{\"column\":\"content\",\"terms\":\"paimon search\","
+ + "\"boost\":2.0,\"fuzziness\":1,\"max_expansions\":50,"
+ + "\"operator\":\"And\",\"prefix_length\":0}}");
+ }
+
+ @Test
+ public void testPhraseBoostMultiMatchAndBooleanJson() {
+ FullTextQuery phrase =
+ FullTextQuery.fromJson(
+ "{\"phrase\":{\"column\":\"content\",\"terms\":\"full text\",\"slop\":1}}");
+ assertThat(phrase).isInstanceOf(FullTextQuery.Phrase.class);
+ assertThat(((FullTextQuery.Phrase) phrase).column()).isEqualTo("content");
+ assertThat(((FullTextQuery.Phrase) phrase).slop()).isEqualTo(1);
+ assertThat(phrase.toJson())
+ .isEqualTo(
+ "{\"match_phrase\":{\"column\":\"content\",\"terms\":\"full text\","
+ + "\"slop\":1}}");
+
+ FullTextQuery boost =
+ FullTextQuery.boost(
+ FullTextQuery.match("paimon", "content"),
+ FullTextQuery.match("vector", "content"),
+ 0.3f);
+ assertThat(boost).isInstanceOf(FullTextQuery.Boost.class);
+ assertThat(((FullTextQuery.Boost) boost).negativeBoost()).isEqualTo(0.3f);
+ assertThat(boost.toJson())
+ .isEqualTo(
+ "{\"boost\":{\"positive\":{\"match\":{\"column\":\"content\",\"terms\":\"paimon\","
+ + "\"boost\":1.0,\"fuzziness\":0,\"max_expansions\":50,"
+ + "\"operator\":\"Or\",\"prefix_length\":0}},\"negative\":{\"match\":"
+ + "{\"column\":\"content\",\"terms\":\"vector\",\"boost\":1.0,"
+ + "\"fuzziness\":0,\"max_expansions\":50,\"operator\":\"Or\","
+ + "\"prefix_length\":0}},\"negative_boost\":0.3}}");
+
+ FullTextQuery multiMatch =
+ FullTextQuery.multiMatch("paimon", Arrays.asList("title", "content"), null, "and");
+ assertThat(multiMatch).isInstanceOf(FullTextQuery.MultiMatch.class);
+ assertThat(multiMatch.columns()).containsExactly("title", "content");
+ assertThat(multiMatch.toJson())
+ .isEqualTo(
+ "{\"multi_match\":{\"query\":\"paimon\",\"columns\":[\"title\",\"content\"],"
+ + "\"boost\":[1.0,1.0],\"operator\":\"And\"}}");
+
+ FullTextQuery.BooleanQuery booleanQuery =
+ new FullTextQuery.BooleanQuery(
+ Collections.singletonList(FullTextQuery.match("lake", "content")),
+ Arrays.asList(
+ FullTextQuery.match("paimon", "content"),
+ FullTextQuery.phrase("full text", "content")),
+ Collections.singletonList(FullTextQuery.match("vector", "content")));
+
+ assertThat(booleanQuery.should()).hasSize(1);
+ assertThat(booleanQuery.must()).hasSize(2);
+ assertThat(booleanQuery.mustNot()).hasSize(1);
+ assertThat(booleanQuery.singleColumn()).isEqualTo("content");
+ assertThat(booleanQuery.toJson()).contains("\"boolean\"");
+ }
+
+ @Test
+ public void testParseBooleanQueriesList() {
+ FullTextQuery query =
+ FullTextQuery.fromJson(
+ "{\"boolean\":{\"queries\":["
+ + "{\"occur\":\"must\",\"query\":{\"match\":{\"column\":\"content\","
+ + "\"terms\":\"paimon\"}}},"
+ + "[\"must_not\",{\"match\":{\"column\":\"content\","
+ + "\"terms\":\"vector\"}}]]}}");
+
+ assertThat(query).isInstanceOf(FullTextQuery.BooleanQuery.class);
+ FullTextQuery.BooleanQuery booleanQuery = (FullTextQuery.BooleanQuery) query;
+ assertThat(booleanQuery.must()).hasSize(1);
+ assertThat(booleanQuery.mustNot()).hasSize(1);
+ assertThat(booleanQuery.should()).isEmpty();
+ }
+
+ @Test
+ public void testParseLanceDbAliases() {
+ FullTextQuery query =
+ FullTextQuery.fromJson(
+ "{\"match\":{\"column\":\"content\",\"query\":\"paimon\","
+ + "\"maxExpansions\":10,\"prefixLength\":1,"
+ + "\"fuzziness\":\"auto\"}}");
+
+ assertThat(query).isInstanceOf(FullTextQuery.Match.class);
+ FullTextQuery.Match match = (FullTextQuery.Match) query;
+ assertThat(match.terms()).isEqualTo("paimon");
+ assertThat(match.maxExpansions()).isEqualTo(10);
+ assertThat(match.prefixLength()).isEqualTo(1);
+ assertThat(match.fuzziness()).isNull();
+
+ FullTextQuery phrase =
+ FullTextQuery.fromJson(
+ "{\"match_phrase\":{\"column\":\"content\",\"query\":\"paimon lake\"}}");
+ assertThat(phrase).isInstanceOf(FullTextQuery.Phrase.class);
+ assertThat(((FullTextQuery.Phrase) phrase).terms()).isEqualTo("paimon lake");
+
+ FullTextQuery boost =
+ FullTextQuery.fromJson(
+ "{\"boost\":{\"positive\":{\"match\":{\"column\":\"content\","
+ + "\"terms\":\"paimon\"}},\"negative\":{\"match\":"
+ + "{\"column\":\"content\",\"terms\":\"vector\"}},"
+ + "\"negativeBoost\":0.2}}");
+ assertThat(((FullTextQuery.Boost) boost).negativeBoost()).isEqualTo(0.2f);
+
+ FullTextQuery multiMatch =
+ FullTextQuery.fromJson(
+ "{\"multi_match\":{\"query\":\"paimon\","
+ + "\"columns\":[\"title\",\"content\"],"
+ + "\"boosts\":[2.0,1.0]}}");
+ assertThat(((FullTextQuery.MultiMatch) multiMatch).boosts()).containsExactly(2.0f, 1.0f);
+ }
+
+ @Test
+ public void testFullTextSearchKeepsStructuredQueryJson() {
+ FullTextSearch search =
+ new FullTextSearch(
+ FullTextQuery.fromJson(
+ "{\"phrase\":{\"column\":\"content\","
+ + "\"terms\":\"paimon lake\",\"slop\":1}}"),
+ 10);
+
+ assertThat(search.fieldName()).isEqualTo("content");
+ assertThat(search.query()).isInstanceOf(FullTextQuery.Phrase.class);
+ assertThat(search.queryJson())
+ .isEqualTo(
+ "{\"match_phrase\":{\"column\":\"content\",\"terms\":\"paimon lake\","
+ + "\"slop\":1}}");
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextReadImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextReadImpl.java
index 0ca65cd05f..e6e93b5211 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextReadImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextReadImpl.java
@@ -31,13 +31,18 @@ import org.apache.paimon.globalindex.io.GlobalIndexFileReader;
import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.index.IndexPathFactory;
+import org.apache.paimon.predicate.FullTextQuery;
import org.apache.paimon.predicate.FullTextSearch;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.types.DataField;
import org.apache.paimon.utils.IOUtils;
+import org.apache.paimon.utils.RoaringNavigableMap64;
import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
@@ -50,26 +55,20 @@ public class FullTextReadImpl implements FullTextRead {
private final FileStoreTable table;
private final int limit;
- private final DataField textColumn;
- private final String queryText;
- private final String queryOperator;
+ private final List<DataField> textColumns;
+ private final FullTextQuery query;
public FullTextReadImpl(
- FileStoreTable table, int limit, DataField textColumn, String queryText) {
- this(table, limit, textColumn, queryText, "or");
+ FileStoreTable table, int limit, DataField textColumn, FullTextQuery query) {
+ this(table, limit, Collections.singletonList(textColumn), query);
}
public FullTextReadImpl(
- FileStoreTable table,
- int limit,
- DataField textColumn,
- String queryText,
- String queryOperator) {
+ FileStoreTable table, int limit, List<DataField> textColumns, FullTextQuery query) {
this.table = table;
this.limit = limit;
- this.textColumn = textColumn;
- this.queryText = queryText;
- this.queryOperator = queryOperator;
+ this.textColumns = Collections.unmodifiableList(new ArrayList<>(textColumns));
+ this.query = query;
}
@Override
@@ -78,7 +77,164 @@ public class FullTextReadImpl implements FullTextRead {
return GlobalIndexResult.createEmpty();
}
- IndexFileMeta firstFile = splits.get(0).fullTextIndexFiles().get(0);
+ IndexPathFactory indexPathFactory = table.store().pathFactory().globalIndexFileFactory();
+
+ int parallelism = table.coreOptions().toConfiguration().get(GLOBAL_INDEX_THREAD_NUM);
+ ExecutorService executor = GlobalIndexReadThreadPool.getExecutorService(parallelism);
+
+ Map<String, DataField> fieldsByName = new HashMap<>();
+ for (DataField textColumn : textColumns) {
+ fieldsByName.put(textColumn.name(), textColumn);
+ }
+
+ Map<String, List<FullTextSearchSplit>> splitsByColumn = new HashMap<>();
+ for (FullTextSearchSplit split : splits) {
+ splitsByColumn.computeIfAbsent(split.columnName(), k -> new ArrayList<>()).add(split);
+ }
+
+ return evalQuery(query, fieldsByName, splitsByColumn, indexPathFactory, executor)
+ .topK(limit);
+ }
+
+ private ScoredGlobalIndexResult evalQuery(
+ FullTextQuery query,
+ Map<String, DataField> fieldsByName,
+ Map<String, List<FullTextSearchSplit>> splitsByColumn,
+ IndexPathFactory indexPathFactory,
+ ExecutorService executor) {
+ if (query instanceof FullTextQuery.Match) {
+ return evalColumnQuery(
+ query,
+ ((FullTextQuery.Match) query).column(),
+ fieldsByName,
+ splitsByColumn,
+ indexPathFactory,
+ executor);
+ }
+ if (query instanceof FullTextQuery.Phrase) {
+ return evalColumnQuery(
+ query,
+ ((FullTextQuery.Phrase) query).column(),
+ fieldsByName,
+ splitsByColumn,
+ indexPathFactory,
+ executor);
+ }
+ if (query instanceof FullTextQuery.MultiMatch) {
+ return evalMultiMatch(
+ (FullTextQuery.MultiMatch) query,
+ fieldsByName,
+ splitsByColumn,
+ indexPathFactory,
+ executor);
+ }
+ if (query instanceof FullTextQuery.Boost) {
+ FullTextQuery.Boost boost = (FullTextQuery.Boost) query;
+ return boost(
+ evalQuery(
+ boost.positive(),
+ fieldsByName,
+ splitsByColumn,
+ indexPathFactory,
+ executor),
+ evalQuery(
+ boost.negative(),
+ fieldsByName,
+ splitsByColumn,
+ indexPathFactory,
+ executor),
+ boost.negativeBoost());
+ }
+ if (query instanceof FullTextQuery.BooleanQuery) {
+ return evalBoolean(
+ (FullTextQuery.BooleanQuery) query,
+ fieldsByName,
+ splitsByColumn,
+ indexPathFactory,
+ executor);
+ }
+ throw new IllegalArgumentException("Unsupported full-text query: " + query);
+ }
+
+ private ScoredGlobalIndexResult evalMultiMatch(
+ FullTextQuery.MultiMatch query,
+ Map<String, DataField> fieldsByName,
+ Map<String, List<FullTextSearchSplit>> splitsByColumn,
+ IndexPathFactory indexPathFactory,
+ ExecutorService executor) {
+ List<String> columns = query.columns();
+ List<Float> boosts = query.boosts();
+ List<ScoredGlobalIndexResult> results = new ArrayList<>(columns.size());
+ for (int i = 0; i < columns.size(); i++) {
+ FullTextQuery.Match match =
+ new FullTextQuery.Match(
+ query.query(),
+ columns.get(i),
+ boosts.get(i),
+ 0,
+ 50,
+ query.operator(),
+ 0);
+ results.add(
+ evalColumnQuery(
+ match,
+ columns.get(i),
+ fieldsByName,
+ splitsByColumn,
+ indexPathFactory,
+ executor));
+ }
+ return or(results).topK(limit);
+ }
+
+ private ScoredGlobalIndexResult evalBoolean(
+ FullTextQuery.BooleanQuery query,
+ Map<String, DataField> fieldsByName,
+ Map<String, List<FullTextSearchSplit>> splitsByColumn,
+ IndexPathFactory indexPathFactory,
+ ExecutorService executor) {
+ ScoredGlobalIndexResult result = null;
+ for (FullTextQuery child : query.must()) {
+ ScoredGlobalIndexResult childResult =
+ evalQuery(child, fieldsByName, splitsByColumn, indexPathFactory, executor);
+ result = result == null ? childResult : and(result, childResult);
+ }
+
+ List<ScoredGlobalIndexResult> shouldResults = new ArrayList<>(query.should().size());
+ for (FullTextQuery child : query.should()) {
+ shouldResults.add(
+ evalQuery(child, fieldsByName, splitsByColumn, indexPathFactory, executor));
+ }
+ if (!shouldResults.isEmpty()) {
+ ScoredGlobalIndexResult shouldResult = or(shouldResults);
+ result = result == null ? shouldResult : andWithBonus(result, shouldResult);
+ }
+
+ if (result == null) {
+ return ScoredGlobalIndexResult.createEmpty();
+ }
+ for (FullTextQuery child : query.mustNot()) {
+ ScoredGlobalIndexResult childResult =
+ evalQuery(child, fieldsByName, splitsByColumn, indexPathFactory, executor);
+ result = andNot(result, childResult);
+ }
+ return result.topK(limit);
+ }
+
+ private ScoredGlobalIndexResult evalColumnQuery(
+ FullTextQuery query,
+ String column,
+ Map<String, DataField> fieldsByName,
+ Map<String, List<FullTextSearchSplit>> splitsByColumn,
+ IndexPathFactory indexPathFactory,
+ ExecutorService executor) {
+ List<FullTextSearchSplit> columnSplits = splitsByColumn.get(column);
+ if (columnSplits == null || columnSplits.isEmpty()) {
+ return ScoredGlobalIndexResult.createEmpty();
+ }
+ DataField textColumn = checkNotNull(fieldsByName.get(column));
+
+ IndexFileMeta firstFile = columnSplits.get(0).fullTextIndexFiles().get(0);
String indexType = firstFile.indexType();
GlobalIndexMeta firstMeta = checkNotNull(firstFile.globalIndexMeta());
GlobalIndexer globalIndexer;
@@ -94,14 +250,10 @@ public class FullTextReadImpl implements FullTextRead {
GlobalIndexerFactoryUtils.load(indexType)
.create(textColumn, table.coreOptions().toConfiguration());
}
- IndexPathFactory indexPathFactory = table.store().pathFactory().globalIndexFileFactory();
-
- int parallelism = table.coreOptions().toConfiguration().get(GLOBAL_INDEX_THREAD_NUM);
- ExecutorService executor = GlobalIndexReadThreadPool.getExecutorService(parallelism);
List<CompletableFuture<Optional<ScoredGlobalIndexResult>>> futures =
- new ArrayList<>(splits.size());
- for (FullTextSearchSplit split : splits) {
+ new ArrayList<>(columnSplits.size());
+ for (FullTextSearchSplit split : columnSplits) {
futures.add(
eval(
globalIndexer,
@@ -109,6 +261,7 @@ public class FullTextReadImpl implements FullTextRead {
split.rowRangeStart(),
split.rowRangeEnd(),
split.fullTextIndexFiles(),
+ query,
executor));
}
@@ -122,7 +275,7 @@ public class FullTextReadImpl implements FullTextRead {
}
}
- return result.topK(limit);
+ return result;
}
private CompletableFuture<Optional<ScoredGlobalIndexResult>> eval(
@@ -131,6 +284,7 @@ public class FullTextReadImpl implements FullTextRead {
long rowRangeStart,
long rowRangeEnd,
List<IndexFileMeta> fullTextIndexFiles,
+ FullTextQuery query,
ExecutorService executor) {
List<GlobalIndexIOMeta> indexIOMetaList = new ArrayList<>();
for (IndexFileMeta indexFile : fullTextIndexFiles) {
@@ -147,9 +301,76 @@ public class FullTextReadImpl implements FullTextRead {
GlobalIndexReader reader =
globalIndexer.createReader(indexFileReader, indexIOMetaList, executor);
FullTextSearch fullTextSearch =
- new FullTextSearch(queryText, limit, textColumn.name(), queryOperator);
+ new FullTextSearch(query, candidateLimit(rowRangeStart, rowRangeEnd));
return new OffsetGlobalIndexReader(reader, rowRangeStart, rowRangeEnd)
.visitFullTextSearch(fullTextSearch)
.whenComplete((r, t) -> IOUtils.closeQuietly(reader));
}
+
+ private static int candidateLimit(long rowRangeStart, long rowRangeEnd) {
+ long size = rowRangeEnd - rowRangeStart + 1;
+ return size > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) size;
+ }
+
+ private static ScoredGlobalIndexResult and(
+ ScoredGlobalIndexResult left, ScoredGlobalIndexResult right) {
+ RoaringNavigableMap64 rowIds = RoaringNavigableMap64.and(left.results(), right.results());
+ return ScoredGlobalIndexResult.create(
+ rowIds,
+ rowId -> left.scoreGetter().score(rowId) + right.scoreGetter().score(rowId));
+ }
+
+ private static ScoredGlobalIndexResult or(List<ScoredGlobalIndexResult> results) {
+ if (results.isEmpty()) {
+ return ScoredGlobalIndexResult.createEmpty();
+ }
+ RoaringNavigableMap64 rowIds = new RoaringNavigableMap64();
+ for (ScoredGlobalIndexResult result : results) {
+ rowIds = RoaringNavigableMap64.or(rowIds, result.results());
+ }
+ final RoaringNavigableMap64 mergedRowIds = rowIds;
+ return ScoredGlobalIndexResult.create(
+ mergedRowIds,
+ rowId -> {
+ float score = 0.0f;
+ for (ScoredGlobalIndexResult result : results) {
+ if (result.results().contains(rowId)) {
+ score += result.scoreGetter().score(rowId);
+ }
+ }
+ return score;
+ });
+ }
+
+ private static ScoredGlobalIndexResult andWithBonus(
+ ScoredGlobalIndexResult base, ScoredGlobalIndexResult bonus) {
+ RoaringNavigableMap64 rowIds = base.results();
+ return ScoredGlobalIndexResult.create(
+ rowIds,
+ rowId ->
+ base.scoreGetter().score(rowId)
+ + (bonus.results().contains(rowId)
+ ? bonus.scoreGetter().score(rowId)
+ : 0.0f));
+ }
+
+ private static ScoredGlobalIndexResult andNot(
+ ScoredGlobalIndexResult left, ScoredGlobalIndexResult right) {
+ RoaringNavigableMap64 rowIds =
+ RoaringNavigableMap64.or(new RoaringNavigableMap64(), left.results());
+ rowIds.andNot(right.results());
+ return ScoredGlobalIndexResult.create(rowIds, left.scoreGetter());
+ }
+
+ private static ScoredGlobalIndexResult boost(
+ ScoredGlobalIndexResult positive,
+ ScoredGlobalIndexResult negative,
+ float negativeBoost) {
+ RoaringNavigableMap64 rowIds = positive.results();
+ return ScoredGlobalIndexResult.create(
+ rowIds,
+ rowId ->
+ positive.scoreGetter().score(rowId)
+ * (negative.results().contains(rowId) ? negativeBoost : 1.0f));
+ }
}
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 9b3c7fba2c..123a8b9980 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
@@ -19,6 +19,8 @@
package org.apache.paimon.table.source;
import org.apache.paimon.Snapshot;
+import org.apache.paimon.globalindex.GlobalIndexerFactory;
+import org.apache.paimon.globalindex.GlobalIndexerFactoryUtils;
import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileHandler;
import org.apache.paimon.index.IndexFileMeta;
@@ -31,10 +33,13 @@ import org.apache.paimon.utils.Filter;
import org.apache.paimon.utils.Range;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Set;
import java.util.stream.Collectors;
import static org.apache.paimon.utils.Preconditions.checkNotNull;
@@ -44,7 +49,7 @@ public class FullTextScanImpl implements FullTextScan {
private final FileStoreTable table;
private final PartitionPredicate partitionFilter;
- private final DataField textColumn;
+ private final List<DataField> textColumns;
public FullTextScanImpl(FileStoreTable table, DataField textColumn) {
this(table, null, textColumn);
@@ -52,14 +57,29 @@ public class FullTextScanImpl implements FullTextScan {
public FullTextScanImpl(
FileStoreTable table, PartitionPredicate partitionFilter, DataField textColumn) {
+ this(table, partitionFilter, Collections.singletonList(textColumn));
+ }
+
+ public FullTextScanImpl(
+ FileStoreTable table, PartitionPredicate partitionFilter, List<DataField> textColumns) {
this.table = table;
this.partitionFilter = partitionFilter;
- this.textColumn = textColumn;
+ this.textColumns = textColumns;
}
@Override
public Plan scan() {
- Objects.requireNonNull(textColumn, "Text column must be set");
+ Objects.requireNonNull(textColumns, "Text columns must be set");
+ if (textColumns.isEmpty()) {
+ return Collections::emptyList;
+ }
+
+ Set<Integer> textColumnIds = new HashSet<>();
+ Map<Integer, String> idToColumn = new HashMap<>();
+ for (DataField textColumn : textColumns) {
+ textColumnIds.add(textColumn.id());
+ idToColumn.put(textColumn.id(), textColumn.name());
+ }
Snapshot snapshot = TimeTravelUtil.tryTravelOrLatest(table);
IndexFileHandler indexFileHandler = table.store().newIndexFileHandler();
@@ -72,7 +92,8 @@ public class FullTextScanImpl implements FullTextScan {
if (globalIndex == null) {
return false;
}
- return textColumn.id() == globalIndex.indexFieldId();
+ return textColumnIds.contains(globalIndex.indexFieldId())
+ && supportsFullTextSearch(entry.indexFile().indexType());
};
List<IndexFileMeta> allIndexFiles =
@@ -80,20 +101,41 @@ public class FullTextScanImpl implements FullTextScan {
.map(IndexManifestEntry::indexFile)
.collect(Collectors.toList());
- // Group full-text index files by (rowRangeStart, rowRangeEnd)
- Map<Range, List<IndexFileMeta>> byRange = new HashMap<>();
+ // Group full-text index files by column and row range.
+ Map<String, Map<Range, List<IndexFileMeta>>> byColumnAndRange = new HashMap<>();
for (IndexFileMeta indexFile : allIndexFiles) {
GlobalIndexMeta meta = checkNotNull(indexFile.globalIndexMeta());
+ String columnName = checkNotNull(idToColumn.get(meta.indexFieldId()));
Range range = new Range(meta.rowRangeStart(), meta.rowRangeEnd());
- byRange.computeIfAbsent(range, k -> new ArrayList<>()).add(indexFile);
+ byColumnAndRange
+ .computeIfAbsent(columnName, k -> new HashMap<>())
+ .computeIfAbsent(range, k -> new ArrayList<>())
+ .add(indexFile);
}
List<FullTextSearchSplit> splits = new ArrayList<>();
- for (Map.Entry<Range, List<IndexFileMeta>> entry : byRange.entrySet()) {
- Range range = entry.getKey();
- splits.add(new FullTextSearchSplit(range.from, range.to, entry.getValue()));
+ for (Map.Entry<String, Map<Range, List<IndexFileMeta>>> columnEntry :
+ byColumnAndRange.entrySet()) {
+ String columnName = columnEntry.getKey();
+ for (Map.Entry<Range, List<IndexFileMeta>> rangeEntry :
+ columnEntry.getValue().entrySet()) {
+ Range range = rangeEntry.getKey();
+ splits.add(
+ new FullTextSearchSplit(
+ columnName, range.from, range.to, rangeEntry.getValue()));
+ }
}
return () -> splits;
}
+
+ private static boolean supportsFullTextSearch(String indexType) {
+ GlobalIndexerFactory factory;
+ try {
+ factory = GlobalIndexerFactoryUtils.load(indexType);
+ } catch (RuntimeException e) {
+ return false;
+ }
+ return factory.supportsFullTextSearch();
+ }
}
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 d7779ca0f5..260316838b 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
@@ -20,6 +20,7 @@ package org.apache.paimon.table.source;
import org.apache.paimon.globalindex.GlobalIndexResult;
import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.predicate.FullTextQuery;
import java.io.Serializable;
@@ -35,14 +36,8 @@ public interface FullTextSearchBuilder extends Serializable {
/** The top k results to return. */
FullTextSearchBuilder withLimit(int limit);
- /** The text column to search. */
- FullTextSearchBuilder withTextColumn(String name);
-
- /** The query text to search. */
- FullTextSearchBuilder withQueryText(String queryText);
-
- /** The default query operator. Supported values are 'or' and 'and'. */
- FullTextSearchBuilder withQueryOperator(String queryOperator);
+ /** The structured full-text query to search. */
+ FullTextSearchBuilder withQuery(FullTextQuery query);
/** Create full-text scan to scan index files. */
FullTextScan newFullTextScan();
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 dea7e8f5d1..f4644005aa 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
@@ -19,10 +19,14 @@
package org.apache.paimon.table.source;
import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.predicate.FullTextQuery;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.InnerTable;
import org.apache.paimon.types.DataField;
+import java.util.ArrayList;
+import java.util.List;
+
import static org.apache.paimon.utils.Preconditions.checkArgument;
import static org.apache.paimon.utils.Preconditions.checkNotNull;
@@ -34,9 +38,7 @@ public class FullTextSearchBuilderImpl implements FullTextSearchBuilder {
private final FileStoreTable table;
private int limit;
- private DataField textColumn;
- private String queryText;
- private String queryOperator = "or";
+ private FullTextQuery query;
private PartitionPredicate partitionFilter;
public FullTextSearchBuilderImpl(InnerTable table) {
@@ -56,34 +58,30 @@ public class FullTextSearchBuilderImpl implements FullTextSearchBuilder {
}
@Override
- public FullTextSearchBuilder withTextColumn(String name) {
- this.textColumn = table.rowType().getField(name);
- return this;
- }
-
- @Override
- public FullTextSearchBuilder withQueryText(String queryText) {
- this.queryText = queryText;
- return this;
- }
-
- @Override
- public FullTextSearchBuilder withQueryOperator(String queryOperator) {
- this.queryOperator = queryOperator;
+ public FullTextSearchBuilder withQuery(FullTextQuery query) {
+ this.query = query;
return this;
}
@Override
public FullTextScan newFullTextScan() {
- checkNotNull(textColumn, "Text column must be set via withTextColumn()");
- return new FullTextScanImpl(table, partitionFilter, textColumn);
+ return new FullTextScanImpl(table, partitionFilter, textColumns());
}
@Override
public FullTextRead newFullTextRead() {
checkArgument(limit > 0, "Limit must be positive, set via withLimit()");
- checkNotNull(textColumn, "Text column must be set via withTextColumn()");
- checkNotNull(queryText, "Query text must be set via withQueryText()");
- return new FullTextReadImpl(table, limit, textColumn, queryText, queryOperator);
+ return new FullTextReadImpl(table, limit, textColumns(), query);
+ }
+
+ private List<DataField> textColumns() {
+ checkNotNull(query, "Query must be set via withQuery()");
+ List<DataField> textColumns = new ArrayList<>();
+ for (String columnName : query.columns()) {
+ DataField textColumn = table.rowType().getField(columnName);
+ checkNotNull(textColumn, "Text column '%s' does not exist.", columnName);
+ textColumns.add(textColumn);
+ }
+ return textColumns;
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchSplit.java b/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchSplit.java
index d230482186..924b95057d 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchSplit.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchSplit.java
@@ -21,6 +21,7 @@ package org.apache.paimon.table.source;
import org.apache.paimon.index.IndexFileMeta;
import java.io.Serializable;
+import java.util.Collections;
import java.util.List;
import java.util.Objects;
@@ -29,15 +30,29 @@ public class FullTextSearchSplit implements Serializable {
private static final long serialVersionUID = 1L;
+ private final String columnName;
private final long rowRangeStart;
private final long rowRangeEnd;
private final List<IndexFileMeta> fullTextIndexFiles;
public FullTextSearchSplit(
long rowRangeStart, long rowRangeEnd, List<IndexFileMeta> fullTextIndexFiles) {
+ this(null, rowRangeStart, rowRangeEnd, fullTextIndexFiles);
+ }
+
+ public FullTextSearchSplit(
+ String columnName,
+ long rowRangeStart,
+ long rowRangeEnd,
+ List<IndexFileMeta> fullTextIndexFiles) {
+ this.columnName = columnName;
this.rowRangeStart = rowRangeStart;
this.rowRangeEnd = rowRangeEnd;
- this.fullTextIndexFiles = fullTextIndexFiles;
+ this.fullTextIndexFiles = Collections.unmodifiableList(fullTextIndexFiles);
+ }
+
+ public String columnName() {
+ return columnName;
}
public long rowRangeStart() {
@@ -58,20 +73,24 @@ public class FullTextSearchSplit implements Serializable {
return false;
}
FullTextSearchSplit that = (FullTextSearchSplit) o;
- return rowRangeStart == that.rowRangeStart
+ return Objects.equals(columnName, that.columnName)
+ && rowRangeStart == that.rowRangeStart
&& rowRangeEnd == that.rowRangeEnd
&& Objects.equals(fullTextIndexFiles, that.fullTextIndexFiles);
}
@Override
public int hashCode() {
- return Objects.hash(rowRangeStart, rowRangeEnd, fullTextIndexFiles);
+ return Objects.hash(columnName, rowRangeStart, rowRangeEnd, fullTextIndexFiles);
}
@Override
public String toString() {
return "FullTextSearchSplit{"
- + "rowRangeStart="
+ + "columnName='"
+ + columnName
+ + '\''
+ + ", rowRangeStart="
+ rowRangeStart
+ ", rowRangeEnd="
+ rowRangeEnd
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
index 4fb6a644ca..01304e1315 100644
--- 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
@@ -63,22 +63,14 @@ public interface HybridSearchBuilder extends Serializable {
}
/** 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);
+ default HybridSearchBuilder addFullTextRoute(String queryJson, int limit, float weight) {
+ return addFullTextRoute(queryJson, 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));
+ String queryJson, int limit, float weight, Map<String, String> options) {
+ return addRoute(HybridSearchRoute.fullText(queryJson, limit, weight, options));
}
/** The final top k ranked results to return. */
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
index 1aa322815c..7e5e328a37 100644
--- 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
@@ -198,9 +198,7 @@ public class HybridSearchBuilderImpl implements HybridSearchBuilder {
protected FullTextSearchBuilder newFullTextSearchBuilder(HybridSearchRoute route) {
FullTextSearchBuilder fullTextSearchBuilder =
table.newFullTextSearchBuilder()
- .withQueryText(route.queryText())
- .withQueryOperator(route.queryOperator())
- .withTextColumn(route.fieldName())
+ .withQuery(route.fullTextQuery())
.withLimit(route.limit());
if (partitionFilter != null) {
fullTextSearchBuilder.withPartitionFilter(partitionFilter);
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 00eccd5dc9..c9178d3dbf 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
@@ -19,6 +19,7 @@
package org.apache.paimon.table.source;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
@@ -28,11 +29,13 @@ import org.apache.paimon.globalindex.GlobalIndexResult;
import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
import org.apache.paimon.globalindex.ResultEntry;
import org.apache.paimon.globalindex.ScoredGlobalIndexResult;
+import org.apache.paimon.globalindex.btree.BTreeGlobalIndexerFactory;
import org.apache.paimon.globalindex.testfulltext.TestFullTextGlobalIndexerFactory;
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.FullTextQuery;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.reader.RecordReader;
@@ -94,9 +97,8 @@ public class FullTextSearchBuilderTest extends TableTestBase {
// Query "Paimon" - should match rows 0, 1, 3
GlobalIndexResult result =
table.newFullTextSearchBuilder()
- .withQueryText("Paimon")
+ .withQuery(FullTextQuery.match("Paimon", TEXT_FIELD_NAME))
.withLimit(3)
- .withTextColumn(TEXT_FIELD_NAME)
.executeLocal();
assertThat(result).isInstanceOf(ScoredGlobalIndexResult.class);
@@ -132,7 +134,10 @@ public class FullTextSearchBuilderTest extends TableTestBase {
ScoredGlobalIndexResult result =
table.newHybridSearchBuilder()
- .addFullTextRoute(TEXT_FIELD_NAME, "Paimon", 3, 1.0f)
+ .addFullTextRoute(
+ "{\"match\":{\"column\":\"content\",\"terms\":\"Paimon\"}}",
+ 3,
+ 1.0f)
.withLimit(3)
.executeLocal();
@@ -150,7 +155,11 @@ public class FullTextSearchBuilderTest extends TableTestBase {
assertThatThrownBy(
() ->
table.newHybridSearchBuilder()
- .addFullTextRoute(TEXT_FIELD_NAME, "Paimon", 3, 1.0f)
+ .addFullTextRoute(
+ "{\"match\":{\"column\":\"content\","
+ + "\"terms\":\"Paimon\"}}",
+ 3,
+ 1.0f)
.withFilter(idFilter)
.withLimit(3)
.routeBuilders())
@@ -176,9 +185,8 @@ public class FullTextSearchBuilderTest extends TableTestBase {
// Query "Paimon search" - row 2 matches both terms, rows 1 matches both, row 0 matches one
GlobalIndexResult result =
table.newFullTextSearchBuilder()
- .withQueryText("Paimon search")
+ .withQuery(FullTextQuery.match("Paimon search", TEXT_FIELD_NAME))
.withLimit(2)
- .withTextColumn(TEXT_FIELD_NAME)
.executeLocal();
assertThat(result).isInstanceOf(ScoredGlobalIndexResult.class);
@@ -196,6 +204,169 @@ public class FullTextSearchBuilderTest extends TableTestBase {
assertThat(ids).contains(1, 2);
}
+ @Test
+ public void testStructuredFullTextSearchPhraseAndBooleanQuery() throws Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+
+ String[] documents = {
+ "Apache Paimon lake format",
+ "Paimon full-text search support",
+ "full-text search in Apache Paimon",
+ "Vector search capabilities",
+ };
+
+ writeDocuments(table, documents);
+ buildAndCommitIndex(table, documents);
+
+ GlobalIndexResult phraseResult =
+ table.newFullTextSearchBuilder()
+ .withQuery(FullTextQuery.phrase("full-text search", TEXT_FIELD_NAME))
+ .withLimit(10)
+ .executeLocal();
+
+ assertThat(readIds(table, phraseResult)).containsExactlyInAnyOrder(1, 2);
+
+ FullTextQuery booleanQuery =
+ new FullTextQuery.BooleanQuery(
+ java.util.Collections.emptyList(),
+ java.util.Arrays.asList(
+ FullTextQuery.match("Paimon", TEXT_FIELD_NAME),
+ FullTextQuery.match("search", TEXT_FIELD_NAME)),
+ java.util.Collections.singletonList(
+ FullTextQuery.match("Vector", TEXT_FIELD_NAME)));
+
+ GlobalIndexResult booleanResult =
+ table.newFullTextSearchBuilder()
+ .withQuery(booleanQuery)
+ .withLimit(10)
+ .executeLocal();
+
+ assertThat(readIds(table, booleanResult)).containsExactlyInAnyOrder(1, 2);
+ }
+
+ @Test
+ public void testMultiMatchFullTextSearchAcrossColumns() throws Exception {
+ FileStoreTable table = createMultiTextTable();
+
+ writeMultiTextDocuments(
+ table,
+ new String[][] {
+ {"paimon overview", "lake format"},
+ {"vector overview", "paimon search"},
+ {"engine notes", "streaming"},
+ {"paimon vector", "paimon lake"}
+ });
+ buildAndCommitIndexForColumn(
+ table,
+ "title",
+ new String[] {
+ "paimon overview", "vector overview", "engine notes", "paimon vector"
+ });
+ buildAndCommitIndexForColumn(
+ table,
+ "body",
+ new String[] {"lake format", "paimon search", "streaming", "paimon lake"});
+
+ GlobalIndexResult result =
+ table.newFullTextSearchBuilder()
+ .withQuery(
+ FullTextQuery.multiMatch("paimon", Arrays.asList("title", "body")))
+ .withLimit(10)
+ .executeLocal();
+
+ assertThat(readIds(table, result)).containsExactlyInAnyOrder(0, 1, 3);
+ }
+
+ @Test
+ public void testMultiMatchFullTextSearchAddsColumnScores() throws Exception {
+ FileStoreTable table = createMultiTextTable();
+
+ writeMultiTextDocuments(
+ table,
+ new String[][] {
+ {"paimon", "paimon"},
+ {"paimon", "engine"},
+ {"engine", "paimon"},
+ });
+ buildAndCommitIndexForColumn(table, "title", new String[] {"paimon", "paimon", "engine"});
+ buildAndCommitIndexForColumn(table, "body", new String[] {"paimon", "engine", "paimon"});
+
+ ScoredGlobalIndexResult result =
+ (ScoredGlobalIndexResult)
+ table.newFullTextSearchBuilder()
+ .withQuery(
+ FullTextQuery.multiMatch(
+ "paimon", Arrays.asList("title", "body")))
+ .withLimit(10)
+ .executeLocal();
+
+ assertThat(result.scoreGetter().score(0)).isEqualTo(2.0f);
+ assertThat(result.scoreGetter().score(1)).isEqualTo(1.0f);
+ assertThat(result.scoreGetter().score(2)).isEqualTo(1.0f);
+ }
+
+ @Test
+ public void testBooleanFullTextSearchAcrossColumns() throws Exception {
+ FileStoreTable table = createMultiTextTable();
+
+ writeMultiTextDocuments(
+ table,
+ new String[][] {
+ {"paimon overview", "lake format"},
+ {"paimon overview", "vector search"},
+ {"spark overview", "lake format"},
+ {"paimon internals", "lake vector"}
+ });
+ buildAndCommitIndexForColumn(
+ table,
+ "title",
+ new String[] {
+ "paimon overview", "paimon overview", "spark overview", "paimon internals"
+ });
+ buildAndCommitIndexForColumn(
+ table,
+ "body",
+ new String[] {"lake format", "vector search", "lake format", "lake vector"});
+
+ FullTextQuery query =
+ new FullTextQuery.BooleanQuery(
+ Collections.emptyList(),
+ Arrays.asList(
+ FullTextQuery.match("paimon", "title"),
+ FullTextQuery.match("lake", "body")),
+ Collections.singletonList(FullTextQuery.match("vector", "body")));
+
+ GlobalIndexResult result =
+ table.newFullTextSearchBuilder().withQuery(query).withLimit(10).executeLocal();
+
+ assertThat(readIds(table, result)).containsExactlyInAnyOrder(0);
+ }
+
+ @Test
+ public void testCompoundFullTextSearchUsesFullLeafCandidatesBeforeFinalTopK() throws Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+
+ String[] documents = {"paimon vector", "paimon", "paimon", "paimon", "paimon", "paimon"};
+
+ writeDocuments(table, documents);
+ buildAndCommitIndex(table, documents);
+
+ ScoredGlobalIndexResult result =
+ (ScoredGlobalIndexResult)
+ table.newFullTextSearchBuilder()
+ .withQuery(
+ FullTextQuery.boost(
+ FullTextQuery.match("paimon", TEXT_FIELD_NAME),
+ FullTextQuery.match("vector", TEXT_FIELD_NAME),
+ 0.1f))
+ .withLimit(3)
+ .executeLocal();
+
+ assertThat(readIds(table, result)).doesNotContain(0);
+ }
+
@Test
public void testFullTextSearchEmptyResult() throws Exception {
createTableDefault();
@@ -207,9 +378,8 @@ public class FullTextSearchBuilderTest extends TableTestBase {
GlobalIndexResult result =
table.newFullTextSearchBuilder()
- .withQueryText("nonexistent")
+ .withQuery(FullTextQuery.match("nonexistent", TEXT_FIELD_NAME))
.withLimit(1)
- .withTextColumn(TEXT_FIELD_NAME)
.executeLocal();
assertThat(result.results().isEmpty()).isTrue();
@@ -231,9 +401,8 @@ public class FullTextSearchBuilderTest extends TableTestBase {
// Search with limit=5
GlobalIndexResult result =
table.newFullTextSearchBuilder()
- .withQueryText("keyword")
+ .withQuery(FullTextQuery.match("keyword", TEXT_FIELD_NAME))
.withLimit(5)
- .withTextColumn(TEXT_FIELD_NAME)
.executeLocal();
ReadBuilder readBuilder = table.newReadBuilder();
@@ -268,9 +437,8 @@ public class FullTextSearchBuilderTest extends TableTestBase {
// Query "Paimon" - results should span across both index files
GlobalIndexResult result =
table.newFullTextSearchBuilder()
- .withQueryText("Paimon")
+ .withQuery(FullTextQuery.match("Paimon", TEXT_FIELD_NAME))
.withLimit(4)
- .withTextColumn(TEXT_FIELD_NAME)
.executeLocal();
assertThat(result).isInstanceOf(ScoredGlobalIndexResult.class);
@@ -289,6 +457,26 @@ public class FullTextSearchBuilderTest extends TableTestBase {
assertThat(ids).containsAnyOf(3, 5);
}
+ @Test
+ public void testFullTextSearchIgnoresOtherIndexTypesOnSameColumn() throws Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+
+ String[] documents = {"Apache Paimon lake format", "vector search"};
+
+ writeDocuments(table, documents);
+ buildAndCommitIndex(table, documents);
+ buildAndCommitBTreeIndex(table, documents);
+
+ GlobalIndexResult result =
+ table.newFullTextSearchBuilder()
+ .withQuery(FullTextQuery.match("Paimon", TEXT_FIELD_NAME))
+ .withLimit(10)
+ .executeLocal();
+
+ assertThat(readIds(table, result)).containsExactly(0);
+ }
+
@Test
public void testFullTextSearchNoMatchingDocuments() throws Exception {
createTableDefault();
@@ -304,9 +492,8 @@ public class FullTextSearchBuilderTest extends TableTestBase {
// Query a term that doesn't exist in any document
GlobalIndexResult result =
table.newFullTextSearchBuilder()
- .withQueryText("nonexistent")
+ .withQuery(FullTextQuery.match("nonexistent", TEXT_FIELD_NAME))
.withLimit(3)
- .withTextColumn(TEXT_FIELD_NAME)
.executeLocal();
assertThat(result.results().isEmpty()).isTrue();
@@ -327,9 +514,8 @@ public class FullTextSearchBuilderTest extends TableTestBase {
// Query lowercase "paimon" should match all three (case-insensitive)
GlobalIndexResult result =
table.newFullTextSearchBuilder()
- .withQueryText("paimon")
+ .withQuery(FullTextQuery.match("paimon", TEXT_FIELD_NAME))
.withLimit(3)
- .withTextColumn(TEXT_FIELD_NAME)
.executeLocal();
assertThat(result).isInstanceOf(ScoredGlobalIndexResult.class);
@@ -360,9 +546,8 @@ public class FullTextSearchBuilderTest extends TableTestBase {
FullTextSearchBuilder searchBuilder =
table.newFullTextSearchBuilder()
- .withQueryText("Paimon")
- .withLimit(2)
- .withTextColumn(TEXT_FIELD_NAME);
+ .withQuery(FullTextQuery.match("Paimon", TEXT_FIELD_NAME))
+ .withLimit(2);
FullTextScan.Plan plan = searchBuilder.newFullTextScan().scan();
assertThat(plan.splits()).isEmpty();
@@ -431,6 +616,126 @@ public class FullTextSearchBuilderTest extends TableTestBase {
}
}
+ private FileStoreTable createMultiTextTable() throws Exception {
+ Identifier identifier = identifier("MultiTextTable");
+ Schema schema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("title", DataTypes.STRING())
+ .column("body", DataTypes.STRING())
+ .option(CoreOptions.BUCKET.key(), "-1")
+ .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true")
+ .build();
+ catalog.createTable(identifier, schema, true);
+ return getTable(identifier);
+ }
+
+ private void writeMultiTextDocuments(FileStoreTable table, String[][] documents)
+ throws Exception {
+ BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = writeBuilder.newWrite();
+ BatchTableCommit commit = writeBuilder.newCommit()) {
+ for (int i = 0; i < documents.length; i++) {
+ write.write(
+ GenericRow.of(
+ i,
+ BinaryString.fromString(documents[i][0]),
+ BinaryString.fromString(documents[i][1])));
+ }
+ commit.commit(write.prepareCommit());
+ }
+ }
+
+ private void buildAndCommitIndexForColumn(
+ FileStoreTable table, String columnName, String[] documents) throws Exception {
+ Options options = table.coreOptions().toConfiguration();
+ DataField textField = table.rowType().getField(columnName);
+
+ GlobalIndexSingleColumnWriter writer =
+ (GlobalIndexSingleColumnWriter)
+ GlobalIndexBuilderUtils.createIndexWriter(
+ table,
+ TestFullTextGlobalIndexerFactory.IDENTIFIER,
+ textField,
+ options);
+ for (int i = 0; i < documents.length; i++) {
+ writer.write(documents[i], i);
+ }
+ List<ResultEntry> entries = writer.finish();
+
+ Range rowRange = new Range(0, documents.length - 1);
+ List<IndexFileMeta> indexFiles =
+ GlobalIndexBuilderUtils.toIndexFileMetas(
+ table.fileIO(),
+ table.store().pathFactory().globalIndexFileFactory(),
+ table.coreOptions(),
+ rowRange,
+ textField.id(),
+ TestFullTextGlobalIndexerFactory.IDENTIFIER,
+ entries);
+
+ DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles);
+ CommitMessage message =
+ new CommitMessageImpl(
+ BinaryRow.EMPTY_ROW,
+ 0,
+ null,
+ dataIncrement,
+ CompactIncrement.emptyIncrement());
+ try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) {
+ commit.commit(Collections.singletonList(message));
+ }
+ }
+
+ private void buildAndCommitBTreeIndex(FileStoreTable table, String[] documents)
+ throws Exception {
+ Options options = table.coreOptions().toConfiguration();
+ DataField textField = table.rowType().getField(TEXT_FIELD_NAME);
+
+ GlobalIndexSingleColumnWriter writer =
+ (GlobalIndexSingleColumnWriter)
+ GlobalIndexBuilderUtils.createIndexWriter(
+ table, BTreeGlobalIndexerFactory.IDENTIFIER, textField, options);
+ for (int i = 0; i < documents.length; i++) {
+ writer.write(BinaryString.fromString(documents[i]), i);
+ }
+ List<ResultEntry> entries = writer.finish();
+
+ Range rowRange = new Range(0, documents.length - 1);
+ List<IndexFileMeta> indexFiles =
+ GlobalIndexBuilderUtils.toIndexFileMetas(
+ table.fileIO(),
+ table.store().pathFactory().globalIndexFileFactory(),
+ table.coreOptions(),
+ rowRange,
+ textField.id(),
+ BTreeGlobalIndexerFactory.IDENTIFIER,
+ entries);
+
+ DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles);
+ CommitMessage message =
+ new CommitMessageImpl(
+ BinaryRow.EMPTY_ROW,
+ 0,
+ null,
+ dataIncrement,
+ CompactIncrement.emptyIncrement());
+ try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) {
+ commit.commit(Collections.singletonList(message));
+ }
+ }
+
+ private List<Integer> readIds(FileStoreTable table, GlobalIndexResult result) throws Exception {
+ ReadBuilder readBuilder = table.newReadBuilder();
+ TableScan.Plan plan = readBuilder.newScan().withGlobalIndexResult(result).plan();
+ List<Integer> ids = new ArrayList<>();
+ try (RecordReader<InternalRow> reader = readBuilder.newRead().createReader(plan)) {
+ reader.forEachRemaining(row -> ids.add(row.getInt(0)));
+ }
+ return ids;
+ }
+
private void buildAndCommitMultipleIndexFiles(FileStoreTable table, String[] documents)
throws Exception {
Options options = table.coreOptions().toConfiguration();
diff --git a/paimon-python/pypaimon/cli/cli_table.py b/paimon-python/pypaimon/cli/cli_table.py
index ba8446fcf9..daee51db8c 100644
--- a/paimon-python/pypaimon/cli/cli_table.py
+++ b/paimon-python/pypaimon/cli/cli_table.py
@@ -271,15 +271,13 @@ def cmd_table_full_text_search(args):
print(f"Error: Failed to get table '{table_identifier}': {e}", file=sys.stderr)
sys.exit(1)
- # Build full-text search
- text_column = args.column
- query_text = args.query
limit = args.limit
try:
+ from pypaimon.globalindex.full_text_query import FullTextQuery
+
builder = table.new_full_text_search_builder()
- builder.with_text_column(text_column)
- builder.with_query_text(query_text)
+ builder.with_query(FullTextQuery.from_json(args.query))
builder.with_limit(limit)
result = builder.execute_local()
except Exception as e:
@@ -1071,15 +1069,10 @@ def add_table_subcommands(table_parser):
'table',
help='Table identifier in format: database.table'
)
- fts_parser.add_argument(
- '--column', '-c',
- required=True,
- help='Text column to search on'
- )
fts_parser.add_argument(
'--query', '-q',
required=True,
- help='Query text to search for'
+ help='LanceDB-style full-text query JSON to search for'
)
fts_parser.add_argument(
'--limit', '-l',
diff --git a/paimon-python/pypaimon/globalindex/__init__.py b/paimon-python/pypaimon/globalindex/__init__.py
index 8cff38b24d..35c043f577 100644
--- a/paimon-python/pypaimon/globalindex/__init__.py
+++ b/paimon-python/pypaimon/globalindex/__init__.py
@@ -18,6 +18,17 @@
from pypaimon.globalindex.global_index_result import GlobalIndexResult
from pypaimon.globalindex.global_index_reader import GlobalIndexReader, FieldRef
from pypaimon.globalindex.vector_search import VectorSearch
+from pypaimon.globalindex.full_text_query import (
+ BooleanQuery,
+ BoostQuery,
+ FullTextOperator,
+ FullTextQuery,
+ FullTextQueryType,
+ MatchQuery,
+ MultiMatchQuery,
+ Occur,
+ PhraseQuery,
+)
from pypaimon.globalindex.full_text_search import FullTextSearch
from pypaimon.globalindex.vector_search_result import (
ScoredGlobalIndexResult,
@@ -40,7 +51,16 @@ __all__ = [
'GlobalIndexReader',
'FieldRef',
'VectorSearch',
+ 'BooleanQuery',
+ 'BoostQuery',
+ 'FullTextOperator',
+ 'FullTextQuery',
+ 'FullTextQueryType',
'FullTextSearch',
+ 'MatchQuery',
+ 'MultiMatchQuery',
+ 'Occur',
+ 'PhraseQuery',
'ScoredGlobalIndexResult',
'DictBasedScoredIndexResult',
'ScoreGetter',
diff --git a/paimon-python/pypaimon/globalindex/full_text_query.py b/paimon-python/pypaimon/globalindex/full_text_query.py
new file mode 100644
index 0000000000..23bb074d60
--- /dev/null
+++ b/paimon-python/pypaimon/globalindex/full_text_query.py
@@ -0,0 +1,512 @@
+# 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.
+
+"""Structured full-text query DSL aligned with LanceDB FTS query JSON."""
+
+import json
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from enum import Enum
+from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple, Union
+
+
+class FullTextQueryType(str, Enum):
+ MATCH = "match"
+ MATCH_PHRASE = "match_phrase"
+ BOOST = "boost"
+ MULTI_MATCH = "multi_match"
+ BOOLEAN = "boolean"
+
+
+class FullTextOperator(str, Enum):
+ AND = "AND"
+ OR = "OR"
+
+ @staticmethod
+ def normalize(value: Optional[Union[str, 'FullTextOperator']]) -> 'FullTextOperator':
+ if value is None:
+ return FullTextOperator.OR
+ if isinstance(value, FullTextOperator):
+ return value
+ normalized = str(value).strip().upper()
+ if normalized == "AND":
+ return FullTextOperator.AND
+ if normalized == "OR":
+ return FullTextOperator.OR
+ raise ValueError("Full-text query operator must be 'or' or 'and', got: %s" % value)
+
+ def json_value(self) -> str:
+ return "And" if self == FullTextOperator.AND else "Or"
+
+
+class Occur(str, Enum):
+ SHOULD = "SHOULD"
+ MUST = "MUST"
+ MUST_NOT = "MUST_NOT"
+
+ @staticmethod
+ def normalize(value: Union[str, 'Occur']) -> 'Occur':
+ if isinstance(value, Occur):
+ return value
+ normalized = str(value).strip().upper()
+ if normalized == "SHOULD":
+ return Occur.SHOULD
+ if normalized == "MUST":
+ return Occur.MUST
+ if normalized in ("MUST_NOT", "MUSTNOT"):
+ return Occur.MUST_NOT
+ raise ValueError("Unknown boolean query occur: %s" % value)
+
+
+class FullTextQuery(ABC):
+ """Base class for LanceDB-style full-text query objects."""
+
+ @abstractmethod
+ def query_type(self) -> FullTextQueryType:
+ pass
+
+ @abstractmethod
+ def query_text(self) -> str:
+ pass
+
+ @abstractmethod
+ def referenced_columns(self) -> List[str]:
+ pass
+
+ @abstractmethod
+ def to_dict(self) -> Dict[str, Any]:
+ pass
+
+ def to_json(self) -> str:
+ return json.dumps(self.to_dict(), separators=(",", ":"))
+
+ def single_column(self) -> str:
+ columns = self.referenced_columns()
+ if len(columns) != 1:
+ raise ValueError(
+ "Full-text query must reference exactly one column, but got: %s" % columns
+ )
+ return columns[0]
+
+ def __and__(self, other: 'FullTextQuery') -> 'BooleanQuery':
+ return BooleanQuery([(Occur.MUST, self), (Occur.MUST, other)])
+
+ def __or__(self, other: 'FullTextQuery') -> 'BooleanQuery':
+ return BooleanQuery([(Occur.SHOULD, self), (Occur.SHOULD, other)])
+
+ @staticmethod
+ def from_json(query_json: str) -> 'FullTextQuery':
+ try:
+ payload = json.loads(query_json)
+ except Exception as e:
+ raise ValueError("Failed to parse full-text query JSON: %s" % query_json) from e
+ return FullTextQuery.from_dict(payload)
+
+ @staticmethod
+ def from_dict(payload: Dict[str, Any]) -> 'FullTextQuery':
+ if not isinstance(payload, dict):
+ raise ValueError("Full-text query JSON must be an object.")
+ if "match" in payload:
+ body = _object_value(payload, "match")
+ return MatchQuery(
+ _text_value_any(body, ("terms", "query")),
+ _text_value(body, "column"),
+ boost=_float_value(body, "boost", 1.0),
+ fuzziness=_fuzziness_value(body),
+ max_expansions=_int_value_any(body, ("max_expansions", "maxExpansions"), 50),
+ operator=FullTextOperator.normalize(body.get("operator")),
+ prefix_length=_int_value_any(body, ("prefix_length", "prefixLength"), 0),
+ )
+ if "match_phrase" in payload or "phrase" in payload:
+ body = _object_value(payload, "match_phrase" if "match_phrase" in payload else "phrase")
+ return PhraseQuery(
+ _text_value_any(body, ("terms", "query")),
+ _text_value(body, "column"),
+ slop=_int_value(body, "slop", 0),
+ )
+ if "boost" in payload:
+ body = _object_value(payload, "boost")
+ return BoostQuery(
+ FullTextQuery.from_dict(_object_value(body, "positive")),
+ FullTextQuery.from_dict(_object_value(body, "negative")),
+ negative_boost=_float_value_any(body, ("negative_boost", "negativeBoost"), 0.5),
+ )
+ if "multi_match" in payload:
+ body = _object_value(payload, "multi_match")
+ return MultiMatchQuery(
+ _text_value(body, "query"),
+ _string_list(body.get("columns"), "columns"),
+ boosts=_optional_float_list(_field_value(body, ("boost", "boosts"))),
+ operator=FullTextOperator.normalize(body.get("operator")),
+ )
+ if "boolean" in payload:
+ body = _object_value(payload, "boolean")
+ queries = []
+ _add_boolean_queries(queries, Occur.SHOULD, body.get("should"))
+ _add_boolean_queries(queries, Occur.MUST, body.get("must"))
+ _add_boolean_queries(queries, Occur.MUST_NOT, body.get("must_not"))
+ if "queries" in body:
+ clauses = body.get("queries")
+ if not isinstance(clauses, list):
+ raise ValueError("Boolean query 'queries' must be an array.")
+ for clause in clauses:
+ if isinstance(clause, (list, tuple)) and len(clause) == 2:
+ occur = Occur.normalize(clause[0])
+ query = FullTextQuery.from_dict(clause[1])
+ elif isinstance(clause, dict):
+ occur = Occur.normalize(clause.get("occur"))
+ query = FullTextQuery.from_dict(_object_value(clause, "query"))
+ else:
+ raise ValueError("Invalid boolean query clause: %s" % (clause,))
+ queries.append((occur, query))
+ return BooleanQuery(queries)
+ raise ValueError("Unknown full-text query JSON: %s" % payload)
+
+
+@dataclass(frozen=True)
+class MatchQuery(FullTextQuery):
+ query: str
+ column: str
+ boost: float = 1.0
+ fuzziness: Optional[int] = 0
+ max_expansions: int = 50
+ operator: Union[str, FullTextOperator] = FullTextOperator.OR
+ prefix_length: int = 0
+
+ def __post_init__(self):
+ _check_terms(self.query)
+ _check_column(self.column)
+ object.__setattr__(self, "query", str(self.query))
+ object.__setattr__(self, "column", str(self.column))
+ _check_positive(self.boost, "Boost factor")
+ if self.fuzziness is not None:
+ if self.fuzziness < 0 or self.fuzziness > 2:
+ raise ValueError("fuzziness must be between 0 and 2, got: %s" % self.fuzziness)
+ if self.max_expansions <= 0:
+ raise ValueError("max_expansions must be positive, got: %s" % self.max_expansions)
+ if self.prefix_length < 0:
+ raise ValueError("prefix_length must be non-negative, got: %s" % self.prefix_length)
+ object.__setattr__(self, "operator", FullTextOperator.normalize(self.operator))
+
+ def query_type(self) -> FullTextQueryType:
+ return FullTextQueryType.MATCH
+
+ def query_text(self) -> str:
+ return self.query
+
+ def referenced_columns(self) -> List[str]:
+ return [self.column]
+
+ def to_dict(self) -> Dict[str, Any]:
+ return {
+ "match": {
+ "column": self.column,
+ "terms": self.query,
+ "boost": self.boost,
+ "fuzziness": self.fuzziness,
+ "max_expansions": self.max_expansions,
+ "operator": self.operator.json_value(),
+ "prefix_length": self.prefix_length,
+ }
+ }
+
+
+@dataclass(frozen=True)
+class PhraseQuery(FullTextQuery):
+ query: str
+ column: str
+ slop: int = 0
+
+ def __post_init__(self):
+ _check_terms(self.query)
+ _check_column(self.column)
+ object.__setattr__(self, "query", str(self.query))
+ object.__setattr__(self, "column", str(self.column))
+ if self.slop < 0:
+ raise ValueError("slop must be non-negative, got: %s" % self.slop)
+
+ def query_type(self) -> FullTextQueryType:
+ return FullTextQueryType.MATCH_PHRASE
+
+ def query_text(self) -> str:
+ return self.query
+
+ def referenced_columns(self) -> List[str]:
+ return [self.column]
+
+ def to_dict(self) -> Dict[str, Any]:
+ return {
+ "match_phrase": {
+ "column": self.column,
+ "terms": self.query,
+ "slop": self.slop,
+ }
+ }
+
+
+@dataclass(frozen=True)
+class BoostQuery(FullTextQuery):
+ positive: FullTextQuery
+ negative: FullTextQuery
+ negative_boost: float = 0.5
+
+ def __post_init__(self):
+ _check_positive(self.negative_boost, "Negative boost")
+
+ def query_type(self) -> FullTextQueryType:
+ return FullTextQueryType.BOOST
+
+ def query_text(self) -> str:
+ return self.positive.query_text()
+
+ def referenced_columns(self) -> List[str]:
+ return _distinct(
+ self.positive.referenced_columns() + self.negative.referenced_columns()
+ )
+
+ def to_dict(self) -> Dict[str, Any]:
+ return {
+ "boost": {
+ "positive": self.positive.to_dict(),
+ "negative": self.negative.to_dict(),
+ "negative_boost": self.negative_boost,
+ }
+ }
+
+
+@dataclass(frozen=True)
+class MultiMatchQuery(FullTextQuery):
+ query: str
+ columns: List[str]
+ boosts: Optional[List[float]] = None
+ operator: Union[str, FullTextOperator] = FullTextOperator.OR
+
+ def __init__(
+ self,
+ query: str,
+ columns: Sequence[str],
+ boosts: Optional[Sequence[float]] = None,
+ operator: Union[str, FullTextOperator] = FullTextOperator.OR,
+ ):
+ object.__setattr__(self, "query", query)
+ object.__setattr__(self, "columns", list(columns) if columns is not None else None)
+ object.__setattr__(
+ self, "boosts", list(boosts) if boosts is not None else None
+ )
+ object.__setattr__(self, "operator", operator)
+ self.__post_init__()
+
+ def __post_init__(self):
+ _check_terms(self.query)
+ if not self.columns:
+ raise ValueError("columns cannot be None or empty")
+ for column in self.columns:
+ _check_column(column)
+ object.__setattr__(self, "query", str(self.query))
+ object.__setattr__(self, "columns", [str(column) for column in self.columns])
+ boosts = self.boosts
+ if boosts is None:
+ boosts = [1.0] * len(self.columns)
+ if len(boosts) != len(self.columns):
+ raise ValueError("The number of boosts must match the number of columns.")
+ for boost in boosts:
+ _check_positive(boost, "Boost")
+ object.__setattr__(self, "boosts", list(boosts))
+ object.__setattr__(self, "operator", FullTextOperator.normalize(self.operator))
+
+ def query_type(self) -> FullTextQueryType:
+ return FullTextQueryType.MULTI_MATCH
+
+ def query_text(self) -> str:
+ return self.query
+
+ def referenced_columns(self) -> List[str]:
+ return list(self.columns)
+
+ def to_dict(self) -> Dict[str, Any]:
+ return {
+ "multi_match": {
+ "query": self.query,
+ "columns": list(self.columns),
+ "boost": list(self.boosts),
+ "operator": self.operator.json_value(),
+ }
+ }
+
+
+@dataclass(frozen=True)
+class BooleanQuery(FullTextQuery):
+ queries: List[Tuple[Occur, FullTextQuery]]
+
+ def __init__(self, queries: Sequence[Tuple[Union[str, Occur], FullTextQuery]]):
+ checked = []
+ for occur, query in queries or []:
+ if not isinstance(query, FullTextQuery):
+ raise ValueError("Boolean query must contain FullTextQuery objects.")
+ checked.append((Occur.normalize(occur), query))
+ if not checked:
+ raise ValueError("Boolean query must contain at least one clause.")
+ object.__setattr__(self, "queries", checked)
+
+ def query_type(self) -> FullTextQueryType:
+ return FullTextQueryType.BOOLEAN
+
+ def query_text(self) -> str:
+ return ""
+
+ def referenced_columns(self) -> List[str]:
+ columns = []
+ for _, query in self.queries:
+ columns.extend(query.referenced_columns())
+ return _distinct(columns)
+
+ def should(self) -> List[FullTextQuery]:
+ return [query for occur, query in self.queries if occur == Occur.SHOULD]
+
+ def must(self) -> List[FullTextQuery]:
+ return [query for occur, query in self.queries if occur == Occur.MUST]
+
+ def must_not(self) -> List[FullTextQuery]:
+ return [query for occur, query in self.queries if occur == Occur.MUST_NOT]
+
+ def to_dict(self) -> Dict[str, Any]:
+ return {
+ "boolean": {
+ "should": [query.to_dict() for query in self.should()],
+ "must": [query.to_dict() for query in self.must()],
+ "must_not": [query.to_dict() for query in self.must_not()],
+ }
+ }
+
+
+def _add_boolean_queries(
+ target: List[Tuple[Occur, FullTextQuery]],
+ occur: Occur,
+ values: Optional[Sequence[Dict[str, Any]]],
+) -> None:
+ if values is None:
+ return
+ if not isinstance(values, list):
+ raise ValueError("Boolean query clauses must be arrays.")
+ for value in values:
+ target.append((occur, FullTextQuery.from_dict(value)))
+
+
+def _object_value(payload: Dict[str, Any], field: str) -> Dict[str, Any]:
+ value = payload.get(field)
+ if not isinstance(value, dict):
+ raise ValueError("Field '%s' must be an object." % field)
+ return value
+
+
+def _field_value(payload: Dict[str, Any], fields: Iterable[str]) -> Any:
+ for field in fields:
+ if field in payload:
+ return payload[field]
+ return None
+
+
+def _text_value(payload: Dict[str, Any], field: str) -> str:
+ value = payload.get(field)
+ if value is None:
+ raise ValueError("Field '%s' cannot be null or empty." % field)
+ return str(value)
+
+
+def _text_value_any(payload: Dict[str, Any], fields: Iterable[str]) -> str:
+ value = _field_value(payload, fields)
+ if value is None:
+ raise ValueError("Query terms cannot be null or empty.")
+ return str(value)
+
+
+def _int_value(payload: Dict[str, Any], field: str, default: int) -> int:
+ value = payload.get(field)
+ if value is None:
+ return default
+ return int(value)
+
+
+def _int_value_any(payload: Dict[str, Any], fields: Iterable[str], default: int) -> int:
+ value = _field_value(payload, fields)
+ if value is None:
+ return default
+ return int(value)
+
+
+def _float_value(payload: Dict[str, Any], field: str, default: float) -> float:
+ value = payload.get(field)
+ if value is None:
+ return default
+ return float(value)
+
+
+def _float_value_any(payload: Dict[str, Any], fields: Iterable[str], default: float) -> float:
+ value = _field_value(payload, fields)
+ if value is None:
+ return default
+ return float(value)
+
+
+def _fuzziness_value(payload: Dict[str, Any]) -> Optional[int]:
+ if "fuzziness" not in payload:
+ return 0
+ value = payload.get("fuzziness")
+ if value is None:
+ return None
+ if isinstance(value, str) and value.lower() == "auto":
+ return None
+ return int(value)
+
+
+def _string_list(value: Any, field: str) -> List[str]:
+ if not isinstance(value, list):
+ raise ValueError("%s must be an array." % field)
+ return [str(item) for item in value]
+
+
+def _optional_float_list(value: Any) -> Optional[List[float]]:
+ if value is None:
+ return None
+ if not isinstance(value, list):
+ raise ValueError("boost must be an array.")
+ return [float(item) for item in value]
+
+
+def _check_column(column: str) -> None:
+ if column is None or str(column) == "":
+ raise ValueError("Column cannot be None or empty.")
+
+
+def _check_terms(terms: str) -> None:
+ if terms is None or str(terms) == "":
+ raise ValueError("Query terms cannot be None or empty.")
+
+
+def _check_positive(value: float, name: str) -> None:
+ if value is None or value <= 0:
+ raise ValueError("%s must be positive, got: %s" % (name, value))
+
+
+def _distinct(values: Sequence[str]) -> List[str]:
+ result = []
+ seen = set()
+ for value in values:
+ if value not in seen:
+ seen.add(value)
+ result.append(value)
+ return result
diff --git a/paimon-python/pypaimon/globalindex/full_text_search.py b/paimon-python/pypaimon/globalindex/full_text_search.py
index 2ec41818d0..d1511df2e3 100644
--- a/paimon-python/pypaimon/globalindex/full_text_search.py
+++ b/paimon-python/pypaimon/globalindex/full_text_search.py
@@ -15,44 +15,44 @@
# specific language governing permissions and limitations
# under the License.
-"""FullTextSearch for performing full-text search on a text column."""
+"""FullTextSearch for performing full-text search with a structured query."""
from concurrent.futures import Future
from dataclasses import dataclass
-from typing import Optional
+from typing import List, Optional
+
+from pypaimon.globalindex.full_text_query import FullTextQuery
@dataclass
class FullTextSearch:
"""
- FullTextSearch to perform full-text search on a text column.
+ FullTextSearch to perform full-text search with a structured query.
Attributes:
- query_text: The query text to search
+ query: The structured full-text query
limit: Maximum number of results to return
- field_name: Name of the text field to search
"""
- query_text: str
+ query: FullTextQuery
limit: int
- field_name: str
- query_operator: str = "or"
def __post_init__(self):
- if not self.query_text:
- raise ValueError("Query text cannot be None or empty")
+ if self.query is None:
+ raise ValueError("Query cannot be None")
if self.limit <= 0:
raise ValueError(f"Limit must be positive, got: {self.limit}")
- if not self.field_name:
- raise ValueError("Field name cannot be null or empty")
- query_operator = (
- "or" if self.query_operator is None else self.query_operator.strip().lower()
- )
- if query_operator not in ("or", "and"):
- raise ValueError(
- "Query operator must be 'or' or 'and', got: %s" % self.query_operator
- )
- self.query_operator = query_operator
+
+ @property
+ def columns(self) -> List[str]:
+ return self.query.referenced_columns()
+
+ @property
+ def field_name(self) -> str:
+ return self.query.single_column()
+
+ def query_json(self) -> str:
+ return self.query.to_json()
def visit(self, visitor: 'GlobalIndexReader') -> 'Future[Optional[ScoredGlobalIndexResult]]':
"""Visit the global index reader with this full-text search."""
@@ -60,6 +60,6 @@ class FullTextSearch:
def __repr__(self) -> str:
return (
- f"FullTextSearch(field={self.field_name}, query='{self.query_text}', "
- f"limit={self.limit}, operator={self.query_operator})"
+ f"FullTextSearch(columns={self.columns}, limit={self.limit}, "
+ f"query_json={self.query_json()})"
)
diff --git a/paimon-python/pypaimon/globalindex/tantivy/tantivy_full_text_global_index_reader.py b/paimon-python/pypaimon/globalindex/tantivy/tantivy_full_text_global_index_reader.py
index 5ed0b9a91c..2da9ebe3bd 100644
--- a/paimon-python/pypaimon/globalindex/tantivy/tantivy_full_text_global_index_reader.py
+++ b/paimon-python/pypaimon/globalindex/tantivy/tantivy_full_text_global_index_reader.py
@@ -401,15 +401,82 @@ class TantivyFullTextGlobalIndexReader(GlobalIndexReader):
index.register_tokenizer(self._index_options.tokenizer_name(), analyzer)
def _parse_query(self, tantivy, full_text_search):
- query_text = full_text_search.query_text
- conjunction_by_default = full_text_search.query_operator == "and"
+ return self._parse_structured_query(tantivy, full_text_search.query)
+
+ def _parse_structured_query(self, tantivy, query):
+ from pypaimon.globalindex.full_text_query import (
+ BooleanQuery,
+ BoostQuery,
+ MatchQuery,
+ MultiMatchQuery,
+ Occur,
+ PhraseQuery,
+ )
+
+ if isinstance(query, MatchQuery):
+ return self._parse_match_query(tantivy, query)
+ if isinstance(query, PhraseQuery):
+ if self._index_options.tokenizer == "jieba":
+ raise ValueError("phrase query is not supported for jieba tokenizer indexes")
+ escaped = query.query.replace("\\", "\\\\").replace('"', '\\"')
+ query_text = '"%s"' % escaped
+ if query.slop != 0:
+ query_text = "%s~%s" % (query_text, query.slop)
+ return self._index.parse_query(query_text, ["text"])
+ if isinstance(query, BooleanQuery):
+ subqueries = []
+ for occur, child in query.queries:
+ if occur == Occur.SHOULD:
+ tantivy_occur = tantivy.Occur.Should
+ elif occur == Occur.MUST:
+ tantivy_occur = tantivy.Occur.Must
+ else:
+ tantivy_occur = getattr(tantivy.Occur, "MustNot", None)
+ if tantivy_occur is None:
+ raise RuntimeError(
+ "PyPaimon Tantivy full-text search requires a tantivy-py "
+ "version with Occur.MustNot support for boolean must_not queries."
+ )
+ subqueries.append(
+ (tantivy_occur, self._parse_structured_query(tantivy, child)))
+ if not subqueries:
+ raise ValueError("boolean query must contain at least one clause")
+ return tantivy.Query.boolean_query(subqueries)
+ if isinstance(query, BoostQuery):
+ raise ValueError(
+ "boost query is not supported by PyPaimon Tantivy full-text reader"
+ )
+ if isinstance(query, MultiMatchQuery):
+ raise ValueError(
+ "multi_match is not supported by single-column Tantivy full-text indexes"
+ )
+ raise ValueError("Unsupported full-text query type: %s" % type(query).__name__)
+
+ def _parse_match_query(self, tantivy, query):
+ if query.boost != 1.0:
+ raise ValueError(
+ "match query boost is not supported by PyPaimon Tantivy full-text reader"
+ )
+ if query.fuzziness not in (None, 0):
+ raise ValueError(
+ "match query fuzziness is not supported by PyPaimon Tantivy full-text reader"
+ )
+ if query.max_expansions != 50:
+ raise ValueError(
+ "match query max_expansions is not supported by PyPaimon Tantivy full-text reader"
+ )
+ if query.prefix_length != 0:
+ raise ValueError(
+ "match query prefix_length is not supported by PyPaimon Tantivy full-text reader"
+ )
+ conjunction_by_default = query.operator.value == "AND"
if self._index_options.tokenizer != "jieba":
if conjunction_by_default:
return self._index.parse_query(
- query_text, ["text"], conjunction_by_default=True)
- return self._index.parse_query(query_text, ["text"])
+ query.query, ["text"], conjunction_by_default=True)
+ return self._index.parse_query(query.query, ["text"])
- tokens = self._jieba_query_tokens(query_text)
+ tokens = self._jieba_query_tokens(query.query)
if not tokens:
return tantivy.Query.empty_query()
@@ -503,7 +570,7 @@ class TantivyFullTextGlobalIndexReader(GlobalIndexReader):
if not hasattr(query, name):
missing.append("Query.%s" % name)
if self._index_options.tokenizer == "jieba" and occur is not None:
- for name in ("Should", "Must"):
+ for name in ("Should", "Must", "MustNot"):
if not hasattr(occur, name):
missing.append("Occur.%s" % name)
if missing:
diff --git a/paimon-python/pypaimon/table/source/full_text_read.py b/paimon-python/pypaimon/table/source/full_text_read.py
index 3257525ad1..e35b081d9f 100644
--- a/paimon-python/pypaimon/table/source/full_text_read.py
+++ b/paimon-python/pypaimon/table/source/full_text_read.py
@@ -19,15 +19,27 @@
from abc import ABC, abstractmethod
from concurrent.futures import wait
-from typing import List
-
+from typing import Dict, List
+
+from pypaimon.globalindex.full_text_query import (
+ BooleanQuery,
+ BoostQuery,
+ FullTextQuery,
+ MatchQuery,
+ MultiMatchQuery,
+ PhraseQuery,
+)
from pypaimon.globalindex.full_text_search import FullTextSearch
from pypaimon.globalindex.global_index_meta import GlobalIndexIOMeta
from pypaimon.globalindex.global_index_result import GlobalIndexResult
from pypaimon.globalindex.offset_global_index_reader import OffsetGlobalIndexReader
-from pypaimon.globalindex.vector_search_result import DictBasedScoredIndexResult
+from pypaimon.globalindex.vector_search_result import (
+ DictBasedScoredIndexResult,
+ ScoredGlobalIndexResult,
+)
from pypaimon.table.source.full_text_search_split import FullTextSearchSplit
from pypaimon.table.source.full_text_scan import FullTextScanPlan
+from pypaimon.utils.roaring_bitmap import RoaringBitmap64
class FullTextRead(ABC):
@@ -48,24 +60,79 @@ class FullTextReadImpl(FullTextRead):
self,
table: 'FileStoreTable',
limit: int,
- text_column: 'DataField',
- query_text: str,
- query_operator: str = "or"
+ text_column,
+ query: FullTextQuery
):
self._table = table
self._limit = limit
- self._text_column = text_column
- self._query_text = query_text
- self._query_operator = query_operator
+ self._text_columns = text_column if isinstance(text_column, list) else [text_column]
+ self._query = query
def read(self, splits: List[FullTextSearchSplit]) -> GlobalIndexResult:
if not splits:
return GlobalIndexResult.create_empty()
+ splits_by_column: Dict[str, List[FullTextSearchSplit]] = {}
+ for split in splits:
+ splits_by_column.setdefault(split.column_name, []).append(split)
+ return self._eval_query(self._query, splits_by_column).top_k(self._limit)
+
+ def _eval_query(
+ self,
+ query: FullTextQuery,
+ splits_by_column: Dict[str, List[FullTextSearchSplit]]
+ ) -> ScoredGlobalIndexResult:
+ if isinstance(query, MatchQuery):
+ return self._eval_column_query(query, query.column, splits_by_column)
+ if isinstance(query, PhraseQuery):
+ return self._eval_column_query(query, query.column, splits_by_column)
+ if isinstance(query, MultiMatchQuery):
+ results = []
+ for column, boost in zip(query.columns, query.boosts):
+ match = MatchQuery(
+ query.query, column, boost=boost, operator=query.operator)
+ results.append(
+ self._eval_column_query(match, column, splits_by_column))
+ return _or(results).top_k(self._limit)
+ if isinstance(query, BoostQuery):
+ positive = self._eval_query(query.positive, splits_by_column)
+ negative = self._eval_query(query.negative, splits_by_column)
+ return _boost(positive, negative, query.negative_boost)
+ if isinstance(query, BooleanQuery):
+ result = None
+ for child in query.must():
+ child_result = self._eval_query(child, splits_by_column)
+ result = child_result if result is None else _and(result, child_result)
+
+ should_results = []
+ for child in query.should():
+ should_results.append(self._eval_query(child, splits_by_column))
+ if should_results:
+ should_result = _or(should_results)
+ result = should_result if result is None else _and_with_bonus(
+ result, should_result)
+
+ if result is None:
+ return ScoredGlobalIndexResult.create_empty()
+ for child in query.must_not():
+ result = _and_not(result, self._eval_query(child, splits_by_column))
+ return result
+ raise ValueError("Unsupported full-text query type: %s" % type(query).__name__)
+
+ def _eval_column_query(
+ self,
+ query: FullTextQuery,
+ column: str,
+ splits_by_column: Dict[str, List[FullTextSearchSplit]]
+ ) -> ScoredGlobalIndexResult:
+ splits = splits_by_column.get(column, [])
+ if not splits:
+ return ScoredGlobalIndexResult.create_empty()
futures = [
self._eval(
split.row_range_start, split.row_range_end,
- split.full_text_index_files
+ split.full_text_index_files,
+ query,
)
for split in splits
]
@@ -83,7 +150,7 @@ class FullTextReadImpl(FullTextRead):
return DictBasedScoredIndexResult(merged_scores).top_k(self._limit)
- def _eval(self, row_range_start, row_range_end, full_text_index_files):
+ def _eval(self, row_range_start, row_range_end, full_text_index_files, query):
index_io_meta_list = []
for index_file in full_text_index_files:
meta = index_file.global_index_meta
@@ -107,10 +174,8 @@ class FullTextReadImpl(FullTextRead):
)
full_text_search = FullTextSearch(
- query_text=self._query_text,
- limit=self._limit,
- field_name=self._text_column.name,
- query_operator=self._query_operator
+ query=query,
+ limit=_candidate_limit(row_range_start, row_range_end),
)
offset_reader = OffsetGlobalIndexReader(reader, row_range_start, row_range_end)
@@ -119,6 +184,63 @@ class FullTextReadImpl(FullTextRead):
return future
+def _and(left: ScoredGlobalIndexResult, right: ScoredGlobalIndexResult):
+ bitmap = RoaringBitmap64.and_(left.results(), right.results())
+ left_score = left.score_getter()
+ right_score = right.score_getter()
+ return ScoredGlobalIndexResult.create(
+ bitmap,
+ lambda row_id: (left_score(row_id) or 0.0) + (right_score(row_id) or 0.0),
+ )
+
+
+def _or(results: List[ScoredGlobalIndexResult]):
+ scores = {}
+ for result in results:
+ score_getter = result.score_getter()
+ for row_id in result.results():
+ scores[row_id] = scores.get(row_id, 0.0) + (score_getter(row_id) or 0.0)
+ return DictBasedScoredIndexResult(scores)
+
+
+def _and_with_bonus(base: ScoredGlobalIndexResult, bonus: ScoredGlobalIndexResult):
+ bitmap = base.results()
+ base_score = base.score_getter()
+ bonus_score = bonus.score_getter()
+ bonus_rows = bonus.results()
+ return ScoredGlobalIndexResult.create(
+ bitmap,
+ lambda row_id: (
+ (base_score(row_id) or 0.0)
+ + ((bonus_score(row_id) or 0.0) if bonus_rows.contains(row_id) else 0.0)
+ ),
+ )
+
+
+def _and_not(left: ScoredGlobalIndexResult, right: ScoredGlobalIndexResult):
+ bitmap = RoaringBitmap64.remove_all(left.results(), right.results())
+ return ScoredGlobalIndexResult.create(bitmap, left.score_getter())
+
+
+def _boost(
+ positive: ScoredGlobalIndexResult,
+ negative: ScoredGlobalIndexResult,
+ negative_boost: float):
+ positive_score = positive.score_getter()
+ negative_rows = negative.results()
+ return ScoredGlobalIndexResult.create(
+ positive.results(),
+ lambda row_id: (
+ (positive_score(row_id) or 0.0)
+ * (negative_boost if negative_rows.contains(row_id) else 1.0)
+ ),
+ )
+
+
+def _candidate_limit(row_range_start: int, row_range_end: int) -> int:
+ return row_range_end - row_range_start + 1
+
+
def _create_full_text_reader(index_type, file_io, index_path, index_io_meta_list):
"""Create a global index reader for full-text search."""
from pypaimon.globalindex.tantivy.tantivy_full_text_global_index_reader import (
diff --git a/paimon-python/pypaimon/table/source/full_text_scan.py b/paimon-python/pypaimon/table/source/full_text_scan.py
index 0da1607677..1324a2830d 100644
--- a/paimon-python/pypaimon/table/source/full_text_scan.py
+++ b/paimon-python/pypaimon/table/source/full_text_scan.py
@@ -21,6 +21,9 @@ from abc import ABC, abstractmethod
from collections import defaultdict
from typing import List
+from pypaimon.globalindex.tantivy.tantivy_full_text_global_index_reader import (
+ TANTIVY_FULLTEXT_IDENTIFIER,
+)
from pypaimon.table.source.full_text_search_split import FullTextSearchSplit
from pypaimon.utils.range import Range
@@ -49,16 +52,20 @@ class FullTextScanImpl(FullTextScan):
def __init__(
self,
table: 'FileStoreTable',
- text_column: 'DataField',
+ text_columns,
partition_filter=None):
self._table = table
- self._text_column = text_column
+ self._text_columns = list(text_columns)
self._partition_filter = partition_filter
def scan(self) -> FullTextScanPlan:
from pypaimon.index.index_file_handler import IndexFileHandler
- text_column = self._text_column
+ if not self._text_columns:
+ return FullTextScanPlan([])
+
+ text_column_ids = {field.id for field in self._text_columns}
+ id_to_column = {field.id: field.name for field in self._text_columns}
from pypaimon.snapshot.time_travel_util import TimeTravelUtil
from pypaimon.common.options.options import Options
@@ -80,21 +87,28 @@ class FullTextScanImpl(FullTextScan):
global_index_meta = entry.index_file.global_index_meta
if global_index_meta is None:
return False
- return text_column.id == global_index_meta.index_field_id
+ return (
+ global_index_meta.index_field_id in text_column_ids
+ and entry.index_file.index_type == TANTIVY_FULLTEXT_IDENTIFIER
+ )
entries = index_file_handler.scan(snapshot, index_file_filter)
all_index_files = [entry.index_file for entry in entries]
- # Group full-text index files by (rowRangeStart, rowRangeEnd)
- by_range = defaultdict(list)
+ # Group full-text index files by column and (rowRangeStart, rowRangeEnd).
+ by_column_and_range = defaultdict(lambda: defaultdict(list))
for index_file in all_index_files:
meta = index_file.global_index_meta
assert meta is not None
range_key = Range(meta.row_range_start, meta.row_range_end)
- by_range[range_key].append(index_file)
+ column_name = id_to_column[meta.index_field_id]
+ by_column_and_range[column_name][range_key].append(index_file)
splits = []
- for range_key, files in by_range.items():
- splits.append(FullTextSearchSplit(range_key.from_, range_key.to, files))
+ for column_name, by_range in by_column_and_range.items():
+ for range_key, files in by_range.items():
+ splits.append(
+ FullTextSearchSplit(
+ column_name, range_key.from_, range_key.to, files))
return FullTextScanPlan(splits)
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 44534a1337..49393d63da 100644
--- a/paimon-python/pypaimon/table/source/full_text_search_builder.py
+++ b/paimon-python/pypaimon/table/source/full_text_search_builder.py
@@ -21,6 +21,7 @@ from abc import ABC, abstractmethod
from typing import Optional
from pypaimon.common.predicate_builder import PredicateBuilder
+from pypaimon.globalindex.full_text_query import FullTextQuery
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
@@ -35,18 +36,8 @@ class FullTextSearchBuilder(ABC):
pass
@abstractmethod
- def with_text_column(self, name: str) -> 'FullTextSearchBuilder':
- """The text column to search."""
- pass
-
- @abstractmethod
- def with_query_text(self, query_text: str) -> 'FullTextSearchBuilder':
- """The query text to search."""
- pass
-
- @abstractmethod
- def with_query_operator(self, query_operator: str) -> 'FullTextSearchBuilder':
- """The default operator for query terms. Supported values are 'or' and 'and'."""
+ def with_query(self, query: FullTextQuery) -> 'FullTextSearchBuilder':
+ """The structured full-text query to search."""
pass
@abstractmethod
@@ -75,28 +66,15 @@ class FullTextSearchBuilderImpl(FullTextSearchBuilder):
def __init__(self, table: 'FileStoreTable'):
self._table = table
self._limit: int = 0
- self._text_column: Optional['DataField'] = None
- self._query_text: Optional[str] = None
- self._query_operator: str = "or"
+ self._query: Optional[FullTextQuery] = None
self._partition_filter = None
def with_limit(self, limit: int) -> 'FullTextSearchBuilder':
self._limit = limit
return self
- def with_text_column(self, name: str) -> 'FullTextSearchBuilder':
- field_dict = {f.name: f for f in self._table.fields}
- if name not in field_dict:
- raise ValueError(f"Text column '{name}' not found in table schema")
- self._text_column = field_dict[name]
- return self
-
- def with_query_text(self, query_text: str) -> 'FullTextSearchBuilder':
- self._query_text = query_text
- return self
-
- def with_query_operator(self, query_operator: str) -> 'FullTextSearchBuilder':
- self._query_operator = query_operator
+ def with_query(self, query: FullTextQuery) -> 'FullTextSearchBuilder':
+ self._query = query
return self
def with_partition_filter(self, partition_filter) -> 'FullTextSearchBuilder':
@@ -135,22 +113,26 @@ class FullTextSearchBuilderImpl(FullTextSearchBuilder):
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,
+ self._text_columns(),
partition_filter=self._partition_filter,
)
def new_full_text_read(self) -> FullTextRead:
if self._limit <= 0:
raise ValueError("Limit must be positive, set via with_limit()")
- if self._text_column is None:
- raise ValueError("Text column must be set via with_text_column()")
- if self._query_text is None:
- raise ValueError("Query text must be set via with_query_text()")
return FullTextReadImpl(
- self._table, self._limit, self._text_column,
- self._query_text, self._query_operator
+ self._table, self._limit, self._text_columns(), self._query
)
+
+ def _text_columns(self):
+ if self._query is None:
+ raise ValueError("Query must be set via with_query()")
+ field_dict = {f.name: f for f in self._table.fields}
+ fields = []
+ for name in self._query.referenced_columns():
+ if name not in field_dict:
+ raise ValueError(f"Text column '{name}' not found in table schema")
+ fields.append(field_dict[name])
+ return fields
diff --git a/paimon-python/pypaimon/table/source/full_text_search_split.py b/paimon-python/pypaimon/table/source/full_text_search_split.py
index b2e44aca7f..dec9c0e7b7 100644
--- a/paimon-python/pypaimon/table/source/full_text_search_split.py
+++ b/paimon-python/pypaimon/table/source/full_text_search_split.py
@@ -27,6 +27,7 @@ from pypaimon.index.index_file_meta import IndexFileMeta
class FullTextSearchSplit:
"""Split of full-text search."""
+ column_name: str
row_range_start: int
row_range_end: int
full_text_index_files: List[IndexFileMeta]
@@ -35,17 +36,23 @@ class FullTextSearchSplit:
if not isinstance(other, FullTextSearchSplit):
return False
return (
- self.row_range_start == other.row_range_start
+ self.column_name == other.column_name
+ and self.row_range_start == other.row_range_start
and self.row_range_end == other.row_range_end
and self.full_text_index_files == other.full_text_index_files
)
def __hash__(self):
- return hash((self.row_range_start, self.row_range_end, tuple(self.full_text_index_files)))
+ return hash((
+ self.column_name,
+ self.row_range_start,
+ self.row_range_end,
+ tuple(self.full_text_index_files)))
def __repr__(self):
return (
f"FullTextSearchSplit("
+ f"column_name={self.column_name}, "
f"row_range_start={self.row_range_start}, "
f"row_range_end={self.row_range_end}, "
f"full_text_index_files={self.full_text_index_files})"
diff --git a/paimon-python/pypaimon/table/source/hybrid_search_builder.py b/paimon-python/pypaimon/table/source/hybrid_search_builder.py
index bb5a32cced..868ea1f44b 100644
--- a/paimon-python/pypaimon/table/source/hybrid_search_builder.py
+++ b/paimon-python/pypaimon/table/source/hybrid_search_builder.py
@@ -23,6 +23,7 @@ from dataclasses import dataclass, field
from typing import Dict, List, Optional
from pypaimon.common.predicate_builder import PredicateBuilder
+from pypaimon.globalindex.full_text_query import FullTextQuery
from pypaimon.globalindex.global_index_result import GlobalIndexResult
from pypaimon.globalindex.vector_search_result import (
DictBasedScoredIndexResult,
@@ -52,8 +53,7 @@ class HybridSearchRoute:
limit: int
weight: float = 1.0
vector: Optional[List[float]] = None
- query_text: Optional[str] = None
- query_operator: str = "or"
+ full_text_query: Optional[FullTextQuery] = None
options: Dict[str, str] = field(default_factory=dict)
VECTOR = "vector"
@@ -66,20 +66,13 @@ class HybridSearchRoute:
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:
+ if self.route_type == self.FULL_TEXT and self.full_text_query is None:
raise ValueError(
- "Query text cannot be None or empty for full-text route")
+ "Query cannot be None 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
@@ -102,17 +95,15 @@ class HybridSearchRoute:
@classmethod
def full_text_route(
cls,
- field_name: str,
- query_text: str,
- query_operator: str,
+ query_json: str,
limit: int,
weight: float = 1.0,
options: Optional[Dict[str, str]] = None) -> 'HybridSearchRoute':
+ query = FullTextQuery.from_json(query_json)
return cls(
route_type=cls.FULL_TEXT,
- field_name=field_name,
- query_text=query_text,
- query_operator=query_operator,
+ field_name=query.referenced_columns()[0],
+ full_text_query=query,
limit=limit,
weight=weight,
options=dict(options or {}),
@@ -124,6 +115,11 @@ class HybridSearchRoute:
def is_full_text(self) -> bool:
return self.route_type == self.FULL_TEXT
+ def columns(self) -> List[str]:
+ if self.is_full_text():
+ return self.full_text_query.referenced_columns()
+ return [self.field_name]
+
@dataclass
class HybridSearchRouteResult:
@@ -192,16 +188,14 @@ class HybridSearchBuilder(ABC):
def add_full_text_route(
self,
- text_column: str,
- query_text: str,
+ query_json: 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))
+ query_json, limit, weight, options))
@abstractmethod
def route_builders(self) -> List[HybridSearchRouteBuilder]:
@@ -349,9 +343,7 @@ class HybridSearchBuilderImpl(HybridSearchBuilder):
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_query(route.full_text_query)
.with_limit(route.limit)
)
if self._partition_filter is not None:
diff --git a/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py b/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
index 691bd16dd9..ad66dd26f1 100644
--- a/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
+++ b/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
@@ -25,6 +25,7 @@ import pyarrow as pa
from parameterized import parameterized
from pypaimon.catalog.catalog_factory import CatalogFactory
from pypaimon.data.generic_variant import GenericVariant
+from pypaimon.globalindex.full_text_query import MatchQuery
from pypaimon.globalindex.global_index_scanner import GlobalIndexScanner
from pypaimon.schema.data_types import VectorType
from pypaimon.schema.schema import Schema
@@ -1051,8 +1052,7 @@ class JavaPyReadWriteTest(unittest.TestCase):
# Use FullTextSearchBuilder to search
builder = table.new_full_text_search_builder()
- builder.with_text_column('content')
- builder.with_query_text('paimon')
+ builder.with_query(MatchQuery('paimon', 'content'))
builder.with_limit(10)
result = builder.execute_local()
@@ -1074,8 +1074,7 @@ class JavaPyReadWriteTest(unittest.TestCase):
# Search for "tantivy" - only row 1
builder2 = table.new_full_text_search_builder()
- builder2.with_text_column('content')
- builder2.with_query_text('tantivy')
+ builder2.with_query(MatchQuery('tantivy', 'content'))
builder2.with_limit(10)
result2 = builder2.execute_local()
@@ -1093,8 +1092,7 @@ class JavaPyReadWriteTest(unittest.TestCase):
# Search for "full-text search" - rows 1, 3
builder3 = table.new_full_text_search_builder()
- builder3.with_text_column('content')
- builder3.with_query_text('full-text search')
+ builder3.with_query(MatchQuery('full-text search', 'content'))
builder3.with_limit(10)
result3 = builder3.execute_local()
@@ -1117,8 +1115,7 @@ class JavaPyReadWriteTest(unittest.TestCase):
# Search for Chinese fragments using the ngram tokenizer metadata written by Java.
ngram_builder = ngram_table.new_full_text_search_builder()
- ngram_builder.with_text_column('content')
- ngram_builder.with_query_text('中文')
+ ngram_builder.with_query(MatchQuery('中文', 'content'))
ngram_builder.with_limit(10)
ngram_result = ngram_builder.execute_local()
@@ -1136,8 +1133,7 @@ class JavaPyReadWriteTest(unittest.TestCase):
['Apache Paimon 支持中文全文检索', '中文索引支持片段查询'])
fragment_builder = ngram_table.new_full_text_search_builder()
- fragment_builder.with_text_column('content')
- fragment_builder.with_query_text('片段')
+ fragment_builder.with_query(MatchQuery('片段', 'content'))
fragment_builder.with_limit(10)
fragment_result = fragment_builder.execute_local()
@@ -1146,9 +1142,7 @@ class JavaPyReadWriteTest(unittest.TestCase):
self.assertEqual(fragment_row_ids, [4])
ngram_and_builder = ngram_table.new_full_text_search_builder()
- ngram_and_builder.with_text_column('content')
- ngram_and_builder.with_query_text('中文 片段')
- ngram_and_builder.with_query_operator('and')
+ ngram_and_builder.with_query(MatchQuery('中文 片段', 'content', operator='and'))
ngram_and_builder.with_limit(10)
ngram_and_result = ngram_and_builder.execute_local()
@@ -1158,8 +1152,7 @@ class JavaPyReadWriteTest(unittest.TestCase):
simple_table = self.catalog.get_table('default.test_tantivy_fulltext_simple')
simple_builder = simple_table.new_full_text_search_builder()
- simple_builder.with_text_column('content')
- simple_builder.with_query_text('running')
+ simple_builder.with_query(MatchQuery('running', 'content'))
simple_builder.with_limit(10)
simple_result = simple_builder.execute_local()
@@ -1171,8 +1164,7 @@ class JavaPyReadWriteTest(unittest.TestCase):
# Search for Chinese words using the jieba tokenizer metadata written by Java.
jieba_builder = jieba_table.new_full_text_search_builder()
- jieba_builder.with_text_column('content')
- jieba_builder.with_query_text('售货员')
+ jieba_builder.with_query(MatchQuery('售货员', 'content'))
jieba_builder.with_limit(10)
jieba_result = jieba_builder.execute_local()
@@ -1190,8 +1182,7 @@ class JavaPyReadWriteTest(unittest.TestCase):
['张华在百货公司当售货员'])
jieba_phrase_builder = jieba_table.new_full_text_search_builder()
- jieba_phrase_builder.with_text_column('content')
- jieba_phrase_builder.with_query_text('自然')
+ jieba_phrase_builder.with_query(MatchQuery('自然', 'content'))
jieba_phrase_builder.with_limit(10)
jieba_phrase_result = jieba_phrase_builder.execute_local()
@@ -1200,9 +1191,7 @@ class JavaPyReadWriteTest(unittest.TestCase):
self.assertEqual(jieba_phrase_row_ids, [3])
jieba_and_builder = jieba_table.new_full_text_search_builder()
- jieba_and_builder.with_text_column('content')
- jieba_and_builder.with_query_text('中文 自然')
- jieba_and_builder.with_query_operator('and')
+ jieba_and_builder.with_query(MatchQuery('中文 自然', 'content', operator='and'))
jieba_and_builder.with_limit(10)
jieba_and_result = jieba_and_builder.execute_local()
diff --git a/paimon-python/pypaimon/tests/vector_search_filter_test.py b/paimon-python/pypaimon/tests/vector_search_filter_test.py
index fe63890a42..e3ccac2f33 100644
--- a/paimon-python/pypaimon/tests/vector_search_filter_test.py
+++ b/paimon-python/pypaimon/tests/vector_search_filter_test.py
@@ -36,6 +36,7 @@ from pypaimon.globalindex.btree.btree_index_meta import BTreeIndexMeta
from pypaimon.globalindex.global_index_meta import GlobalIndexIOMeta, GlobalIndexMeta
from pypaimon.globalindex.global_index_reader import _completed_future
from pypaimon.globalindex.global_index_result import GlobalIndexResult
+from pypaimon.globalindex.full_text_query import MatchQuery
from pypaimon.globalindex.vector_search import VectorSearch
from pypaimon.globalindex.vector_search_result import ScoredGlobalIndexResult
from pypaimon.index.index_file_meta import IndexFileMeta
@@ -291,6 +292,7 @@ class _FakeQuery:
class _FakeOccur:
Should = "should"
Must = "must"
+ MustNot = "must_not"
class _FakeSearchResults:
@@ -586,7 +588,7 @@ class TantivyFullTextIndexOptionsTest(unittest.TestCase):
prefix_only=True, lower_case=True))])
try:
result = reader.visit_full_text_search(
- FullTextSearch("中文", 10, "content")).result()
+ FullTextSearch(MatchQuery("中文", "content"), 10)).result()
finally:
reader.close()
finally:
@@ -650,7 +652,7 @@ class TantivyFullTextIndexOptionsTest(unittest.TestCase):
[GlobalIndexIOMeta(file_name="ft.index", file_size=1)])
try:
reader.visit_full_text_search(
- FullTextSearch("hello", 5, "content")).result()
+ FullTextSearch(MatchQuery("hello", "content"), 5)).result()
finally:
reader.close()
finally:
@@ -687,7 +689,7 @@ class TantivyFullTextIndexOptionsTest(unittest.TestCase):
with_position=False))])
try:
reader.visit_full_text_search(
- FullTextSearch("running", 10, "content")).result()
+ FullTextSearch(MatchQuery("running", "content"), 10)).result()
finally:
reader.close()
finally:
@@ -727,7 +729,7 @@ class TantivyFullTextIndexOptionsTest(unittest.TestCase):
with self.assertRaisesRegex(
RuntimeError, "ngram tokenizer support"):
reader.visit_full_text_search(
- FullTextSearch("中文", 10, "content")).result()
+ FullTextSearch(MatchQuery("中文", "content"), 10)).result()
finally:
if old_tantivy is None:
sys.modules.pop("tantivy", None)
@@ -762,7 +764,7 @@ class TantivyFullTextIndexOptionsTest(unittest.TestCase):
metadata=_java_tantivy_meta(tokenizer="jieba"))])
try:
result = reader.visit_full_text_search(
- FullTextSearch("售货员", 10, "content", "and")).result()
+ FullTextSearch(MatchQuery("售货员", "content", operator="and"), 10)).result()
finally:
reader.close()
finally:
@@ -813,7 +815,7 @@ class TantivyFullTextIndexOptionsTest(unittest.TestCase):
try:
with self.assertRaisesRegex(RuntimeError, "pip install jieba"):
reader.visit_full_text_search(
- FullTextSearch("售货员", 10, "content")).result()
+ FullTextSearch(MatchQuery("售货员", "content"), 10)).result()
finally:
reader.close()
finally:
@@ -827,6 +829,299 @@ class TantivyFullTextIndexOptionsTest(unittest.TestCase):
sys.modules["jieba"] = old_jieba
+class FullTextQueryDslTest(unittest.TestCase):
+
+ def test_lancedb_style_query_json_round_trip(self):
+ from pypaimon.globalindex.full_text_query import (
+ BooleanQuery,
+ BoostQuery,
+ FullTextQuery,
+ FullTextOperator,
+ MatchQuery,
+ MultiMatchQuery,
+ Occur,
+ PhraseQuery,
+ )
+
+ query = FullTextQuery.from_json(
+ '{"match":{"column":"content","query":"paimon",'
+ '"maxExpansions":10,"prefixLength":1,"fuzziness":"auto"}}')
+ self.assertIsInstance(query, MatchQuery)
+ self.assertEqual("paimon", query.query)
+ self.assertIsNone(query.fuzziness)
+ self.assertEqual(10, query.max_expansions)
+ self.assertEqual(1, query.prefix_length)
+
+ phrase = PhraseQuery("paimon lake", "content", slop=1)
+ self.assertEqual(
+ '{"match_phrase":{"column":"content","terms":"paimon lake","slop":1}}',
+ phrase.to_json())
+ self.assertEqual("match_phrase", phrase.query_type().value)
+
+ boost = BoostQuery(
+ MatchQuery("paimon", "content"),
+ MatchQuery("vector", "content"),
+ negative_boost=0.2)
+ self.assertEqual(0.2, boost.negative_boost)
+ self.assertIn('"negative_boost":0.2', boost.to_json())
+
+ multi_match = MultiMatchQuery(
+ "paimon", ["title", "content"], boosts=[2.0, 1.0],
+ operator=FullTextOperator.AND)
+ self.assertEqual(["title", "content"], multi_match.columns)
+ self.assertEqual(
+ '{"multi_match":{"query":"paimon","columns":["title","content"],'
+ '"boost":[2.0,1.0],"operator":"And"}}',
+ multi_match.to_json())
+
+ boolean = BooleanQuery([
+ (Occur.MUST, MatchQuery("paimon", "content")),
+ ("must_not", MatchQuery("vector", "content")),
+ ])
+ self.assertEqual(["content"], boolean.referenced_columns())
+ self.assertEqual(1, len(boolean.must()))
+ self.assertEqual(1, len(boolean.must_not()))
+
+ def test_lancedb_boolean_queries_alias(self):
+ from pypaimon.globalindex.full_text_query import FullTextQuery
+
+ query = FullTextQuery.from_json(
+ '{"boolean":{"queries":[{"occur":"must","query":{"match":'
+ '{"column":"content","terms":"paimon"}}},["must_not",{"match":'
+ '{"column":"content","terms":"vector"}}]]}}')
+
+ self.assertEqual(1, len(query.must()))
+ self.assertEqual(1, len(query.must_not()))
+ self.assertEqual("content", query.single_column())
+
+
+class FullTextSearchBuilderDslTest(unittest.TestCase):
+
+ def tearDown(self):
+ mock.patch.stopall()
+
+ def test_full_text_builder_threads_structured_query_to_reader(self):
+ from pypaimon.table.source.full_text_search_builder import (
+ FullTextSearchBuilderImpl,
+ )
+ from pypaimon.table.source.full_text_search_split import (
+ FullTextSearchSplit,
+ )
+
+ text_field = _field(1, "content", "STRING")
+ entry = _entry(
+ None, field_id=1, index_type="tantivy-fulltext",
+ file_name="ft.index", row_range_start=0, row_range_end=9)
+ table = _StubTable(fields=[text_field], entries=[entry])
+ _patch_snapshot(self, [entry])
+ captured_searches = []
+
+ def _fake_create(index_type, file_io, index_path, index_io_meta_list):
+ class _FakeReader:
+ def visit_full_text_search(self_inner, fts):
+ captured_searches.append(fts)
+ return _completed_future(None)
+
+ def close(self_inner):
+ pass
+ return _FakeReader()
+
+ split = FullTextSearchSplit(
+ column_name="content",
+ row_range_start=0, row_range_end=9,
+ full_text_index_files=[entry.index_file])
+
+ with mock.patch(
+ "pypaimon.table.source.full_text_read._create_full_text_reader",
+ side_effect=_fake_create):
+ (
+ FullTextSearchBuilderImpl(table)
+ .with_query(MatchQuery("paimon", "content", operator="and"))
+ .with_limit(10)
+ .new_full_text_read()
+ .read([split])
+ )
+
+ self.assertEqual(1, len(captured_searches))
+ self.assertEqual("content", captured_searches[0].field_name)
+ self.assertEqual(10, captured_searches[0].limit)
+ self.assertEqual(
+ '{"match":{"column":"content","terms":"paimon","boost":1.0,'
+ '"fuzziness":0,"max_expansions":50,"operator":"And",'
+ '"prefix_length":0}}',
+ captured_searches[0].query_json())
+
+ def test_full_text_builder_sends_leaf_candidate_limit_for_compound_queries(self):
+ from pypaimon.globalindex.full_text_query import BoostQuery
+ from pypaimon.table.source.full_text_search_builder import (
+ FullTextSearchBuilderImpl,
+ )
+ from pypaimon.table.source.full_text_search_split import (
+ FullTextSearchSplit,
+ )
+
+ text_field = _field(1, "content", "STRING")
+ entry = _entry(
+ None, field_id=1, index_type="tantivy-fulltext",
+ file_name="ft.index", row_range_start=0, row_range_end=99)
+ table = _StubTable(fields=[text_field], entries=[entry])
+ _patch_snapshot(self, [entry])
+ captured_limits = []
+
+ def _fake_create(index_type, file_io, index_path, index_io_meta_list):
+ class _FakeReader:
+ def visit_full_text_search(self_inner, fts):
+ captured_limits.append(fts.limit)
+ return _completed_future(None)
+
+ def close(self_inner):
+ pass
+ return _FakeReader()
+
+ split = FullTextSearchSplit(
+ column_name="content",
+ row_range_start=0,
+ row_range_end=99,
+ full_text_index_files=[entry.index_file])
+
+ with mock.patch(
+ "pypaimon.table.source.full_text_read._create_full_text_reader",
+ side_effect=_fake_create):
+ (
+ FullTextSearchBuilderImpl(table)
+ .with_query(
+ BoostQuery(
+ MatchQuery("paimon", "content"),
+ MatchQuery("vector", "content"),
+ negative_boost=0.1))
+ .with_limit(3)
+ .new_full_text_read()
+ .read([split])
+ )
+
+ self.assertEqual([100, 100], captured_limits)
+
+ def test_full_text_builder_supports_multi_match_across_columns(self):
+ from pypaimon.globalindex.full_text_query import MultiMatchQuery
+ from pypaimon.globalindex.vector_search_result import (
+ DictBasedScoredIndexResult,
+ )
+ from pypaimon.table.source.full_text_search_builder import (
+ FullTextSearchBuilderImpl,
+ )
+
+ title_field = _field(1, "title", "STRING")
+ body_field = _field(2, "body", "STRING")
+ title_entry = _entry(
+ None, field_id=1, index_type="tantivy-fulltext",
+ file_name="title.index", row_range_start=0, row_range_end=9)
+ body_entry = _entry(
+ None, field_id=2, index_type="tantivy-fulltext",
+ file_name="body.index", row_range_start=0, row_range_end=9)
+ table = _StubTable(
+ fields=[title_field, body_field], entries=[title_entry, body_entry])
+ _patch_snapshot(self, [title_entry, body_entry])
+ captured = []
+
+ def _fake_create(index_type, file_io, index_path, index_io_meta_list):
+ file_name = index_io_meta_list[0].file_name
+
+ class _FakeReader:
+ def visit_full_text_search(self_inner, fts):
+ captured.append(fts.query_json())
+ if file_name == "title.index":
+ return _completed_future(DictBasedScoredIndexResult({1: 2.0}))
+ return _completed_future(DictBasedScoredIndexResult({2: 1.0}))
+
+ def close(self_inner):
+ pass
+ return _FakeReader()
+
+ with mock.patch(
+ "pypaimon.table.source.full_text_read._create_full_text_reader",
+ side_effect=_fake_create):
+ result = (
+ FullTextSearchBuilderImpl(table)
+ .with_query(MultiMatchQuery("paimon", ["title", "body"]))
+ .with_limit(10)
+ .execute_local()
+ )
+
+ self.assertEqual({1, 2}, set(result.results().to_list()))
+ self.assertEqual(2, len(captured))
+
+ def test_full_text_multi_match_adds_column_scores(self):
+ from pypaimon.globalindex.full_text_query import MultiMatchQuery
+ from pypaimon.globalindex.vector_search_result import (
+ DictBasedScoredIndexResult,
+ )
+ from pypaimon.table.source.full_text_search_builder import (
+ FullTextSearchBuilderImpl,
+ )
+
+ title_field = _field(1, "title", "STRING")
+ body_field = _field(2, "body", "STRING")
+ title_entry = _entry(
+ None, field_id=1, index_type="tantivy-fulltext",
+ file_name="title.index", row_range_start=0, row_range_end=9)
+ body_entry = _entry(
+ None, field_id=2, index_type="tantivy-fulltext",
+ file_name="body.index", row_range_start=0, row_range_end=9)
+ table = _StubTable(
+ fields=[title_field, body_field], entries=[title_entry, body_entry])
+ _patch_snapshot(self, [title_entry, body_entry])
+
+ def _fake_create(index_type, file_io, index_path, index_io_meta_list):
+ file_name = index_io_meta_list[0].file_name
+
+ class _FakeReader:
+ def visit_full_text_search(self_inner, fts):
+ if file_name == "title.index":
+ return _completed_future(
+ DictBasedScoredIndexResult({1: 2.0, 3: 1.0}))
+ return _completed_future(
+ DictBasedScoredIndexResult({1: 5.0, 2: 1.0}))
+
+ def close(self_inner):
+ pass
+ return _FakeReader()
+
+ with mock.patch(
+ "pypaimon.table.source.full_text_read._create_full_text_reader",
+ side_effect=_fake_create):
+ result = (
+ FullTextSearchBuilderImpl(table)
+ .with_query(MultiMatchQuery("paimon", ["title", "body"]))
+ .with_limit(10)
+ .execute_local()
+ )
+
+ self.assertEqual({1, 2, 3}, set(result.results().to_list()))
+ self.assertEqual(7.0, result.score_getter()(1))
+ self.assertEqual(1.0, result.score_getter()(2))
+ self.assertEqual(1.0, result.score_getter()(3))
+
+ def test_full_text_scan_ignores_other_index_types_on_same_column(self):
+ from pypaimon.table.source.full_text_scan import FullTextScanImpl
+
+ text_field = _field(1, "content", "STRING")
+ ft_entry = _entry(
+ None, field_id=1, index_type="tantivy-fulltext",
+ file_name="ft.index", row_range_start=0, row_range_end=9)
+ btree_entry = _entry(
+ None, field_id=1, index_type="btree",
+ file_name="btree.index", row_range_start=0, row_range_end=9)
+ table = _StubTable(fields=[text_field], entries=[ft_entry, btree_entry])
+ _patch_snapshot(self, [ft_entry, btree_entry])
+
+ splits = FullTextScanImpl(table, [text_field]).scan().splits()
+
+ self.assertEqual(1, len(splits))
+ self.assertEqual(
+ ["ft.index"],
+ [f.file_name for f in splits[0].full_text_index_files])
+
+
class VectorSearchFilterTest(unittest.TestCase):
"""Non-partitioned wiring: scan + read + external_path plumbing."""
@@ -1572,16 +1867,8 @@ class HybridSearchBuilderTest(unittest.TestCase):
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))
+ def with_query(self, query):
+ self.calls.append(("query", query))
return self
def with_limit(self, limit):
@@ -1617,8 +1904,10 @@ class HybridSearchBuilderTest(unittest.TestCase):
"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")
+ '{"match":{"column":"content","terms":"paimon search",'
+ '"operator":"and"}}',
+ 10,
+ weight=1.0)
.with_limit(2)
.with_rrf_ranker()
.execute_local()
@@ -1629,7 +1918,8 @@ class HybridSearchBuilderTest(unittest.TestCase):
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)
+ self.assertEqual("paimon search",
+ captured_builders[1].calls[0][1].query)
def test_hybrid_search_rejects_data_filter_with_full_text_route(self):
from pypaimon.table.source.hybrid_search_builder import (
@@ -1643,7 +1933,8 @@ class HybridSearchBuilderTest(unittest.TestCase):
builder = (
HybridSearchBuilderImpl(table)
- .add_full_text_route("content", "paimon search", 10)
+ .add_full_text_route(
+ '{"match":{"column":"content","terms":"paimon search"}}', 10)
.with_filter(pb.equal("id", 1))
.with_limit(5)
)
@@ -1679,7 +1970,8 @@ class HybridSearchBuilderTest(unittest.TestCase):
pb = PredicateBuilder(table.fields)
route_builders = (
HybridSearchBuilderImpl(table)
- .add_full_text_route("content", "paimon search", 10)
+ .add_full_text_route(
+ '{"match":{"column":"content","terms":"paimon search"}}', 10)
.with_filter(pb.equal("pt", 2))
.with_limit(5)
.route_builders()
@@ -1945,6 +2237,7 @@ class FullTextSearchManySplitsTest(unittest.TestCase):
return _FakeReader()
split = FullTextSearchSplit(
+ column_name="content",
row_range_start=0, row_range_end=9,
full_text_index_files=[entry.index_file])
@@ -1953,7 +2246,7 @@ class FullTextSearchManySplitsTest(unittest.TestCase):
side_effect=_fake_create):
reader = FullTextReadImpl(
table, limit=10, text_column=text_field,
- query_text="test")
+ query=MatchQuery("test", "content"))
reader.read([split])
self.assertEqual(1, len(captured_io_metas))
@@ -1996,6 +2289,7 @@ class FullTextSearchManySplitsTest(unittest.TestCase):
splits = [
FullTextSearchSplit(
+ column_name="content",
row_range_start=i, row_range_end=i,
full_text_index_files=[entries[i].index_file])
for i in range(num_splits)
@@ -2006,7 +2300,7 @@ class FullTextSearchManySplitsTest(unittest.TestCase):
side_effect=_fake_create):
reader = FullTextReadImpl(
table, limit=10, text_column=text_field,
- query_text="test")
+ query=MatchQuery("test", "content"))
result = reader.read(splits)
self.assertLessEqual(result.results().cardinality(), 10)
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 b4453288cf..a5f79df34b 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
@@ -129,9 +129,7 @@ abstract class PaimonBaseScan(table: InnerTable)
val fullTextSearch = pushedFullTextSearch.get
val ftBuilder = table
.newFullTextSearchBuilder()
- .withQueryText(fullTextSearch.queryText())
- .withQueryOperator(fullTextSearch.queryOperator())
- .withTextColumn(fullTextSearch.fieldName())
+ .withQuery(fullTextSearch.query())
.withLimit(fullTextSearch.limit())
if (pushedPartitionFilters.nonEmpty) {
ftBuilder.withPartitionFilter(PartitionPredicate.and(pushedPartitionFilters.asJava))
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 05f70bcb37..4cafdc0b20 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
@@ -20,7 +20,7 @@ package org.apache.paimon.spark.catalyst.plans.logical
import org.apache.paimon.CoreOptions
import org.apache.paimon.globalindex.HybridSearchRanker
-import org.apache.paimon.predicate.{FullTextSearch, HybridSearch, HybridSearchRoute, VectorSearch}
+import org.apache.paimon.predicate.{FullTextQuery, 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, HybridSearchTable, InnerTable, VectorSearchTable}
@@ -455,8 +455,7 @@ case class VectorSearchQuery(override val args: Seq[Expression])
* Usage: hybrid_search(table_name, vector_routes, full_text_routes, limit[, ranker])
* - table_name: the Paimon table to search
* - 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
+ * - full_text_routes: route config array with query, limit, weight, and options fields
* - limit: the final number of ranked top results to return
* - ranker: optional ranker for combining results from multiple routes
*/
@@ -488,10 +487,12 @@ case class HybridSearchQuery(override val args: Seq[Expression])
(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.columns().asScala.foreach {
+ columnName =>
+ if (!innerTable.rowType().containsField(columnName)) {
+ throw new RuntimeException(
+ s"Column $columnName does not exist in table ${innerTable.name()}")
+ }
}
route
}.toList
@@ -582,9 +583,7 @@ case class HybridSearchQuery(override val args: Seq[Expression])
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 query: Option[String] = None
var limit: Option[Int] = None
var weight: Option[Float] = None
var options = Map.empty[String, String]
@@ -592,12 +591,8 @@ case class HybridSearchQuery(override val args: Seq[Expression])
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 "query" =>
+ query = Some(VectorSearchQuery(Seq.empty).extractString(valueExpr))
case "limit" =>
limit = Some(parsePositiveLimit(valueExpr.eval()))
case "weight" =>
@@ -605,22 +600,16 @@ case class HybridSearchQuery(override val args: Seq[Expression])
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.")
+ throw new IllegalArgumentException(
+ s"Unsupported full-text route field '$key'. " +
+ "Supported fields are query, 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"),
+ query.getOrElse(throw new IllegalArgumentException("Full-text route must define query.")),
limit.getOrElse(defaultLimit),
weight.getOrElse(1.0f),
options.asJava
@@ -648,14 +637,13 @@ case class HybridSearchQuery(override val args: Seq[Expression])
/**
* Plan for the [[FULL_TEXT_SEARCH]] table-valued function.
*
- * Usage: full_text_search(table_name, column_name, query_text, limit[, query_operator])
+ * Usage: full_text_search(table_name, query_json, limit)
* - table_name: the Paimon table to search
- * - column_name: the text column name
- * - query_text: the query text string
+ * - query_json: the LanceDB-style full-text query JSON string
* - limit: the number of top results to return
- * - query_operator: optional query operator, supported values are 'or' and 'and'
*
- * Example: SELECT * FROM full_text_search('T', 'content', 'hello world', 10, 'and')
+ * Example: SELECT * FROM full_text_search('T', '{"match":{"column":"content","terms":"hello"}}',
+ * 10)
*/
case class FullTextSearchQuery(override val args: Seq[Expression])
extends PaimonTableValueFunction(FULL_TEXT_SEARCH) {
@@ -668,27 +656,23 @@ case class FullTextSearchQuery(override val args: Seq[Expression])
def createFullTextSearch(
innerTable: InnerTable,
argsWithoutTable: Seq[Expression]): FullTextSearch = {
- if (argsWithoutTable.size != 3 && argsWithoutTable.size != 4) {
+ if (argsWithoutTable.size != 2) {
throw new RuntimeException(
- s"$FULL_TEXT_SEARCH needs three or four parameters after table_name: " +
- s"column_name, query_text, limit[, query_operator]. " +
+ s"$FULL_TEXT_SEARCH needs two parameters after table_name: " +
+ s"query_json, limit. " +
s"Got ${argsWithoutTable.size} parameters after table_name."
)
}
- val columnName = argsWithoutTable.head.eval().toString
- if (!innerTable.rowType().containsField(columnName)) {
- throw new RuntimeException(
- s"Column $columnName does not exist in table ${innerTable.name()}"
- )
+ val query = FullTextQuery.fromJson(argsWithoutTable.head.eval().toString)
+ query.columns().asScala.foreach {
+ columnName =>
+ if (!innerTable.rowType().containsField(columnName)) {
+ throw new RuntimeException(
+ s"Column $columnName does not exist in table ${innerTable.name()}"
+ )
+ }
}
- val queryText = argsWithoutTable(1).eval().toString
- val limit = parsePositiveLimit(argsWithoutTable(2).eval())
- val queryOperator =
- if (argsWithoutTable.size == 4) {
- VectorSearchQuery(Seq.empty).extractString(argsWithoutTable(3))
- } else {
- "or"
- }
- new FullTextSearch(queryText, limit, columnName, queryOperator)
+ val limit = parsePositiveLimit(argsWithoutTable(1).eval())
+ new FullTextSearch(query, limit)
}
}
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 23427a54e6..4a13fbc56b 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
@@ -109,12 +109,8 @@ class VectorSearchQueryTest extends AnyFunSuite {
CreateArray(
Seq(
CreateNamedStruct(Seq(
- Literal("field"),
- Literal("content"),
- Literal("query_text"),
- Literal("paimon lake"),
- Literal("query_operator"),
- Literal("and"),
+ Literal("query"),
+ Literal("""{"match":{"column":"content","terms":"paimon lake","operator":"And"}}"""),
Literal("limit"),
Literal(20),
Literal("weight"),
@@ -131,42 +127,57 @@ class VectorSearchQueryTest extends AnyFunSuite {
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).fullTextQuery().queryText() == "paimon lake")
+ assert(search.routes().get(0).fullTextQuery().toJson.contains("\"operator\":\"And\""))
assert(search.routes().get(0).limit() == 20)
assert(search.routes().get(0).weight() == 1.5f)
}
- test("create full-text search with default query operator") {
+ test("create full-text search") {
val search = FullTextSearchQuery(Seq.empty).createFullTextSearch(
innerTable,
- Seq(Literal("content"), Literal("paimon lake"), Literal(10)))
+ Seq(Literal("""{"match":{"column":"content","terms":"paimon lake"}}"""), Literal(10)))
assert(search.fieldName() == "content")
- assert(search.queryText() == "paimon lake")
+ assert(search.query().queryText() == "paimon lake")
assert(search.limit() == 10)
- assert(search.queryOperator() == "or")
}
test("create full-text search with explicit query operator") {
val search = FullTextSearchQuery(Seq.empty).createFullTextSearch(
innerTable,
- Seq(Literal("content"), Literal("paimon lake"), Literal(10), Literal("and")))
+ Seq(
+ Literal("""{"match":{"column":"content","terms":"paimon lake","operator":"And"}}"""),
+ Literal(10)))
assert(search.fieldName() == "content")
- assert(search.queryText() == "paimon lake")
+ assert(search.query().queryText() == "paimon lake")
assert(search.limit() == 10)
- assert(search.queryOperator() == "and")
+ assert(search.query().toJson.contains("\"operator\":\"And\""))
+ }
+
+ test("create multi-column full-text search") {
+ val search = FullTextSearchQuery(Seq.empty).createFullTextSearch(
+ innerTable,
+ Seq(
+ Literal(
+ """{"multi_match":{"query":"paimon","columns":["title","content"],"boost":[2.0,1.0]}}"""),
+ Literal(10)))
+
+ assert(search.columns().size() == 2)
+ assert(search.columns().contains("title"))
+ assert(search.columns().contains("content"))
+ assert(search.query().toJson.contains("\"multi_match\""))
}
- test("reject invalid full-text search query operator") {
- val exception = intercept[IllegalArgumentException] {
+ test("reject full-text search column that does not exist") {
+ val exception = intercept[RuntimeException] {
FullTextSearchQuery(Seq.empty).createFullTextSearch(
innerTable,
- Seq(Literal("content"), Literal("paimon lake"), Literal(10), Literal("xor")))
+ Seq(Literal("""{"match":{"column":"missing","terms":"paimon lake"}}"""), Literal(10)))
}
- assert(exception.getMessage.contains("Query operator must be 'or' or 'and'"))
+ assert(exception.getMessage.contains("Column missing does not exist"))
}
test("reject hybrid search query map") {
@@ -227,8 +238,9 @@ class VectorSearchQueryTest extends AnyFunSuite {
new ArrayType(DataTypes.FLOAT()),
new ArrayType(DataTypes.FLOAT()),
new ArrayType(DataTypes.FLOAT()),
+ DataTypes.STRING(),
DataTypes.STRING()),
- Array[String]("v", "title_vec", "body_vec", "content")
+ Array[String]("v", "title_vec", "body_vec", "title", "content")
)
override def invoke(proxy: Any, method: Method, args: Array[AnyRef]): AnyRef = {
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FullTextSearchTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FullTextSearchTest.scala
index 179cb32645..896a61016a 100644
--- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FullTextSearchTest.scala
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FullTextSearchTest.scala
@@ -91,7 +91,7 @@ class FullTextSearchTest extends PaimonSparkTestBase {
val result = spark
.sql("""
- |SELECT * FROM full_text_search('T', 'content', 'paimon', 5)
+ |SELECT * FROM full_text_search('T', '{"match":{"column":"content","terms":"paimon"}}', 5)
|""".stripMargin)
.collect()
assert(result.length == 5)
@@ -122,7 +122,7 @@ class FullTextSearchTest extends PaimonSparkTestBase {
// Test with k=1
var result = spark
.sql("""
- |SELECT * FROM full_text_search('T', 'content', 'paimon', 1)
+ |SELECT * FROM full_text_search('T', '{"match":{"column":"content","terms":"paimon"}}', 1)
|""".stripMargin)
.collect()
assert(result.length == 1)
@@ -130,7 +130,7 @@ class FullTextSearchTest extends PaimonSparkTestBase {
// Test with k=10
result = spark
.sql("""
- |SELECT * FROM full_text_search('T', 'content', 'paimon', 10)
+ |SELECT * FROM full_text_search('T', '{"match":{"column":"content","terms":"paimon"}}', 10)
|""".stripMargin)
.collect()
assert(result.length == 10)
@@ -164,7 +164,7 @@ class FullTextSearchTest extends PaimonSparkTestBase {
val defaultOrResult = spark
.sql("""
- |SELECT id FROM full_text_search('T', 'content', 'Paimon search', 5)
+ |SELECT id FROM full_text_search('T', '{"match":{"column":"content","terms":"Paimon search"}}', 5)
|ORDER BY id
|""".stripMargin)
.collect()
@@ -173,7 +173,7 @@ class FullTextSearchTest extends PaimonSparkTestBase {
val explicitAndResult = spark
.sql("""
- |SELECT id FROM full_text_search('T', 'content', 'Paimon search', 5, 'and')
+ |SELECT id FROM full_text_search('T', '{"match":{"column":"content","terms":"Paimon search","operator":"And"}}', 5)
|ORDER BY id
|""".stripMargin)
.collect()
@@ -182,6 +182,53 @@ class FullTextSearchTest extends PaimonSparkTestBase {
}
}
+ test("full-text search - structured JSON DSL") {
+ 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, 'Apache Paimon lake format'),
+ | (1, 'Paimon supports full-text search'),
+ | (2, 'full-text search in Apache Paimon'),
+ | (3, 'vector similarity search')
+ |""".stripMargin)
+
+ spark
+ .sql(
+ s"CALL sys.create_global_index(table => 'test.T', index_column => 'content', index_type => '$indexType')")
+ .collect()
+
+ val phraseQuery = """{"phrase":{"column":"content","terms":"full-text search"}}"""
+ val phraseResult = spark
+ .sql(s"""
+ |SELECT id FROM full_text_search('T', '$phraseQuery', 10)
+ |ORDER BY id
+ |""".stripMargin)
+ .collect()
+
+ assert(phraseResult.map(_.getInt(0)).toSeq == Seq(1, 2))
+
+ val booleanQuery =
+ """{"boolean":{"must":[{"match":{"column":"content","terms":"Paimon"}},{"match":{"column":"content","terms":"search"}}],"must_not":[{"match":{"column":"content","terms":"vector"}}]}}"""
+ val booleanResult = spark
+ .sql(s"""
+ |SELECT id FROM full_text_search('T', '$booleanQuery', 10)
+ |ORDER BY id
+ |""".stripMargin)
+ .collect()
+
+ assert(booleanResult.map(_.getInt(0)).toSeq == Seq(1, 2))
+ }
+ }
+
// ========== Integration Tests ==========
test("end-to-end: write, index, search cycle") {
@@ -218,7 +265,7 @@ class FullTextSearchTest extends PaimonSparkTestBase {
val searchResult = spark
.sql("""
- |SELECT id, title FROM full_text_search('T', 'content', 'paimon', 10)
+ |SELECT id, title FROM full_text_search('T', '{"match":{"column":"content","terms":"paimon"}}', 10)
|""".stripMargin)
.collect()
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
index bfc39cad3a..3366a41d50 100644
--- 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
@@ -195,9 +195,7 @@ class HybridSearchTest extends PaimonSparkTestBase {
| 'options', map())),
| array(
| named_struct(
- | 'field', 'content',
- | 'query_text', 'paimon',
- | 'query_operator', 'or',
+ | 'query', '{"match":{"column":"content","terms":"paimon"}}',
| 'limit', 2,
| 'weight', 1.0f,
| 'options', map())),
@@ -244,9 +242,7 @@ class HybridSearchTest extends PaimonSparkTestBase {
| array(),
| array(
| named_struct(
- | 'field', 'content',
- | 'query_text', 'paimon search',
- | 'query_operator', 'or',
+ | 'query', '{"match":{"column":"content","terms":"paimon search"}}',
| 'limit', 1,
| 'weight', 1.0f,
| 'options', map())),
@@ -290,9 +286,7 @@ class HybridSearchTest extends PaimonSparkTestBase {
| array(),
| array(
| named_struct(
- | 'field', 'content',
- | 'query_text', 'paimon',
- | 'query_operator', 'or',
+ | 'query', '{"match":{"column":"content","terms":"paimon"}}',
| 'limit', 1,
| 'weight', 1.0f,
| 'options', map())),
diff --git a/paimon-tantivy/paimon-tantivy-index/README.md b/paimon-tantivy/paimon-tantivy-index/README.md
index 0b93d15875..3378fa6893 100644
--- a/paimon-tantivy/paimon-tantivy-index/README.md
+++ b/paimon-tantivy/paimon-tantivy-index/README.md
@@ -165,7 +165,11 @@ PyPaimon can query `jieba` indexes when the Python `jieba` package is installed.
### Search
```sql
-SELECT * FROM full_text_search('my_table', 'content', 'search query', 10);
+SELECT * FROM full_text_search(
+ 'my_table',
+ '{"match":{"column":"content","terms":"search query"}}',
+ 10
+);
```
### Java API
@@ -174,9 +178,8 @@ SELECT * FROM full_text_search('my_table', 'content', 'search query', 10);
Table table = catalog.getTable(identifier);
GlobalIndexResult result = table.newFullTextSearchBuilder()
- .withQueryText("search query")
+ .withQuery(FullTextQuery.match("search query", "content"))
.withLimit(10)
- .withTextColumn("content")
.executeLocal();
ReadBuilder readBuilder = table.newReadBuilder();
diff --git a/paimon-tantivy/paimon-tantivy-index/src/main/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexReader.java b/paimon-tantivy/paimon-tantivy-index/src/main/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexReader.java
index 67a0762a8e..4c45df6111 100644
--- a/paimon-tantivy/paimon-tantivy-index/src/main/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexReader.java
+++ b/paimon-tantivy/paimon-tantivy-index/src/main/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexReader.java
@@ -94,10 +94,8 @@ public class TantivyFullTextGlobalIndexReader implements GlobalIndexReader {
try {
ensureLoaded();
SearchResult result =
- borrowed.searcher.search(
- fullTextSearch.queryText(),
- fullTextSearch.limit(),
- fullTextSearch.queryOperator());
+ borrowed.searcher.searchJson(
+ fullTextSearch.queryJson(), fullTextSearch.limit());
return Optional.of(toScoredResult(result));
} catch (IOException e) {
throw new RuntimeException("Failed to search Tantivy full-text index", e);
diff --git a/paimon-tantivy/paimon-tantivy-index/src/main/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexerFactory.java b/paimon-tantivy/paimon-tantivy-index/src/main/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexerFactory.java
index 4fc0e9cf6b..641a779a81 100644
--- a/paimon-tantivy/paimon-tantivy-index/src/main/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexerFactory.java
+++ b/paimon-tantivy/paimon-tantivy-index/src/main/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexerFactory.java
@@ -38,6 +38,11 @@ public class TantivyFullTextGlobalIndexerFactory implements GlobalIndexerFactory
return IDENTIFIER;
}
+ @Override
+ public boolean supportsFullTextSearch() {
+ return true;
+ }
+
@Override
public GlobalIndexer create(DataField field, Options options) {
if (searcherPool == null) {
diff --git a/paimon-tantivy/paimon-tantivy-index/src/test/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexTest.java b/paimon-tantivy/paimon-tantivy-index/src/test/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexTest.java
index 5986ce70f3..f259646557 100644
--- a/paimon-tantivy/paimon-tantivy-index/src/test/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexTest.java
+++ b/paimon-tantivy/paimon-tantivy-index/src/test/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexTest.java
@@ -29,6 +29,7 @@ import org.apache.paimon.globalindex.ScoredGlobalIndexResult;
import org.apache.paimon.globalindex.io.GlobalIndexFileReader;
import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
import org.apache.paimon.options.Options;
+import org.apache.paimon.predicate.FullTextQuery;
import org.apache.paimon.predicate.FullTextSearch;
import org.apache.paimon.tantivy.NativeLoader;
import org.apache.paimon.utils.RoaringNavigableMap64;
@@ -137,7 +138,7 @@ public class TantivyFullTextGlobalIndexTest {
GlobalIndexFileReader fileReader = createFileReader();
try (TantivyFullTextGlobalIndexReader reader = createReader(fileReader, metas)) {
- FullTextSearch search = new FullTextSearch("paimon", 10, "text");
+ FullTextSearch search = new FullTextSearch(FullTextQuery.match("paimon", "text"), 10);
Optional<ScoredGlobalIndexResult> searchResult =
reader.visitFullTextSearch(search).join();
assertThat(searchResult).isPresent();
@@ -206,7 +207,7 @@ public class TantivyFullTextGlobalIndexTest {
GlobalIndexFileReader fileReader = createFileReader();
try (TantivyFullTextGlobalIndexReader reader = createReader(fileReader, metas)) {
- FullTextSearch search = new FullTextSearch("售货员", 10, "text");
+ FullTextSearch search = new FullTextSearch(FullTextQuery.match("售货员", "text"), 10);
Optional<ScoredGlobalIndexResult> searchResult =
reader.visitFullTextSearch(search).join();
assertThat(searchResult).isPresent();
@@ -230,7 +231,8 @@ public class TantivyFullTextGlobalIndexTest {
GlobalIndexFileReader fileReader = createFileReader();
try (TantivyFullTextGlobalIndexReader reader = createReader(fileReader, metas)) {
- FullTextSearch search = new FullTextSearch("nonexistent", 10, "text");
+ FullTextSearch search =
+ new FullTextSearch(FullTextQuery.match("nonexistent", "text"), 10);
Optional<ScoredGlobalIndexResult> searchResult =
reader.visitFullTextSearch(search).join();
assertThat(searchResult).isPresent();
@@ -256,7 +258,7 @@ public class TantivyFullTextGlobalIndexTest {
GlobalIndexFileReader fileReader = createFileReader();
try (TantivyFullTextGlobalIndexReader reader = createReader(fileReader, metas)) {
- FullTextSearch search = new FullTextSearch("paimon", 10, "text");
+ FullTextSearch search = new FullTextSearch(FullTextQuery.match("paimon", "text"), 10);
Optional<ScoredGlobalIndexResult> searchResult =
reader.visitFullTextSearch(search).join();
assertThat(searchResult).isPresent();
@@ -302,7 +304,8 @@ public class TantivyFullTextGlobalIndexTest {
try (TantivyFullTextGlobalIndexReader reader = createReader(fileReader, metas)) {
// Search for the special keyword — should match every 10th doc
- FullTextSearch search = new FullTextSearch("special_keyword", 1000, "text");
+ FullTextSearch search =
+ new FullTextSearch(FullTextQuery.match("special_keyword", "text"), 1000);
Optional<ScoredGlobalIndexResult> searchResult =
reader.visitFullTextSearch(search).join();
assertThat(searchResult).isPresent();
@@ -332,7 +335,7 @@ public class TantivyFullTextGlobalIndexTest {
try (TantivyFullTextGlobalIndexReader reader = createReader(fileReader, metas)) {
// Limit to 5 results
- FullTextSearch search = new FullTextSearch("paimon", 5, "text");
+ FullTextSearch search = new FullTextSearch(FullTextQuery.match("paimon", "text"), 5);
Optional<ScoredGlobalIndexResult> searchResult =
reader.visitFullTextSearch(search).join();
assertThat(searchResult).isPresent();
@@ -352,7 +355,7 @@ public class TantivyFullTextGlobalIndexTest {
List<ResultEntry> results = writer.finish();
List<GlobalIndexIOMeta> metas = toIOMetas(results, indexPath);
GlobalIndexFileReader fileReader = createFileReader();
- FullTextSearch search = new FullTextSearch("paimon", 10, "text");
+ FullTextSearch search = new FullTextSearch(FullTextQuery.match("paimon", "text"), 10);
// First query: pool miss, searcher is loaded and returned to pool on close.
try (TantivyFullTextGlobalIndexReader reader = createReader(fileReader, metas)) {
@@ -389,7 +392,7 @@ public class TantivyFullTextGlobalIndexTest {
try (TantivyFullTextGlobalIndexReader reader =
(TantivyFullTextGlobalIndexReader)
indexer.createReader(fileReader, metas, newDirectExecutorService())) {
- FullTextSearch search = new FullTextSearch("indexer", 10, "text");
+ FullTextSearch search = new FullTextSearch(FullTextQuery.match("indexer", "text"), 10);
Optional<ScoredGlobalIndexResult> searchResult =
reader.visitFullTextSearch(search).join();
assertThat(searchResult).isPresent();
@@ -397,4 +400,34 @@ public class TantivyFullTextGlobalIndexTest {
assertThat(searchResult.get().results().contains(0L)).isTrue();
}
}
+
+ @Test
+ public void testStructuredBoostQueryDemotesNegativeMatches() throws IOException {
+ GlobalIndexFileWriter fileWriter = createFileWriter(indexPath);
+ TantivyFullTextGlobalIndexWriter writer = new TantivyFullTextGlobalIndexWriter(fileWriter);
+
+ writer.write(BinaryString.fromString("paimon lake"), 0);
+ writer.write(BinaryString.fromString("paimon vector"), 1);
+
+ List<ResultEntry> results = writer.finish();
+ List<GlobalIndexIOMeta> metas = toIOMetas(results, indexPath);
+ GlobalIndexFileReader fileReader = createFileReader();
+
+ try (TantivyFullTextGlobalIndexReader reader = createReader(fileReader, metas)) {
+ FullTextSearch search =
+ new FullTextSearch(
+ FullTextQuery.boost(
+ FullTextQuery.match("paimon", "text"),
+ FullTextQuery.match("vector", "text"),
+ 0.5f),
+ 10);
+ Optional<ScoredGlobalIndexResult> searchResult =
+ reader.visitFullTextSearch(search).join();
+ assertThat(searchResult).isPresent();
+ assertThat(searchResult.get().results().contains(0L)).isTrue();
+ assertThat(searchResult.get().results().contains(1L)).isTrue();
+ assertThat(searchResult.get().scoreGetter().score(0L))
+ .isGreaterThan(searchResult.get().scoreGetter().score(1L));
+ }
+ }
}
diff --git a/paimon-tantivy/paimon-tantivy-jni/rust/src/lib.rs b/paimon-tantivy/paimon-tantivy-jni/rust/src/lib.rs
index ef8983837d..7f111efcf8 100644
--- a/paimon-tantivy/paimon-tantivy-jni/rust/src/lib.rs
+++ b/paimon-tantivy/paimon-tantivy-jni/rust/src/lib.rs
@@ -20,10 +20,12 @@ mod jni_directory;
use jni::objects::{JClass, JObject, JString, JValue};
use jni::sys::{jboolean, jfloat, jint, jlong, jobject};
use jni::JNIEnv;
-use serde::Deserialize;
+use serde::de::{Error as DeError, SeqAccess, Visitor};
+use serde::{Deserialize, Deserializer};
+use std::fmt;
use std::ptr;
use tantivy::collector::TopDocs;
-use tantivy::query::QueryParser;
+use tantivy::query::{BooleanQuery, BoostQuery, Occur, Query, QueryParser};
use tantivy::schema::{
Field, IndexRecordOption, NumericOptions, Schema, TextFieldIndexing, TextOptions,
};
@@ -32,7 +34,7 @@ use tantivy::tokenizer::{
SimpleTokenizer, Stemmer, StopWordFilter, TextAnalyzer, TextAnalyzerBuilder,
WhitespaceTokenizer,
};
-use tantivy::{Index, IndexReader, IndexWriter, ReloadPolicy};
+use tantivy::{DocAddress, Index, IndexReader, IndexWriter, ReloadPolicy, Score};
use tantivy_jieba::JiebaTokenizer;
use crate::jni_directory::JniDirectory;
@@ -61,6 +63,188 @@ struct TantivySearcherHandle {
text_field: Field,
}
+#[derive(Deserialize)]
+#[serde(rename_all = "snake_case")]
+enum FullTextQueryJson {
+ Match(MatchQueryJson),
+ #[serde(alias = "match_phrase")]
+ Phrase(PhraseQueryJson),
+ Boost(BoostQueryJson),
+ Boolean(BooleanQueryJson),
+ #[allow(dead_code)]
+ MultiMatch(serde_json::Value),
+}
+
+#[derive(Deserialize)]
+struct MatchQueryJson {
+ #[allow(dead_code)]
+ column: Option<String>,
+ #[serde(alias = "query")]
+ terms: String,
+ #[serde(default = "default_boost")]
+ boost: f32,
+ #[serde(default = "default_fuzziness")]
+ #[serde(deserialize_with = "deserialize_fuzziness")]
+ fuzziness: Option<u8>,
+ #[serde(default = "default_max_expansions")]
+ #[serde(alias = "maxExpansions")]
+ max_expansions: usize,
+ #[serde(default)]
+ operator: QueryOperatorJson,
+ #[serde(default)]
+ #[serde(alias = "prefixLength")]
+ prefix_length: u32,
+}
+
+#[derive(Deserialize)]
+struct PhraseQueryJson {
+ #[allow(dead_code)]
+ column: Option<String>,
+ #[serde(alias = "query")]
+ terms: String,
+ #[serde(default)]
+ slop: u32,
+}
+
+#[derive(Deserialize)]
+struct BoostQueryJson {
+ positive: Box<FullTextQueryJson>,
+ negative: Box<FullTextQueryJson>,
+ #[serde(default = "default_negative_boost")]
+ #[serde(alias = "negativeBoost")]
+ negative_boost: f32,
+}
+
+#[derive(Deserialize)]
+struct BooleanQueryJson {
+ #[serde(default)]
+ should: Vec<FullTextQueryJson>,
+ #[serde(default)]
+ must: Vec<FullTextQueryJson>,
+ #[serde(default)]
+ must_not: Vec<FullTextQueryJson>,
+ #[serde(default)]
+ queries: Vec<BooleanClauseJson>,
+}
+
+#[derive(Deserialize)]
+#[serde(untagged)]
+enum BooleanClauseJson {
+ Tuple(BooleanClauseTupleJson),
+ Object {
+ occur: BooleanOccurJson,
+ query: FullTextQueryJson,
+ },
+}
+
+struct BooleanClauseTupleJson(BooleanOccurJson, FullTextQueryJson);
+
+impl<'de> Deserialize<'de> for BooleanClauseTupleJson {
+ fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+ where
+ D: Deserializer<'de>,
+ {
+ struct TupleVisitor;
+
+ impl<'de> Visitor<'de> for TupleVisitor {
+ type Value = BooleanClauseTupleJson;
+
+ fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
+ formatter.write_str("a two-element boolean clause tuple")
+ }
+
+ fn visit_seq<A>(self, mut seq: A) -> Result<Self::Value, A::Error>
+ where
+ A: SeqAccess<'de>,
+ {
+ let occur = seq
+ .next_element()?
+ .ok_or_else(|| DeError::invalid_length(0, &self))?;
+ let query = seq
+ .next_element()?
+ .ok_or_else(|| DeError::invalid_length(1, &self))?;
+ Ok(BooleanClauseTupleJson(occur, query))
+ }
+ }
+
+ deserializer.deserialize_tuple(2, TupleVisitor)
+ }
+}
+
+#[derive(Clone, Copy, Deserialize)]
+#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
+enum BooleanOccurJson {
+ Should,
+ Must,
+ MustNot,
+}
+
+enum QueryOperatorJson {
+ Or,
+ And,
+}
+
+impl<'de> Deserialize<'de> for QueryOperatorJson {
+ fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+ where
+ D: Deserializer<'de>,
+ {
+ let value = String::deserialize(deserializer)?;
+ match value.as_str() {
+ "Or" | "or" | "OR" => Ok(Self::Or),
+ "And" | "and" | "AND" => Ok(Self::And),
+ _ => Err(DeError::custom(format!(
+ "invalid full-text query operator: {}",
+ value
+ ))),
+ }
+ }
+}
+
+impl Default for QueryOperatorJson {
+ fn default() -> Self {
+ Self::Or
+ }
+}
+
+fn default_boost() -> f32 {
+ 1.0
+}
+
+fn default_fuzziness() -> Option<u8> {
+ Some(0)
+}
+
+fn default_negative_boost() -> f32 {
+ 0.5
+}
+
+fn default_max_expansions() -> usize {
+ 50
+}
+
+fn deserialize_fuzziness<'de, D>(deserializer: D) -> Result<Option<u8>, D::Error>
+where
+ D: Deserializer<'de>,
+{
+ let value = serde_json::Value::deserialize(deserializer)?;
+ if value.is_null() {
+ return Ok(None);
+ }
+ if let Some(text) = value.as_str() {
+ if text.eq_ignore_ascii_case("auto") {
+ return Ok(None);
+ }
+ }
+ if let Some(number) = value.as_u64() {
+ return Ok(Some(number as u8));
+ }
+ Err(serde::de::Error::custom(format!(
+ "invalid match query fuzziness: {}",
+ value
+ )))
+}
+
#[derive(Clone, Deserialize)]
#[serde(default)]
struct TokenizerConfig {
@@ -298,6 +482,340 @@ fn build_schema(config: &TokenizerConfig) -> (Schema, Field, Field) {
(builder.build(), row_id_field, text_field)
}
+fn query_parser_for(
+ searcher: &tantivy::Searcher,
+ field: Field,
+ operator: QueryOperatorJson,
+ fuzziness: Option<u8>,
+ prefix_length: u32,
+) -> QueryParser {
+ let mut query_parser = QueryParser::for_index(&searcher.index(), vec![field]);
+ if matches!(operator, QueryOperatorJson::And) {
+ query_parser.set_conjunction_by_default();
+ }
+ if let Some(distance) = fuzziness {
+ if distance > 0 {
+ query_parser.set_field_fuzzy(field, prefix_length > 0, distance, true);
+ }
+ }
+ query_parser
+}
+
+fn build_query_from_json(
+ searcher: &tantivy::Searcher,
+ field: Field,
+ query_json: FullTextQueryJson,
+) -> Result<Box<dyn Query>, String> {
+ match query_json {
+ FullTextQueryJson::Match(query) => {
+ if query.terms.is_empty() {
+ return Err("match query terms cannot be empty".to_string());
+ }
+ if query.boost <= 0.0 {
+ return Err(format!(
+ "match query boost must be positive, got {}",
+ query.boost
+ ));
+ }
+ if query.max_expansions == 0 {
+ return Err("match query max_expansions must be positive".to_string());
+ }
+ if let Some(distance) = query.fuzziness {
+ if distance > 2 {
+ return Err(format!(
+ "match query fuzziness must be <= 2 for Tantivy, got {}",
+ distance
+ ));
+ }
+ }
+ if query.max_expansions != default_max_expansions() {
+ return Err(
+ "match query max_expansions is not supported by Tantivy 0.22".to_string(),
+ );
+ }
+ if query.prefix_length != 0 {
+ return Err(
+ "match query prefix_length is not supported by Tantivy 0.22".to_string()
+ );
+ }
+ let parser = query_parser_for(searcher, field, query.operator, query.fuzziness, 0);
+ let parsed = parser
+ .parse_query(&query.terms)
+ .map_err(|e| format!("Failed to parse match query '{}': {}", query.terms, e))?;
+ if (query.boost - 1.0).abs() > f32::EPSILON {
+ Ok(Box::new(BoostQuery::new(parsed, query.boost)))
+ } else {
+ Ok(parsed)
+ }
+ }
+ FullTextQueryJson::Phrase(query) => {
+ if query.terms.is_empty() {
+ return Err("phrase query terms cannot be empty".to_string());
+ }
+ let parser = query_parser_for(searcher, field, QueryOperatorJson::Or, None, 0);
+ let escaped = query.terms.replace('\\', "\\\\").replace('"', "\\\"");
+ let query_string = if query.slop == 0 {
+ format!("\"{}\"", escaped)
+ } else {
+ format!("\"{}\"~{}", escaped, query.slop)
+ };
+ parser
+ .parse_query(&query_string)
+ .map_err(|e| format!("Failed to parse phrase query '{}': {}", query.terms, e))
+ }
+ FullTextQueryJson::Boost(query) => build_boost_query_from_json(searcher, field, query),
+ FullTextQueryJson::Boolean(query) => {
+ let mut subqueries: Vec<(Occur, Box<dyn Query>)> = Vec::new();
+ for child in query.should {
+ subqueries.push((
+ Occur::Should,
+ build_query_from_json(searcher, field, child)?,
+ ));
+ }
+ for child in query.must {
+ subqueries.push((Occur::Must, build_query_from_json(searcher, field, child)?));
+ }
+ for child in query.must_not {
+ subqueries.push((
+ Occur::MustNot,
+ build_query_from_json(searcher, field, child)?,
+ ));
+ }
+ for clause in query.queries {
+ let (occur, child) = match clause {
+ BooleanClauseJson::Tuple(BooleanClauseTupleJson(occur, child)) => {
+ (occur, child)
+ }
+ BooleanClauseJson::Object { occur, query } => (occur, query),
+ };
+ let occur = match occur {
+ BooleanOccurJson::Should => Occur::Should,
+ BooleanOccurJson::Must => Occur::Must,
+ BooleanOccurJson::MustNot => Occur::MustNot,
+ };
+ subqueries.push((occur, build_query_from_json(searcher, field, child)?));
+ }
+ if subqueries.is_empty() {
+ return Err("boolean query must contain at least one clause".to_string());
+ }
+ Ok(Box::new(BooleanQuery::new(subqueries)))
+ }
+ FullTextQueryJson::MultiMatch(_) => Err(
+ "multi_match is not supported by single-column Tantivy full-text indexes".to_string(),
+ ),
+ }
+}
+
+struct DemoteQuery {
+ positive: Box<dyn Query>,
+ negative: Box<dyn Query>,
+ negative_boost: f32,
+}
+
+impl Clone for DemoteQuery {
+ fn clone(&self) -> Self {
+ Self {
+ positive: self.positive.box_clone(),
+ negative: self.negative.box_clone(),
+ negative_boost: self.negative_boost,
+ }
+ }
+}
+
+impl std::fmt::Debug for DemoteQuery {
+ fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ formatter
+ .debug_struct("DemoteQuery")
+ .field("negative_boost", &self.negative_boost)
+ .finish()
+ }
+}
+
+impl Query for DemoteQuery {
+ fn weight(
+ &self,
+ enable_scoring: tantivy::query::EnableScoring<'_>,
+ ) -> tantivy::Result<Box<dyn tantivy::query::Weight>> {
+ let positive = self.positive.weight(enable_scoring.clone())?;
+ let negative = self.negative.weight(enable_scoring)?;
+ Ok(Box::new(DemoteWeight {
+ positive,
+ negative,
+ negative_boost: self.negative_boost,
+ }))
+ }
+}
+
+struct DemoteWeight {
+ positive: Box<dyn tantivy::query::Weight>,
+ negative: Box<dyn tantivy::query::Weight>,
+ negative_boost: f32,
+}
+
+impl tantivy::query::Weight for DemoteWeight {
+ fn scorer(
+ &self,
+ reader: &tantivy::SegmentReader,
+ boost: Score,
+ ) -> tantivy::Result<Box<dyn tantivy::query::Scorer>> {
+ let positive = self.positive.scorer(reader, boost)?;
+ let negative = self.negative.scorer(reader, boost)?;
+ Ok(Box::new(DemoteScorer {
+ positive,
+ negative,
+ negative_boost: self.negative_boost,
+ }))
+ }
+
+ fn explain(
+ &self,
+ reader: &tantivy::SegmentReader,
+ doc: tantivy::DocId,
+ ) -> tantivy::Result<tantivy::query::Explanation> {
+ self.positive.explain(reader, doc)
+ }
+}
+
+struct DemoteScorer {
+ positive: Box<dyn tantivy::query::Scorer>,
+ negative: Box<dyn tantivy::query::Scorer>,
+ negative_boost: f32,
+}
+
+impl tantivy::DocSet for DemoteScorer {
+ fn advance(&mut self) -> tantivy::DocId {
+ self.positive.advance()
+ }
+
+ fn seek(&mut self, target: tantivy::DocId) -> tantivy::DocId {
+ self.positive.seek(target)
+ }
+
+ fn doc(&self) -> tantivy::DocId {
+ self.positive.doc()
+ }
+
+ fn size_hint(&self) -> u32 {
+ self.positive.size_hint()
+ }
+}
+
+impl tantivy::query::Scorer for DemoteScorer {
+ fn score(&mut self) -> Score {
+ let mut score = self.positive.score();
+ let doc = self.positive.doc();
+ if self.negative.doc() < doc {
+ self.negative.seek(doc);
+ }
+ if self.negative.doc() == doc {
+ score *= self.negative_boost;
+ }
+ score
+ }
+}
+
+fn build_boost_query_from_json(
+ searcher: &tantivy::Searcher,
+ field: Field,
+ query: BoostQueryJson,
+) -> Result<Box<dyn Query>, String> {
+ if query.negative_boost <= 0.0 {
+ return Err(format!(
+ "boost query negative_boost must be positive, got {}",
+ query.negative_boost
+ ));
+ }
+ Ok(Box::new(DemoteQuery {
+ positive: build_query_from_json(searcher, field, *query.positive)?,
+ negative: build_query_from_json(searcher, field, *query.negative)?,
+ negative_boost: query.negative_boost,
+ }))
+}
+
+fn search_with_query(
+ env: &mut JNIEnv,
+ handle: &TantivySearcherHandle,
+ query: Box<dyn Query>,
+ limit: jint,
+) -> jobject {
+ if limit <= 0 {
+ return throw_and_return_null(env, &format!("Limit must be positive, got: {}", limit));
+ }
+ let searcher = handle.reader.searcher();
+ let top_docs = match searcher.search(&query, &TopDocs::with_limit(limit as usize)) {
+ Ok(d) => d,
+ Err(e) => return throw_and_return_null(env, &format!("Search failed: {}", e)),
+ };
+
+ build_search_result(env, &searcher, &top_docs)
+}
+
+fn build_search_result(
+ env: &mut JNIEnv,
+ searcher: &tantivy::Searcher,
+ top_docs: &[(Score, DocAddress)],
+) -> jobject {
+ let count = top_docs.len();
+
+ let row_id_array = match env.new_long_array(count as i32) {
+ Ok(a) => a,
+ Err(e) => {
+ return throw_and_return_null(env, &format!("Failed to create long array: {}", e))
+ }
+ };
+ let score_array = match env.new_float_array(count as i32) {
+ Ok(a) => a,
+ Err(e) => {
+ return throw_and_return_null(env, &format!("Failed to create float array: {}", e))
+ }
+ };
+
+ let mut row_ids: Vec<jlong> = Vec::with_capacity(count);
+ let mut scores: Vec<jfloat> = Vec::with_capacity(count);
+
+ for (score, doc_address) in top_docs {
+ let segment_reader = searcher.segment_reader(doc_address.segment_ord);
+ let fast_fields = match segment_reader.fast_fields().u64("row_id") {
+ Ok(f) => f,
+ Err(e) => {
+ return throw_and_return_null(env, &format!("Failed to get fast field: {}", e))
+ }
+ };
+ let row_id = fast_fields.first(doc_address.doc_id).unwrap_or(0) as jlong;
+ row_ids.push(row_id);
+ scores.push(*score as jfloat);
+ }
+
+ if let Err(e) = env.set_long_array_region(&row_id_array, 0, &row_ids) {
+ return throw_and_return_null(env, &format!("Failed to set long array: {}", e));
+ }
+ if let Err(e) = env.set_float_array_region(&score_array, 0, &scores) {
+ return throw_and_return_null(env, &format!("Failed to set float array: {}", e));
+ }
+
+ let class = match env.find_class("org/apache/paimon/tantivy/SearchResult") {
+ Ok(c) => c,
+ Err(e) => {
+ return throw_and_return_null(env, &format!("Failed to find SearchResult class: {}", e))
+ }
+ };
+ let obj = match env.new_object(
+ class,
+ "([J[F)V",
+ &[
+ JValue::Object(&JObject::from(row_id_array)),
+ JValue::Object(&JObject::from(score_array)),
+ ],
+ ) {
+ Ok(o) => o,
+ Err(e) => {
+ return throw_and_return_null(env, &format!("Failed to create SearchResult: {}", e))
+ }
+ };
+
+ obj.into_raw()
+}
+
fn create_index_internal(mut env: JNIEnv, index_path: JString, config: TokenizerConfig) -> jlong {
let path: String = match env.get_string(&index_path) {
Ok(s) => s.into(),
@@ -727,79 +1245,39 @@ pub extern "system" fn Java_org_apache_paimon_tantivy_TantivySearcher_searchInde
)
}
};
- let top_docs = match searcher.search(&query, &TopDocs::with_limit(limit as usize)) {
- Ok(d) => d,
- Err(e) => return throw_and_return_null(&mut env, &format!("Search failed: {}", e)),
- };
-
- let count = top_docs.len();
+ search_with_query(&mut env, handle, query, limit)
+}
- // Build Java long[] and float[]
- let row_id_array = match env.new_long_array(count as i32) {
- Ok(a) => a,
- Err(e) => {
- return throw_and_return_null(&mut env, &format!("Failed to create long array: {}", e))
- }
- };
- let score_array = match env.new_float_array(count as i32) {
- Ok(a) => a,
+#[no_mangle]
+pub extern "system" fn Java_org_apache_paimon_tantivy_TantivySearcher_searchIndexJson(
+ mut env: JNIEnv,
+ _class: JClass,
+ searcher_ptr: jlong,
+ query_json: JString,
+ limit: jint,
+) -> jobject {
+ let handle = unsafe { &*(searcher_ptr as *const TantivySearcherHandle) };
+ let query_json_str: String = match env.get_string(&query_json) {
+ Ok(s) => s.into(),
Err(e) => {
- return throw_and_return_null(&mut env, &format!("Failed to create float array: {}", e))
+ return throw_and_return_null(&mut env, &format!("Failed to get query JSON: {}", e))
}
};
-
- let mut row_ids: Vec<jlong> = Vec::with_capacity(count);
- let mut scores: Vec<jfloat> = Vec::with_capacity(count);
-
- // Use fast field reader for efficient row_id retrieval
- for (score, doc_address) in &top_docs {
- let segment_reader = searcher.segment_reader(doc_address.segment_ord);
- let fast_fields = match segment_reader.fast_fields().u64("row_id") {
- Ok(f) => f,
- Err(e) => {
- return throw_and_return_null(&mut env, &format!("Failed to get fast field: {}", e))
- }
- };
- let row_id = fast_fields.first(doc_address.doc_id).unwrap_or(0) as jlong;
- row_ids.push(row_id);
- scores.push(*score as jfloat);
- }
-
- if let Err(e) = env.set_long_array_region(&row_id_array, 0, &row_ids) {
- return throw_and_return_null(&mut env, &format!("Failed to set long array: {}", e));
- }
- if let Err(e) = env.set_float_array_region(&score_array, 0, &scores) {
- return throw_and_return_null(&mut env, &format!("Failed to set float array: {}", e));
- }
-
- // Construct SearchResult object
- let class = match env.find_class("org/apache/paimon/tantivy/SearchResult") {
- Ok(c) => c,
+ let query_json = match serde_json::from_str::<FullTextQueryJson>(&query_json_str) {
+ Ok(q) => q,
Err(e) => {
return throw_and_return_null(
&mut env,
- &format!("Failed to find SearchResult class: {}", e),
+ &format!("Failed to parse query JSON '{}': {}", query_json_str, e),
)
}
};
- let obj = match env.new_object(
- class,
- "([J[F)V",
- &[
- JValue::Object(&JObject::from(row_id_array)),
- JValue::Object(&JObject::from(score_array)),
- ],
- ) {
- Ok(o) => o,
- Err(e) => {
- return throw_and_return_null(
- &mut env,
- &format!("Failed to create SearchResult: {}", e),
- )
- }
+ let searcher = handle.reader.searcher();
+ let query = match build_query_from_json(&searcher, handle.text_field, query_json) {
+ Ok(q) => q,
+ Err(e) => return throw_and_return_null(&mut env, &e),
};
-
- obj.into_raw()
+ search_with_query(&mut env, handle, query, limit)
}
#[no_mangle]
diff --git a/paimon-tantivy/paimon-tantivy-jni/src/main/java/org/apache/paimon/tantivy/TantivySearcher.java b/paimon-tantivy/paimon-tantivy-jni/src/main/java/org/apache/paimon/tantivy/TantivySearcher.java
index bc04632bb0..8962e5b99b 100644
--- a/paimon-tantivy/paimon-tantivy-jni/src/main/java/org/apache/paimon/tantivy/TantivySearcher.java
+++ b/paimon-tantivy/paimon-tantivy-jni/src/main/java/org/apache/paimon/tantivy/TantivySearcher.java
@@ -121,6 +121,11 @@ public class TantivySearcher implements AutoCloseable {
return searchIndex(searcherPtr, queryString, limit, queryOperator);
}
+ public SearchResult searchJson(String queryJson, int limit) {
+ checkNotClosed();
+ return searchIndexJson(searcherPtr, queryJson, limit);
+ }
+
@Override
public void close() {
if (!closed) {
@@ -178,6 +183,8 @@ public class TantivySearcher implements AutoCloseable {
static native SearchResult searchIndex(
long searcherPtr, String queryString, int limit, String queryOperator);
+ static native SearchResult searchIndexJson(long searcherPtr, String queryJson, int limit);
+
static native void freeSearcher(long searcherPtr);
}