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 4fbb308aff [spark][flink][docs] Expose primary-key full-text search
(#8659)
4fbb308aff is described below
commit 4fbb308aff4f1911908328434e747e2af76d919a
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Jul 15 22:25:11 2026 +0800
[spark][flink][docs] Expose primary-key full-text search (#8659)
Expose primary-key full-text search through Spark SQL, a Flink
procedure, and the Java search path. Hybrid search can now fuse
primary-key Vector and Full Text results against one snapshot while
preserving physical-position semantics.
---
docs/docs/flink/procedures.md | 29 ++
docs/docs/primary-key-table/global-index.mdx | 165 ++++++++++-
.../table/source/FullTextSearchBuilderImpl.java | 12 +-
.../table/source/HybridSearchBuilderImpl.java | 183 +++++++++++-
.../table/source/PrimaryKeyFullTextScan.java | 37 ++-
.../table/source/PrimaryKeyFullTextScanTest.java | 5 +-
.../table/source/PrimaryKeyFullTextSearchTest.java | 21 +-
.../table/source/PrimaryKeyHybridSearchTest.java | 185 ++++++++++++
.../flink/procedure/FullTextSearchProcedure.java | 322 +++++++++++++++++++++
.../services/org.apache.paimon.factories.Factory | 1 +
.../procedure/FullTextSearchProcedureITCase.java | 150 ++++++++++
...TestPrimaryKeyFullTextGlobalIndexerFactory.java | 30 ++
....apache.paimon.globalindex.GlobalIndexerFactory | 16 +
paimon-full-text/README.md | 51 +++-
...TestPrimaryKeyFullTextGlobalIndexerFactory.java | 30 ++
....apache.paimon.globalindex.GlobalIndexerFactory | 16 +
.../paimon/spark/sql/FullTextSearchTest.scala | 46 +++
.../apache/paimon/spark/sql/HybridSearchTest.scala | 58 ++++
18 files changed, 1319 insertions(+), 38 deletions(-)
diff --git a/docs/docs/flink/procedures.md b/docs/docs/flink/procedures.md
index a8ab2abb65..eb91819554 100644
--- a/docs/docs/flink/procedures.md
+++ b/docs/docs/flink/procedures.md
@@ -1140,6 +1140,35 @@ All available procedures are listed below.
`dry_run` => true)
</td>
</tr>
+ <tr>
+ <td>full_text_search</td>
+ <td>
+ CALL [catalog.]sys.full_text_search(<br/>
+ `table` => 'identifier',<br/>
+ `column` => 'columnName',<br/>
+ query => 'queryJson',<br/>
+ top_k => topK,<br/>
+ projection => 'col1,col2,__paimon_search_score',<br/>
+ options => 'key1=value1;key2=value2')<br/>
+ </td>
+ <td>
+ To perform full-text search on a table and return deterministically
ordered JSON rows. Arguments:
+ <li>table(required): the target table identifier.</li>
+ <li>column(required): the character column to search.</li>
+ <li>query(required): native full-text query JSON.</li>
+ <li>top_k(required): the maximum number of results, from 1 through
10,000.</li>
+ <li>projection(optional): comma-separated result columns. Add
<code>__paimon_search_score</code> to return the search relevance score.</li>
+ <li>options(optional): additional dynamic options of the table.
The query authorization setting cannot be overridden.</li>
+ </td>
+ <td>
+ CALL sys.full_text_search(<br/>
+ `table` => 'default.articles',<br/>
+ `column` => 'content',<br/>
+ query => '{"match":{"column":"content","terms":"paimon
lake"}}',<br/>
+ top_k => 10,<br/>
+ projection => 'id,content,__paimon_search_score')
+ </td>
+ </tr>
<tr>
<td>vector_search</td>
<td>
diff --git a/docs/docs/primary-key-table/global-index.mdx
b/docs/docs/primary-key-table/global-index.mdx
index 6dd0c5bc8e..a8f3dea2f0 100644
--- a/docs/docs/primary-key-table/global-index.mdx
+++ b/docs/docs/primary-key-table/global-index.mdx
@@ -27,8 +27,8 @@ under the License.
# Primary-Key Indexes
-Primary-key tables can maintain Vector, BTree, and Bitmap indexes together
with compact data
-files. These indexes are bucket-local and source-backed: every index group
records its source data
+Primary-key tables can maintain Vector, Full Text, BTree, and Bitmap indexes
together with compact
+data files. These indexes are bucket-local and source-backed: every index
group records its source data
files and maps matches back to physical row positions. Deletion vectors are
applied when indexed
rows are read, so updates and deletes remain exact.
@@ -73,10 +73,22 @@ For an append-only or Data Evolution table whose vector
index is built independe
</TabItem>
+<TabItem value="full-text" label="Full Text">
+
+Use Full Text for native ranked search on `CHAR`, `VARCHAR`, or `STRING`
content that can be
+updated or deleted with the primary-key table. The fixed `full-text`
implementation is selected
+automatically. Search is exposed through Spark SQL, a Flink procedure, and the
Java API.
+
+For an append-only or Data Evolution table whose full-text index is built
independently, see
+[Global Index](../multimodal-table/global-index).
+
+</TabItem>
+
</Tabs>
Different columns in one table can use different index families. One column
can occur in at most
-one of `pk-vector.index.columns`, `pk-btree.index.columns`, and
`pk-bitmap.index.columns`.
+one of `pk-vector.index.columns`, `pk-full-text.index.columns`,
`pk-btree.index.columns`, and
+`pk-bitmap.index.columns`.
## Requirements
@@ -118,12 +130,27 @@ Exactly one vector column is currently supported per
table.
</TabItem>
+<TabItem value="full-text" label="Full Text">
+
+Full Text additionally requires:
+
+- A `CHAR`, `VARCHAR`, or `STRING` column.
+- A merge engine of `deduplicate`, `partial-update`, `aggregation`, or
`first-row`.
+- `deletion-vectors.enabled = true`, except for `first-row`, where it must be
`false`.
+- `deletion-vectors.merge-on-read = false`.
+- The `paimon-full-text` module and its native implementation on every writer,
compactor, and
+ reader classpath.
+
+Exactly one full-text column is currently supported per table.
+
+</TabItem>
+
</Tabs>
## Create a Table
-The following table uses all three families on different columns: Vector for
`embedding`, BTree
-for `amount`, and Bitmap for `status`.
+The following table uses all four families on different columns: Vector for
`embedding`, Full
+Text for `content`, BTree for `amount`, and Bitmap for `status`.
<Tabs groupId="primary-key-index-create-table">
@@ -134,6 +161,7 @@ CREATE TABLE items (
id BIGINT,
status STRING,
amount DECIMAL(12, 2),
+ content STRING,
embedding ARRAY<FLOAT> COMMENT '__VECTOR_FIELD;3',
PRIMARY KEY (id) NOT ENFORCED
) WITH (
@@ -143,6 +171,8 @@ CREATE TABLE items (
'fields.embedding.pk-vector.index.type' = 'ivf-flat',
'fields.embedding.pk-vector.distance.metric' = 'cosine',
'fields.embedding.pk-vector.index.options' = '{"nlist":"256"}',
+ 'pk-full-text.index.columns' = 'content',
+ 'fields.content.pk-full-text.index.options' = '{"tokenizer":"jieba"}',
'pk-btree.index.columns' = 'amount',
'fields.amount.pk-btree.index.options' = '{"block-size":"64 kb"}',
'pk-bitmap.index.columns' = 'status',
@@ -163,6 +193,7 @@ CREATE TABLE items (
id BIGINT,
status STRING,
amount DECIMAL(12, 2),
+ content STRING,
embedding ARRAY<FLOAT> COMMENT '__VECTOR_FIELD;3'
) USING paimon
TBLPROPERTIES (
@@ -173,6 +204,8 @@ TBLPROPERTIES (
'fields.embedding.pk-vector.index.type' = 'ivf-flat',
'fields.embedding.pk-vector.distance.metric' = 'cosine',
'fields.embedding.pk-vector.index.options' = '{"nlist":"256"}',
+ 'pk-full-text.index.columns' = 'content',
+ 'fields.content.pk-full-text.index.options' = '{"tokenizer":"jieba"}',
'pk-btree.index.columns' = 'amount',
'fields.amount.pk-btree.index.options' = '{"block-size":"64 kb"}',
'pk-bitmap.index.columns' = 'status',
@@ -195,17 +228,20 @@ schema validation.
| `fields.<column>.pk-vector.index.type` | Required | ANN implementation, such
as `ivf-flat`, `ivf-pq`, `ivf-hnsw-flat`, `ivf-hnsw-sq`, or `lumina`. |
| `fields.<column>.pk-vector.distance.metric` | `inner_product` | Distance
metric: `l2`, `cosine`, or `inner_product`. |
| `fields.<column>.pk-vector.index.options` | Not set | JSON object containing
build options for the selected ANN implementation. |
+| `pk-full-text.index.columns` | Not set | Character column to index. Exactly
one full-text column is currently supported. |
+| `fields.<column>.pk-full-text.index.options` | Not set | JSON object
containing native analyzer options. Unqualified keys are scoped to `full-text`.
|
| `pk-btree.index.columns` | Not set | Comma-separated columns which own
independent BTree indexes. |
| `fields.<column>.pk-btree.index.options` | Not set | JSON object containing
BTree build options. Unqualified keys are scoped to `btree-index`. |
| `pk-bitmap.index.columns` | Not set | Comma-separated columns which own
independent Bitmap indexes. |
| `fields.<column>.pk-bitmap.index.options` | Not set | JSON object containing
Bitmap build options. Unqualified keys are scoped to `bitmap-index`. |
-| `fields.<column>.pk-index.compaction.level-fanout` | `5` | Number of
similarly sized index groups which triggers a rebuild and maximum row-count
ratio within one size tier. Shared by all three families. Must be greater than
`1`. |
-| `fields.<column>.pk-index.compaction.stale-ratio-threshold` | `0.2` | Ratio
of rows from inactive source files which triggers a rebuild. Shared by all
three families. Must be in `(0, 1]`. |
+| `fields.<column>.pk-index.compaction.level-fanout` | `5` | Number of
similarly sized index groups which triggers a rebuild and maximum row-count
ratio within one size tier. Shared by all four families. Must be greater than
`1`. |
+| `fields.<column>.pk-index.compaction.stale-ratio-threshold` | `0.2` | Ratio
of rows from inactive source files which triggers a rebuild. Shared by all four
families. Must be in `(0, 1]`. |
For algorithm-specific options, see the corresponding
[BTree](../multimodal-table/global-index/btree),
[Bitmap](../multimodal-table/global-index/bitmap), or
-[Vector](../multimodal-table/global-index/vector) index page.
+[Vector](../multimodal-table/global-index/vector) index page. Full-text
analyzer options are listed
+in the [`paimon-full-text`
README](https://github.com/apache/paimon/blob/master/paimon-full-text/README.md).
## Maintenance and Coverage
@@ -238,12 +274,17 @@ Index construction can execute asynchronously inside the
writer. A writer which
compaction also waits for active index maintenance; a non-blocking writer can
complete maintenance
in a later commit. Coverage can therefore be temporarily partial.
-Partial coverage affects acceleration, not correctness:
+For scalar and vector indexes, partial coverage affects acceleration, not
correctness:
- BTree and Bitmap scans read uncovered files through the ordinary data path.
- Vector search evaluates files without an active ANN group exactly.
- The original scalar predicate and deletion vectors are applied after index
pruning.
+Primary-key Full Text currently supports only `global-index.search-mode =
fast`. It searches
+persistent archives and ignores uncovered files; `full` and `detail` are
rejected because a
+merge-aware logical-row fallback is not implemented. Consequently, newly
appended Level-0 rows
+become full-text searchable only after compaction publishes an eligible data
file and archive.
+
## BTree and Bitmap Queries
BTree and Bitmap indexes are applied automatically to snapshot-scoped batch
scans. They can
@@ -350,17 +391,112 @@ Only ANN candidates can win the rerank, so a larger
factor can improve recall bu
guarantee the exact global Top-K. It also increases ANN work and data-file
I/O. Uncovered files are
searched exactly and merged separately with the ANN candidates.
+## Full-Text Search
+
+A primary-key full-text search captures one table snapshot, searches active
archives in the
+selected buckets, applies deletion vectors, and merges native relevance scores
into a global
+Top-K. The score is preserved as `__paimon_search_score`; it is not rewritten
by RRF unless the
+full-text route is later combined by Hybrid search.
+
+<Tabs groupId="primary-key-full-text-search-api">
+
+<TabItem value="spark-sql" label="Spark SQL">
+
+```sql
+SELECT id, content, __paimon_search_score
+FROM full_text_search(
+ 'items',
+ 'content',
+ '{"match":{"column":"content","terms":"paimon lake"}}',
+ 10
+)
+ORDER BY __paimon_search_score DESC;
+```
+
+</TabItem>
+
+<TabItem value="flink-sql" label="Flink SQL">
+
+Flink returns JSON-serialized rows. Add `__paimon_search_score` to
`projection` when the native
+relevance score is required. `top_k` must be between 1 and 10,000.
+
+```sql
+CALL sys.full_text_search(
+ `table` => 'default.items',
+ `column` => 'content',
+ query => '{"match":{"column":"content","terms":"paimon lake"}}',
+ top_k => 10,
+ projection => 'id,content,__paimon_search_score'
+);
+```
+
+</TabItem>
+
+<TabItem value="java-api" label="Java API">
+
+```java
+GlobalIndexResult result = table.newFullTextSearchBuilder()
+ .withQuery("content", queryJson)
+ .withLimit(10)
+ .executeLocal();
+
+ReadBuilder readBuilder = table.newReadBuilder();
+TableScan.Plan plan =
readBuilder.newScan().withGlobalIndexResult(result).plan();
+try (RecordReader<InternalRow> reader =
readBuilder.newRead().createReader(plan)) {
+ reader.forEachRemaining(row -> consume(row));
+}
+```
+
+</TabItem>
+
+</Tabs>
+
+Partition predicates can prune buckets before ranking. Arbitrary row
predicates cannot be pushed
+into a full-text Top-K route.
+
+## Hybrid Search
+
+Spark `hybrid_search` can fuse primary-key Vector and Full Text routes by
physical data-file
+position. Hybrid captures one snapshot for all physical routes, rejects a mix
of physical and
+global row-ID routes, and deduplicates their source files before reading rows.
Supported rankers
+are `rrf`, `weighted_score`, and `mrr`; route weights must be finite and
positive.
+
+```sql
+SELECT id, __paimon_search_score
+FROM hybrid_search(
+ 'items',
+ array(named_struct(
+ 'field', 'embedding',
+ 'query_vector', array(0.1f, 0.2f, 0.3f),
+ 'limit', 20,
+ 'weight', 1.0f,
+ 'options', map())),
+ array(named_struct(
+ 'column', 'content',
+ 'query', '{"match":{"column":"content","terms":"paimon lake"}}',
+ 'limit', 20,
+ 'weight', 1.0f,
+ 'options', map())),
+ 10,
+ 'rrf'
+)
+ORDER BY __paimon_search_score DESC;
+```
+
## Merge-Engine Behavior
-Vector maintenance follows the table's merge engine:
+Vector and Full Text maintenance follow the table's merge engine:
- `deduplicate`: an update indexes the latest row and the deletion vector
hides the replaced
- physical row. A delete removes the old row from search results through the
deletion vector.
+ physical row. A delete removes the old row from both search indexes through
the deletion vector.
- `partial-update`: the vector index is built from the lookup-completed
compact-output row.
- `aggregation`: the vector index is built from the aggregated compact-output
row.
- `first-row`: the retained first row is indexed. Deletion vectors must be
disabled because later
rows with the same primary key are ignored rather than deleting the retained
row.
+Full Text indexes the lookup-completed, aggregated, or retained compact-output
text using the
+same rules. Null text consumes a physical row ordinal but is not added to the
native term index.
+
## Schema Evolution
An indexed column cannot be renamed, dropped, or have its type changed while
its definition is
@@ -372,9 +508,14 @@ create a table with the desired definition and migrate the
data.
- BTree and Bitmap definitions are single-column indexes.
- Exactly one vector index column is currently supported per table.
+- Exactly one full-text index column is currently supported per table.
- Only `FLOAT` vectors are supported.
- Indexes are built from eligible compact output, not directly from Level-0
appends.
- Index acceleration and vector search are snapshot-scoped batch operations;
continuous streaming
- and lateral vector search are not supported.
+ and lateral Vector or Full Text search are not supported.
- Flink vector search returns rows but does not expose the ANN score as a
separate column.
+- Primary-key Full Text supports only FAST search mode and excludes uncovered
files until
+ compaction creates persistent archives.
+- Full Text routes support partition pruning but not arbitrary row predicates
before Top-K.
+- Hybrid search cannot mix source-backed physical routes with global row-ID
routes.
- Online replacement between two definitions on the same column is not
supported.
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 bfd2c31199..873e73b2f4 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchBuilderImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchBuilderImpl.java
@@ -18,6 +18,7 @@
package org.apache.paimon.table.source;
+import org.apache.paimon.Snapshot;
import org.apache.paimon.index.pk.PrimaryKeyIndexDefinition;
import org.apache.paimon.index.pk.PrimaryKeyIndexDefinitions;
import org.apache.paimon.partition.PartitionPredicate;
@@ -25,6 +26,8 @@ import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.InnerTable;
import org.apache.paimon.types.DataField;
+import javax.annotation.Nullable;
+
import java.util.Collections;
import java.util.Optional;
@@ -42,6 +45,7 @@ public class FullTextSearchBuilderImpl implements
FullTextSearchBuilder {
private String fieldName;
private String query;
private PartitionPredicate partitionFilter;
+ @Nullable private Snapshot pinnedSnapshot;
public FullTextSearchBuilderImpl(InnerTable table) {
this.table = (FileStoreTable) table;
@@ -71,7 +75,8 @@ public class FullTextSearchBuilderImpl implements
FullTextSearchBuilder {
DataField textColumn = textColumn();
Optional<PrimaryKeyIndexDefinition> definition =
primaryKeyFullTextDefinition(textColumn);
return definition.isPresent()
- ? new PrimaryKeyFullTextScan(table, definition.get(),
partitionFilter)
+ ? new PrimaryKeyFullTextScan(
+ table, definition.get(), partitionFilter,
pinnedSnapshot)
: new DataEvolutionFullTextScan(
table, partitionFilter,
Collections.singletonList(textColumn));
}
@@ -115,4 +120,9 @@ public class FullTextSearchBuilderImpl implements
FullTextSearchBuilder {
}
return Optional.empty();
}
+
+ FullTextSearchBuilderImpl withSnapshot(Snapshot snapshot) {
+ this.pinnedSnapshot = snapshot;
+ return this;
+ }
}
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 48531ca358..7d3725444d 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
@@ -18,22 +18,32 @@
package org.apache.paimon.table.source;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.globalindex.GlobalIndexResult;
import org.apache.paimon.globalindex.HybridSearchRanker;
+import org.apache.paimon.globalindex.IndexedSplit;
import org.apache.paimon.globalindex.ScoredGlobalIndexResult;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.predicate.HybridSearchRoute;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.InnerTable;
+import org.apache.paimon.table.source.snapshot.TimeTravelUtil;
import org.apache.paimon.utils.Pair;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
import java.util.Objects;
import java.util.Optional;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
/** Implementation for {@link HybridSearchBuilder}. */
public class HybridSearchBuilderImpl implements HybridSearchBuilder {
@@ -120,12 +130,27 @@ public class HybridSearchBuilderImpl implements
HybridSearchBuilder {
public List<Route> routeBuilders() {
validateSearch();
+ Snapshot snapshot = null;
+ if (table instanceof FileStoreTable) {
+ FileStoreTable fileStoreTable = (FileStoreTable) table;
+ if
(!TimeTravelUtil.tryTravelToSnapshot(fileStoreTable).isPresent()) {
+ snapshot = fileStoreTable.latestSnapshot().orElse(null);
+ }
+ }
List<Route> routeBuilders = new ArrayList<>(routes.size());
for (HybridSearchRoute route : routes) {
if (route.isVector()) {
- routeBuilders.add(new Route(route,
newVectorSearchBuilder(route)));
+ VectorSearchBuilder builder = newVectorSearchBuilder(route);
+ if (snapshot != null && builder instanceof
VectorSearchBuilderImpl) {
+ ((VectorSearchBuilderImpl) builder).withSnapshot(snapshot);
+ }
+ routeBuilders.add(new Route(route, builder));
} else {
- routeBuilders.add(new Route(route,
newFullTextSearchBuilder(route)));
+ FullTextSearchBuilder builder =
newFullTextSearchBuilder(route);
+ if (snapshot != null && builder instanceof
FullTextSearchBuilderImpl) {
+ ((FullTextSearchBuilderImpl)
builder).withSnapshot(snapshot);
+ }
+ routeBuilders.add(new Route(route, builder));
}
}
return routeBuilders;
@@ -154,6 +179,24 @@ public class HybridSearchBuilderImpl implements
HybridSearchBuilder {
public ScoredGlobalIndexResult rank(List<RouteResult> routeResults) {
validateSearch();
+ boolean hasPhysical = false;
+ boolean hasGlobal = false;
+ for (RouteResult routeResult : routeResults) {
+ if (routeResult.result() instanceof PrimaryKeyScoredResult) {
+ hasPhysical = true;
+ } else {
+ hasGlobal = true;
+ }
+ }
+ if (hasPhysical && hasGlobal) {
+ throw new UnsupportedOperationException(
+ "Hybrid search cannot combine physical primary-key
positions and global "
+ + "row-id address spaces.");
+ }
+ if (hasPhysical) {
+ return rankPhysical(routeResults);
+ }
+
List<HybridSearchRanker.WeightedResult> weightedResults =
new ArrayList<>(routeResults.size());
for (RouteResult routeResult : routeResults) {
@@ -166,9 +209,138 @@ public class HybridSearchBuilderImpl implements
HybridSearchBuilder {
return HybridSearchRanker.rank(ranker, weightedResults, limit);
}
+ private PrimaryKeyScoredResult rankPhysical(List<RouteResult>
routeResults) {
+ Long snapshotId = null;
+ List<PrimaryKeySearchRanker.Ranking> rankings = new
ArrayList<>(routeResults.size());
+ List<PrimaryKeyScoredResult> physicalResults = new
ArrayList<>(routeResults.size());
+ for (RouteResult routeResult : routeResults) {
+ PrimaryKeyScoredResult result = (PrimaryKeyScoredResult)
routeResult.result();
+ if (snapshotId == null) {
+ snapshotId = result.snapshotId();
+ } else {
+ checkArgument(
+ snapshotId == result.snapshotId(),
+ "Primary-key hybrid routes must use the same snapshot,
but found %s and %s.",
+ snapshotId,
+ result.snapshotId());
+ }
+ physicalResults.add(result);
+ if (!result.positions().isEmpty()) {
+ rankings.add(
+ new PrimaryKeySearchRanker.Ranking(
+ result.positions(),
routeResult.route().weight()));
+ }
+ }
+
+ List<PrimaryKeySearchPosition> positions;
+ if (HybridSearchRanker.WEIGHTED_SCORE_RANKER.equals(ranker)) {
+ positions = PrimaryKeySearchRanker.weightedScore(rankings, limit);
+ } else if (HybridSearchRanker.MRR_RANKER.equals(ranker)) {
+ positions = PrimaryKeySearchRanker.weightedMrr(rankings, limit);
+ } else {
+ positions = PrimaryKeySearchRanker.weightedRrf(rankings, limit);
+ }
+ return new PrimaryKeyScoredResult(
+ snapshotId, physicalSources(physicalResults, positions),
positions);
+ }
+
+ private static List<DataSplit> physicalSources(
+ List<PrimaryKeyScoredResult> results,
List<PrimaryKeySearchPosition> positions) {
+ Map<PhysicalFileKey, DataSplit> available = new LinkedHashMap<>();
+ for (PrimaryKeyScoredResult result : results) {
+ for (IndexedSplit indexedSplit : result.splits()) {
+ DataSplit source = indexedSplit.dataSplit();
+ checkArgument(
+ source.dataFiles().size() == 1,
+ "Primary-key scored source split must contain exactly
one data file.");
+ PhysicalFileKey key =
+ new PhysicalFileKey(
+ source.partition(),
+ source.bucket(),
+ source.dataFiles().get(0).fileName());
+ DataSplit previous = available.putIfAbsent(key, source);
+ if (previous != null) {
+ checkArgument(
+ previous.snapshotId() == source.snapshotId()
+ &&
previous.bucketPath().equals(source.bucketPath())
+ && Objects.equals(
+ previous.totalBuckets(),
source.totalBuckets())
+ && previous.dataFiles().get(0).fileSize()
+ ==
source.dataFiles().get(0).fileSize()
+ && previous.dataFiles().get(0).rowCount()
+ ==
source.dataFiles().get(0).rowCount()
+ && Objects.equals(deletionFile(previous),
deletionFile(source)),
+ "Primary-key hybrid routes contain inconsistent
metadata for data file %s.",
+ key.dataFileName);
+ }
+ }
+ }
+
+ Map<PhysicalFileKey, DataSplit> selected = new LinkedHashMap<>();
+ for (PrimaryKeySearchPosition position : positions) {
+ PhysicalFileKey key = PhysicalFileKey.from(position);
+ DataSplit source = available.get(key);
+ checkArgument(
+ source != null,
+ "Primary-key hybrid result references unknown data file
%s.",
+ position.dataFileName());
+ selected.putIfAbsent(key, source);
+ }
+ return Collections.unmodifiableList(new
ArrayList<>(selected.values()));
+ }
+
+ private static DeletionFile deletionFile(DataSplit split) {
+ if (!split.deletionFiles().isPresent()) {
+ return null;
+ }
+ checkArgument(
+ split.deletionFiles().get().size() == 1,
+ "Primary-key scored source split must contain exactly one
deletion-file entry.");
+ return split.deletionFiles().get().get(0);
+ }
+
+ private static class PhysicalFileKey {
+
+ private final BinaryRow partition;
+ private final int bucket;
+ private final String dataFileName;
+
+ private PhysicalFileKey(BinaryRow partition, int bucket, String
dataFileName) {
+ this.partition = partition.copy();
+ this.bucket = bucket;
+ this.dataFileName = dataFileName;
+ }
+
+ private static PhysicalFileKey from(PrimaryKeySearchPosition position)
{
+ return new PhysicalFileKey(
+ position.partition(), position.bucket(),
position.dataFileName());
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof PhysicalFileKey)) {
+ return false;
+ }
+ PhysicalFileKey that = (PhysicalFileKey) o;
+ return bucket == that.bucket
+ && partition.equals(that.partition)
+ && dataFileName.equals(that.dataFileName);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(partition, bucket, dataFileName);
+ }
+ }
+
@Override
public RouteResult toRouteResult(Route route, GlobalIndexResult result) {
- if (result instanceof ScoredGlobalIndexResult) {
+ if (result instanceof PrimaryKeyVectorResult) {
+ return new RouteResult(route.route(), ((PrimaryKeyVectorResult)
result).scoredResult());
+ } else if (result instanceof ScoredGlobalIndexResult) {
return new RouteResult(route.route(), (ScoredGlobalIndexResult)
result);
} else if (result.results().isEmpty()) {
return new RouteResult(route.route(),
ScoredGlobalIndexResult.createEmpty());
@@ -200,11 +372,6 @@ public class HybridSearchBuilderImpl implements
HybridSearchBuilder {
table.newFullTextSearchBuilder()
.withQuery(route.fieldName(), route.fullTextQuery())
.withLimit(route.limit());
- if (fullTextSearchBuilder.newFullTextScan() instanceof
PrimaryKeyFullTextScan) {
- throw new UnsupportedOperationException(
- "Hybrid search does not support primary-key full-text
indexes because their "
- + "results use physical file positions instead of
global row ids.");
- }
if (partitionFilter != null) {
fullTextSearchBuilder.withPartitionFilter(partitionFilter);
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyFullTextScan.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyFullTextScan.java
index d160523510..f4b91a3e68 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyFullTextScan.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyFullTextScan.java
@@ -18,6 +18,7 @@
package org.apache.paimon.table.source;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.Snapshot;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.globalindex.IndexedSplit;
@@ -54,23 +55,34 @@ public class PrimaryKeyFullTextScan implements FullTextScan
{
private final FileStoreTable table;
private final PrimaryKeyIndexDefinition definition;
@Nullable private final PartitionPredicate partitionFilter;
+ @Nullable private final Snapshot pinnedSnapshot;
public PrimaryKeyFullTextScan(
FileStoreTable table,
PrimaryKeyIndexDefinition definition,
@Nullable PartitionPredicate partitionFilter) {
+ this(table, definition, partitionFilter, null);
+ }
+
+ PrimaryKeyFullTextScan(
+ FileStoreTable table,
+ PrimaryKeyIndexDefinition definition,
+ @Nullable PartitionPredicate partitionFilter,
+ @Nullable Snapshot pinnedSnapshot) {
checkArgument(
definition.family() ==
PrimaryKeyIndexDefinition.Family.FULL_TEXT,
"Primary-key full-text scan requires a full-text index
definition.");
this.table = table;
this.definition = definition;
this.partitionFilter = partitionFilter;
+ this.pinnedSnapshot = pinnedSnapshot;
}
@Override
public Plan scan() {
- SnapshotReader snapshotReader = table.newSnapshotReader().keepStats();
- DataTableScan dataScan = table.newScan(ignored -> snapshotReader);
+ FileStoreTable scanTable = scanTable();
+ SnapshotReader snapshotReader =
scanTable.newSnapshotReader().keepStats();
+ DataTableScan dataScan = scanTable.newScan(ignored -> snapshotReader);
checkArgument(
dataScan instanceof PrimaryKeyBatchScan,
"Primary-key full-text search requires a primary-key batch
scan.");
@@ -89,8 +101,16 @@ public class PrimaryKeyFullTextScan implements FullTextScan
{
if (snapshotPlan.snapshotId() == null) {
return new Plan(0, Collections.emptyList());
}
- Snapshot snapshot =
snapshotReader.snapshotManager().snapshot(snapshotPlan.snapshotId());
+ Snapshot snapshot =
+ pinnedSnapshot == null
+ ?
snapshotReader.snapshotManager().snapshot(snapshotPlan.snapshotId())
+ : pinnedSnapshot;
checkArgument(snapshot != null, "Primary-key full-text snapshot does
not exist.");
+ checkArgument(
+ snapshot.id() == snapshotPlan.snapshotId(),
+ "Primary-key full-text plan snapshot %s does not match pinned
snapshot %s.",
+ snapshotPlan.snapshotId(),
+ snapshot.id());
IndexFileHandler indexFileHandler = snapshotReader.indexFileHandler();
checkArgument(
@@ -105,6 +125,17 @@ public class PrimaryKeyFullTextScan implements
FullTextScan {
return plan(snapshot.id(), snapshotPlan.splits(), payloadEntries,
definition.fieldId());
}
+ private FileStoreTable scanTable() {
+ if (pinnedSnapshot == null) {
+ return table;
+ }
+ return (FileStoreTable)
+ table.copy(
+ Collections.singletonMap(
+ CoreOptions.SCAN_SNAPSHOT_ID.key(),
+ String.valueOf(pinnedSnapshot.id())));
+ }
+
private boolean matchesDefinition(IndexManifestEntry entry) {
IndexFileMeta payload = entry.indexFile();
GlobalIndexMeta globalMeta = payload.globalIndexMeta();
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyFullTextScanTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyFullTextScanTest.java
index ca656e82b2..ad290b1362 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyFullTextScanTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyFullTextScanTest.java
@@ -131,6 +131,8 @@ class PrimaryKeyFullTextScanTest {
Options tableOptions = new Options();
tableOptions.set(CoreOptions.PK_FULL_TEXT_INDEX_COLUMNS, "content");
when(table.coreOptions()).thenReturn(new CoreOptions(tableOptions));
+
when(table.copy(Collections.singletonMap(CoreOptions.SCAN_SNAPSHOT_ID.key(),
"11")))
+ .thenReturn(table);
SnapshotReader reader = mock(SnapshotReader.class, RETURNS_SELF);
SnapshotReader.Plan snapshotPlan = mock(SnapshotReader.Plan.class,
CALLS_REAL_METHODS);
@@ -169,7 +171,7 @@ class PrimaryKeyFullTextScanTest {
configureBatchScan(table, reader, snapshot);
PrimaryKeyFullTextScan.Plan plan =
- new PrimaryKeyFullTextScan(table, definition,
partitionFilter).scan();
+ new PrimaryKeyFullTextScan(table, definition, partitionFilter,
snapshot).scan();
assertThat(plan.snapshotId()).isEqualTo(11);
PrimaryKeyFullTextSearchSplit split = (PrimaryKeyFullTextSearchSplit)
plan.splits().get(0);
@@ -177,6 +179,7 @@ class PrimaryKeyFullTextScanTest {
.extracting(IndexFileMeta::fileName)
.containsExactly("current");
assertThat(split.uncoveredDataFiles()).isEmpty();
+
verify(table).copy(Collections.singletonMap(CoreOptions.SCAN_SNAPSHOT_ID.key(),
"11"));
verify(reader).withPartitionFilter(partitionFilter);
verify(reader).indexFileHandler();
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyFullTextSearchTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyFullTextSearchTest.java
index 47a3d5344d..e3faa6a38d 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyFullTextSearchTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyFullTextSearchTest.java
@@ -32,7 +32,6 @@ import java.util.HashMap;
import java.util.Map;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -79,18 +78,18 @@ class PrimaryKeyFullTextSearchTest {
}
@Test
- void testHybridRouteRejectsPrimaryKeyFullText() {
+ void testHybridRouteUsesPrimaryKeyFullText() {
FileStoreTable table = table(false);
- assertThatThrownBy(
- () ->
- new HybridSearchBuilderImpl(table)
- .addFullTextRoute("content", "hello",
10, 1F)
- .withLimit(10)
- .routeBuilders())
- .isInstanceOf(UnsupportedOperationException.class)
- .hasMessageContaining(
- "Hybrid search does not support primary-key full-text
indexes");
+ HybridSearchBuilder.Route route =
+ new HybridSearchBuilderImpl(table)
+ .addFullTextRoute("content", "hello", 10, 1F)
+ .withLimit(10)
+ .routeBuilders()
+ .get(0);
+
+ assertThat(route.fullTextSearchBuilder().newFullTextScan())
+ .isInstanceOf(PrimaryKeyFullTextScan.class);
}
private static FileStoreTable table(boolean dataEvolution) {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyHybridSearchTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyHybridSearchTest.java
new file mode 100644
index 0000000000..43670ce8ac
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyHybridSearchTest.java
@@ -0,0 +1,185 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.table.source;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.globalindex.ScoredGlobalIndexResult;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.predicate.HybridSearchRoute;
+import org.apache.paimon.stats.SimpleStats;
+import org.apache.paimon.utils.RoaringNavigableMap64;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for snapshot-scoped hybrid search on primary-key physical positions.
*/
+class PrimaryKeyHybridSearchTest {
+
+ @Test
+ void testRrfFusesPhysicalRoutesAndDeduplicatesSourceFiles() {
+ PrimaryKeySearchPosition a = position("a", 0, 10F);
+ PrimaryKeySearchPosition b = position("b", 0, 9F);
+ PrimaryKeySearchPosition c = position("c", 0, 8F);
+ PrimaryKeyScoredResult vectorResult =
+ result(7, new String[] {"a", "b"}, new
PrimaryKeySearchPosition[] {a, b});
+ PrimaryKeyScoredResult textResult =
+ result(
+ 7,
+ new String[] {"b", "c"},
+ new PrimaryKeySearchPosition[] {b.withScore(100F), c});
+ HybridSearchRoute vectorRoute = new HybridSearchRoute("vector", new
float[] {1F}, 2, 1F);
+ HybridSearchRoute textRoute = HybridSearchRoute.fullText("text",
"query", 2, 2F, null);
+ HybridSearchBuilderImpl builder =
+ (HybridSearchBuilderImpl)
+ new HybridSearchBuilderImpl(null)
+ .addRoute(vectorRoute)
+ .addRoute(textRoute)
+ .withLimit(3)
+ .withRrfRanker();
+
+ ScoredGlobalIndexResult ranked =
+ builder.rank(
+ Arrays.asList(
+ new
HybridSearchBuilder.RouteResult(vectorRoute, vectorResult),
+ new HybridSearchBuilder.RouteResult(textRoute,
textResult)));
+
+ assertThat(ranked).isInstanceOf(PrimaryKeyScoredResult.class);
+ PrimaryKeyScoredResult physical = (PrimaryKeyScoredResult) ranked;
+ assertThat(physical.snapshotId()).isEqualTo(7);
+ assertThat(physical.positions())
+ .extracting(PrimaryKeySearchPosition::dataFileName)
+ .containsExactly("b", "c", "a");
+ assertThat(physical.splits()).hasSize(3);
+ assertThat(physical.positions().get(0).score())
+ .isCloseTo(
+ (float) (1D / 62D + 2D / 61D),
+ org.assertj.core.data.Offset.offset(0.000001F));
+ }
+
+ @Test
+ void testRejectsPhysicalRoutesFromDifferentSnapshots() {
+ PrimaryKeyScoredResult first =
+ result(
+ 7,
+ new String[] {"a"},
+ new PrimaryKeySearchPosition[] {position("a", 0, 1F)});
+ PrimaryKeyScoredResult second =
+ result(
+ 8,
+ new String[] {"a"},
+ new PrimaryKeySearchPosition[] {position("a", 0, 1F)});
+ HybridSearchRoute firstRoute = new HybridSearchRoute("first", new
float[] {1F}, 1, 1F);
+ HybridSearchRoute secondRoute = new HybridSearchRoute("second", new
float[] {1F}, 1, 1F);
+ HybridSearchBuilderImpl builder =
+ (HybridSearchBuilderImpl)
+ new HybridSearchBuilderImpl(null)
+ .addRoute(firstRoute)
+ .addRoute(secondRoute)
+ .withLimit(1);
+
+ assertThatThrownBy(
+ () ->
+ builder.rank(
+ Arrays.asList(
+ new
HybridSearchBuilder.RouteResult(
+ firstRoute, first),
+ new
HybridSearchBuilder.RouteResult(
+ secondRoute, second))))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("same snapshot");
+ }
+
+ @Test
+ void testRejectsMixedPhysicalAndGlobalAddressSpaces() {
+ PrimaryKeyScoredResult physical =
+ result(
+ 7,
+ new String[] {"a"},
+ new PrimaryKeySearchPosition[] {position("a", 0, 1F)});
+ RoaringNavigableMap64 rowIds = new RoaringNavigableMap64();
+ rowIds.add(1L);
+ ScoredGlobalIndexResult global =
ScoredGlobalIndexResult.create(rowIds, ignored -> 1F);
+ HybridSearchRoute physicalRoute =
+ new HybridSearchRoute("physical", new float[] {1F}, 1, 1F);
+ HybridSearchRoute globalRoute = new HybridSearchRoute("global", new
float[] {1F}, 1, 1F);
+ HybridSearchBuilderImpl builder =
+ (HybridSearchBuilderImpl)
+ new HybridSearchBuilderImpl(null)
+ .addRoute(physicalRoute)
+ .addRoute(globalRoute)
+ .withLimit(1);
+
+ assertThatThrownBy(
+ () ->
+ builder.rank(
+ Arrays.asList(
+ new
HybridSearchBuilder.RouteResult(
+ physicalRoute,
physical),
+ new
HybridSearchBuilder.RouteResult(
+ globalRoute, global))))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining("address spaces");
+ }
+
+ private static PrimaryKeyScoredResult result(
+ long snapshotId, String[] files, PrimaryKeySearchPosition[]
positions) {
+ DataSplit source =
+ DataSplit.builder()
+ .withSnapshot(snapshotId)
+ .withPartition(BinaryRow.EMPTY_ROW)
+ .withBucket(0)
+ .withBucketPath("bucket-0")
+ .withTotalBuckets(1)
+ .withDataFiles(
+ Arrays.stream(files)
+
.map(PrimaryKeyHybridSearchTest::dataFile)
+
.collect(java.util.stream.Collectors.toList()))
+ .build();
+ return new PrimaryKeyScoredResult(
+ snapshotId, Collections.singletonList(source),
Arrays.asList(positions));
+ }
+
+ private static PrimaryKeySearchPosition position(String file, long row,
float score) {
+ return new PrimaryKeySearchPosition(BinaryRow.EMPTY_ROW, 0, file, row,
score);
+ }
+
+ private static DataFileMeta dataFile(String fileName) {
+ return DataFileMeta.forAppend(
+ fileName,
+ 100,
+ 5,
+ SimpleStats.EMPTY_STATS,
+ 0,
+ 1,
+ 1,
+ Collections.emptyList(),
+ null,
+ FileSource.COMPACT,
+ null,
+ null,
+ null,
+ null);
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/FullTextSearchProcedure.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/FullTextSearchProcedure.java
new file mode 100644
index 0000000000..7f0ca15527
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/FullTextSearchProcedure.java
@@ -0,0 +1,322 @@
+/*
+ * 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.flink.procedure;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.format.json.JsonFormatWriter;
+import org.apache.paimon.format.json.JsonOptions;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.globalindex.GlobalIndexResult;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.reader.ScoreRecordIterator;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.table.source.TableScan;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.InternalRowUtils;
+import org.apache.paimon.utils.StringUtils;
+
+import org.apache.flink.table.annotation.ArgumentHint;
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.ProcedureHint;
+import org.apache.flink.table.procedure.ProcedureContext;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/** Procedure for local full-text search with optional JSON projection and
search scores. */
+public class FullTextSearchProcedure extends ProcedureBase {
+
+ public static final String IDENTIFIER = "full_text_search";
+ public static final String SEARCH_SCORE = "__paimon_search_score";
+
+ private static final int MAX_TOP_K = 10_000;
+
+ private static final DataField SEARCH_SCORE_FIELD =
+ new DataField(Integer.MAX_VALUE, SEARCH_SCORE, DataTypes.FLOAT());
+
+ @ProcedureHint(
+ argument = {
+ @ArgumentHint(name = "table", type = @DataTypeHint("STRING")),
+ @ArgumentHint(name = "column", type = @DataTypeHint("STRING")),
+ @ArgumentHint(name = "query", type = @DataTypeHint("STRING")),
+ @ArgumentHint(name = "top_k", type = @DataTypeHint("INT")),
+ @ArgumentHint(
+ name = "projection",
+ type = @DataTypeHint("STRING"),
+ isOptional = true),
+ @ArgumentHint(name = "options", type =
@DataTypeHint("STRING"), isOptional = true)
+ })
+ public String[] call(
+ ProcedureContext procedureContext,
+ String tableId,
+ String column,
+ String query,
+ Integer topK,
+ String projection,
+ String options)
+ throws Exception {
+ validateSearch(column, query, topK);
+
+ Table table = table(tableId);
+ Map<String, String> optionsMap = optionalConfigMap(options);
+ String queryAuthOption = CoreOptions.QUERY_AUTH_ENABLED.key();
+ if (optionsMap.containsKey(queryAuthOption)) {
+ throw new IllegalArgumentException(
+ String.format("Option '%s' is not allowed",
queryAuthOption));
+ }
+ if (!optionsMap.isEmpty()) {
+ table = table.copy(optionsMap);
+ }
+
+ Projection parsedProjection = Projection.parse(projection,
table.rowType());
+ GlobalIndexResult result =
+ table.newFullTextSearchBuilder()
+ .withQuery(column, query)
+ .withLimit(topK)
+ .executeLocal();
+
+ ReadBuilder readBuilder = table.newReadBuilder();
+ if (parsedProjection.dataProjection != null) {
+ readBuilder.withProjection(parsedProjection.dataProjection);
+ }
+ TableScan.Plan plan =
readBuilder.newScan().withGlobalIndexResult(result).plan();
+ return readRows(readBuilder, plan, parsedProjection);
+ }
+
+ private static void validateSearch(String column, String query, Integer
topK) {
+ if (StringUtils.isNullOrWhitespaceOnly(column)) {
+ throw new IllegalArgumentException("column must not be blank");
+ }
+ if (StringUtils.isNullOrWhitespaceOnly(query)) {
+ throw new IllegalArgumentException("query must not be blank");
+ }
+ if (topK == null || topK <= 0) {
+ throw new IllegalArgumentException("top_k must be positive");
+ }
+ if (topK > MAX_TOP_K) {
+ throw new IllegalArgumentException("top_k must not exceed " +
MAX_TOP_K);
+ }
+ }
+
+ private static String[] readRows(
+ ReadBuilder readBuilder, TableScan.Plan plan, Projection
projection)
+ throws IOException {
+ ByteArrayOutputStream byteOut = new ByteArrayOutputStream(1024);
+ JsonOptions jsonOptions = new JsonOptions(new Options());
+ List<Float> scores = new ArrayList<>();
+ try (JsonFormatWriter jsonWriter =
+ new JsonFormatWriter(
+ new ByteArrayPositionOutputStream(byteOut),
+ projection.outputType,
+ jsonOptions,
+ "none");
+ RecordReader<InternalRow> reader =
readBuilder.newRead().createReader(plan)) {
+ RecordReader.RecordIterator<InternalRow> batch;
+ while ((batch = reader.readBatch()) != null) {
+ try {
+ if (!(batch instanceof ScoreRecordIterator)) {
+ throw new IllegalStateException(
+ "Full-text search reader did not expose search
scores.");
+ }
+ ScoreRecordIterator<InternalRow> scoredBatch =
+ (ScoreRecordIterator<InternalRow>) batch;
+ InternalRow row;
+ while ((row = scoredBatch.next()) != null) {
+ float score = scoredBatch.returnedScore();
+ if (Float.isNaN(score)) {
+ throw new IllegalStateException(
+ "Full-text search reader returned a
missing score.");
+ }
+ jsonWriter.addElement(projection.output(row, score));
+ scores.add(score);
+ }
+ } finally {
+ batch.releaseBatch();
+ }
+ }
+ }
+
+ String[] lines =
+ StringUtils.split(byteOut.toString("UTF-8"),
jsonOptions.getLineDelimiter());
+ List<ScoredJson> rows = new ArrayList<>(lines.length);
+ int scoreIndex = 0;
+ for (String line : lines) {
+ String trimmed = line.trim();
+ if (!trimmed.isEmpty()) {
+ if (scoreIndex >= scores.size()) {
+ throw new IllegalStateException(
+ "Full-text JSON rows and scores are misaligned.");
+ }
+ rows.add(new ScoredJson(trimmed, scores.get(scoreIndex++)));
+ }
+ }
+ if (scoreIndex != scores.size()) {
+ throw new IllegalStateException("Full-text JSON rows and scores
are misaligned.");
+ }
+ Collections.sort(
+ rows,
+ (left, right) -> {
+ int scoreOrder = Float.compare(right.score, left.score);
+ return scoreOrder != 0 ? scoreOrder :
left.json.compareTo(right.json);
+ });
+ return rows.stream().map(row -> row.json).toArray(String[]::new);
+ }
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+
+ private static class Projection {
+
+ private final RowType readType;
+ private final RowType outputType;
+ private final int[] dataProjection;
+ private final int[] outputToRead;
+
+ private Projection(
+ RowType readType, RowType outputType, int[] dataProjection,
int[] outputToRead) {
+ this.readType = readType;
+ this.outputType = outputType;
+ this.dataProjection = dataProjection;
+ this.outputToRead = outputToRead;
+ }
+
+ private static Projection parse(String projection, RowType tableType) {
+ if (StringUtils.isNullOrWhitespaceOnly(projection)) {
+ return new Projection(tableType, tableType, null, null);
+ }
+
+ String[] names = projection.split(",", -1);
+ List<Integer> dataIndices = new ArrayList<>();
+ List<DataField> outputFields = new ArrayList<>();
+ int[] outputToRead = new int[names.length];
+ Set<String> seen = new HashSet<>();
+ for (int i = 0; i < names.length; i++) {
+ String name = names[i].trim();
+ if (name.isEmpty()) {
+ throw new IllegalArgumentException("Projection column must
not be blank");
+ }
+ if (!seen.add(name)) {
+ throw new IllegalArgumentException("Duplicate projection
column: " + name);
+ }
+ if (SEARCH_SCORE.equals(name)) {
+ outputFields.add(SEARCH_SCORE_FIELD);
+ outputToRead[i] = -1;
+ continue;
+ }
+
+ int tableIndex = tableType.getFieldIndex(name);
+ if (tableIndex < 0) {
+ throw new IllegalArgumentException("Unknown projection
column: " + name);
+ }
+ outputFields.add(tableType.getFields().get(tableIndex));
+ outputToRead[i] = dataIndices.size();
+ dataIndices.add(tableIndex);
+ }
+
+ int[] dataProjection =
dataIndices.stream().mapToInt(Integer::intValue).toArray();
+ return new Projection(
+ tableType.project(dataProjection),
+ new RowType(outputFields),
+ dataProjection,
+ outputToRead);
+ }
+
+ private InternalRow output(InternalRow row, float score) {
+ if (outputToRead == null) {
+ return row;
+ }
+ GenericRow output = new GenericRow(outputToRead.length);
+ for (int i = 0; i < outputToRead.length; i++) {
+ int readIndex = outputToRead[i];
+ output.setField(
+ i,
+ readIndex < 0
+ ? score
+ : InternalRowUtils.get(
+ row, readIndex,
readType.getTypeAt(readIndex)));
+ }
+ return output;
+ }
+ }
+
+ private static class ScoredJson {
+
+ private final String json;
+ private final float score;
+
+ private ScoredJson(String json, float score) {
+ this.json = json;
+ this.score = score;
+ }
+ }
+
+ /** A {@link PositionOutputStream} wrapping a {@link
ByteArrayOutputStream}. */
+ private static class ByteArrayPositionOutputStream extends
PositionOutputStream {
+
+ private final ByteArrayOutputStream out;
+
+ private ByteArrayPositionOutputStream(ByteArrayOutputStream out) {
+ this.out = out;
+ }
+
+ @Override
+ public long getPos() {
+ return out.size();
+ }
+
+ @Override
+ public void write(int b) {
+ out.write(b);
+ }
+
+ @Override
+ public void write(byte[] b) throws IOException {
+ out.write(b);
+ }
+
+ @Override
+ public void write(byte[] b, int off, int len) {
+ out.write(b, off, len);
+ }
+
+ @Override
+ public void flush() throws IOException {
+ out.flush();
+ }
+
+ @Override
+ public void close() throws IOException {
+ out.close();
+ }
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
b/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
index b9696067ae..7cf7aefbf3 100644
---
a/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
+++
b/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
@@ -109,4 +109,5 @@
org.apache.paimon.flink.procedure.DataEvolutionMergeIntoProcedure
org.apache.paimon.flink.procedure.ReassignRowIdProcedure
org.apache.paimon.flink.procedure.CreateGlobalIndexProcedure
org.apache.paimon.flink.procedure.VectorSearchProcedure
+org.apache.paimon.flink.procedure.FullTextSearchProcedure
org.apache.paimon.flink.procedure.DropGlobalIndexProcedure
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/FullTextSearchProcedureITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/FullTextSearchProcedureITCase.java
new file mode 100644
index 0000000000..b621d0bc9f
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/FullTextSearchProcedureITCase.java
@@ -0,0 +1,150 @@
+/*
+ * 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.flink.procedure;
+
+import org.apache.paimon.flink.CatalogITCaseBase;
+import org.apache.paimon.index.pkfulltext.PkFullTextIndexFile;
+
+import org.apache.flink.table.api.config.TableConfigOptions;
+import org.apache.flink.types.Row;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** IT cases for {@link FullTextSearchProcedure}. */
+public class FullTextSearchProcedureITCase extends CatalogITCaseBase {
+
+ @Test
+ public void testPrimaryKeyFullTextSearchWithScoreProjection() throws
Exception {
+ createPrimaryKeyFullTextTable("T");
+ sql("INSERT INTO T VALUES (0, 'lake format')");
+ sql(
+ "INSERT INTO T VALUES "
+ + "(1, 'paimon full text search'), "
+ + "(2, 'apache paimon storage')");
+
+ tEnv.getConfig().set(TableConfigOptions.TABLE_DML_SYNC, true);
+ sql("CALL sys.compact(`table` => 'default.T')");
+
+ assertThat(
+
paimonTable("T").store().newIndexFileHandler().scanEntries().stream()
+ .filter(
+ entry ->
+
PkFullTextIndexFile.INDEX_TYPE.equals(
+
entry.indexFile().indexType())))
+ .isNotEmpty();
+
+ List<String> rows =
+ sql(
+ "CALL sys.full_text_search("
+ + "`table` => 'default.T', "
+ + "`column` => 'content', "
+ + "query =>
'{\"match\":{\"column\":\"content\",\"terms\":\"paimon storage\"}}', "
+ + "top_k => 10, "
+ + "projection =>
'id,__paimon_search_score')")
+ .stream()
+ .map(row -> row.getField(0).toString())
+ .collect(Collectors.toList());
+
+ assertThat(rows).hasSize(2);
+ assertThat(rows.get(0)).contains("\"id\":\"2\"");
+ assertThat(rows.get(1)).contains("\"id\":\"1\"");
+ assertThat(rows).allMatch(row ->
row.contains("\"__paimon_search_score\":\""));
+ }
+
+ @Test
+ public void testValidatesQueryLimitAndProjection() {
+ createPrimaryKeyFullTextTable("VALIDATION_T");
+
+ assertThatThrownBy(
+ () ->
+ search(
+ "VALIDATION_T",
+
"{\"match\":{\"column\":\"content\",\"terms\":\"paimon\"}}",
+ 0,
+ "id"))
+ .hasStackTraceContaining("top_k must be positive");
+ assertThatThrownBy(() -> search("VALIDATION_T", " ", 1, "id"))
+ .hasStackTraceContaining("query must not be blank");
+ assertThatThrownBy(
+ () ->
+ search(
+ "VALIDATION_T",
+
"{\"match\":{\"column\":\"content\",\"terms\":\"paimon\"}}",
+ 1,
+ "missing"))
+ .hasStackTraceContaining("Unknown projection column");
+ assertThatThrownBy(
+ () ->
+ search(
+ "VALIDATION_T",
+
"{\"match\":{\"column\":\"content\",\"terms\":\"paimon\"}}",
+ 10_001,
+ "id"))
+ .hasStackTraceContaining("top_k must not exceed 10000");
+ }
+
+ @Test
+ public void testRejectsQueryAuthorizationOverride() {
+ createPrimaryKeyFullTextTable("AUTH_T");
+
+ assertThatThrownBy(
+ () ->
+ sql(
+ "CALL sys.full_text_search("
+ + "`table` =>
'default.AUTH_T', "
+ + "`column` => 'content', "
+ + "query =>
'{\"match\":{\"column\":\"content\",\"terms\":\"paimon\"}}', "
+ + "top_k => 1, "
+ + "projection => 'id', "
+ + "options =>
'query-auth.enabled=false')"))
+ .hasStackTraceContaining("Option 'query-auth.enabled' is not
allowed");
+ }
+
+ private List<Row> search(String table, String query, int topK, String
projection) {
+ return sql(
+ "CALL sys.full_text_search("
+ + "`table` => 'default.%s', "
+ + "`column` => 'content', "
+ + "query => '%s', "
+ + "top_k => %d, "
+ + "projection => '%s')",
+ table, query, topK, projection);
+ }
+
+ private void createPrimaryKeyFullTextTable(String tableName) {
+ sql(
+ "CREATE TABLE %s ("
+ + "id INT, "
+ + "content STRING, "
+ + "PRIMARY KEY (id) NOT ENFORCED"
+ + ") WITH ("
+ + "'bucket' = '1', "
+ + "'file.format' = 'json', "
+ + "'file.compression' = 'none', "
+ + "'deletion-vectors.enabled' = 'true', "
+ + "'pk-full-text.index.columns' = 'content'"
+ + ")",
+ tableName);
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/TestPrimaryKeyFullTextGlobalIndexerFactory.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/TestPrimaryKeyFullTextGlobalIndexerFactory.java
new file mode 100644
index 0000000000..77c33453e2
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/TestPrimaryKeyFullTextGlobalIndexerFactory.java
@@ -0,0 +1,30 @@
+/*
+ * 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.flink.procedure;
+
+import
org.apache.paimon.globalindex.testfulltext.TestFullTextGlobalIndexerFactory;
+
+/** Test-only non-native backend for the fixed primary-key full-text SPI
identifier. */
+public class TestPrimaryKeyFullTextGlobalIndexerFactory extends
TestFullTextGlobalIndexerFactory {
+
+ @Override
+ public String identifier() {
+ return "full-text";
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/test/resources/META-INF/services/org.apache.paimon.globalindex.GlobalIndexerFactory
b/paimon-flink/paimon-flink-common/src/test/resources/META-INF/services/org.apache.paimon.globalindex.GlobalIndexerFactory
new file mode 100644
index 0000000000..620e29dfd0
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/resources/META-INF/services/org.apache.paimon.globalindex.GlobalIndexerFactory
@@ -0,0 +1,16 @@
+# 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.
+
+org.apache.paimon.flink.procedure.TestPrimaryKeyFullTextGlobalIndexerFactory
diff --git a/paimon-full-text/README.md b/paimon-full-text/README.md
index 10e4f2a66b..26a08ac999 100644
--- a/paimon-full-text/README.md
+++ b/paimon-full-text/README.md
@@ -4,7 +4,10 @@ Full-text search global index for Apache Paimon, backed by the
native `paimon-fu
## Overview
-This module provides full-text search capabilities for Paimon's Data Evolution
(append) tables through the Global Index framework. It contains only the Paimon
integration layer. Native full-text access, JNI, FFI, index archive handling,
and query parsing are provided by the separate `paimon-full-text-index`
dependency.
+This module provides full-text search for both Data Evolution (append) tables
through the Global
+Index framework and compaction-visible primary-key tables through file-aligned
index archives. It
+contains only the Paimon integration layer. Native full-text access, JNI, FFI,
index archive
+handling, and query parsing are provided by the separate
`paimon-full-text-index` dependency.
### Architecture
@@ -93,7 +96,51 @@ All integers are **big-endian**.
## Usage
-### Build Index
+### Primary-Key Tables
+
+Primary-key full-text indexing uses the fixed `full-text` SPI automatically.
Configure the text
+column directly on the table; no implementation selector is needed.
+
+```sql
+CREATE TABLE articles (
+ id BIGINT,
+ content STRING,
+ PRIMARY KEY (id) NOT ENFORCED
+) WITH (
+ 'bucket' = '16',
+ 'deletion-vectors.enabled' = 'true',
+ 'pk-full-text.index.columns' = 'content',
+ 'fields.content.pk-full-text.index.options' = '{"tokenizer":"jieba"}'
+);
+```
+
+Paimon creates native archives from complete Level-1-or-higher `COMPACT` data
files and
+incrementally consolidates them with the shared primary-key index LSM policy.
One archive can
+cover multiple ordered source files; its row IDs concatenate their physical
row positions. The
+shared `fields.<column>.pk-index.compaction.level-fanout` and
+`fields.<column>.pk-index.compaction.stale-ratio-threshold` options control
size-tier and stale-
+source rebuilds.
+
+Primary-key full-text search currently supports only
`global-index.search-mode=fast`. Level-0 and
+other uncovered files are not searched; their rows become searchable after
compaction publishes
+an eligible data file and persistent archive. Search applies each source
file's deletion vector,
+preserves native relevance scores, and selects a global Top-K. Only Hybrid
search rewrites route
+scores through its configured `rrf`, `weighted_score`, or `mrr` ranker.
+
+```sql
+CALL sys.full_text_search(
+ `table` => 'default.articles',
+ `column` => 'content',
+ query => '{"match":{"column":"content","terms":"paimon lake"}}',
+ top_k => 10,
+ projection => 'id,content,__paimon_search_score'
+);
+```
+
+See [Primary-Key Indexes](../docs/docs/primary-key-table/global-index.mdx) for
requirements,
+Spark and Java examples, Hybrid search, and current limitations.
+
+### Build a Global Index
```sql
CALL sys.create_global_index(
diff --git
a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/sql/TestPrimaryKeyFullTextGlobalIndexerFactory.java
b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/sql/TestPrimaryKeyFullTextGlobalIndexerFactory.java
new file mode 100644
index 0000000000..633c78365e
--- /dev/null
+++
b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/sql/TestPrimaryKeyFullTextGlobalIndexerFactory.java
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.sql;
+
+import
org.apache.paimon.globalindex.testfulltext.TestFullTextGlobalIndexerFactory;
+
+/** Test-only non-native backend for the fixed primary-key full-text SPI
identifier. */
+public class TestPrimaryKeyFullTextGlobalIndexerFactory extends
TestFullTextGlobalIndexerFactory {
+
+ @Override
+ public String identifier() {
+ return "full-text";
+ }
+}
diff --git
a/paimon-spark/paimon-spark-ut/src/test/resources/META-INF/services/org.apache.paimon.globalindex.GlobalIndexerFactory
b/paimon-spark/paimon-spark-ut/src/test/resources/META-INF/services/org.apache.paimon.globalindex.GlobalIndexerFactory
new file mode 100644
index 0000000000..c4589f2002
--- /dev/null
+++
b/paimon-spark/paimon-spark-ut/src/test/resources/META-INF/services/org.apache.paimon.globalindex.GlobalIndexerFactory
@@ -0,0 +1,16 @@
+# 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.
+
+org.apache.paimon.spark.sql.TestPrimaryKeyFullTextGlobalIndexerFactory
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 64b9115829..5777f867fb 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
@@ -68,6 +68,52 @@ class FullTextSearchTest extends PaimonSparkTestBase {
// ========== Index Read/Search Tests ==========
+ test("primary-key full-text search uses physical splits and exposes scores")
{
+ withTable("T") {
+ spark.sql("""
+ |CREATE TABLE T (id INT, content STRING)
+ |TBLPROPERTIES (
+ | 'primary-key' = 'id',
+ | 'bucket' = '1',
+ | 'deletion-vectors.enabled' = 'true',
+ | 'pk-full-text.index.columns' = 'content')
+ |""".stripMargin)
+
+ spark.sql("INSERT INTO T VALUES (0, 'lake format')")
+ spark.sql("""
+ |INSERT INTO T VALUES
+ | (1, 'paimon full text search'),
+ | (2, 'apache paimon storage')
+ |""".stripMargin)
+ spark.sql("CALL sys.compact(table => 'T')")
+
+ val compactedFiles = spark.sql("SELECT level FROM `T$files`").collect()
+ assert(compactedFiles.exists(_.getInt(0) > 0))
+ val payloads = loadTable("T")
+ .store()
+ .newIndexFileHandler()
+ .scanEntries()
+ .asScala
+ .filter(_.indexFile().indexType() == "full-text")
+ assert(payloads.nonEmpty)
+
+ val rows = spark
+ .sql("""
+ |SELECT id, __paimon_search_score
+ |FROM full_text_search(
+ | 'T',
+ | 'content',
+ | '{"match":{"column":"content","terms":"paimon"}}',
+ | 10)
+ |ORDER BY id
+ |""".stripMargin)
+ .collect()
+
+ assert(rows.map(_.getInt(0)).toSeq == Seq(1, 2))
+ assert(rows.forall(row => !row.isNullAt(1) && row.getFloat(1) > 0.0f))
+ }
+ }
+
test("full-text search - basic search") {
withTable("T") {
spark.sql("""
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 944559e5a5..254bf50c77 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
@@ -24,6 +24,64 @@ import org.apache.paimon.spark.PaimonSparkTestBase
/** Tests for hybrid search. */
class HybridSearchTest extends PaimonSparkTestBase {
+ test("primary-key hybrid search fuses vector and full-text physical
positions") {
+ withTable("T") {
+ spark.sql(
+ s"""
+ |CREATE TABLE T (id INT, content STRING, vec ARRAY<FLOAT>)
+ |TBLPROPERTIES (
+ | 'primary-key' = 'id',
+ | 'bucket' = '1',
+ | 'deletion-vectors.enabled' = 'true',
+ | 'pk-full-text.index.columns' = 'content',
+ | 'vector-field' = 'vec',
+ | 'field.vec.vector-dim' = '2',
+ | 'pk-vector.index.columns' = 'vec',
+ | 'fields.vec.pk-vector.index.type' =
'${TestVectorGlobalIndexerFactory.IDENTIFIER}',
+ | 'fields.vec.pk-vector.distance.metric' = 'l2',
+ | 'test.vector.dimension' = '2',
+ | 'test.vector.metric' = 'l2')
+ |""".stripMargin)
+
+ spark.sql("""
+ |INSERT INTO T VALUES
+ | (0, 'lake format', array(1.0f, 0.0f)),
+ | (1, 'paimon hybrid search', array(0.9f, 0.1f)),
+ | (2, 'paimon full text', array(0.0f, 1.0f))
+ |""".stripMargin)
+ spark.sql("CALL sys.compact(table => 'T')")
+
+ val rows = spark
+ .sql("""
+ |SELECT id, __paimon_search_score
+ |FROM hybrid_search(
+ | 'T',
+ | array(
+ | named_struct(
+ | 'field', 'vec',
+ | 'query_vector', array(1.0f, 0.0f),
+ | 'limit', 2,
+ | 'weight', 1.0f,
+ | 'options', map())),
+ | array(
+ | named_struct(
+ | 'column', 'content',
+ | 'query',
'{"match":{"column":"content","terms":"paimon"}}',
+ | 'limit', 2,
+ | 'weight', 1.0f,
+ | 'options', map())),
+ | 3,
+ | 'rrf')
+ |ORDER BY __paimon_search_score DESC, id
+ |""".stripMargin)
+ .collect()
+
+ assert(rows.length == 3)
+ assert(rows.head.getInt(0) == 1)
+ assert(rows.forall(row => !row.isNullAt(1) && row.getFloat(1) > 0.0f))
+ }
+ }
+
test("hybrid search ranks results from multiple vector columns") {
withTable("T") {
spark.sql("""
