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 f4c9efe2e0 [core] Support full mode for full-text search (#8316)
f4c9efe2e0 is described below
commit f4c9efe2e05ed682e9f512521054ab11ec916695
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Jun 22 15:41:02 2026 +0800
[core] Support full mode for full-text search (#8316)
Support non-FAST global index search modes for full-text search by
searching both existing full-text index files and raw row ranges that
are not covered by those indexes. `FULL` uses the snapshot row-id range
and `DETAIL` uses data-file row-id ranges through `GlobalIndexCoverage`,
while `FAST` remains index-only. Python full-text search now fails fast
for non-FAST raw full-text fallback because it would need to rebuild
temporary full-text indexes, which is not implemented there yet.
---
.../paimon/globalindex/GlobalIndexCoverage.java | 36 +-
.../paimon/table/source/FullTextReadImpl.java | 102 +++-
.../paimon/table/source/FullTextScanImpl.java | 12 +-
.../table/source/FullTextSearchBuilderImpl.java | 2 +-
.../paimon/table/source/FullTextSearchSplit.java | 78 +--
...rchSplit.java => IndexFullTextSearchSplit.java} | 59 ++-
.../paimon/table/source/RawFullTextReadImpl.java | 521 +++++++++++++++++++++
.../table/source/RawFullTextSearchSplit.java | 61 +++
.../table/source/FullTextSearchBuilderTest.java | 242 +++++++++-
.../pypaimon/globalindex/global_index_coverage.py | 14 +-
.../pypaimon/table/source/full_text_scan.py | 15 +
paimon-python/pypaimon/tests/global_index_test.py | 16 +
.../pypaimon/tests/vector_search_filter_test.py | 29 +-
.../index/TantivyFullTextGlobalIndexerFactory.java | 7 +-
.../TantivyFullTextGlobalIndexerFactoryTest.java | 53 +++
15 files changed, 1109 insertions(+), 138 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexCoverage.java
b/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexCoverage.java
index 6ef52eee2d..5191d0f942 100644
---
a/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexCoverage.java
+++
b/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexCoverage.java
@@ -82,24 +82,7 @@ public class GlobalIndexCoverage {
return unindexedRanges(Collections.singleton(fieldId));
}
- private void addCoverage(int fieldId, Range range) {
- coverageByField.computeIfAbsent(fieldId, k -> new
ArrayList<>()).add(range);
- }
-
- private List<Range> indexedRanges(Collection<Integer> fieldIds) {
- List<Range> ranges = null;
- for (Integer fieldId : fieldIds) {
- List<Range> fieldRanges = coverageByField.get(fieldId);
- if (fieldRanges == null || fieldRanges.isEmpty()) {
- return Collections.emptyList();
- }
- fieldRanges = Range.sortAndMergeOverlap(fieldRanges, true);
- ranges = ranges == null ? fieldRanges : Range.and(ranges,
fieldRanges);
- }
- return ranges == null ? Collections.emptyList() :
Range.sortAndMergeOverlap(ranges, true);
- }
-
- private List<Range> unindexedRanges(Collection<Integer> fieldIds) {
+ public List<Range> unindexedRanges(Collection<Integer> fieldIds) {
GlobalIndexSearchMode searchMode =
table.coreOptions().globalIndexSearchMode();
if (searchMode == GlobalIndexSearchMode.FAST) {
return Collections.emptyList();
@@ -124,6 +107,23 @@ public class GlobalIndexCoverage {
return Range.sortAndMergeOverlap(unindexedRanges, true);
}
+ private void addCoverage(int fieldId, Range range) {
+ coverageByField.computeIfAbsent(fieldId, k -> new
ArrayList<>()).add(range);
+ }
+
+ private List<Range> indexedRanges(Collection<Integer> fieldIds) {
+ List<Range> ranges = null;
+ for (Integer fieldId : fieldIds) {
+ List<Range> fieldRanges = coverageByField.get(fieldId);
+ if (fieldRanges == null || fieldRanges.isEmpty()) {
+ return Collections.emptyList();
+ }
+ fieldRanges = Range.sortAndMergeOverlap(fieldRanges, true);
+ ranges = ranges == null ? fieldRanges : Range.and(ranges,
fieldRanges);
+ }
+ return ranges == null ? Collections.emptyList() :
Range.sortAndMergeOverlap(ranges, true);
+ }
+
private List<Range> dataRangesByDataFiles() {
SnapshotReader snapshotReader =
table.newSnapshotReader()
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextReadImpl.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextReadImpl.java
index e6e93b5211..97609ba564 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextReadImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextReadImpl.java
@@ -18,7 +18,6 @@
package org.apache.paimon.table.source;
-import org.apache.paimon.fs.FileIO;
import org.apache.paimon.globalindex.GlobalIndexIOMeta;
import org.apache.paimon.globalindex.GlobalIndexReadThreadPool;
import org.apache.paimon.globalindex.GlobalIndexReader;
@@ -31,13 +30,17 @@ import
org.apache.paimon.globalindex.io.GlobalIndexFileReader;
import org.apache.paimon.index.GlobalIndexMeta;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.index.IndexPathFactory;
+import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.predicate.FullTextQuery;
import org.apache.paimon.predicate.FullTextSearch;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.types.DataField;
import org.apache.paimon.utils.IOUtils;
+import org.apache.paimon.utils.Range;
import org.apache.paimon.utils.RoaringNavigableMap64;
+import javax.annotation.Nullable;
+
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -54,18 +57,29 @@ import static
org.apache.paimon.utils.Preconditions.checkNotNull;
public class FullTextReadImpl implements FullTextRead {
private final FileStoreTable table;
+ @Nullable private final PartitionPredicate partitionFilter;
private final int limit;
private final List<DataField> textColumns;
private final FullTextQuery query;
public FullTextReadImpl(
FileStoreTable table, int limit, DataField textColumn,
FullTextQuery query) {
- this(table, limit, Collections.singletonList(textColumn), query);
+ this(table, null, limit, Collections.singletonList(textColumn), query);
}
public FullTextReadImpl(
FileStoreTable table, int limit, List<DataField> textColumns,
FullTextQuery query) {
+ this(table, null, limit, textColumns, query);
+ }
+
+ public FullTextReadImpl(
+ FileStoreTable table,
+ @Nullable PartitionPredicate partitionFilter,
+ int limit,
+ List<DataField> textColumns,
+ FullTextQuery query) {
this.table = table;
+ this.partitionFilter = partitionFilter;
this.limit = limit;
this.textColumns = Collections.unmodifiableList(new
ArrayList<>(textColumns));
this.query = query;
@@ -87,20 +101,43 @@ public class FullTextReadImpl implements FullTextRead {
fieldsByName.put(textColumn.name(), textColumn);
}
- Map<String, List<FullTextSearchSplit>> splitsByColumn = new
HashMap<>();
+ Map<String, List<IndexFullTextSearchSplit>> splitsByColumn = new
HashMap<>();
+ List<Range> rawRowRanges = new ArrayList<>();
for (FullTextSearchSplit split : splits) {
- splitsByColumn.computeIfAbsent(split.columnName(), k -> new
ArrayList<>()).add(split);
+ if (split instanceof IndexFullTextSearchSplit) {
+ IndexFullTextSearchSplit indexSplit =
(IndexFullTextSearchSplit) split;
+ splitsByColumn
+ .computeIfAbsent(indexSplit.columnName(), k -> new
ArrayList<>())
+ .add(indexSplit);
+ } else if (split instanceof RawFullTextSearchSplit) {
+ rawRowRanges.addAll(((RawFullTextSearchSplit)
split).rowRanges());
+ }
}
- return evalQuery(query, fieldsByName, splitsByColumn,
indexPathFactory, executor)
- .topK(limit);
+ GlobalIndexFileReader indexFileReader = m ->
table.fileIO().newInputStream(m.filePath());
+ ScoredGlobalIndexResult result =
+ evalQuery(
+ query,
+ fieldsByName,
+ splitsByColumn,
+ indexPathFactory,
+ indexFileReader,
+ executor);
+ if (!rawRowRanges.isEmpty()) {
+ result =
+ new RawFullTextReadImpl(table, partitionFilter, limit,
query, this::evalQuery)
+ .withRawSearch(
+ result, rawRowRanges, fieldsByName,
splitsByColumn, executor);
+ }
+ return result.topK(limit);
}
- private ScoredGlobalIndexResult evalQuery(
+ ScoredGlobalIndexResult evalQuery(
FullTextQuery query,
Map<String, DataField> fieldsByName,
- Map<String, List<FullTextSearchSplit>> splitsByColumn,
+ Map<String, List<IndexFullTextSearchSplit>> splitsByColumn,
IndexPathFactory indexPathFactory,
+ GlobalIndexFileReader indexFileReader,
ExecutorService executor) {
if (query instanceof FullTextQuery.Match) {
return evalColumnQuery(
@@ -109,6 +146,7 @@ public class FullTextReadImpl implements FullTextRead {
fieldsByName,
splitsByColumn,
indexPathFactory,
+ indexFileReader,
executor);
}
if (query instanceof FullTextQuery.Phrase) {
@@ -118,6 +156,7 @@ public class FullTextReadImpl implements FullTextRead {
fieldsByName,
splitsByColumn,
indexPathFactory,
+ indexFileReader,
executor);
}
if (query instanceof FullTextQuery.MultiMatch) {
@@ -126,6 +165,7 @@ public class FullTextReadImpl implements FullTextRead {
fieldsByName,
splitsByColumn,
indexPathFactory,
+ indexFileReader,
executor);
}
if (query instanceof FullTextQuery.Boost) {
@@ -136,12 +176,14 @@ public class FullTextReadImpl implements FullTextRead {
fieldsByName,
splitsByColumn,
indexPathFactory,
+ indexFileReader,
executor),
evalQuery(
boost.negative(),
fieldsByName,
splitsByColumn,
indexPathFactory,
+ indexFileReader,
executor),
boost.negativeBoost());
}
@@ -151,6 +193,7 @@ public class FullTextReadImpl implements FullTextRead {
fieldsByName,
splitsByColumn,
indexPathFactory,
+ indexFileReader,
executor);
}
throw new IllegalArgumentException("Unsupported full-text query: " +
query);
@@ -159,8 +202,9 @@ public class FullTextReadImpl implements FullTextRead {
private ScoredGlobalIndexResult evalMultiMatch(
FullTextQuery.MultiMatch query,
Map<String, DataField> fieldsByName,
- Map<String, List<FullTextSearchSplit>> splitsByColumn,
+ Map<String, List<IndexFullTextSearchSplit>> splitsByColumn,
IndexPathFactory indexPathFactory,
+ GlobalIndexFileReader indexFileReader,
ExecutorService executor) {
List<String> columns = query.columns();
List<Float> boosts = query.boosts();
@@ -182,6 +226,7 @@ public class FullTextReadImpl implements FullTextRead {
fieldsByName,
splitsByColumn,
indexPathFactory,
+ indexFileReader,
executor));
}
return or(results).topK(limit);
@@ -190,20 +235,33 @@ public class FullTextReadImpl implements FullTextRead {
private ScoredGlobalIndexResult evalBoolean(
FullTextQuery.BooleanQuery query,
Map<String, DataField> fieldsByName,
- Map<String, List<FullTextSearchSplit>> splitsByColumn,
+ Map<String, List<IndexFullTextSearchSplit>> splitsByColumn,
IndexPathFactory indexPathFactory,
+ GlobalIndexFileReader indexFileReader,
ExecutorService executor) {
ScoredGlobalIndexResult result = null;
for (FullTextQuery child : query.must()) {
ScoredGlobalIndexResult childResult =
- evalQuery(child, fieldsByName, splitsByColumn,
indexPathFactory, executor);
+ evalQuery(
+ child,
+ fieldsByName,
+ splitsByColumn,
+ indexPathFactory,
+ indexFileReader,
+ executor);
result = result == null ? childResult : and(result, childResult);
}
List<ScoredGlobalIndexResult> shouldResults = new
ArrayList<>(query.should().size());
for (FullTextQuery child : query.should()) {
shouldResults.add(
- evalQuery(child, fieldsByName, splitsByColumn,
indexPathFactory, executor));
+ evalQuery(
+ child,
+ fieldsByName,
+ splitsByColumn,
+ indexPathFactory,
+ indexFileReader,
+ executor));
}
if (!shouldResults.isEmpty()) {
ScoredGlobalIndexResult shouldResult = or(shouldResults);
@@ -215,7 +273,13 @@ public class FullTextReadImpl implements FullTextRead {
}
for (FullTextQuery child : query.mustNot()) {
ScoredGlobalIndexResult childResult =
- evalQuery(child, fieldsByName, splitsByColumn,
indexPathFactory, executor);
+ evalQuery(
+ child,
+ fieldsByName,
+ splitsByColumn,
+ indexPathFactory,
+ indexFileReader,
+ executor);
result = andNot(result, childResult);
}
return result.topK(limit);
@@ -225,10 +289,11 @@ public class FullTextReadImpl implements FullTextRead {
FullTextQuery query,
String column,
Map<String, DataField> fieldsByName,
- Map<String, List<FullTextSearchSplit>> splitsByColumn,
+ Map<String, List<IndexFullTextSearchSplit>> splitsByColumn,
IndexPathFactory indexPathFactory,
+ GlobalIndexFileReader indexFileReader,
ExecutorService executor) {
- List<FullTextSearchSplit> columnSplits = splitsByColumn.get(column);
+ List<IndexFullTextSearchSplit> columnSplits =
splitsByColumn.get(column);
if (columnSplits == null || columnSplits.isEmpty()) {
return ScoredGlobalIndexResult.createEmpty();
}
@@ -253,7 +318,7 @@ public class FullTextReadImpl implements FullTextRead {
List<CompletableFuture<Optional<ScoredGlobalIndexResult>>> futures =
new ArrayList<>(columnSplits.size());
- for (FullTextSearchSplit split : columnSplits) {
+ for (IndexFullTextSearchSplit split : columnSplits) {
futures.add(
eval(
globalIndexer,
@@ -262,6 +327,7 @@ public class FullTextReadImpl implements FullTextRead {
split.rowRangeEnd(),
split.fullTextIndexFiles(),
query,
+ indexFileReader,
executor));
}
@@ -285,6 +351,7 @@ public class FullTextReadImpl implements FullTextRead {
long rowRangeEnd,
List<IndexFileMeta> fullTextIndexFiles,
FullTextQuery query,
+ GlobalIndexFileReader indexFileReader,
ExecutorService executor) {
List<GlobalIndexIOMeta> indexIOMetaList = new ArrayList<>();
for (IndexFileMeta indexFile : fullTextIndexFiles) {
@@ -295,9 +362,6 @@ public class FullTextReadImpl implements FullTextRead {
indexFile.fileSize(),
meta.indexMeta()));
}
- @SuppressWarnings("resource")
- FileIO fileIO = table.fileIO();
- GlobalIndexFileReader indexFileReader = m ->
fileIO.newInputStream(m.filePath());
GlobalIndexReader reader =
globalIndexer.createReader(indexFileReader, indexIOMetaList,
executor);
FullTextSearch fullTextSearch =
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextScanImpl.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextScanImpl.java
index 123a8b9980..5fdebadf70 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextScanImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextScanImpl.java
@@ -19,6 +19,7 @@
package org.apache.paimon.table.source;
import org.apache.paimon.Snapshot;
+import org.apache.paimon.globalindex.GlobalIndexCoverage;
import org.apache.paimon.globalindex.GlobalIndexerFactory;
import org.apache.paimon.globalindex.GlobalIndexerFactoryUtils;
import org.apache.paimon.index.GlobalIndexMeta;
@@ -121,11 +122,20 @@ public class FullTextScanImpl implements FullTextScan {
columnEntry.getValue().entrySet()) {
Range range = rangeEntry.getKey();
splits.add(
- new FullTextSearchSplit(
+ new IndexFullTextSearchSplit(
columnName, range.from, range.to,
rangeEntry.getValue()));
}
}
+ if (!allIndexFiles.isEmpty()) {
+ List<Range> rawRowRanges =
+ new GlobalIndexCoverage(table, snapshot, partitionFilter,
allIndexFiles)
+ .unindexedRanges(textColumnIds);
+ if (!rawRowRanges.isEmpty()) {
+ splits.add(new RawFullTextSearchSplit(rawRowRanges));
+ }
+ }
+
return () -> splits;
}
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 f4644005aa..6d136d5ee2 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
@@ -71,7 +71,7 @@ public class FullTextSearchBuilderImpl implements
FullTextSearchBuilder {
@Override
public FullTextRead newFullTextRead() {
checkArgument(limit > 0, "Limit must be positive, set via
withLimit()");
- return new FullTextReadImpl(table, limit, textColumns(), query);
+ return new FullTextReadImpl(table, partitionFilter, limit,
textColumns(), query);
}
private List<DataField> textColumns() {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchSplit.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchSplit.java
index 924b95057d..a6e7f86cba 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchSplit.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchSplit.java
@@ -18,84 +18,10 @@
package org.apache.paimon.table.source;
-import org.apache.paimon.index.IndexFileMeta;
-
import java.io.Serializable;
-import java.util.Collections;
-import java.util.List;
-import java.util.Objects;
-/** Split of full-text search. */
-public class FullTextSearchSplit implements Serializable {
+/** Base split of full-text search. */
+public abstract class FullTextSearchSplit implements Serializable {
private static final long serialVersionUID = 1L;
-
- private final String columnName;
- private final long rowRangeStart;
- private final long rowRangeEnd;
- private final List<IndexFileMeta> fullTextIndexFiles;
-
- public FullTextSearchSplit(
- long rowRangeStart, long rowRangeEnd, List<IndexFileMeta>
fullTextIndexFiles) {
- this(null, rowRangeStart, rowRangeEnd, fullTextIndexFiles);
- }
-
- public FullTextSearchSplit(
- String columnName,
- long rowRangeStart,
- long rowRangeEnd,
- List<IndexFileMeta> fullTextIndexFiles) {
- this.columnName = columnName;
- this.rowRangeStart = rowRangeStart;
- this.rowRangeEnd = rowRangeEnd;
- this.fullTextIndexFiles =
Collections.unmodifiableList(fullTextIndexFiles);
- }
-
- public String columnName() {
- return columnName;
- }
-
- public long rowRangeStart() {
- return rowRangeStart;
- }
-
- public long rowRangeEnd() {
- return rowRangeEnd;
- }
-
- public List<IndexFileMeta> fullTextIndexFiles() {
- return fullTextIndexFiles;
- }
-
- @Override
- public boolean equals(Object o) {
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- FullTextSearchSplit that = (FullTextSearchSplit) o;
- return Objects.equals(columnName, that.columnName)
- && rowRangeStart == that.rowRangeStart
- && rowRangeEnd == that.rowRangeEnd
- && Objects.equals(fullTextIndexFiles, that.fullTextIndexFiles);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(columnName, rowRangeStart, rowRangeEnd,
fullTextIndexFiles);
- }
-
- @Override
- public String toString() {
- return "FullTextSearchSplit{"
- + "columnName='"
- + columnName
- + '\''
- + ", rowRangeStart="
- + rowRangeStart
- + ", rowRangeEnd="
- + rowRangeEnd
- + ", fullTextIndexFiles="
- + fullTextIndexFiles
- + '}';
- }
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchSplit.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/IndexFullTextSearchSplit.java
similarity index 55%
copy from
paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchSplit.java
copy to
paimon-core/src/main/java/org/apache/paimon/table/source/IndexFullTextSearchSplit.java
index 924b95057d..976620e015 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextSearchSplit.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/IndexFullTextSearchSplit.java
@@ -19,28 +19,39 @@
package org.apache.paimon.table.source;
import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.IndexFileMetaSerializer;
+import org.apache.paimon.io.DataInputViewStreamWrapper;
+import org.apache.paimon.io.DataOutputViewStreamWrapper;
-import java.io.Serializable;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
-/** Split of full-text search. */
-public class FullTextSearchSplit implements Serializable {
+/** Split to read full-text index files. */
+public class IndexFullTextSearchSplit extends FullTextSearchSplit {
private static final long serialVersionUID = 1L;
- private final String columnName;
- private final long rowRangeStart;
- private final long rowRangeEnd;
- private final List<IndexFileMeta> fullTextIndexFiles;
+ private static final int VERSION = 1;
- public FullTextSearchSplit(
+ private static final ThreadLocal<IndexFileMetaSerializer> INDEX_SERIALIZER
=
+ ThreadLocal.withInitial(IndexFileMetaSerializer::new);
+
+ private String columnName;
+ private long rowRangeStart;
+ private long rowRangeEnd;
+ private transient List<IndexFileMeta> fullTextIndexFiles;
+
+ public IndexFullTextSearchSplit(
long rowRangeStart, long rowRangeEnd, List<IndexFileMeta>
fullTextIndexFiles) {
this(null, rowRangeStart, rowRangeEnd, fullTextIndexFiles);
}
- public FullTextSearchSplit(
+ public IndexFullTextSearchSplit(
String columnName,
long rowRangeStart,
long rowRangeEnd,
@@ -48,7 +59,7 @@ public class FullTextSearchSplit implements Serializable {
this.columnName = columnName;
this.rowRangeStart = rowRangeStart;
this.rowRangeEnd = rowRangeEnd;
- this.fullTextIndexFiles =
Collections.unmodifiableList(fullTextIndexFiles);
+ this.fullTextIndexFiles = Collections.unmodifiableList(new
ArrayList<>(fullTextIndexFiles));
}
public String columnName() {
@@ -67,15 +78,35 @@ public class FullTextSearchSplit implements Serializable {
return fullTextIndexFiles;
}
+ private void writeObject(ObjectOutputStream out) throws IOException {
+ out.defaultWriteObject();
+ out.writeInt(VERSION);
+ IndexFileMetaSerializer serializer = INDEX_SERIALIZER.get();
+ DataOutputViewStreamWrapper view = new
DataOutputViewStreamWrapper(out);
+ serializer.serializeList(fullTextIndexFiles, view);
+ }
+
+ private void readObject(ObjectInputStream in) throws IOException,
ClassNotFoundException {
+ in.defaultReadObject();
+ int version = in.readInt();
+ if (version != VERSION) {
+ throw new IOException("Unsupported IndexFullTextSearchSplit
version: " + version);
+ }
+ IndexFileMetaSerializer serializer = INDEX_SERIALIZER.get();
+ DataInputViewStreamWrapper view = new DataInputViewStreamWrapper(in);
+ this.fullTextIndexFiles =
+ Collections.unmodifiableList(new
ArrayList<>(serializer.deserializeList(view)));
+ }
+
@Override
public boolean equals(Object o) {
if (o == null || getClass() != o.getClass()) {
return false;
}
- FullTextSearchSplit that = (FullTextSearchSplit) o;
- return Objects.equals(columnName, that.columnName)
- && rowRangeStart == that.rowRangeStart
+ IndexFullTextSearchSplit that = (IndexFullTextSearchSplit) o;
+ return rowRangeStart == that.rowRangeStart
&& rowRangeEnd == that.rowRangeEnd
+ && Objects.equals(columnName, that.columnName)
&& Objects.equals(fullTextIndexFiles, that.fullTextIndexFiles);
}
@@ -86,7 +117,7 @@ public class FullTextSearchSplit implements Serializable {
@Override
public String toString() {
- return "FullTextSearchSplit{"
+ return "IndexFullTextSearchSplit{"
+ "columnName='"
+ columnName
+ '\''
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/RawFullTextReadImpl.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/RawFullTextReadImpl.java
new file mode 100644
index 0000000000..6508dd5751
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/RawFullTextReadImpl.java
@@ -0,0 +1,521 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.table.source;
+
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.SeekableInputStream;
+import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
+import org.apache.paimon.globalindex.GlobalIndexWriter;
+import org.apache.paimon.globalindex.GlobalIndexer;
+import org.apache.paimon.globalindex.GlobalIndexerFactoryUtils;
+import org.apache.paimon.globalindex.ResultEntry;
+import org.apache.paimon.globalindex.ScoredGlobalIndexResult;
+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.IndexPathFactory;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.predicate.FullTextQuery;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.SpecialFields;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.CloseableIterator;
+import org.apache.paimon.utils.IOUtils;
+import org.apache.paimon.utils.Range;
+import org.apache.paimon.utils.RoaringNavigableMap64;
+
+import javax.annotation.Nullable;
+
+import java.io.ByteArrayOutputStream;
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ExecutorService;
+
+import static org.apache.paimon.utils.Preconditions.checkNotNull;
+
+/** Implementation for raw full-text search. */
+class RawFullTextReadImpl {
+
+ private final FileStoreTable table;
+ @Nullable private final PartitionPredicate partitionFilter;
+ private final int limit;
+ private final FullTextQuery query;
+ private final IndexSearch indexSearch;
+
+ RawFullTextReadImpl(
+ FileStoreTable table,
+ @Nullable PartitionPredicate partitionFilter,
+ int limit,
+ FullTextQuery query,
+ IndexSearch indexSearch) {
+ this.table = table;
+ this.partitionFilter = partitionFilter;
+ this.limit = limit;
+ this.query = query;
+ this.indexSearch = indexSearch;
+ }
+
+ ScoredGlobalIndexResult withRawSearch(
+ ScoredGlobalIndexResult indexedResult,
+ List<Range> rawRowRanges,
+ Map<String, DataField> fieldsByName,
+ Map<String, List<IndexFullTextSearchSplit>> splitsByColumn,
+ ExecutorService executor) {
+ rawRowRanges = Range.sortAndMergeOverlap(rawRowRanges, true);
+ if (rawRowRanges.isEmpty()) {
+ return indexedResult;
+ }
+
+ ScoredGlobalIndexResult rawResult =
+ readRawSearch(rawRowRanges, fieldsByName, splitsByColumn,
executor);
+ return overrideWithRawSearch(indexedResult, rawRowRanges, rawResult);
+ }
+
+ private ScoredGlobalIndexResult readRawSearch(
+ List<Range> rawRowRanges,
+ Map<String, DataField> fieldsByName,
+ Map<String, List<IndexFullTextSearchSplit>> splitsByColumn,
+ ExecutorService executor) {
+ RowType readType = SpecialFields.rowTypeWithRowId(table.rowType());
+ TableScan.Plan plan =
rawReadBuilder(readType).withRowRanges(rawRowRanges).newScan().plan();
+ ReadBuilder readBuilder = rawReadBuilder(readType);
+ int rowIdIndex = readType.getFieldIndex(SpecialFields.ROW_ID.name());
+ Map<String, RawFullTextIndex> rawIndexes =
+ createRawFullTextIndexes(fieldsByName, splitsByColumn,
readType, rawRowRanges);
+
+ try {
+ try (RecordReader<InternalRow> reader =
readBuilder.newRead().createReader(plan);
+ CloseableIterator<InternalRow> iterator =
reader.toCloseableIterator()) {
+ while (iterator.hasNext()) {
+ InternalRow row = iterator.next();
+ for (RawFullTextIndex rawIndex : rawIndexes.values()) {
+ long rowId = row.getLong(rowIdIndex);
+ rawIndex.write(row, rowId);
+ }
+ }
+ }
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to read raw rows for full-text
search.", e);
+ }
+
+ try {
+ Map<String, List<IndexFullTextSearchSplit>> rawSplitsByColumn =
new HashMap<>();
+ for (Map.Entry<String, RawFullTextIndex> entry :
rawIndexes.entrySet()) {
+ IndexFullTextSearchSplit split = entry.getValue().finish();
+ if (split != null) {
+ rawSplitsByColumn.put(entry.getKey(),
Collections.singletonList(split));
+ }
+ }
+
+ return indexSearch
+ .eval(
+ query,
+ fieldsByName,
+ rawSplitsByColumn,
+ new RawFullTextIndexPathFactory(rawIndexes),
+ m ->
+ new MemorySeekableInputStream(
+ rawFileBytes(rawIndexes,
m.filePath().getName())),
+ executor)
+ .topK(limit);
+ } finally {
+ IOUtils.closeAllQuietly(rawIndexes.values());
+ }
+ }
+
+ private static ScoredGlobalIndexResult overrideWithRawSearch(
+ ScoredGlobalIndexResult indexedResult,
+ List<Range> rawRowRanges,
+ ScoredGlobalIndexResult rawResult) {
+ RoaringNavigableMap64 rawRows = new RoaringNavigableMap64();
+ for (Range range : rawRowRanges) {
+ rawRows.addRange(range);
+ }
+ RoaringNavigableMap64 filteredIndexedRows =
+ RoaringNavigableMap64.or(new RoaringNavigableMap64(),
indexedResult.results());
+ filteredIndexedRows.andNot(rawRows);
+ RoaringNavigableMap64 resultRows =
+ RoaringNavigableMap64.or(filteredIndexedRows,
rawResult.results());
+ return ScoredGlobalIndexResult.create(
+ resultRows,
+ rowId ->
+ rawResult.results().contains(rowId)
+ ? rawResult.scoreGetter().score(rowId)
+ : indexedResult.scoreGetter().score(rowId));
+ }
+
+ private Map<String, RawFullTextIndex> createRawFullTextIndexes(
+ Map<String, DataField> fieldsByName,
+ Map<String, List<IndexFullTextSearchSplit>> splitsByColumn,
+ RowType readType,
+ List<Range> rawRowRanges) {
+ Map<String, RawFullTextIndex> rawIndexes = new HashMap<>();
+ long rowRangeStart = rawRowRanges.get(0).from;
+ long rowRangeEnd = rawRowRanges.get(rawRowRanges.size() - 1).to;
+ String fallbackIndexType = firstIndexType(splitsByColumn);
+ for (String column : query.columns()) {
+ if (rawIndexes.containsKey(column)) {
+ continue;
+ }
+ DataField textColumn = checkNotNull(fieldsByName.get(column));
+ String indexType = indexType(column, splitsByColumn);
+ if (indexType == null) {
+ indexType = checkNotNull(fallbackIndexType);
+ }
+ GlobalIndexer globalIndexer =
+ GlobalIndexerFactoryUtils.load(indexType)
+ .create(textColumn, rawSearchOptions());
+ try {
+ RawFullTextIndexFileWriter fileWriter = new
RawFullTextIndexFileWriter(column);
+ GlobalIndexWriter indexWriter =
globalIndexer.createWriter(fileWriter);
+ if (!(indexWriter instanceof GlobalIndexSingleColumnWriter)) {
+ throw new IllegalArgumentException(
+ "Full-text raw search requires a single-column
global index writer.");
+ }
+ rawIndexes.put(
+ column,
+ new RawFullTextIndex(
+ column,
+ readColumnIndex(textColumn, readType),
+ indexType,
+ textColumn.id(),
+ rowRangeStart,
+ rowRangeEnd,
+ (GlobalIndexSingleColumnWriter) indexWriter,
+ fileWriter));
+ } catch (IOException e) {
+ throw new RuntimeException(
+ "Failed to create raw full-text index writer for
column: " + column, e);
+ }
+ }
+ return rawIndexes;
+ }
+
+ @Nullable
+ private static String indexType(
+ String column, Map<String, List<IndexFullTextSearchSplit>>
splitsByColumn) {
+ List<IndexFullTextSearchSplit> splits = splitsByColumn.get(column);
+ if (splits == null || splits.isEmpty() ||
splits.get(0).fullTextIndexFiles().isEmpty()) {
+ return null;
+ }
+ return splits.get(0).fullTextIndexFiles().get(0).indexType();
+ }
+
+ @Nullable
+ private static String firstIndexType(
+ Map<String, List<IndexFullTextSearchSplit>> splitsByColumn) {
+ for (List<IndexFullTextSearchSplit> splits : splitsByColumn.values()) {
+ if (splits != null
+ && !splits.isEmpty()
+ && !splits.get(0).fullTextIndexFiles().isEmpty()) {
+ return splits.get(0).fullTextIndexFiles().get(0).indexType();
+ }
+ }
+ return null;
+ }
+
+ private static int readColumnIndex(DataField textColumn, RowType readType)
{
+ return readType.getFieldIndexByFieldId(textColumn.id());
+ }
+
+ private static byte[] rawFileBytes(Map<String, RawFullTextIndex>
rawIndexes, String fileName) {
+ for (RawFullTextIndex rawIndex : rawIndexes.values()) {
+ if (rawIndex.containsFile(fileName)) {
+ return rawIndex.fileBytes(fileName);
+ }
+ }
+ throw new IllegalArgumentException("Unknown raw full-text index file:
" + fileName);
+ }
+
+ private ReadBuilder rawReadBuilder(RowType readType) {
+ ReadBuilder readBuilder =
table.newReadBuilder().withReadType(readType);
+ if (partitionFilter != null) {
+ readBuilder.withPartitionFilter(partitionFilter);
+ }
+ return readBuilder;
+ }
+
+ private Options rawSearchOptions() {
+ Options options = new
Options(table.coreOptions().toConfiguration().toMap());
+ options.setString("tantivy.searcher-pool.max-size", "0");
+ return options;
+ }
+
+ interface IndexSearch {
+
+ ScoredGlobalIndexResult eval(
+ FullTextQuery query,
+ Map<String, DataField> fieldsByName,
+ Map<String, List<IndexFullTextSearchSplit>> splitsByColumn,
+ IndexPathFactory indexPathFactory,
+ GlobalIndexFileReader indexFileReader,
+ ExecutorService executor);
+ }
+
+ private static class RawFullTextIndex implements Closeable {
+
+ private final String column;
+ private final int columnIndex;
+ private final String indexType;
+ private final int fieldId;
+ private final long rowRangeStart;
+ private final long rowRangeEnd;
+ private final GlobalIndexSingleColumnWriter writer;
+ private final RawFullTextIndexFileWriter fileWriter;
+ private final RoaringNavigableMap64 indexedRows = new
RoaringNavigableMap64();
+
+ private RawFullTextIndex(
+ String column,
+ int columnIndex,
+ String indexType,
+ int fieldId,
+ long rowRangeStart,
+ long rowRangeEnd,
+ GlobalIndexSingleColumnWriter writer,
+ RawFullTextIndexFileWriter fileWriter) {
+ this.column = column;
+ this.columnIndex = columnIndex;
+ this.indexType = indexType;
+ this.fieldId = fieldId;
+ this.rowRangeStart = rowRangeStart;
+ this.rowRangeEnd = rowRangeEnd;
+ this.writer = writer;
+ this.fileWriter = fileWriter;
+ }
+
+ private void write(InternalRow row, long rowId) {
+ if (row.isNullAt(columnIndex)) {
+ return;
+ }
+ writer.write(row.getString(columnIndex), rowId - rowRangeStart);
+ indexedRows.add(rowId);
+ }
+
+ @Nullable
+ private IndexFullTextSearchSplit finish() {
+ List<ResultEntry> resultEntries = writer.finish();
+ if (resultEntries.isEmpty() || indexedRows.isEmpty()) {
+ return null;
+ }
+
+ List<IndexFileMeta> indexFiles = new
ArrayList<>(resultEntries.size());
+ for (ResultEntry entry : resultEntries) {
+ byte[] bytes = fileWriter.fileBytes(entry.fileName());
+ GlobalIndexMeta meta =
+ new GlobalIndexMeta(
+ rowRangeStart, rowRangeEnd, fieldId, null,
entry.meta());
+ indexFiles.add(
+ new IndexFileMeta(
+ indexType,
+ entry.fileName(),
+ bytes.length,
+ entry.rowCount(),
+ meta,
+ null));
+ }
+ return new IndexFullTextSearchSplit(column, rowRangeStart,
rowRangeEnd, indexFiles);
+ }
+
+ private Path path(String fileName) {
+ return fileWriter.path(fileName);
+ }
+
+ private byte[] fileBytes(String fileName) {
+ return fileWriter.fileBytes(fileName);
+ }
+
+ private boolean containsFile(String fileName) {
+ return fileWriter.containsFile(fileName);
+ }
+
+ @Override
+ public void close() {
+ if (writer instanceof AutoCloseable) {
+ IOUtils.closeQuietly((AutoCloseable) writer);
+ }
+ }
+ }
+
+ private static class RawFullTextIndexFileWriter implements
GlobalIndexFileWriter {
+
+ private final String column;
+ private final String id = UUID.randomUUID().toString();
+ private final Map<String, byte[]> files = new HashMap<>();
+
+ private RawFullTextIndexFileWriter(String column) {
+ this.column = column;
+ }
+
+ @Override
+ public String newFileName(String prefix) {
+ return "raw-" + column + "-" + prefix + "-" + id + "-" +
files.size() + ".index";
+ }
+
+ @Override
+ public PositionOutputStream newOutputStream(String fileName) {
+ return new MemoryPositionOutputStream(bytes -> files.put(fileName,
bytes));
+ }
+
+ private Path path(String fileName) {
+ return new Path("memory://full-text-raw/" + fileName);
+ }
+
+ private byte[] fileBytes(String fileName) {
+ return checkNotNull(files.get(fileName));
+ }
+
+ private boolean containsFile(String fileName) {
+ return files.containsKey(fileName);
+ }
+ }
+
+ private static class RawFullTextIndexPathFactory implements
IndexPathFactory {
+
+ private final Map<String, RawFullTextIndex> rawIndexes;
+
+ private RawFullTextIndexPathFactory(Map<String, RawFullTextIndex>
rawIndexes) {
+ this.rawIndexes = rawIndexes;
+ }
+
+ @Override
+ public Path toPath(String fileName) {
+ RawFullTextIndex rawIndex = rawIndex(fileName);
+ return rawIndex.path(fileName);
+ }
+
+ @Override
+ public Path newPath() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean isExternalPath() {
+ return false;
+ }
+
+ private RawFullTextIndex rawIndex(String fileName) {
+ for (RawFullTextIndex rawIndex : rawIndexes.values()) {
+ if (rawIndex.containsFile(fileName)) {
+ return rawIndex;
+ }
+ }
+ throw new IllegalArgumentException("Unknown raw full-text index
file: " + fileName);
+ }
+ }
+
+ private static class MemoryPositionOutputStream extends
PositionOutputStream {
+
+ private final ByteArrayOutputStream out = new ByteArrayOutputStream();
+ private final FileCommitter committer;
+
+ private MemoryPositionOutputStream(FileCommitter committer) {
+ this.committer = committer;
+ }
+
+ @Override
+ public long getPos() {
+ return out.size();
+ }
+
+ @Override
+ public void write(int b) {
+ out.write(b);
+ }
+
+ @Override
+ public void write(byte[] b) throws IOException {
+ out.write(b);
+ }
+
+ @Override
+ public void write(byte[] b, int off, int len) {
+ out.write(b, off, len);
+ }
+
+ @Override
+ public void flush() {}
+
+ @Override
+ public void close() {
+ committer.commit(out.toByteArray());
+ }
+ }
+
+ private static class MemorySeekableInputStream extends SeekableInputStream
{
+
+ private final byte[] bytes;
+ private int pos;
+
+ private MemorySeekableInputStream(byte[] bytes) {
+ this.bytes = bytes;
+ }
+
+ @Override
+ public void seek(long desired) {
+ if (desired < 0 || desired > bytes.length) {
+ throw new IllegalArgumentException("Cannot seek to position: "
+ desired);
+ }
+ pos = (int) desired;
+ }
+
+ @Override
+ public long getPos() {
+ return pos;
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) {
+ if (pos >= bytes.length) {
+ return -1;
+ }
+ int read = Math.min(len, bytes.length - pos);
+ System.arraycopy(bytes, pos, b, off, read);
+ pos += read;
+ return read;
+ }
+
+ @Override
+ public void close() {}
+
+ @Override
+ public int read() {
+ if (pos >= bytes.length) {
+ return -1;
+ }
+ return bytes[pos++] & 0xFF;
+ }
+ }
+
+ private interface FileCommitter {
+
+ void commit(byte[] bytes);
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/RawFullTextSearchSplit.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/RawFullTextSearchSplit.java
new file mode 100644
index 0000000000..a95ea76255
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/RawFullTextSearchSplit.java
@@ -0,0 +1,61 @@
+/*
+ * 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.utils.Range;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+/** Split to scan raw rows for full-text search. */
+public class RawFullTextSearchSplit extends FullTextSearchSplit {
+
+ private static final long serialVersionUID = 1L;
+
+ private final List<Range> rowRanges;
+
+ public RawFullTextSearchSplit(List<Range> rowRanges) {
+ this.rowRanges = Collections.unmodifiableList(new
ArrayList<>(rowRanges));
+ }
+
+ public List<Range> rowRanges() {
+ return rowRanges;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ RawFullTextSearchSplit that = (RawFullTextSearchSplit) o;
+ return Objects.equals(rowRanges, that.rowRanges);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(rowRanges);
+ }
+
+ @Override
+ public String toString() {
+ return "RawFullTextSearchSplit{" + "rowRanges=" + rowRanges + '}';
+ }
+}
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 c9178d3dbf..5f61bc5de6 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
@@ -24,6 +24,7 @@ import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.serializer.InternalRowSerializer;
import org.apache.paimon.globalindex.GlobalIndexBuilderUtils;
import org.apache.paimon.globalindex.GlobalIndexResult;
import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
@@ -35,6 +36,7 @@ import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.io.CompactIncrement;
import org.apache.paimon.io.DataIncrement;
import org.apache.paimon.options.Options;
+import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.predicate.FullTextQuery;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
@@ -49,10 +51,15 @@ import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.CommitMessageImpl;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.Range;
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;
@@ -118,6 +125,162 @@ public class FullTextSearchBuilderTest extends
TableTestBase {
assertThat(ids).containsAnyOf(0, 1, 3);
}
+ @Test
+ public void testFullTextSearchNonFastModesScanUnindexedData() throws
Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+
+ String[] indexedDocuments = {
+ "Apache Paimon is a lake format", "Paimon supports full-text
search"
+ };
+ writeDocuments(table, indexedDocuments);
+ buildAndCommitIndex(table, indexedDocuments);
+ writeDocuments(
+ table,
+ new String[] {
+ "Vector search is also supported", "Fresh Paimon documents
should be searchable"
+ });
+
+ GlobalIndexResult fastResult =
+ table.newFullTextSearchBuilder()
+ .withQuery(FullTextQuery.match("Fresh",
TEXT_FIELD_NAME))
+ .withLimit(10)
+ .executeLocal();
+ assertThat(readIds(table, fastResult)).isEmpty();
+
+ for (String searchMode : Arrays.asList("full", "detail")) {
+ FileStoreTable nonFastModeTable =
+ (FileStoreTable)
+ table.copy(
+ Collections.singletonMap(
+
CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key(),
+ searchMode));
+ GlobalIndexResult result =
+ nonFastModeTable
+ .newFullTextSearchBuilder()
+ .withQuery(FullTextQuery.match("Fresh",
TEXT_FIELD_NAME))
+ .withLimit(10)
+ .executeLocal();
+
+ assertThat(readIds(nonFastModeTable, result)).containsExactly(1);
+ }
+ }
+
+ @Test
+ public void testFullTextSearchRawSearchRespectsPartitionFilter() throws
Exception {
+ Identifier identifier = identifier("PartitionedTextTable");
+ Schema schema =
+ Schema.newBuilder()
+ .column("pt", DataTypes.INT())
+ .column("id", DataTypes.INT())
+ .column(TEXT_FIELD_NAME, DataTypes.STRING())
+ .partitionKeys("pt")
+ .option(CoreOptions.BUCKET.key(), "-1")
+ .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(),
"true")
+ .build();
+ catalog.createTable(identifier, schema, false);
+ FileStoreTable table = getTable(identifier);
+
+ RowType partitionType = RowType.of(DataTypes.INT());
+ InternalRowSerializer serializer = new
InternalRowSerializer(partitionType);
+ BinaryRow partition1 = serializer.toBinaryRow(GenericRow.of(1)).copy();
+
+ writePartitionedDocuments(
+ table, 1, new String[] {"indexed Paimon document", "another
document"});
+ buildAndCommitIndexForColumn(
+ table,
+ TEXT_FIELD_NAME,
+ new String[] {"indexed Paimon document", "another document"},
+ partition1);
+ writePartitionedDocuments(table, 2, new String[] {"fresh Paimon
document"});
+
+ PartitionPredicate partitionFilter =
+ PartitionPredicate.fromMultiple(
+ partitionType, Collections.singletonList(partition1));
+ FileStoreTable fullModeTable =
+ (FileStoreTable)
+ table.copy(
+ Collections.singletonMap(
+
CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key(), "full"));
+
+ GlobalIndexResult result =
+ fullModeTable
+ .newFullTextSearchBuilder()
+ .withPartitionFilter(partitionFilter)
+ .withQuery(FullTextQuery.match("fresh",
TEXT_FIELD_NAME))
+ .withLimit(10)
+ .executeLocal();
+
+ assertThat(readIds(fullModeTable, result)).isEmpty();
+ }
+
+ @Test
+ public void testFullTextSearchRawSearchOverridesPartiallyIndexedRows()
throws Exception {
+ FileStoreTable table = createMultiTextTable();
+ writeMultiTextDocuments(
+ table,
+ new String[][] {
+ {"paimon title", "other body"},
+ {"other title", "paimon body"},
+ {"paimon title", "paimon body"}
+ });
+ buildAndCommitIndexForColumn(
+ table, "title", new String[] {"paimon title", "other title",
"paimon title"});
+ buildAndCommitIndexForColumn(table, "body", new String[] {"other
body", "paimon body"});
+
+ FileStoreTable fullModeTable =
+ (FileStoreTable)
+ table.copy(
+ Collections.singletonMap(
+
CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key(), "full"));
+ ScoredGlobalIndexResult result =
+ (ScoredGlobalIndexResult)
+ fullModeTable
+ .newFullTextSearchBuilder()
+ .withQuery(
+ FullTextQuery.multiMatch(
+ "paimon",
Arrays.asList("title", "body")))
+ .withLimit(10)
+ .executeLocal();
+
+ assertThat(result.results()).contains(2L);
+ assertThat(result.scoreGetter().score(2L)).isEqualTo(2.0f);
+ }
+
+ @Test
+ public void testFullTextSearchRawSearchRemovesPartialIndexedRows() throws
Exception {
+ FileStoreTable table = createMultiTextTable();
+ writeMultiTextDocuments(
+ table,
+ new String[][] {
+ {"paimon title", "paimon body"},
+ {"paimon title", "other body"}
+ });
+ buildAndCommitIndexForColumn(table, "title", new String[] {"paimon
title", "paimon title"});
+ buildAndCommitIndexForColumn(table, "body", new String[] {"paimon
body"});
+
+ FileStoreTable fullModeTable =
+ (FileStoreTable)
+ table.copy(
+ Collections.singletonMap(
+
CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key(), "full"));
+ GlobalIndexResult result =
+ fullModeTable
+ .newFullTextSearchBuilder()
+ .withQuery(
+ new FullTextQuery.BooleanQuery(
+ Collections.emptyList(),
+ Arrays.asList(
+ FullTextQuery.match("paimon",
"title"),
+ FullTextQuery.match("paimon",
"body")),
+ Collections.emptyList()))
+ .withLimit(10)
+ .executeLocal();
+
+ assertThat(result.results()).containsExactly(0L);
+ }
+
@Test
public void testHybridSearchBuilderWithFullTextRoute() throws Exception {
createTableDefault();
@@ -554,6 +717,61 @@ public class FullTextSearchBuilderTest extends
TableTestBase {
assertThat(searchBuilder.executeLocal().results().isEmpty()).isTrue();
}
+ @Test
+ public void testFullTextSearchSplitSerialization() throws Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+
+ String[] documents = {"Apache Paimon", "full-text search"};
+ writeDocuments(table, documents);
+ buildAndCommitIndex(table, documents);
+
+ FullTextScan.Plan plan =
+ table.newFullTextSearchBuilder()
+ .withQuery(FullTextQuery.match("Paimon",
TEXT_FIELD_NAME))
+ .withLimit(2)
+ .newFullTextScan()
+ .scan();
+
+ assertThat(plan.splits()).hasSize(1);
+ IndexFullTextSearchSplit original = (IndexFullTextSearchSplit)
plan.splits().get(0);
+
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ try (ObjectOutputStream out = new ObjectOutputStream(bos)) {
+ out.writeObject(original);
+ }
+
+ IndexFullTextSearchSplit deserialized;
+ try (ObjectInputStream in =
+ new ObjectInputStream(new
ByteArrayInputStream(bos.toByteArray()))) {
+ deserialized = (IndexFullTextSearchSplit) in.readObject();
+ }
+
+ assertThat(deserialized.columnName()).isEqualTo(original.columnName());
+
assertThat(deserialized.rowRangeStart()).isEqualTo(original.rowRangeStart());
+
assertThat(deserialized.rowRangeEnd()).isEqualTo(original.rowRangeEnd());
+
assertThat(deserialized.fullTextIndexFiles()).hasSize(original.fullTextIndexFiles().size());
+ for (int i = 0; i < original.fullTextIndexFiles().size(); i++) {
+ assertThat(deserialized.fullTextIndexFiles().get(i).fileName())
+
.isEqualTo(original.fullTextIndexFiles().get(i).fileName());
+ }
+
+ RawFullTextSearchSplit rawOriginal =
+ new RawFullTextSearchSplit(Collections.singletonList(new
Range(2, 3)));
+ bos = new ByteArrayOutputStream();
+ try (ObjectOutputStream out = new ObjectOutputStream(bos)) {
+ out.writeObject(rawOriginal);
+ }
+
+ RawFullTextSearchSplit rawDeserialized;
+ try (ObjectInputStream in =
+ new ObjectInputStream(new
ByteArrayInputStream(bos.toByteArray()))) {
+ rawDeserialized = (RawFullTextSearchSplit) in.readObject();
+ }
+
+
assertThat(rawDeserialized.rowRanges()).isEqualTo(rawOriginal.rowRanges());
+ }
+
// ====================== Helper methods ======================
private void writeDocuments(FileStoreTable table, String[] documents)
throws Exception {
@@ -567,6 +785,18 @@ public class FullTextSearchBuilderTest extends
TableTestBase {
}
}
+ private void writePartitionedDocuments(FileStoreTable table, int
partition, String[] documents)
+ throws Exception {
+ BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = writeBuilder.newWrite();
+ BatchTableCommit commit = writeBuilder.newCommit()) {
+ for (int i = 0; i < documents.length; i++) {
+ write.write(GenericRow.of(partition, i,
BinaryString.fromString(documents[i])));
+ }
+ commit.commit(write.prepareCommit());
+ }
+ }
+
private void buildAndCommitIndex(FileStoreTable table, String[] documents)
throws Exception {
buildAndCommitIndexWithFields(
table,
@@ -649,6 +879,12 @@ public class FullTextSearchBuilderTest extends
TableTestBase {
private void buildAndCommitIndexForColumn(
FileStoreTable table, String columnName, String[] documents)
throws Exception {
+ buildAndCommitIndexForColumn(table, columnName, documents,
BinaryRow.EMPTY_ROW);
+ }
+
+ private void buildAndCommitIndexForColumn(
+ FileStoreTable table, String columnName, String[] documents,
BinaryRow partition)
+ throws Exception {
Options options = table.coreOptions().toConfiguration();
DataField textField = table.rowType().getField(columnName);
@@ -678,11 +914,7 @@ public class FullTextSearchBuilderTest extends
TableTestBase {
DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles);
CommitMessage message =
new CommitMessageImpl(
- BinaryRow.EMPTY_ROW,
- 0,
- null,
- dataIncrement,
- CompactIncrement.emptyIncrement());
+ partition, 0, null, dataIncrement,
CompactIncrement.emptyIncrement());
try (BatchTableCommit commit =
table.newBatchWriteBuilder().newCommit()) {
commit.commit(Collections.singletonList(message));
}
diff --git a/paimon-python/pypaimon/globalindex/global_index_coverage.py
b/paimon-python/pypaimon/globalindex/global_index_coverage.py
index 9f94f89495..d0a05cb50b 100644
--- a/paimon-python/pypaimon/globalindex/global_index_coverage.py
+++ b/paimon-python/pypaimon/globalindex/global_index_coverage.py
@@ -53,11 +53,13 @@ class GlobalIndexCoverage:
def unindexed_ranges(
self,
- fields_or_field_id: Union[List[DataField], int],
+ fields_or_field_id: Union[List[DataField], Collection[int], int],
predicate: Optional[Predicate] = None,
) -> List[Range]:
if isinstance(fields_or_field_id, int):
field_ids = {fields_or_field_id}
+ elif _is_field_id_collection(fields_or_field_id):
+ field_ids = set(fields_or_field_id)
else:
field_by_name = {f.name: f for f in fields_or_field_id}
field_ids = set()
@@ -143,3 +145,13 @@ def _global_index_search_mode(table):
if hasattr(options, "global_index_search_mode"):
return options.global_index_search_mode()
return CoreOptions(Options.from_none()).global_index_search_mode()
+
+
+def _is_field_id_collection(fields_or_field_id):
+ if isinstance(fields_or_field_id, (str, bytes)):
+ return False
+ try:
+ iterator = iter(fields_or_field_id)
+ except TypeError:
+ return False
+ return all(isinstance(field_id, int) for field_id in iterator)
diff --git a/paimon-python/pypaimon/table/source/full_text_scan.py
b/paimon-python/pypaimon/table/source/full_text_scan.py
index 1324a2830d..84968e7b65 100644
--- a/paimon-python/pypaimon/table/source/full_text_scan.py
+++ b/paimon-python/pypaimon/table/source/full_text_scan.py
@@ -21,6 +21,7 @@ from abc import ABC, abstractmethod
from collections import defaultdict
from typing import List
+from pypaimon.globalindex.global_index_coverage import GlobalIndexCoverage
from pypaimon.globalindex.tantivy.tantivy_full_text_global_index_reader import
(
TANTIVY_FULLTEXT_IDENTIFIER,
)
@@ -111,4 +112,18 @@ class FullTextScanImpl(FullTextScan):
FullTextSearchSplit(
column_name, range_key.from_, range_key.to, files))
+ if all_index_files:
+ raw_row_ranges = GlobalIndexCoverage(
+ self._table,
+ snapshot,
+ partition_filter,
+ all_index_files,
+ ).unindexed_ranges(list(text_column_ids))
+ if raw_row_ranges:
+ raise NotImplementedError(
+ "Python full-text search does not support "
+ "global-index.search-mode=full/detail for uncovered row "
+ "ranges yet. Raw full-text search requires rebuilding "
+ "temporary full-text indexes.")
+
return FullTextScanPlan(splits)
diff --git a/paimon-python/pypaimon/tests/global_index_test.py
b/paimon-python/pypaimon/tests/global_index_test.py
index 60ca559cda..78d0f9a3d7 100644
--- a/paimon-python/pypaimon/tests/global_index_test.py
+++ b/paimon-python/pypaimon/tests/global_index_test.py
@@ -126,6 +126,22 @@ class GlobalIndexCoverageTest(unittest.TestCase):
self.assertEqual([Range(5, 9)], coverage.unindexed_ranges(0))
+ def test_full_mode_accepts_multiple_field_ids(self):
+ from pypaimon.globalindex.global_index_coverage import
GlobalIndexCoverage
+
+ table = _CoverageTable("full")
+ coverage = GlobalIndexCoverage(
+ table,
+ _CoverageSnapshot(10),
+ None,
+ [
+ _coverage_index_file(0, 0, 9),
+ _coverage_index_file(1, 0, 4),
+ ],
+ )
+
+ self.assertEqual([Range(5, 9)], coverage.unindexed_ranges([0, 1]))
+
def test_full_mode_intersects_coverage_for_all_predicate_fields(self):
from pypaimon.globalindex.global_index_coverage import
GlobalIndexCoverage
diff --git a/paimon-python/pypaimon/tests/vector_search_filter_test.py
b/paimon-python/pypaimon/tests/vector_search_filter_test.py
index e3ccac2f33..0e29e5c1ad 100644
--- a/paimon-python/pypaimon/tests/vector_search_filter_test.py
+++ b/paimon-python/pypaimon/tests/vector_search_filter_test.py
@@ -121,7 +121,7 @@ def _entry(partition_row, field_id, index_type, file_name,
index_file=index_file)
-def _patch_snapshot(testcase, entries):
+def _patch_snapshot(testcase, entries, snapshot=None):
"""Stub IndexFileHandler.scan + snapshot resolution."""
mock.patch.stopall()
@@ -144,7 +144,7 @@ def _patch_snapshot(testcase, entries):
testcase._scan_patch.start()
testcase._travel_patch = mock.patch(
"pypaimon.snapshot.time_travel_util.TimeTravelUtil.try_travel_to_snapshot",
- return_value=object())
+ return_value=snapshot if snapshot is not None else object())
testcase._travel_patch.start()
@@ -1121,6 +1121,31 @@ class FullTextSearchBuilderDslTest(unittest.TestCase):
["ft.index"],
[f.file_name for f in splits[0].full_text_index_files])
+ def test_full_text_scan_rejects_full_mode_raw_search(self):
+ from pypaimon.common.options.core_options import CoreOptions
+ from pypaimon.common.options.options import Options
+ from pypaimon.table.source.full_text_scan import FullTextScanImpl
+
+ class _Options:
+ options = Options({"global-index.search-mode": "full"})
+
+ def global_index_search_mode(self_inner):
+ return
CoreOptions(self_inner.options).global_index_search_mode()
+
+ text_field = _field(1, "content", "STRING")
+ entry = _entry(
+ None, field_id=1, index_type="tantivy-fulltext",
+ file_name="ft.index", row_range_start=0, row_range_end=4)
+ table = _StubTable(fields=[text_field], entries=[entry])
+ table.options = _Options()
+ _patch_snapshot(self, [entry], types.SimpleNamespace(next_row_id=10))
+
+ with self.assertRaises(NotImplementedError) as ctx:
+ FullTextScanImpl(table, [text_field]).scan()
+
+ self.assertIn("global-index.search-mode=full/detail",
str(ctx.exception))
+ self.assertIn("rebuilding temporary full-text indexes",
str(ctx.exception))
+
class VectorSearchFilterTest(unittest.TestCase):
"""Non-partitioned wiring: scan + read + external_path plumbing."""
diff --git
a/paimon-tantivy/paimon-tantivy-index/src/main/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexerFactory.java
b/paimon-tantivy/paimon-tantivy-index/src/main/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexerFactory.java
index 641a779a81..127d22c657 100644
---
a/paimon-tantivy/paimon-tantivy-index/src/main/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexerFactory.java
+++
b/paimon-tantivy/paimon-tantivy-index/src/main/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexerFactory.java
@@ -45,10 +45,15 @@ public class TantivyFullTextGlobalIndexerFactory implements
GlobalIndexerFactory
@Override
public GlobalIndexer create(DataField field, Options options) {
+ int maxSize =
options.get(TantivyFullTextIndexOptions.SEARCHER_POOL_MAX_SIZE);
+ if (maxSize <= 0) {
+ return new TantivyFullTextGlobalIndexer(
+ new TantivySearcherPool(0),
+ new
TantivyFullTextIndexOptions(removeTantivyPrefix(options)));
+ }
if (searcherPool == null) {
synchronized (this) {
if (searcherPool == null) {
- int maxSize =
options.get(TantivyFullTextIndexOptions.SEARCHER_POOL_MAX_SIZE);
searcherPool = new TantivySearcherPool(maxSize);
}
}
diff --git
a/paimon-tantivy/paimon-tantivy-index/src/test/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexerFactoryTest.java
b/paimon-tantivy/paimon-tantivy-index/src/test/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexerFactoryTest.java
new file mode 100644
index 0000000000..3bd9bb0459
--- /dev/null
+++
b/paimon-tantivy/paimon-tantivy-index/src/test/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexerFactoryTest.java
@@ -0,0 +1,53 @@
+/*
+ * 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.tantivy.index;
+
+import org.apache.paimon.globalindex.GlobalIndexer;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Field;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link TantivyFullTextGlobalIndexerFactory}. */
+public class TantivyFullTextGlobalIndexerFactoryTest {
+
+ @Test
+ public void testFactoryDisablesSearcherPoolForZeroMaxSize() throws
Exception {
+ TantivyFullTextGlobalIndexerFactory factory = new
TantivyFullTextGlobalIndexerFactory();
+ DataField field = new DataField(0, "text", DataTypes.STRING());
+
+ GlobalIndexer pooled = factory.create(field, new Options());
+ Options disabledOptions = new Options();
+
disabledOptions.set(TantivyFullTextIndexOptions.SEARCHER_POOL_MAX_SIZE, 0);
+ GlobalIndexer disabled = factory.create(field, disabledOptions);
+
+ assertThat(searcherPool(pooled)).isNotSameAs(searcherPool(disabled));
+ }
+
+ private static TantivySearcherPool searcherPool(GlobalIndexer indexer)
throws Exception {
+ Field field =
TantivyFullTextGlobalIndexer.class.getDeclaredField("searcherPool");
+ field.setAccessible(true);
+ return (TantivySearcherPool) field.get(indexer);
+ }
+}
