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 16af7fd137 [core] Search primary-key full-text indexes (#8652)
16af7fd137 is described below
commit 16af7fd137767350c03962c7dddf5ff892d48ef9
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Jul 15 18:10:12 2026 +0800
[core] Search primary-key full-text indexes (#8652)
Build on #8649 and #8651 by adding the core read path for primary-key
full-text indexes.
---
.../pkfulltext/PrimaryKeyFullTextBucketSearch.java | 262 +++++++++++++
...eadImpl.java => DataEvolutionFullTextRead.java} | 9 +-
...canImpl.java => DataEvolutionFullTextScan.java} | 12 +-
.../table/source/FullTextSearchBuilderImpl.java | 43 ++-
.../table/source/HybridSearchBuilderImpl.java | 5 +
.../table/source/PrimaryKeyFullTextRead.java | 224 +++++++++++
.../table/source/PrimaryKeyFullTextScan.java | 303 +++++++++++++++
.../source/PrimaryKeyFullTextSearchSplit.java | 167 +++++++++
.../table/source/PrimaryKeySearchRanker.java | 29 ++
.../PrimaryKeyFullTextBucketSearchTest.java | 416 +++++++++++++++++++++
.../table/source/FullTextSearchBuilderTest.java | 83 +++-
.../table/source/PrimaryKeyFullTextReadTest.java | 162 ++++++++
.../table/source/PrimaryKeyFullTextScanTest.java | 307 +++++++++++++++
.../table/source/PrimaryKeyFullTextSearchTest.java | 120 ++++++
.../index/NativePrimaryKeyFullTextIndexTest.java | 267 +++++++++++++
15 files changed, 2393 insertions(+), 16 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/pkfulltext/PrimaryKeyFullTextBucketSearch.java
b/paimon-core/src/main/java/org/apache/paimon/index/pkfulltext/PrimaryKeyFullTextBucketSearch.java
new file mode 100644
index 0000000000..31f95a8839
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/index/pkfulltext/PrimaryKeyFullTextBucketSearch.java
@@ -0,0 +1,262 @@
+/*
+ * 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.index.pkfulltext;
+
+import org.apache.paimon.deletionvectors.DeletionVector;
+import org.apache.paimon.globalindex.GlobalIndexReader;
+import org.apache.paimon.globalindex.ScoredGlobalIndexResult;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceMeta;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.predicate.FullTextSearch;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.PrimaryKeyFullTextSearchSplit;
+import org.apache.paimon.table.source.PrimaryKeySearchPosition;
+import org.apache.paimon.table.source.PrimaryKeySearchRanker;
+import org.apache.paimon.utils.IOUtils;
+import org.apache.paimon.utils.Range;
+import org.apache.paimon.utils.RoaringNavigableMap64;
+
+import javax.annotation.Nullable;
+
+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 static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Searches source-backed full-text payloads and merges their scored
candidates. */
+public class PrimaryKeyFullTextBucketSearch {
+
+ private final ReaderFactory readerFactory;
+
+ public PrimaryKeyFullTextBucketSearch(ReaderFactory readerFactory) {
+ this.readerFactory = readerFactory;
+ }
+
+ public List<PrimaryKeySearchPosition> search(
+ PrimaryKeyFullTextSearchSplit split,
+ Map<String, DeletionVector> deletionVectors,
+ String column,
+ String query,
+ int limit) {
+ List<List<PrimaryKeySearchPosition>> localRankings =
+ searchRankings(split, deletionVectors, column, query, limit);
+ if (localRankings.isEmpty()) {
+ return Collections.emptyList();
+ }
+ return PrimaryKeySearchRanker.topKByScore(localRankings, limit);
+ }
+
+ public List<List<PrimaryKeySearchPosition>> searchRankings(
+ PrimaryKeyFullTextSearchSplit split,
+ Map<String, DeletionVector> deletionVectors,
+ String column,
+ String query,
+ int limit) {
+ checkArgument(limit > 0, "Full-text search limit must be positive:
%s.", limit);
+ DataSplit dataSplit = split.dataSplit();
+ Map<String, DataFileMeta> files = new HashMap<>();
+ for (DataFileMeta file : dataSplit.dataFiles()) {
+ checkArgument(
+ files.put(file.fileName(), file) == null,
+ "Duplicate full-text source file %s.",
+ file.fileName());
+ }
+
+ List<PayloadRequest> requests = new ArrayList<>();
+ for (IndexFileMeta payload : split.payloadFiles()) {
+ List<SourceRange> sourceRanges = new ArrayList<>();
+ boolean needsInclude = false;
+ long totalRowCount = 0;
+ for (PrimaryKeyIndexSourceFile source :
+
PrimaryKeyIndexSourceMeta.fromIndexFile(payload).sourceFiles()) {
+ DataFileMeta file = files.get(source.fileName());
+ if (file != null) {
+ checkArgument(
+ source.rowCount() == file.rowCount(),
+ "Full-text payload %s source row count does not
match data file %s.",
+ payload.fileName(),
+ source.fileName());
+ }
+ DeletionVector deletionVector =
deletionVectors.get(source.fileName());
+ needsInclude |=
+ file == null || (deletionVector != null &&
!deletionVector.isEmpty());
+ sourceRanges.add(
+ new SourceRange(
+ source.fileName(), totalRowCount,
source.rowCount(), file != null));
+ totalRowCount = Math.addExact(totalRowCount,
source.rowCount());
+ }
+ RoaringNavigableMap64 include =
+ needsInclude ? liveRows(sourceRanges, deletionVectors) :
null;
+ if (include != null && include.isEmpty()) {
+ continue;
+ }
+ GlobalIndexReader reader = readerFactory.create(payload);
+ CompletableFuture<Optional<ScoredGlobalIndexResult>> future;
+ try {
+ FullTextSearch predicate = new FullTextSearch(column, query,
limit);
+ if (include != null) {
+ predicate.withIncludeRowIds(include);
+ }
+ future =
+ reader.visitFullTextSearch(predicate)
+ .whenComplete((ignored, error) ->
IOUtils.closeQuietly(reader));
+ } catch (RuntimeException | Error t) {
+ IOUtils.closeQuietly(reader);
+ throw t;
+ }
+ requests.add(new PayloadRequest(sourceRanges, totalRowCount,
include, future));
+ }
+
+ CompletableFuture.allOf(
+ requests.stream()
+ .map(request -> request.future)
+ .toArray(CompletableFuture[]::new))
+ .join();
+ List<List<PrimaryKeySearchPosition>> localRankings = new
ArrayList<>(requests.size());
+ for (PayloadRequest request : requests) {
+ Optional<ScoredGlobalIndexResult> result = request.future.join();
+ if (!result.isPresent()) {
+ continue;
+ }
+ ScoredGlobalIndexResult scored = result.get();
+ List<PrimaryKeySearchPosition> ranking = new ArrayList<>();
+ for (long rowId : scored.results()) {
+ checkArgument(
+ rowId >= 0 && rowId < request.totalRowCount,
+ "Full-text index returned archive row position %s
outside row count %s.",
+ rowId,
+ request.totalRowCount);
+ if (request.include != null &&
!request.include.contains(rowId)) {
+ continue;
+ }
+ SourceRange source = request.source(rowId);
+ checkArgument(
+ source != null && source.active,
+ "Full-text index returned row position %s from an
inactive source.",
+ rowId);
+ ranking.add(
+ new PrimaryKeySearchPosition(
+ dataSplit.partition(),
+ dataSplit.bucket(),
+ source.fileName,
+ rowId - source.offset,
+ scored.scoreGetter().score(rowId)));
+ }
+ ranking.sort(
+ (left, right) -> {
+ int scoreOrder = Float.compare(right.score(),
left.score());
+ if (scoreOrder != 0) {
+ return scoreOrder;
+ }
+ int fileOrder =
left.dataFileName().compareTo(right.dataFileName());
+ return fileOrder != 0
+ ? fileOrder
+ : Long.compare(left.rowPosition(),
right.rowPosition());
+ });
+ localRankings.add(ranking);
+ }
+ return Collections.unmodifiableList(localRankings);
+ }
+
+ private static RoaringNavigableMap64 liveRows(
+ List<SourceRange> sourceRanges, Map<String, DeletionVector>
deletionVectors) {
+ RoaringNavigableMap64 include = new RoaringNavigableMap64();
+ RoaringNavigableMap64 deleted = new RoaringNavigableMap64();
+ for (SourceRange source : sourceRanges) {
+ if (!source.active) {
+ continue;
+ }
+ if (source.rowCount > 0) {
+ include.addRange(
+ new Range(
+ source.offset, Math.addExact(source.offset,
source.rowCount) - 1));
+ }
+ DeletionVector deletionVector =
deletionVectors.get(source.fileName);
+ if (deletionVector != null && !deletionVector.isEmpty()) {
+ deletionVector.forEachDeletedPosition(
+ position -> {
+ checkArgument(
+ position >= 0 && position <
source.rowCount,
+ "Deletion vector contains invalid row
position %s.",
+ position);
+ deleted.add(Math.addExact(source.offset,
position));
+ });
+ }
+ }
+ include.andNot(deleted);
+ return include;
+ }
+
+ /** Creates one independently closeable reader for an immutable payload
archive. */
+ @FunctionalInterface
+ public interface ReaderFactory {
+ GlobalIndexReader create(IndexFileMeta payload);
+ }
+
+ private static class PayloadRequest {
+
+ private final List<SourceRange> sourceRanges;
+ private final long totalRowCount;
+ @Nullable private final RoaringNavigableMap64 include;
+ private final CompletableFuture<Optional<ScoredGlobalIndexResult>>
future;
+
+ private PayloadRequest(
+ List<SourceRange> sourceRanges,
+ long totalRowCount,
+ @Nullable RoaringNavigableMap64 include,
+ CompletableFuture<Optional<ScoredGlobalIndexResult>> future) {
+ this.sourceRanges = sourceRanges;
+ this.totalRowCount = totalRowCount;
+ this.include = include;
+ this.future = future;
+ }
+
+ @Nullable
+ private SourceRange source(long rowId) {
+ for (SourceRange source : sourceRanges) {
+ if (rowId >= source.offset && rowId < source.offset +
source.rowCount) {
+ return source;
+ }
+ }
+ return null;
+ }
+ }
+
+ private static class SourceRange {
+
+ private final String fileName;
+ private final long offset;
+ private final long rowCount;
+ private final boolean active;
+
+ private SourceRange(String fileName, long offset, long rowCount,
boolean active) {
+ this.fileName = fileName;
+ this.offset = offset;
+ this.rowCount = rowCount;
+ this.active = active;
+ }
+ }
+}
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/DataEvolutionFullTextRead.java
similarity index 97%
rename from
paimon-core/src/main/java/org/apache/paimon/table/source/FullTextReadImpl.java
rename to
paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextRead.java
index 0a2134c7a5..bf74d2e036 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/DataEvolutionFullTextRead.java
@@ -53,7 +53,7 @@ import static
org.apache.paimon.CoreOptions.GLOBAL_INDEX_THREAD_NUM;
import static org.apache.paimon.utils.Preconditions.checkNotNull;
/** Implementation for {@link FullTextRead}. */
-public class FullTextReadImpl implements FullTextRead {
+public class DataEvolutionFullTextRead implements FullTextRead {
private final FileStoreTable table;
@Nullable private final PartitionPredicate partitionFilter;
@@ -61,16 +61,17 @@ public class FullTextReadImpl implements FullTextRead {
private final DataField textColumn;
private final String query;
- public FullTextReadImpl(FileStoreTable table, int limit, DataField
textColumn, String query) {
+ public DataEvolutionFullTextRead(
+ FileStoreTable table, int limit, DataField textColumn, String
query) {
this(table, null, limit, Collections.singletonList(textColumn), query);
}
- public FullTextReadImpl(
+ public DataEvolutionFullTextRead(
FileStoreTable table, int limit, List<DataField> textColumns,
String query) {
this(table, null, limit, textColumns, query);
}
- public FullTextReadImpl(
+ public DataEvolutionFullTextRead(
FileStoreTable table,
@Nullable PartitionPredicate partitionFilter,
int limit,
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/DataEvolutionFullTextScan.java
similarity index 97%
rename from
paimon-core/src/main/java/org/apache/paimon/table/source/FullTextScanImpl.java
rename to
paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextScan.java
index ecf6719f74..bb250dbe60 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/DataEvolutionFullTextScan.java
@@ -50,22 +50,22 @@ import java.util.stream.Collectors;
import static org.apache.paimon.utils.Preconditions.checkNotNull;
/** Implementation for {@link FullTextScan}. */
-public class FullTextScanImpl implements FullTextScan {
+public class DataEvolutionFullTextScan implements FullTextScan {
private final FileStoreTable table;
private final PartitionPredicate partitionFilter;
private final List<DataField> textColumns;
- public FullTextScanImpl(FileStoreTable table, DataField textColumn) {
+ public DataEvolutionFullTextScan(FileStoreTable table, DataField
textColumn) {
this(table, null, textColumn);
}
- public FullTextScanImpl(
+ public DataEvolutionFullTextScan(
FileStoreTable table, PartitionPredicate partitionFilter,
DataField textColumn) {
this(table, partitionFilter, Collections.singletonList(textColumn));
}
- public FullTextScanImpl(
+ public DataEvolutionFullTextScan(
FileStoreTable table, PartitionPredicate partitionFilter,
List<DataField> textColumns) {
this.table = table;
this.partitionFilter = partitionFilter;
@@ -94,7 +94,7 @@ public class FullTextScanImpl implements FullTextScan {
return false;
}
GlobalIndexMeta globalIndex =
entry.indexFile().globalIndexMeta();
- if (globalIndex == null) {
+ if (globalIndex == null || globalIndex.sourceMeta() !=
null) {
return false;
}
return !matchedTextColumnIds(globalIndex,
textColumnIds).isEmpty()
@@ -216,7 +216,7 @@ public class FullTextScanImpl implements FullTextScan {
List<Long> sortedBoundaries = new ArrayList<>(boundaries);
Map<IndexRangeCandidate, List<Range>> assigned = new
LinkedHashMap<>();
TreeSet<IndexRangeCandidate> active =
- new TreeSet<>(FullTextScanImpl::compareCandidates);
+ new
TreeSet<>(DataEvolutionFullTextScan::compareCandidates);
for (int i = 0; i < sortedBoundaries.size(); i++) {
long from = sortedBoundaries.get(i);
List<IndexRangeCandidate> ending = endingAt.get(from);
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 a46d174f04..bfd2c31199 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,13 +18,15 @@
package org.apache.paimon.table.source;
+import org.apache.paimon.index.pk.PrimaryKeyIndexDefinition;
+import org.apache.paimon.index.pk.PrimaryKeyIndexDefinitions;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.InnerTable;
import org.apache.paimon.types.DataField;
import java.util.Collections;
-import java.util.List;
+import java.util.Optional;
import static org.apache.paimon.utils.Preconditions.checkArgument;
import static org.apache.paimon.utils.Preconditions.checkNotNull;
@@ -66,20 +68,51 @@ public class FullTextSearchBuilderImpl implements
FullTextSearchBuilder {
@Override
public FullTextScan newFullTextScan() {
- return new FullTextScanImpl(table, partitionFilter, textColumns());
+ DataField textColumn = textColumn();
+ Optional<PrimaryKeyIndexDefinition> definition =
primaryKeyFullTextDefinition(textColumn);
+ return definition.isPresent()
+ ? new PrimaryKeyFullTextScan(table, definition.get(),
partitionFilter)
+ : new DataEvolutionFullTextScan(
+ table, partitionFilter,
Collections.singletonList(textColumn));
}
@Override
public FullTextRead newFullTextRead() {
checkArgument(limit > 0, "Limit must be positive, set via
withLimit()");
- return new FullTextReadImpl(table, partitionFilter, limit,
textColumns(), query);
+ DataField textColumn = textColumn();
+ Optional<PrimaryKeyIndexDefinition> definition =
primaryKeyFullTextDefinition(textColumn);
+ return definition.isPresent()
+ ? new PrimaryKeyFullTextRead(table, definition.get(),
textColumn, query, limit)
+ : new DataEvolutionFullTextRead(
+ table,
+ partitionFilter,
+ limit,
+ Collections.singletonList(textColumn),
+ query);
}
- private List<DataField> textColumns() {
+ private DataField textColumn() {
checkNotNull(query, "Query must be set via withQuery()");
checkNotNull(fieldName, "Field name must be set via withQuery()");
DataField textColumn = table.rowType().getField(fieldName);
checkNotNull(textColumn, "Text column '%s' does not exist.",
fieldName);
- return Collections.singletonList(textColumn);
+ return textColumn;
+ }
+
+ private Optional<PrimaryKeyIndexDefinition>
primaryKeyFullTextDefinition(DataField textColumn) {
+ if (table.coreOptions().dataEvolutionEnabled()) {
+ return Optional.empty();
+ }
+ if (table.coreOptions().primaryKeyFullTextIndexColumns().isEmpty()) {
+ return Optional.empty();
+ }
+ for (PrimaryKeyIndexDefinition definition :
+
PrimaryKeyIndexDefinitions.create(table.schema()).definitions()) {
+ if (definition.family() ==
PrimaryKeyIndexDefinition.Family.FULL_TEXT
+ && definition.fieldId() == textColumn.id()) {
+ return Optional.of(definition);
+ }
+ }
+ return Optional.empty();
}
}
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 6f5983ef15..48531ca358 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
@@ -200,6 +200,11 @@ 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/PrimaryKeyFullTextRead.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyFullTextRead.java
new file mode 100644
index 0000000000..3374141a33
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyFullTextRead.java
@@ -0,0 +1,224 @@
+/*
+ * 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.CoreOptions.GlobalIndexSearchMode;
+import org.apache.paimon.deletionvectors.DeletionVector;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.globalindex.GlobalIndexIOMeta;
+import org.apache.paimon.globalindex.GlobalIndexReadThreadPool;
+import org.apache.paimon.globalindex.GlobalIndexReader;
+import org.apache.paimon.globalindex.GlobalIndexer;
+import org.apache.paimon.globalindex.io.GlobalIndexFileReader;
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.index.IndexFileHandler;
+import org.apache.paimon.index.pk.PrimaryKeyIndexDefinition;
+import org.apache.paimon.index.pkfulltext.PkFullTextIndexFile;
+import org.apache.paimon.index.pkfulltext.PrimaryKeyFullTextBucketSearch;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.types.DataField;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+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.ExecutorService;
+
+import static org.apache.paimon.CoreOptions.GLOBAL_INDEX_THREAD_NUM;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+
+/** Reads compaction-visible primary-key full-text payloads in fast search
mode. */
+public class PrimaryKeyFullTextRead implements FullTextRead {
+
+ private final int limit;
+ private final BucketRankingSearch indexedSearch;
+
+ public PrimaryKeyFullTextRead(
+ FileStoreTable table,
+ PrimaryKeyIndexDefinition definition,
+ DataField textField,
+ String query,
+ int limit) {
+ checkArgument(limit > 0, "Full-text search limit must be positive:
%s.", limit);
+ checkArgument(
+ definition.family() ==
PrimaryKeyIndexDefinition.Family.FULL_TEXT,
+ "Primary-key full-text read requires a full-text definition.");
+ checkArgument(
+ definition.fieldId() == textField.id(),
+ "Full-text definition does not match field %s.",
+ textField.name());
+ checkFastSearchMode(table.coreOptions().globalIndexSearchMode());
+ ProductionSearch production =
+ new ProductionSearch(table, definition, textField, query,
limit);
+ this.limit = limit;
+ this.indexedSearch = production::searchIndexed;
+ }
+
+ PrimaryKeyFullTextRead(
+ GlobalIndexSearchMode searchMode, int limit, BucketRankingSearch
indexedSearch) {
+ checkArgument(limit > 0, "Full-text search limit must be positive:
%s.", limit);
+ checkFastSearchMode(searchMode);
+ this.limit = limit;
+ this.indexedSearch = indexedSearch;
+ }
+
+ private static void checkFastSearchMode(GlobalIndexSearchMode searchMode) {
+ if (searchMode != GlobalIndexSearchMode.FAST) {
+ throw new UnsupportedOperationException(
+ "Primary-key full-text search only supports the FAST
global-index search mode; "
+ + "FULL and DETAIL require merge-aware logical-row
fallback.");
+ }
+ }
+
+ @Override
+ public PrimaryKeyScoredResult read(FullTextScan.Plan plan) {
+ checkArgument(
+ plan instanceof PrimaryKeyFullTextScan.Plan,
+ "Primary-key full-text read requires a PrimaryKeyFullTextScan
plan.");
+ PrimaryKeyFullTextScan.Plan primaryKeyPlan =
(PrimaryKeyFullTextScan.Plan) plan;
+ return read(primaryKeyPlan.snapshotId(), primaryKeyPlan.splits());
+ }
+
+ @Override
+ public PrimaryKeyScoredResult read(List<FullTextSearchSplit> splits) {
+ if (splits.isEmpty()) {
+ return new PrimaryKeyScoredResult(0, Collections.emptyList(),
Collections.emptyList());
+ }
+ checkArgument(
+ splits.get(0) instanceof PrimaryKeyFullTextSearchSplit,
+ "Primary-key full-text read requires primary-key full-text
splits.");
+ long snapshotId = ((PrimaryKeyFullTextSearchSplit)
splits.get(0)).dataSplit().snapshotId();
+ return read(snapshotId, splits);
+ }
+
+ private PrimaryKeyScoredResult read(long snapshotId,
List<FullTextSearchSplit> splits) {
+ List<DataSplit> sourceSplits = new ArrayList<>(splits.size());
+ List<List<PrimaryKeySearchPosition>> rankings = new ArrayList<>();
+ for (FullTextSearchSplit searchSplit : splits) {
+ checkArgument(
+ searchSplit instanceof PrimaryKeyFullTextSearchSplit,
+ "Primary-key full-text read received an incompatible
split.");
+ PrimaryKeyFullTextSearchSplit split =
(PrimaryKeyFullTextSearchSplit) searchSplit;
+ checkArgument(
+ split.dataSplit().snapshotId() == snapshotId,
+ "Full-text bucket split snapshot does not match its
plan.");
+ sourceSplits.add(split.dataSplit());
+ rankings.addAll(indexedSearch.search(split));
+ }
+ List<PrimaryKeySearchPosition> positions =
+ rankings.isEmpty()
+ ? Collections.emptyList()
+ : PrimaryKeySearchRanker.topKByScore(rankings, limit);
+ return new PrimaryKeyScoredResult(snapshotId, sourceSplits, positions);
+ }
+
+ @FunctionalInterface
+ interface BucketRankingSearch {
+ List<List<PrimaryKeySearchPosition>>
search(PrimaryKeyFullTextSearchSplit split);
+ }
+
+ private static class ProductionSearch {
+
+ private final DataField textField;
+ private final String query;
+ private final int limit;
+ private final FileIO fileIO;
+ private final IndexFileHandler indexFileHandler;
+ private final ExecutorService executor;
+ private final GlobalIndexer indexer;
+ private final GlobalIndexFileReader archiveReader;
+
+ private ProductionSearch(
+ FileStoreTable table,
+ PrimaryKeyIndexDefinition definition,
+ DataField textField,
+ String query,
+ int limit) {
+ this.textField = textField;
+ this.query = checkNotNull(query, "Full-text query must not be
null.");
+ this.limit = limit;
+ this.fileIO = table.fileIO();
+ this.indexFileHandler = table.store().newIndexFileHandler();
+ this.executor =
+ GlobalIndexReadThreadPool.getExecutorService(
+
table.coreOptions().toConfiguration().get(GLOBAL_INDEX_THREAD_NUM));
+ this.indexer =
+ GlobalIndexer.create(
+ PkFullTextIndexFile.INDEX_TYPE, textField,
definition.options());
+ this.archiveReader = meta ->
fileIO.newInputStream(meta.filePath());
+ }
+
+ private List<List<PrimaryKeySearchPosition>> searchIndexed(
+ PrimaryKeyFullTextSearchSplit split) {
+ if (split.payloadFiles().isEmpty()) {
+ return Collections.emptyList();
+ }
+ return bucketSearch(split)
+ .searchRankings(
+ split,
+ deletionVectors(split.dataSplit()),
+ textField.name(),
+ query,
+ limit);
+ }
+
+ private PrimaryKeyFullTextBucketSearch
bucketSearch(PrimaryKeyFullTextSearchSplit split) {
+ PkFullTextIndexFile indexFile =
+ indexFileHandler.pkFullTextIndex(
+ split.dataSplit().partition(),
split.dataSplit().bucket());
+ return new PrimaryKeyFullTextBucketSearch(
+ payload -> {
+ GlobalIndexMeta meta =
checkNotNull(payload.globalIndexMeta());
+ GlobalIndexIOMeta ioMeta =
+ new GlobalIndexIOMeta(
+ indexFile.path(payload),
+ payload.fileSize(),
+ meta.indexMeta());
+ GlobalIndexReader reader =
+ indexer.createReader(
+ archiveReader,
Collections.singletonList(ioMeta), executor);
+ return reader;
+ });
+ }
+
+ private Map<String, DeletionVector> deletionVectors(DataSplit split) {
+ try {
+ DeletionVector.Factory factory =
+ DeletionVector.factory(
+ fileIO, split.dataFiles(),
split.deletionFiles().orElse(null));
+ Map<String, DeletionVector> result = new HashMap<>();
+ for (DataFileMeta file : split.dataFiles()) {
+ Optional<DeletionVector> deletionVector =
factory.create(file.fileName());
+ if (deletionVector.isPresent()) {
+ result.put(file.fileName(), deletionVector.get());
+ }
+ }
+ return result;
+ } catch (IOException e) {
+ throw new UncheckedIOException("Failed to read full-text
deletion vectors.", e);
+ }
+ }
+ }
+}
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
new file mode 100644
index 0000000000..d160523510
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyFullTextScan.java
@@ -0,0 +1,303 @@
+/*
+ * 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.Snapshot;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.globalindex.IndexedSplit;
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.index.IndexFileHandler;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.pk.PrimaryKeyIndexDefinition;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourcePolicy;
+import org.apache.paimon.index.pkfulltext.PkFullTextBucketIndexState;
+import org.apache.paimon.index.pkfulltext.PkFullTextIndexFile;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.source.snapshot.SnapshotReader;
+import org.apache.paimon.utils.Pair;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Plans compaction-visible full-text inputs from one captured primary-key
snapshot. */
+public class PrimaryKeyFullTextScan implements FullTextScan {
+
+ private final FileStoreTable table;
+ private final PrimaryKeyIndexDefinition definition;
+ @Nullable private final PartitionPredicate partitionFilter;
+
+ public PrimaryKeyFullTextScan(
+ FileStoreTable table,
+ PrimaryKeyIndexDefinition definition,
+ @Nullable PartitionPredicate partitionFilter) {
+ 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;
+ }
+
+ @Override
+ public Plan scan() {
+ SnapshotReader snapshotReader = table.newSnapshotReader().keepStats();
+ DataTableScan dataScan = table.newScan(ignored -> snapshotReader);
+ checkArgument(
+ dataScan instanceof PrimaryKeyBatchScan,
+ "Primary-key full-text search requires a primary-key batch
scan.");
+ PrimaryKeyBatchScan batchScan = (PrimaryKeyBatchScan) dataScan;
+ if (partitionFilter != null) {
+ batchScan.withPartitionFilter(partitionFilter);
+ }
+ TableScan.Plan tablePlan = batchScan.planWithoutAuth();
+ if (!(tablePlan instanceof SnapshotReader.Plan)) {
+ checkArgument(
+ tablePlan.splits().isEmpty(),
+ "Primary-key full-text search requires a snapshot plan.");
+ return new Plan(0, Collections.emptyList());
+ }
+ SnapshotReader.Plan snapshotPlan = (SnapshotReader.Plan) tablePlan;
+ if (snapshotPlan.snapshotId() == null) {
+ return new Plan(0, Collections.emptyList());
+ }
+ Snapshot snapshot =
snapshotReader.snapshotManager().snapshot(snapshotPlan.snapshotId());
+ checkArgument(snapshot != null, "Primary-key full-text snapshot does
not exist.");
+
+ IndexFileHandler indexFileHandler = snapshotReader.indexFileHandler();
+ checkArgument(
+ indexFileHandler != null, "Primary-key full-text index handler
is unavailable.");
+ List<IndexManifestEntry> payloadEntries =
+ indexFileHandler.scan(
+ snapshot,
+ entry ->
+ matchesDefinition(entry)
+ && (partitionFilter == null
+ ||
partitionFilter.test(entry.partition())));
+ return plan(snapshot.id(), snapshotPlan.splits(), payloadEntries,
definition.fieldId());
+ }
+
+ private boolean matchesDefinition(IndexManifestEntry entry) {
+ IndexFileMeta payload = entry.indexFile();
+ GlobalIndexMeta globalMeta = payload.globalIndexMeta();
+ if (!PkFullTextIndexFile.INDEX_TYPE.equals(payload.indexType())
+ || globalMeta == null
+ || globalMeta.sourceMeta() == null
+ || globalMeta.indexFieldId() != definition.fieldId()) {
+ return false;
+ }
+ return true;
+ }
+
+ static Plan plan(
+ long snapshotId,
+ List<? extends Split> dataSplits,
+ List<IndexManifestEntry> payloadEntries,
+ int textFieldId) {
+ Map<Pair<BinaryRow, Integer>, List<IndexFileMeta>> payloads = new
LinkedHashMap<>();
+ for (IndexManifestEntry entry : payloadEntries) {
+ checkArgument(
+ entry.kind() == FileKind.ADD,
+ "Primary-key full-text index file %s is not active.",
+ entry.indexFile().fileName());
+ Pair<BinaryRow, Integer> key = Pair.of(entry.partition(),
entry.bucket());
+ payloads.computeIfAbsent(key, ignored -> new
ArrayList<>()).add(entry.indexFile());
+ }
+
+ Map<Pair<BinaryRow, Integer>, BucketAccumulator> buckets = new
LinkedHashMap<>();
+ for (Split split : dataSplits) {
+ DataSplit dataSplit = unwrapDataSplit(split);
+ checkArgument(
+ dataSplit.snapshotId() == snapshotId,
+ "Data split snapshot %s does not match full-text scan
snapshot %s.",
+ dataSplit.snapshotId(),
+ snapshotId);
+ checkArgument(
+ !dataSplit.isStreaming(),
+ "Primary-key full-text search requires a batch split.");
+ if (dataSplit.bucket() < 0) {
+ continue;
+ }
+ Pair<BinaryRow, Integer> key = Pair.of(dataSplit.partition(),
dataSplit.bucket());
+ BucketAccumulator accumulator = buckets.get(key);
+ if (accumulator == null) {
+ accumulator = new BucketAccumulator(dataSplit);
+ buckets.put(key, accumulator);
+ }
+ accumulator.add(dataSplit);
+ }
+
+ List<PrimaryKeyFullTextSearchSplit> result = new ArrayList<>();
+ for (Map.Entry<Pair<BinaryRow, Integer>, BucketAccumulator> entry :
buckets.entrySet()) {
+ BucketAccumulator bucket = entry.getValue();
+ if (bucket.isEmpty()) {
+ continue;
+ }
+ PkFullTextBucketIndexState state =
+ PkFullTextBucketIndexState.fromActivePayloads(
+ textFieldId,
+ payloads.getOrDefault(entry.getKey(),
Collections.emptyList()));
+ Set<String> activeSources = bucket.dataFileNames();
+ Map<String, IndexFileMeta> currentPayloads = new LinkedHashMap<>();
+ Set<String> covered = new HashSet<>();
+ for (Map.Entry<String, IndexFileMeta> payload :
+ state.payloadBySourceFile().entrySet()) {
+ if (activeSources.contains(payload.getKey())) {
+ currentPayloads.put(payload.getValue().fileName(),
payload.getValue());
+ covered.add(payload.getKey());
+ }
+ }
+ List<String> uncovered = new ArrayList<>();
+ for (DataFileMeta dataFile : bucket.dataFiles()) {
+ if (!covered.contains(dataFile.fileName())) {
+ uncovered.add(dataFile.fileName());
+ }
+ }
+ result.add(
+ new PrimaryKeyFullTextSearchSplit(
+ bucket.build(), new
ArrayList<>(currentPayloads.values()), uncovered));
+ }
+ return new Plan(snapshotId, result);
+ }
+
+ private static DataSplit unwrapDataSplit(Split split) {
+ if (split instanceof IndexedSplit) {
+ return ((IndexedSplit) split).dataSplit();
+ }
+ checkArgument(
+ split instanceof DataSplit,
+ "Unsupported primary-key full-text source split: %s.",
+ split.getClass().getName());
+ return (DataSplit) split;
+ }
+
+ /** Immutable snapshot full-text plan. */
+ public static class Plan implements FullTextScan.Plan {
+
+ private final long snapshotId;
+ private final List<FullTextSearchSplit> splits;
+
+ private Plan(long snapshotId, List<PrimaryKeyFullTextSearchSplit>
splits) {
+ this.snapshotId = snapshotId;
+ this.splits = Collections.unmodifiableList(new
ArrayList<FullTextSearchSplit>(splits));
+ }
+
+ public long snapshotId() {
+ return snapshotId;
+ }
+
+ @Override
+ public List<FullTextSearchSplit> splits() {
+ return splits;
+ }
+ }
+
+ private static class BucketAccumulator {
+
+ private final long snapshotId;
+ private final BinaryRow partition;
+ private final int bucket;
+ private final String bucketPath;
+ @Nullable private final Integer totalBuckets;
+ private final List<DataFileMeta> dataFiles = new ArrayList<>();
+ private final List<DeletionFile> deletionFiles = new ArrayList<>();
+ private final Set<String> dataFileNames = new HashSet<>();
+ private boolean hasDeletionMetadata;
+
+ private BucketAccumulator(DataSplit split) {
+ this.snapshotId = split.snapshotId();
+ this.partition = split.partition();
+ this.bucket = split.bucket();
+ this.bucketPath = split.bucketPath();
+ this.totalBuckets = split.totalBuckets();
+ }
+
+ private void add(DataSplit split) {
+ checkArgument(
+ snapshotId == split.snapshotId()
+ && partition.equals(split.partition())
+ && bucket == split.bucket()
+ && bucketPath.equals(split.bucketPath()),
+ "Cannot combine data splits from different snapshot
buckets.");
+ checkArgument(
+ totalBuckets == null
+ ? split.totalBuckets() == null
+ : totalBuckets.equals(split.totalBuckets()),
+ "Bucket split total-bucket metadata is inconsistent.");
+ List<DeletionFile> splitDeletions =
split.deletionFiles().orElse(null);
+ checkArgument(
+ splitDeletions == null || splitDeletions.size() ==
split.dataFiles().size(),
+ "Deletion files must align with data files in a full-text
bucket split.");
+ hasDeletionMetadata |= splitDeletions != null;
+ for (int i = 0; i < split.dataFiles().size(); i++) {
+ DataFileMeta file = split.dataFiles().get(i);
+ if (!PrimaryKeyIndexSourcePolicy.shouldRead(file)) {
+ continue;
+ }
+ checkArgument(
+ dataFileNames.add(file.fileName()),
+ "Data file %s appears more than once in full-text
bucket planning.",
+ file.fileName());
+ dataFiles.add(file);
+ deletionFiles.add(splitDeletions == null ? null :
splitDeletions.get(i));
+ }
+ }
+
+ private boolean isEmpty() {
+ return dataFiles.isEmpty();
+ }
+
+ private List<DataFileMeta> dataFiles() {
+ return dataFiles;
+ }
+
+ private Set<String> dataFileNames() {
+ return dataFileNames;
+ }
+
+ private DataSplit build() {
+ DataSplit.Builder builder =
+ DataSplit.builder()
+ .withSnapshot(snapshotId)
+ .withPartition(partition)
+ .withBucket(bucket)
+ .withBucketPath(bucketPath)
+ .withTotalBuckets(totalBuckets)
+ .withDataFiles(dataFiles)
+ .isStreaming(false)
+ .rawConvertible(false);
+ if (hasDeletionMetadata) {
+ builder.withDataDeletionFiles(deletionFiles);
+ }
+ return builder.build();
+ }
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyFullTextSearchSplit.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyFullTextSearchSplit.java
new file mode 100644
index 0000000000..8a5bd09963
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyFullTextSearchSplit.java
@@ -0,0 +1,167 @@
+/*
+ * 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.index.IndexFileMeta;
+import org.apache.paimon.index.IndexFileMetaSerializer;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceMeta;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataInputViewStreamWrapper;
+import org.apache.paimon.io.DataOutputViewStreamWrapper;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Compaction-visible data files and full-text payloads for one snapshot
bucket. */
+public class PrimaryKeyFullTextSearchSplit extends FullTextSearchSplit {
+
+ private static final long serialVersionUID = 1L;
+ private static final int VERSION = 1;
+
+ private DataSplit dataSplit;
+ private transient List<IndexFileMeta> payloadFiles;
+ private List<String> uncoveredDataFiles;
+
+ public PrimaryKeyFullTextSearchSplit(
+ DataSplit dataSplit,
+ List<IndexFileMeta> payloadFiles,
+ List<String> uncoveredDataFiles) {
+ checkArgument(
+ !dataSplit.isStreaming(), "Primary-key full-text search
requires a batch split.");
+ Set<String> sourceFiles = new HashSet<>();
+ for (DataFileMeta dataFile : dataSplit.dataFiles()) {
+ checkArgument(
+ sourceFiles.add(dataFile.fileName()),
+ "Data file %s appears more than once in a full-text bucket
split.",
+ dataFile.fileName());
+ }
+
+ Set<String> covered = new HashSet<>();
+ for (IndexFileMeta payload : payloadFiles) {
+ boolean coversActiveSource = false;
+ for (PrimaryKeyIndexSourceFile source :
+
PrimaryKeyIndexSourceMeta.fromIndexFile(payload).sourceFiles()) {
+ if (!sourceFiles.contains(source.fileName())) {
+ continue;
+ }
+ coversActiveSource = true;
+ checkArgument(
+ covered.add(source.fileName()),
+ "Data file %s is covered by more than one full-text
payload.",
+ source.fileName());
+ }
+ checkArgument(
+ coversActiveSource,
+ "Full-text payload %s does not cover an active data file
in its bucket split.",
+ payload.fileName());
+ }
+
+ Set<String> uncovered = new HashSet<>();
+ for (String source : uncoveredDataFiles) {
+ checkArgument(
+ sourceFiles.contains(source),
+ "Uncovered full-text data file %s is outside its bucket
split.",
+ source);
+ checkArgument(
+ uncovered.add(source),
+ "Uncovered full-text data file %s appears more than once.",
+ source);
+ checkArgument(
+ !covered.contains(source),
+ "Data file %s cannot be both indexed and uncovered.",
+ source);
+ }
+ checkArgument(
+ covered.size() + uncovered.size() == sourceFiles.size(),
+ "Every full-text source file must be indexed or explicitly
uncovered.");
+
+ this.dataSplit = dataSplit;
+ this.payloadFiles = Collections.unmodifiableList(new
ArrayList<>(payloadFiles));
+ this.uncoveredDataFiles = Collections.unmodifiableList(new
ArrayList<>(uncoveredDataFiles));
+ }
+
+ public DataSplit dataSplit() {
+ return dataSplit;
+ }
+
+ public List<IndexFileMeta> payloadFiles() {
+ return payloadFiles;
+ }
+
+ public List<String> uncoveredDataFiles() {
+ return uncoveredDataFiles;
+ }
+
+ private void writeObject(ObjectOutputStream out) throws IOException {
+ out.defaultWriteObject();
+ out.writeInt(VERSION);
+ out.writeInt(payloadFiles.size());
+ IndexFileMetaSerializer serializer = new IndexFileMetaSerializer();
+ for (IndexFileMeta payloadFile : payloadFiles) {
+ serializer.serialize(payloadFile, new
DataOutputViewStreamWrapper(out));
+ }
+ }
+
+ private void readObject(ObjectInputStream in) throws IOException,
ClassNotFoundException {
+ in.defaultReadObject();
+ int version = in.readInt();
+ if (version != VERSION) {
+ throw new IOException("Unsupported PrimaryKeyFullTextSearchSplit
version: " + version);
+ }
+ int payloadFileCount = in.readInt();
+ if (payloadFileCount < 0) {
+ throw new IOException("Negative primary-key full-text payload file
count.");
+ }
+ List<IndexFileMeta> payloads = new ArrayList<>(payloadFileCount);
+ IndexFileMetaSerializer serializer = new IndexFileMetaSerializer();
+ for (int i = 0; i < payloadFileCount; i++) {
+ payloads.add(serializer.deserialize(new
DataInputViewStreamWrapper(in)));
+ }
+ this.payloadFiles = Collections.unmodifiableList(payloads);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ PrimaryKeyFullTextSearchSplit that = (PrimaryKeyFullTextSearchSplit) o;
+ return Objects.equals(dataSplit, that.dataSplit)
+ && Objects.equals(payloadFiles, that.payloadFiles)
+ && Objects.equals(uncoveredDataFiles, that.uncoveredDataFiles);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(dataSplit, payloadFiles, uncoveredDataFiles);
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeySearchRanker.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeySearchRanker.java
index 779a64a400..d4fab665eb 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeySearchRanker.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeySearchRanker.java
@@ -44,6 +44,35 @@ public final class PrimaryKeySearchRanker {
private PrimaryKeySearchRanker() {}
+ /** Selects globally highest-scored physical positions without rewriting
their scores. */
+ public static List<PrimaryKeySearchPosition> topKByScore(
+ List<List<PrimaryKeySearchPosition>> rankings, int limit) {
+ checkArgument(limit > 0, "Search result limit must be positive: %s.",
limit);
+ Map<PrimaryKeySearchPosition, PrimaryKeySearchPosition> unique = new
HashMap<>();
+ for (List<PrimaryKeySearchPosition> ranking : rankings) {
+ for (PrimaryKeySearchPosition position : ranking) {
+ PrimaryKeySearchPosition previous = unique.get(position);
+ if (previous == null || LOCAL_BEST_FIRST.compare(position,
previous) < 0) {
+ unique.put(position, position);
+ }
+ }
+ }
+
+ PriorityQueue<PrimaryKeySearchPosition> topK =
+ new PriorityQueue<>(limit, LOCAL_BEST_FIRST.reversed());
+ for (PrimaryKeySearchPosition position : unique.values()) {
+ if (topK.size() < limit) {
+ topK.add(position);
+ } else if (LOCAL_BEST_FIRST.compare(position, topK.peek()) < 0) {
+ topK.poll();
+ topK.add(position);
+ }
+ }
+ List<PrimaryKeySearchPosition> result = new ArrayList<>(topK);
+ result.sort(LOCAL_BEST_FIRST);
+ return Collections.unmodifiableList(result);
+ }
+
public static List<PrimaryKeySearchPosition> rrf(
List<List<PrimaryKeySearchPosition>> rankings, int limit) {
List<Ranking> weighted = new ArrayList<>(rankings.size());
diff --git
a/paimon-core/src/test/java/org/apache/paimon/index/pkfulltext/PrimaryKeyFullTextBucketSearchTest.java
b/paimon-core/src/test/java/org/apache/paimon/index/pkfulltext/PrimaryKeyFullTextBucketSearchTest.java
new file mode 100644
index 0000000000..09d0df4a15
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/index/pkfulltext/PrimaryKeyFullTextBucketSearchTest.java
@@ -0,0 +1,416 @@
+/*
+ * 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.index.pkfulltext;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.deletionvectors.BitmapDeletionVector;
+import org.apache.paimon.deletionvectors.DeletionVector;
+import org.apache.paimon.globalindex.GlobalIndexReader;
+import org.apache.paimon.globalindex.GlobalIndexResult;
+import org.apache.paimon.globalindex.ScoredGlobalIndexResult;
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceMeta;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.predicate.FieldRef;
+import org.apache.paimon.predicate.FullTextSearch;
+import org.apache.paimon.stats.SimpleStats;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.PrimaryKeyFullTextSearchSplit;
+import org.apache.paimon.table.source.PrimaryKeySearchPosition;
+import org.apache.paimon.utils.RoaringNavigableMap64;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assertions.tuple;
+import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
+
+/** Tests payload-local full-text search and cross-payload score merging. */
+class PrimaryKeyFullTextBucketSearchTest {
+
+ @Test
+ void testFiltersDeletedRowsAndSelectsGlobalScores() {
+ PrimaryKeyFullTextSearchSplit split = split();
+ BitmapDeletionVector deletionVector = new BitmapDeletionVector();
+ deletionVector.delete(0);
+ Map<String, DeletionVector> deletionVectors =
Collections.singletonMap("a", deletionVector);
+ Map<String, AtomicInteger> closes = new HashMap<>();
+
+ PrimaryKeyFullTextBucketSearch search =
+ new PrimaryKeyFullTextBucketSearch(
+ payload -> {
+ String source =
+
PrimaryKeyIndexSourceMeta.fromIndexFile(payload)
+ .sourceFile()
+ .fileName();
+ closes.put(source, new AtomicInteger());
+ if (source.equals("a")) {
+ return reader(scores(0, 10F, 1, 9F, 2, 8F),
closes.get(source));
+ }
+ return reader(scores(0, 4F), closes.get(source));
+ });
+
+ List<PrimaryKeySearchPosition> result =
+ search.search(split, deletionVectors, "content", "hello", 3);
+
+ assertThat(result)
+ .extracting(
+ PrimaryKeySearchPosition::dataFileName,
+ PrimaryKeySearchPosition::rowPosition)
+ .containsExactly(tuple("a", 1L), tuple("a", 2L), tuple("b",
0L));
+
assertThat(result).extracting(PrimaryKeySearchPosition::score).containsExactly(9F,
8F, 4F);
+ assertThat(closes.get("a")).hasValue(1);
+ assertThat(closes.get("b")).hasValue(1);
+ }
+
+ @Test
+ void testOrdersEachPayloadByScore() {
+ AtomicInteger closes = new AtomicInteger();
+ PrimaryKeyFullTextBucketSearch search =
+ new PrimaryKeyFullTextBucketSearch(
+ payload -> reader(scores(0, 1F, 2, 10F), closes));
+ PrimaryKeyFullTextSearchSplit split =
+ new PrimaryKeyFullTextSearchSplit(
+ dataSplit(Collections.singletonList(dataFile("a"))),
+ Collections.singletonList(payload("a", "index-a")),
+ Collections.emptyList());
+
+ List<List<PrimaryKeySearchPosition>> rankings =
+ search.searchRankings(split, Collections.emptyMap(),
"content", "hello", 2);
+
+ assertThat(rankings).hasSize(1);
+ assertThat(rankings.get(0))
+ .extracting(PrimaryKeySearchPosition::rowPosition,
PrimaryKeySearchPosition::score)
+ .containsExactly(tuple(2L, 10F), tuple(0L, 1F));
+ assertThat(closes).hasValue(1);
+ }
+
+ @Test
+ void testMapsMultiSourceRowsAndShiftsDeletionVectors() {
+ BitmapDeletionVector deletionVector = new BitmapDeletionVector();
+ deletionVector.delete(1);
+ AtomicInteger closes = new AtomicInteger();
+ PrimaryKeyFullTextBucketSearch search =
+ new PrimaryKeyFullTextBucketSearch(
+ ignored -> reader(scores(0, 10F, 3, 9F, 4, 8F, 5, 7F),
closes));
+ PrimaryKeyFullTextSearchSplit split =
+ new PrimaryKeyFullTextSearchSplit(
+ dataSplit(Arrays.asList(dataFile("a"), dataFile("b"))),
+ Collections.singletonList(payload(Arrays.asList("a",
"b"), "index-ab")),
+ Collections.emptyList());
+
+ List<List<PrimaryKeySearchPosition>> rankings =
+ search.searchRankings(
+ split,
+ Collections.singletonMap("b", deletionVector),
+ "content",
+ "hello",
+ 4);
+
+ assertThat(rankings).hasSize(1);
+ assertThat(rankings.get(0))
+ .extracting(
+ PrimaryKeySearchPosition::dataFileName,
+ PrimaryKeySearchPosition::rowPosition,
+ PrimaryKeySearchPosition::score)
+ .containsExactly(tuple("a", 0L, 10F), tuple("b", 0L, 9F),
tuple("b", 2L, 7F));
+ assertThat(closes).hasValue(1);
+ }
+
+ @Test
+ void testBuildsDeletionVectorIncludeWithoutEnumeratingLiveRows() {
+ long rowCount = 1_000_000_000L;
+ BitmapDeletionVector deletionVector = new BitmapDeletionVector();
+ deletionVector.delete(1);
+ AtomicInteger closes = new AtomicInteger();
+ PrimaryKeyFullTextBucketSearch search =
+ new PrimaryKeyFullTextBucketSearch(
+ ignored -> reader(Collections.emptyMap(), closes));
+ PrimaryKeyFullTextSearchSplit split =
+ new PrimaryKeyFullTextSearchSplit(
+ dataSplit(Collections.singletonList(dataFile("large",
rowCount))),
+ Collections.singletonList(payload("large",
"index-large", rowCount)),
+ Collections.emptyList());
+
+ assertTimeoutPreemptively(
+ Duration.ofSeconds(1),
+ () ->
+ search.searchRankings(
+ split,
+ Collections.singletonMap("large",
deletionVector),
+ "content",
+ "hello",
+ 1));
+ assertThat(closes).hasValue(1);
+ }
+
+ @Test
+ void testClosesReaderAfterFailedSearch() {
+ AtomicInteger closes = new AtomicInteger();
+ PrimaryKeyFullTextBucketSearch search =
+ new PrimaryKeyFullTextBucketSearch(
+ payload ->
+ new FullTextOnlyReader() {
+ @Override
+ public
CompletableFuture<Optional<ScoredGlobalIndexResult>>
+ visitFullTextSearch(FullTextSearch
fullTextSearch) {
+
CompletableFuture<Optional<ScoredGlobalIndexResult>>
+ failed = new
CompletableFuture<>();
+ failed.completeExceptionally(new
IOException("broken"));
+ return failed;
+ }
+
+ @Override
+ public void close() {
+ closes.incrementAndGet();
+ }
+ });
+
+ assertThatThrownBy(
+ () ->
+ search.search(
+ new PrimaryKeyFullTextSearchSplit(
+
dataSplit(Collections.singletonList(dataFile("a"))),
+
Collections.singletonList(payload("a", "index-a")),
+ Collections.emptyList()),
+ Collections.emptyMap(),
+ "content",
+ "hello",
+ 1))
+ .isInstanceOf(CompletionException.class)
+ .hasRootCauseMessage("broken");
+ assertThat(closes).hasValue(1);
+ }
+
+ private static GlobalIndexReader reader(Map<Long, Float> scores,
AtomicInteger closeCounter) {
+ return new FullTextOnlyReader() {
+ @Override
+ public CompletableFuture<Optional<ScoredGlobalIndexResult>>
visitFullTextSearch(
+ FullTextSearch fullTextSearch) {
+ RoaringNavigableMap64 rows = new RoaringNavigableMap64();
+ RoaringNavigableMap64 include = fullTextSearch.includeRowIds();
+ for (Long row : scores.keySet()) {
+ if (include == null || include.contains(row)) {
+ rows.add(row);
+ }
+ }
+ return CompletableFuture.completedFuture(
+ Optional.of(ScoredGlobalIndexResult.create(rows,
scores::get)));
+ }
+
+ @Override
+ public void close() {
+ closeCounter.incrementAndGet();
+ }
+ };
+ }
+
+ private abstract static class FullTextOnlyReader implements
GlobalIndexReader {
+
+ private CompletableFuture<Optional<GlobalIndexResult>> emptyResult() {
+ return CompletableFuture.completedFuture(Optional.empty());
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>>
visitIsNotNull(FieldRef fieldRef) {
+ return emptyResult();
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>>
visitIsNull(FieldRef fieldRef) {
+ return emptyResult();
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitStartsWith(
+ FieldRef fieldRef, Object literal) {
+ return emptyResult();
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitEndsWith(
+ FieldRef fieldRef, Object literal) {
+ return emptyResult();
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitContains(
+ FieldRef fieldRef, Object literal) {
+ return emptyResult();
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitLike(
+ FieldRef fieldRef, Object literal) {
+ return emptyResult();
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitLessThan(
+ FieldRef fieldRef, Object literal) {
+ return emptyResult();
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>>
visitGreaterOrEqual(
+ FieldRef fieldRef, Object literal) {
+ return emptyResult();
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitNotEqual(
+ FieldRef fieldRef, Object literal) {
+ return emptyResult();
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitLessOrEqual(
+ FieldRef fieldRef, Object literal) {
+ return emptyResult();
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitEqual(
+ FieldRef fieldRef, Object literal) {
+ return emptyResult();
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitGreaterThan(
+ FieldRef fieldRef, Object literal) {
+ return emptyResult();
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitIn(
+ FieldRef fieldRef, List<Object> literals) {
+ return emptyResult();
+ }
+
+ @Override
+ public CompletableFuture<Optional<GlobalIndexResult>> visitNotIn(
+ FieldRef fieldRef, List<Object> literals) {
+ return emptyResult();
+ }
+ }
+
+ private static Map<Long, Float> scores(Object... pairs) {
+ Map<Long, Float> scores = new LinkedHashMap<>();
+ for (int i = 0; i < pairs.length; i += 2) {
+ scores.put(((Number) pairs[i]).longValue(), ((Number) pairs[i +
1]).floatValue());
+ }
+ return scores;
+ }
+
+ private static PrimaryKeyFullTextSearchSplit split() {
+ List<DataFileMeta> dataFiles = Arrays.asList(dataFile("a"),
dataFile("b"));
+ return new PrimaryKeyFullTextSearchSplit(
+ dataSplit(dataFiles),
+ Arrays.asList(payload("a", "index-a"), payload("b",
"index-b")),
+ Collections.emptyList());
+ }
+
+ private static DataSplit dataSplit(List<DataFileMeta> dataFiles) {
+ return DataSplit.builder()
+ .withSnapshot(11)
+ .withPartition(BinaryRow.EMPTY_ROW)
+ .withBucket(0)
+ .withBucketPath("bucket-0")
+ .withTotalBuckets(1)
+ .withDataFiles(dataFiles)
+ .build();
+ }
+
+ private static DataFileMeta dataFile(String name) {
+ return dataFile(name, 3);
+ }
+
+ private static DataFileMeta dataFile(String name, long rowCount) {
+ return DataFileMeta.create(
+ name,
+ 100,
+ rowCount,
+ BinaryRow.EMPTY_ROW,
+ BinaryRow.EMPTY_ROW,
+ SimpleStats.EMPTY_STATS,
+ SimpleStats.EMPTY_STATS,
+ 0,
+ 0,
+ 0,
+ 1,
+ Collections.emptyList(),
+ 0L,
+ null,
+ FileSource.COMPACT,
+ null,
+ null,
+ null,
+ null);
+ }
+
+ private static IndexFileMeta payload(String source, String name) {
+ return payload(Collections.singletonList(source), name);
+ }
+
+ private static IndexFileMeta payload(String source, String name, long
rowCount) {
+ byte[] sourceMeta =
+ new PrimaryKeyIndexSourceMeta(new
PrimaryKeyIndexSourceFile(source, rowCount))
+ .serialize();
+ return new IndexFileMeta(
+ "full-text",
+ name,
+ 100,
+ rowCount,
+ new GlobalIndexMeta(0, rowCount - 1, 7, null, null,
sourceMeta),
+ null);
+ }
+
+ private static IndexFileMeta payload(List<String> sources, String name) {
+ List<PrimaryKeyIndexSourceFile> sourceFiles = new
java.util.ArrayList<>();
+ for (String source : sources) {
+ sourceFiles.add(new PrimaryKeyIndexSourceFile(source, 3));
+ }
+ byte[] sourceMeta = new
PrimaryKeyIndexSourceMeta(sourceFiles).serialize();
+ long rowCount = 3L * sourceFiles.size();
+ return new IndexFileMeta(
+ "full-text",
+ name,
+ 100,
+ rowCount,
+ new GlobalIndexMeta(0, rowCount - 1, 7, null, null,
sourceMeta),
+ null);
+ }
+}
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 c092932ef7..c351f8691a 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
@@ -34,6 +34,8 @@ import
org.apache.paimon.globalindex.btree.BTreeGlobalIndexerFactory;
import
org.apache.paimon.globalindex.testfulltext.TestFullTextGlobalIndexerFactory;
import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceMeta;
import org.apache.paimon.io.CompactIncrement;
import org.apache.paimon.io.DataIncrement;
import org.apache.paimon.options.Options;
@@ -718,7 +720,8 @@ public class FullTextSearchBuilderTest extends
TableTestBase {
// The reader's projected row layout follows this order, so these are
different index
// definitions even though they contain the same field-id set.
- assertThat(FullTextScanImpl.sameIndexIdentity(titleThenBody,
bodyThenTitle)).isFalse();
+ assertThat(DataEvolutionFullTextScan.sameIndexIdentity(titleThenBody,
bodyThenTitle))
+ .isFalse();
}
@Test
@@ -743,6 +746,25 @@ public class FullTextSearchBuilderTest extends
TableTestBase {
assertThat(searchBuilder.executeLocal().results().isEmpty()).isTrue();
}
+ @Test
+ public void testOrdinaryFullTextScanSkipsSourceBackedPrimaryKeyArchive()
throws Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+
+ String[] documents = {"Apache Paimon", "vector search"};
+ writeDocuments(table, documents);
+ buildAndCommitSourceBackedIndex(table, documents);
+
+ FullTextScan.Plan plan =
+ table.newFullTextSearchBuilder()
+ .withQuery(TEXT_FIELD_NAME, matchQuery("Paimon"))
+ .withLimit(2)
+ .newFullTextScan()
+ .scan();
+
+ assertThat(plan.splits()).isEmpty();
+ }
+
@Test
public void testFullTextSearchSplitSerialization() throws Exception {
createTableDefault();
@@ -904,6 +926,65 @@ public class FullTextSearchBuilderTest extends
TableTestBase {
}
}
+ private void buildAndCommitSourceBackedIndex(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,
+ TestFullTextGlobalIndexerFactory.IDENTIFIER,
+ textField,
+ options);
+ for (int i = 0; i < documents.length; i++) {
+ writer.write(documents[i], i);
+ }
+
+ List<IndexFileMeta> indexFiles =
+ GlobalIndexBuilderUtils.toIndexFileMetas(
+ table.fileIO(),
+ table.store().pathFactory().globalIndexFileFactory(),
+ table.coreOptions(),
+ new Range(0, documents.length - 1),
+ Collections.singletonList(textField),
+ TestFullTextGlobalIndexerFactory.IDENTIFIER,
+ writer.finish());
+ byte[] sourceMeta =
+ new PrimaryKeyIndexSourceMeta(
+ new PrimaryKeyIndexSourceFile("data-file",
documents.length))
+ .serialize();
+ List<IndexFileMeta> sourceBackedFiles = new ArrayList<>();
+ for (IndexFileMeta indexFile : indexFiles) {
+ GlobalIndexMeta meta = indexFile.globalIndexMeta();
+ sourceBackedFiles.add(
+ new IndexFileMeta(
+ indexFile.indexType(),
+ indexFile.fileName(),
+ indexFile.fileSize(),
+ indexFile.rowCount(),
+ new GlobalIndexMeta(
+ meta.rowRangeStart(),
+ meta.rowRangeEnd(),
+ meta.indexFieldId(),
+ meta.extraFieldIds(),
+ meta.indexMeta(),
+ sourceMeta),
+ indexFile.externalPath()));
+ }
+
+ CommitMessage message =
+ new CommitMessageImpl(
+ BinaryRow.EMPTY_ROW,
+ 0,
+ null,
+ DataIncrement.indexIncrement(sourceBackedFiles),
+ CompactIncrement.emptyIncrement());
+ try (BatchTableCommit commit =
table.newBatchWriteBuilder().newCommit()) {
+ commit.commit(Collections.singletonList(message));
+ }
+ }
+
/**
* Builds and commits a single full-text index file covering rows
[rowStart, rowStart+N-1].
* {@code indexFields} determines the index identity (primary = first
field, the rest are extra
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyFullTextReadTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyFullTextReadTest.java
new file mode 100644
index 0000000000..1be9fe7c10
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyFullTextReadTest.java
@@ -0,0 +1,162 @@
+/*
+ * 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.CoreOptions.GlobalIndexSearchMode;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.globalindex.IndexedSplit;
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceMeta;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.stats.SimpleStats;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assertions.tuple;
+
+/** Tests PK full-text search modes and physical scored results. */
+class PrimaryKeyFullTextReadTest {
+
+ @Test
+ void testFastPropagatesScores() {
+ PrimaryKeyFullTextRead read =
+ new PrimaryKeyFullTextRead(
+ GlobalIndexSearchMode.FAST,
+ 10,
+ split ->
+ Collections.singletonList(
+
Collections.singletonList(position("indexed", 1, 9F))));
+
+ PrimaryKeyScoredResult result =
read.read(Collections.singletonList(split()));
+
+ assertThat(result.snapshotId()).isEqualTo(11);
+ assertThat(result.positions())
+ .extracting(
+ PrimaryKeySearchPosition::dataFileName,
+ PrimaryKeySearchPosition::rowPosition)
+ .containsExactly(tuple("indexed", 1L));
+ assertThat(result.positions().get(0).score()).isEqualTo(9F);
+ IndexedSplit indexedSplit = result.splits().get(0);
+ assertThat(indexedSplit.scores()).containsExactly(9F);
+ }
+
+ @Test
+ void testSelectsGlobalTopKByFullTextScore() {
+ PrimaryKeyFullTextRead read =
+ new PrimaryKeyFullTextRead(
+ GlobalIndexSearchMode.FAST,
+ 2,
+ split ->
+ Arrays.asList(
+ Arrays.asList(
+ position("indexed", 0, 100F),
+ position("indexed", 1, 99F)),
+
Collections.singletonList(position("raw", 0, 1F))));
+
+ PrimaryKeyScoredResult result =
read.read(Collections.singletonList(split()));
+
+ assertThat(result.positions())
+ .extracting(
+ PrimaryKeySearchPosition::dataFileName,
+ PrimaryKeySearchPosition::rowPosition,
+ PrimaryKeySearchPosition::score)
+ .containsExactly(tuple("indexed", 0L, 100F), tuple("indexed",
1L, 99F));
+ }
+
+ @ParameterizedTest
+ @EnumSource(
+ value = GlobalIndexSearchMode.class,
+ names = {"FULL", "DETAIL"})
+ void testRejectsIncompleteSearchModes(GlobalIndexSearchMode mode) {
+ assertThatThrownBy(
+ () ->
+ new PrimaryKeyFullTextRead(
+ mode, 10, split ->
Collections.emptyList()))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining("only supports the FAST global-index
search mode");
+ }
+
+ private static PrimaryKeySearchPosition position(
+ String dataFile, long rowPosition, float score) {
+ return new PrimaryKeySearchPosition(BinaryRow.EMPTY_ROW, 0, dataFile,
rowPosition, score);
+ }
+
+ private static PrimaryKeyFullTextSearchSplit split() {
+ List<DataFileMeta> dataFiles = Arrays.asList(dataFile("indexed"),
dataFile("raw"));
+ DataSplit dataSplit =
+ DataSplit.builder()
+ .withSnapshot(11)
+ .withPartition(BinaryRow.EMPTY_ROW)
+ .withBucket(0)
+ .withBucketPath("bucket-0")
+ .withTotalBuckets(1)
+ .withDataFiles(dataFiles)
+ .build();
+ return new PrimaryKeyFullTextSearchSplit(
+ dataSplit,
+ Collections.singletonList(payload("indexed")),
+ Collections.singletonList("raw"));
+ }
+
+ private static DataFileMeta dataFile(String name) {
+ return DataFileMeta.create(
+ name,
+ 100,
+ 2,
+ BinaryRow.EMPTY_ROW,
+ BinaryRow.EMPTY_ROW,
+ SimpleStats.EMPTY_STATS,
+ SimpleStats.EMPTY_STATS,
+ 0,
+ 0,
+ 0,
+ 1,
+ Collections.emptyList(),
+ 0L,
+ null,
+ FileSource.COMPACT,
+ null,
+ null,
+ null,
+ null);
+ }
+
+ private static IndexFileMeta payload(String source) {
+ byte[] sourceMeta =
+ new PrimaryKeyIndexSourceMeta(new
PrimaryKeyIndexSourceFile(source, 2)).serialize();
+ return new IndexFileMeta(
+ "full-text",
+ "index-" + source,
+ 100,
+ 2,
+ new GlobalIndexMeta(0, 1, 7, null, null, sourceMeta),
+ null);
+ }
+}
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
new file mode 100644
index 0000000000..ca656e82b2
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyFullTextScanTest.java
@@ -0,0 +1,307 @@
+/*
+ * 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.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.index.IndexFileHandler;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.pk.PrimaryKeyIndexDefinition;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceMeta;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.manifest.IndexManifestEntry;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.stats.SimpleStats;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.source.snapshot.SnapshotReader;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Filter;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Answers.CALLS_REAL_METHODS;
+import static org.mockito.Answers.RETURNS_SELF;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/** Tests snapshot-consistent planning for primary-key full-text search. */
+class PrimaryKeyFullTextScanTest {
+
+ private static final int FIELD_ID = 7;
+
+ @Test
+ void testSeparatesCurrentCoverageAndExcludesIneligibleFiles() {
+ DataFileMeta indexed = dataFile("indexed", 1, FileSource.COMPACT);
+ DataFileMeta uncovered = dataFile("uncovered", 2, FileSource.COMPACT);
+ DataFileMeta levelZero = dataFile("level-zero", 0, FileSource.COMPACT);
+ DataFileMeta appended = dataFile("appended", 1, FileSource.APPEND);
+ DeletionFile indexedDv = new DeletionFile("indexed.dv", 0, 10, 1L);
+ DeletionFile uncoveredDv = new DeletionFile("uncovered.dv", 10, 10,
1L);
+ DataSplit source =
+ dataSplit(
+ Arrays.asList(indexed, uncovered, levelZero, appended),
+ Arrays.asList(indexedDv, uncoveredDv, null, null));
+
+ List<IndexManifestEntry> payloads =
+ Arrays.asList(
+ payloadEntry("indexed", FIELD_ID, "current"),
+ payloadEntry("uncovered", 8, "other-field"));
+
+ PrimaryKeyFullTextScan.Plan plan =
+ PrimaryKeyFullTextScan.plan(
+ 11, Collections.singletonList(source), payloads,
FIELD_ID);
+
+ assertThat(plan.snapshotId()).isEqualTo(11);
+ assertThat(plan.splits()).hasSize(1);
+ PrimaryKeyFullTextSearchSplit split = (PrimaryKeyFullTextSearchSplit)
plan.splits().get(0);
+ assertThat(split.dataSplit().dataFiles()).containsExactly(indexed,
uncovered);
+ assertThat(split.dataSplit().deletionFiles()).isPresent();
+
assertThat(split.dataSplit().deletionFiles().get()).containsExactly(indexedDv,
uncoveredDv);
+ assertThat(split.payloadFiles())
+ .extracting(IndexFileMeta::fileName)
+ .containsExactly("current");
+ assertThat(split.uncoveredDataFiles()).containsExactly("uncovered");
+ }
+
+ @Test
+ void testPlansMultiSourceArchiveOnce() {
+ DataFileMeta first = dataFile("data-1", 1, FileSource.COMPACT);
+ DataFileMeta second = dataFile("data-2", 1, FileSource.COMPACT);
+ IndexManifestEntry payload =
+ payloadEntry(Arrays.asList("data-1", "data-2"), FIELD_ID,
"multi-source");
+
+ PrimaryKeyFullTextScan.Plan plan =
+ PrimaryKeyFullTextScan.plan(
+ 11,
+
Collections.singletonList(dataSplit(Arrays.asList(first, second), null)),
+ Collections.singletonList(payload),
+ FIELD_ID);
+
+ PrimaryKeyFullTextSearchSplit split = (PrimaryKeyFullTextSearchSplit)
plan.splits().get(0);
+ assertThat(split.payloadFiles()).containsExactly(payload.indexFile());
+ assertThat(split.uncoveredDataFiles()).isEmpty();
+ }
+
+ @Test
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ void testCapturesOneSnapshotAndPrunesPartitions() {
+ FileStoreTable table = mock(FileStoreTable.class);
+ Snapshot snapshot = mock(Snapshot.class);
+ when(snapshot.id()).thenReturn(11L);
+ Options tableOptions = new Options();
+ tableOptions.set(CoreOptions.PK_FULL_TEXT_INDEX_COLUMNS, "content");
+ when(table.coreOptions()).thenReturn(new CoreOptions(tableOptions));
+
+ SnapshotReader reader = mock(SnapshotReader.class, RETURNS_SELF);
+ SnapshotReader.Plan snapshotPlan = mock(SnapshotReader.Plan.class,
CALLS_REAL_METHODS);
+ when(snapshotPlan.snapshotId()).thenReturn(11L);
+ when(snapshotPlan.splits())
+ .thenReturn(
+ Collections.singletonList(
+ dataSplit(
+ Collections.singletonList(
+ dataFile("indexed", 1,
FileSource.COMPACT)),
+ null)));
+ when(reader.read()).thenReturn(snapshotPlan);
+ when(table.newSnapshotReader()).thenReturn(reader);
+
+ PartitionPredicate partitionFilter = mock(PartitionPredicate.class);
+ when(partitionFilter.test(BinaryRow.EMPTY_ROW)).thenReturn(true);
+ PrimaryKeyIndexDefinition definition = definition();
+ List<IndexManifestEntry> entries =
+ Arrays.asList(
+ payloadEntry("indexed", FIELD_ID, "current"),
+ payloadEntry("indexed", FIELD_ID + 1, "other-field"));
+ IndexFileHandler indexFileHandler = mock(IndexFileHandler.class);
+ when(indexFileHandler.scan(eq(snapshot), any(Filter.class)))
+ .thenAnswer(
+ invocation -> {
+ Filter<IndexManifestEntry> filter =
invocation.getArgument(1);
+ List<IndexManifestEntry> filtered = new
ArrayList<>();
+ for (IndexManifestEntry entry : entries) {
+ if (filter.test(entry)) {
+ filtered.add(entry);
+ }
+ }
+ return filtered;
+ });
+ when(reader.indexFileHandler()).thenReturn(indexFileHandler);
+ configureBatchScan(table, reader, snapshot);
+
+ PrimaryKeyFullTextScan.Plan plan =
+ new PrimaryKeyFullTextScan(table, definition,
partitionFilter).scan();
+
+ assertThat(plan.snapshotId()).isEqualTo(11);
+ PrimaryKeyFullTextSearchSplit split = (PrimaryKeyFullTextSearchSplit)
plan.splits().get(0);
+ assertThat(split.payloadFiles())
+ .extracting(IndexFileMeta::fileName)
+ .containsExactly("current");
+ assertThat(split.uncoveredDataFiles()).isEmpty();
+ verify(reader).withPartitionFilter(partitionFilter);
+ verify(reader).indexFileHandler();
+ }
+
+ @Test
+ void testSplitSerialization() throws Exception {
+ PrimaryKeyFullTextSearchSplit split =
+ new PrimaryKeyFullTextSearchSplit(
+ dataSplit(
+ Arrays.asList(
+ dataFile("indexed", 1,
FileSource.COMPACT),
+ dataFile("raw", 1,
FileSource.COMPACT)),
+ Arrays.asList(new DeletionFile("indexed.dv",
0, 10, 1L), null)),
+ Collections.singletonList(
+ payloadEntry("indexed", FIELD_ID,
"current").indexFile()),
+ Collections.singletonList("raw"));
+
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ try (ObjectOutputStream output = new ObjectOutputStream(bytes)) {
+ output.writeObject(split);
+ }
+ PrimaryKeyFullTextSearchSplit restored;
+ try (ObjectInputStream input =
+ new ObjectInputStream(new
ByteArrayInputStream(bytes.toByteArray()))) {
+ restored = (PrimaryKeyFullTextSearchSplit) input.readObject();
+ }
+
+ assertThat(restored).isEqualTo(split);
+ }
+
+ private static PrimaryKeyIndexDefinition definition() {
+ return new PrimaryKeyIndexDefinition(
+ "content",
+ FIELD_ID,
+ "full-text",
+ new Options(),
+ PrimaryKeyIndexDefinition.Family.FULL_TEXT,
+ 2,
+ 0.5);
+ }
+
+ private static DataSplit dataSplit(
+ List<DataFileMeta> dataFiles, List<DeletionFile> deletionFiles) {
+ DataSplit.Builder builder =
+ DataSplit.builder()
+ .withSnapshot(11)
+ .withPartition(BinaryRow.EMPTY_ROW)
+ .withBucket(0)
+ .withBucketPath("bucket-0")
+ .withTotalBuckets(1)
+ .withDataFiles(dataFiles);
+ if (deletionFiles != null) {
+ builder.withDataDeletionFiles(deletionFiles);
+ }
+ return builder.build();
+ }
+
+ private static IndexManifestEntry payloadEntry(
+ String sourceFile, int fieldId, String payloadFile) {
+ return payloadEntry(Collections.singletonList(sourceFile), fieldId,
payloadFile);
+ }
+
+ private static IndexManifestEntry payloadEntry(
+ List<String> sourceFiles, int fieldId, String payloadFile) {
+ List<PrimaryKeyIndexSourceFile> sources = new ArrayList<>();
+ for (String sourceFile : sourceFiles) {
+ sources.add(new PrimaryKeyIndexSourceFile(sourceFile, 2));
+ }
+ byte[] sourceMeta = new PrimaryKeyIndexSourceMeta(sources).serialize();
+ long rowCount = 2L * sources.size();
+ return new IndexManifestEntry(
+ FileKind.ADD,
+ BinaryRow.EMPTY_ROW,
+ 0,
+ new IndexFileMeta(
+ "full-text",
+ payloadFile,
+ 100,
+ rowCount,
+ new GlobalIndexMeta(0, rowCount - 1, fieldId, null,
null, sourceMeta),
+ null));
+ }
+
+ private static void configureBatchScan(
+ FileStoreTable table, SnapshotReader snapshotReader, Snapshot
snapshot) {
+ TableSchema schema = mock(TableSchema.class);
+ when(schema.primaryKeys()).thenReturn(Collections.singletonList("id"));
+ when(schema.logicalRowType())
+ .thenReturn(RowType.of(new DataField(1, "id",
DataTypes.INT().notNull())));
+ when(table.schema()).thenReturn(schema);
+ when(table.schemaManager()).thenReturn(mock(SchemaManager.class));
+ SnapshotManager snapshotManager = mock(SnapshotManager.class);
+ when(snapshotManager.latestSnapshot()).thenReturn(snapshot);
+ when(snapshotManager.snapshot(snapshot.id())).thenReturn(snapshot);
+ when(snapshotReader.snapshotManager()).thenReturn(snapshotManager);
+ when(table.newScan(any(FileStoreTable.SnapshotReaderFactory.class)))
+ .thenAnswer(
+ invocation -> {
+ FileStoreTable.SnapshotReaderFactory factory =
+ invocation.getArgument(0);
+ return new PrimaryKeyBatchScan(
+ table, factory.create(table),
mock(TableQueryAuth.class), null);
+ });
+ }
+
+ private static DataFileMeta dataFile(String fileName, int level,
FileSource fileSource) {
+ return DataFileMeta.create(
+ fileName,
+ 100,
+ 2,
+ BinaryRow.EMPTY_ROW,
+ BinaryRow.EMPTY_ROW,
+ SimpleStats.EMPTY_STATS,
+ SimpleStats.EMPTY_STATS,
+ 0,
+ 0,
+ 0,
+ level,
+ Collections.emptyList(),
+ 0L,
+ null,
+ fileSource,
+ null,
+ null,
+ null,
+ null);
+ }
+}
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
new file mode 100644
index 0000000000..47a3d5344d
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyFullTextSearchTest.java
@@ -0,0 +1,120 @@
+/*
+ * 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.CoreOptions;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/** Tests Java API dispatch for configured primary-key full-text indexes. */
+class PrimaryKeyFullTextSearchTest {
+
+ @Test
+ void testDispatchesConfiguredFieldToPrimaryKeyScan() {
+ FileStoreTable table = table(false);
+
+ FullTextScan scan =
+ new FullTextSearchBuilderImpl(table)
+ .withQuery("content", "hello")
+ .withLimit(10)
+ .newFullTextScan();
+
+ assertThat(scan).isInstanceOf(PrimaryKeyFullTextScan.class);
+ }
+
+ @Test
+ void testOtherFieldRetainsGlobalFullTextPath() {
+ FileStoreTable table = table(false);
+
+ FullTextScan scan =
+ new FullTextSearchBuilderImpl(table)
+ .withQuery("other", "hello")
+ .withLimit(10)
+ .newFullTextScan();
+
+ assertThat(scan).isInstanceOf(DataEvolutionFullTextScan.class);
+ }
+
+ @Test
+ void testDataEvolutionRetainsGlobalFullTextPath() {
+ FileStoreTable table = table(true);
+
+ FullTextScan scan =
+ new FullTextSearchBuilderImpl(table)
+ .withQuery("content", "hello")
+ .withLimit(10)
+ .newFullTextScan();
+
+ assertThat(scan).isInstanceOf(DataEvolutionFullTextScan.class);
+ }
+
+ @Test
+ void testHybridRouteRejectsPrimaryKeyFullText() {
+ 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");
+ }
+
+ private static FileStoreTable table(boolean dataEvolution) {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.BUCKET.key(), "2");
+ options.put(CoreOptions.PK_FULL_TEXT_INDEX_COLUMNS.key(), "content");
+ options.put(CoreOptions.DATA_EVOLUTION_ENABLED.key(),
Boolean.toString(dataEvolution));
+ TableSchema schema =
+ new TableSchema(
+ 1,
+ Arrays.asList(
+ new DataField(0, "id",
DataTypes.INT().notNull()),
+ new DataField(1, "content",
DataTypes.STRING()),
+ new DataField(2, "other", DataTypes.STRING())),
+ 2,
+ Collections.emptyList(),
+ Collections.singletonList("id"),
+ options,
+ null);
+ FileStoreTable table = mock(FileStoreTable.class);
+ when(table.schema()).thenReturn(schema);
+ when(table.rowType()).thenReturn(schema.logicalRowType());
+ when(table.coreOptions()).thenReturn(new CoreOptions(options));
+ when(table.newFullTextSearchBuilder()).thenReturn(new
FullTextSearchBuilderImpl(table));
+ return table;
+ }
+}
diff --git
a/paimon-full-text/src/test/java/org/apache/paimon/fulltext/index/NativePrimaryKeyFullTextIndexTest.java
b/paimon-full-text/src/test/java/org/apache/paimon/fulltext/index/NativePrimaryKeyFullTextIndexTest.java
new file mode 100644
index 0000000000..567cb2a56e
--- /dev/null
+++
b/paimon-full-text/src/test/java/org/apache/paimon/fulltext/index/NativePrimaryKeyFullTextIndexTest.java
@@ -0,0 +1,267 @@
+/*
+ * 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.fulltext.index;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.deletionvectors.BitmapDeletionVector;
+import org.apache.paimon.deletionvectors.DeletionVector;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.globalindex.GlobalIndexIOMeta;
+import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
+import org.apache.paimon.globalindex.GlobalIndexer;
+import org.apache.paimon.globalindex.ResultEntry;
+import org.apache.paimon.globalindex.io.GlobalIndexFileReader;
+import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceMeta;
+import org.apache.paimon.index.pkfulltext.PrimaryKeyFullTextBucketSearch;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.stats.SimpleStats;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.PrimaryKeyFullTextSearchSplit;
+import org.apache.paimon.table.source.PrimaryKeySearchPosition;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.utils.JsonSerdeUtil;
+
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.type.TypeReference;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import static
org.apache.paimon.shade.guava30.com.google.common.util.concurrent.MoreExecutors.newDirectExecutorService;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.tuple;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+/** Native SPI tests for file-aligned primary-key full-text archives. */
+class NativePrimaryKeyFullTextIndexTest {
+
+ private static final DataField TEXT_FIELD = new DataField(7, "content",
DataTypes.STRING());
+
+ @TempDir java.nio.file.Path tempDir;
+
+ private FileIO fileIO;
+ private Path indexPath;
+
+ @BeforeEach
+ void before() {
+ assumeTrue(isNativeAvailable(), "Native full-text library not
available, skipping tests");
+ fileIO = LocalFileIO.create();
+ indexPath = new Path(tempDir.toUri());
+ }
+
+ @AfterEach
+ void after() throws IOException {
+ if (fileIO != null) {
+ fileIO.delete(indexPath, true);
+ }
+ }
+
+ @Test
+ void testPreservesNullOrdinalFiltersDeletedRowsAndOrdersByNativeScore()
throws Exception {
+ Options options = new Options();
+ IndexFileMeta archive =
+ buildArchive(
+ Arrays.asList(
+ BinaryString.fromString("paimon vector"),
+ null,
+ BinaryString.fromString("paimon lake"),
+ BinaryString.fromString("paimon storage")),
+ options);
+
+ assertThat(archive.indexType()).isEqualTo("full-text");
+ assertThat(archive.rowCount()).isEqualTo(4);
+ assertThat(archive.globalIndexMeta().rowRangeStart()).isZero();
+ assertThat(archive.globalIndexMeta().rowRangeEnd()).isEqualTo(3);
+
assertThat(archive.globalIndexMeta().indexFieldId()).isEqualTo(TEXT_FIELD.id());
+ PrimaryKeyIndexSourceMeta sourceMeta =
PrimaryKeyIndexSourceMeta.fromIndexFile(archive);
+ assertThat(sourceMeta.sourceFile().fileName()).isEqualTo("data-1");
+ assertThat(sourceMeta.sourceFile().rowCount()).isEqualTo(4);
+
+ DataFileMeta dataFile = dataFile(4);
+ PrimaryKeyFullTextSearchSplit split =
+ new PrimaryKeyFullTextSearchSplit(
+ DataSplit.builder()
+ .withSnapshot(1)
+ .withPartition(BinaryRow.EMPTY_ROW)
+ .withBucket(0)
+ .withBucketPath(indexPath.toString())
+ .withTotalBuckets(1)
+
.withDataFiles(Collections.singletonList(dataFile))
+ .build(),
+ Collections.singletonList(archive),
+ Collections.emptyList());
+ BitmapDeletionVector deletionVector = new BitmapDeletionVector();
+ deletionVector.delete(2);
+ Map<String, DeletionVector> deletionVectors =
+ Collections.singletonMap(dataFile.fileName(), deletionVector);
+ GlobalIndexer indexer = GlobalIndexer.create("full-text", TEXT_FIELD,
options);
+ PrimaryKeyFullTextBucketSearch search =
+ new PrimaryKeyFullTextBucketSearch(
+ payload ->
+ indexer.createReader(
+ fileReader(),
+
Collections.singletonList(toIOMeta(payload)),
+ newDirectExecutorService()));
+
+ List<List<PrimaryKeySearchPosition>> rankings =
+ search.searchRankings(
+ split, deletionVectors, "content",
boostQuery("paimon", "vector", 0.1F), 2);
+
+ assertThat(rankings).hasSize(1);
+ assertThat(rankings.get(0))
+ .extracting(
+ PrimaryKeySearchPosition::rowPosition,
+ PrimaryKeySearchPosition::dataFileName)
+ .containsExactly(tuple(3L, "data-1"), tuple(0L, "data-1"));
+
assertThat(rankings.get(0).get(0).score()).isGreaterThan(rankings.get(0).get(1).score());
+ }
+
+ @Test
+ void testPersistsTokenizerOptionsInPkArchiveMetadata() throws Exception {
+ Options options = new Options();
+ options.set("full-text.tokenizer", "ngram");
+ options.set("full-text.ngram.min-gram", "2");
+ options.set("full-text.ngram.max-gram", "2");
+
+ IndexFileMeta archive =
+
buildArchive(Collections.singletonList(BinaryString.fromString("中文全文检索")),
options);
+
+ assertThat(serializedOptions(archive.globalIndexMeta().indexMeta()))
+ .containsEntry("tokenizer", "ngram")
+ .containsEntry("ngram.min-gram", "2")
+ .containsEntry("ngram.max-gram", "2");
+ }
+
+ private IndexFileMeta buildArchive(List<BinaryString> texts, Options
options) throws Exception {
+ GlobalIndexer indexer = GlobalIndexer.create("full-text", TEXT_FIELD,
options);
+ GlobalIndexSingleColumnWriter writer =
+ (GlobalIndexSingleColumnWriter)
indexer.createWriter(fileWriter());
+ for (int i = 0; i < texts.size(); i++) {
+ writer.write(texts.get(i), i);
+ }
+ List<ResultEntry> results = writer.finish();
+ assertThat(results).hasSize(1);
+ ResultEntry result = results.get(0);
+ Path path = new Path(indexPath, result.fileName());
+ return new IndexFileMeta(
+ "full-text",
+ result.fileName(),
+ fileIO.getFileSize(path),
+ result.rowCount(),
+ new GlobalIndexMeta(
+ 0,
+ texts.size() - 1,
+ TEXT_FIELD.id(),
+ null,
+ result.meta(),
+ new PrimaryKeyIndexSourceMeta(
+ new
PrimaryKeyIndexSourceFile("data-1", texts.size()))
+ .serialize()),
+ null);
+ }
+
+ private GlobalIndexFileWriter fileWriter() {
+ return new GlobalIndexFileWriter() {
+ @Override
+ public String newFileName(String prefix) {
+ return prefix + "-" + UUID.randomUUID();
+ }
+
+ @Override
+ public PositionOutputStream newOutputStream(String fileName)
throws IOException {
+ return fileIO.newOutputStream(new Path(indexPath, fileName),
false);
+ }
+ };
+ }
+
+ private GlobalIndexFileReader fileReader() {
+ return meta -> fileIO.newInputStream(meta.filePath());
+ }
+
+ private GlobalIndexIOMeta toIOMeta(IndexFileMeta archive) {
+ return new GlobalIndexIOMeta(
+ new Path(indexPath, archive.fileName()),
+ archive.fileSize(),
+ archive.globalIndexMeta().indexMeta());
+ }
+
+ private static DataFileMeta dataFile(long rowCount) {
+ return DataFileMeta.forAppend(
+ "data-1",
+ 100,
+ rowCount,
+ SimpleStats.EMPTY_STATS,
+ 0,
+ 1,
+ 1,
+ Collections.emptyList(),
+ null,
+ FileSource.COMPACT,
+ null,
+ null,
+ null,
+ null);
+ }
+
+ private static Map<String, String> serializedOptions(byte[] metadata) {
+ return JsonSerdeUtil.fromJson(
+ new String(metadata, StandardCharsets.UTF_8),
+ new TypeReference<Map<String, String>>() {});
+ }
+
+ private static String boostQuery(String positive, String negative, float
negativeBoost) {
+ return "{\"boost\":{\"positive\":"
+ + matchQuery(positive)
+ + ",\"negative\":"
+ + matchQuery(negative)
+ + ",\"negative_boost\":"
+ + negativeBoost
+ + "}}";
+ }
+
+ private static String matchQuery(String terms) {
+ return "{\"match\":{\"query\":\"" + terms + "\"}}";
+ }
+
+ private static boolean isNativeAvailable() {
+ String path = System.getenv("PAIMON_FTINDEX_JNI_LIB_PATH");
+ return path != null && !path.isEmpty() && new File(path).isFile();
+ }
+}
