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 1155e12082 [core] Support full-text indexes on DV tables (#9031)
1155e12082 is described below
commit 1155e1208262b925b05bd37ef671d3c04fa042f0
Author: XiaoHongbo <[email protected]>
AuthorDate: Wed Aug 5 11:27:32 2026 +0800
[core] Support full-text indexes on DV tables (#9031)
---
.../table/source/DataEvolutionFullTextRead.java | 21 ++++-
.../table/source/DataEvolutionFullTextScan.java | 40 +++++++-
.../apache/paimon/table/source/FullTextScan.java | 10 ++
.../table/source/FullTextSearchBuilderImpl.java | 5 +-
.../paimon/table/source/RawFullTextReadImpl.java | 24 ++++-
.../table/source/FullTextSearchBuilderTest.java | 101 +++++++++++++++++++++
6 files changed, 192 insertions(+), 9 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextRead.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextRead.java
index d9100c2838..780155f837 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextRead.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextRead.java
@@ -18,6 +18,7 @@
package org.apache.paimon.table.source;
+import org.apache.paimon.Snapshot;
import org.apache.paimon.globalindex.GlobalIndexIOMeta;
import org.apache.paimon.globalindex.GlobalIndexReadThreadPool;
import org.apache.paimon.globalindex.GlobalIndexReader;
@@ -79,6 +80,16 @@ public class DataEvolutionFullTextRead implements
FullTextRead {
@Override
public GlobalIndexResult read(List<FullTextSearchSplit> splits) {
+ return read(splits, null);
+ }
+
+ @Override
+ public GlobalIndexResult read(FullTextScan.Plan plan) {
+ return read(plan.splits(), plan.snapshot());
+ }
+
+ private GlobalIndexResult read(
+ List<FullTextSearchSplit> splits, @Nullable Snapshot planSnapshot)
{
if (splits.isEmpty()) {
return GlobalIndexResult.createEmpty();
}
@@ -102,13 +113,19 @@ public class DataEvolutionFullTextRead implements
FullTextRead {
}
GlobalIndexFileReader indexFileReader = m ->
table.fileIO().newInputStream(m.filePath());
- RoaringNavigableMap64 liveRows =
GlobalIndexLiveRowFilter.liveRows(table, partitionFilter);
+ RoaringNavigableMap64 liveRows =
+ GlobalIndexLiveRowFilter.liveRows(table, planSnapshot,
partitionFilter, null);
ScoredGlobalIndexResult result =
evalQuery(splitsByColumn, indexPathFactory, indexFileReader,
executor, liveRows);
if (!rawRowRanges.isEmpty()) {
result =
new RawFullTextReadImpl(
- table, partitionFilter, limit, textColumn,
this::evalQuery)
+ table,
+ planSnapshot,
+ partitionFilter,
+ limit,
+ textColumn,
+ this::evalQuery)
.withRawSearch(result, rawRowRanges,
splitsByColumn, executor);
}
return result.topK(limit);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextScan.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextScan.java
index 3dc1f36812..fbd1bd83d1 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextScan.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextScan.java
@@ -34,6 +34,8 @@ import org.apache.paimon.types.DataField;
import org.apache.paimon.utils.Filter;
import org.apache.paimon.utils.Range;
+import javax.annotation.Nullable;
+
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -53,19 +55,33 @@ import static
org.apache.paimon.utils.Preconditions.checkNotNull;
public class DataEvolutionFullTextScan implements FullTextScan {
private final FileStoreTable table;
- private final PartitionPredicate partitionFilter;
+ @Nullable private final PartitionPredicate partitionFilter;
private final List<DataField> textColumns;
+ @Nullable private final Snapshot pinnedSnapshot;
public DataEvolutionFullTextScan(
- FileStoreTable table, PartitionPredicate partitionFilter,
DataField textColumn) {
+ FileStoreTable table,
+ @Nullable PartitionPredicate partitionFilter,
+ DataField textColumn) {
this(table, partitionFilter, Collections.singletonList(textColumn));
}
public DataEvolutionFullTextScan(
- FileStoreTable table, PartitionPredicate partitionFilter,
List<DataField> textColumns) {
+ FileStoreTable table,
+ @Nullable PartitionPredicate partitionFilter,
+ List<DataField> textColumns) {
+ this(table, partitionFilter, textColumns, null);
+ }
+
+ public DataEvolutionFullTextScan(
+ FileStoreTable table,
+ @Nullable PartitionPredicate partitionFilter,
+ List<DataField> textColumns,
+ @Nullable Snapshot pinnedSnapshot) {
this.table = table;
this.partitionFilter = partitionFilter;
this.textColumns = textColumns;
+ this.pinnedSnapshot = pinnedSnapshot;
}
@Override
@@ -82,7 +98,9 @@ public class DataEvolutionFullTextScan implements
FullTextScan {
idToColumn.put(textColumn.id(), textColumn.name());
}
- Snapshot snapshot = TimeTravelUtil.tryTravelOrLatest(table);
+ @Nullable
+ Snapshot snapshot =
+ pinnedSnapshot != null ? pinnedSnapshot :
TimeTravelUtil.tryTravelOrLatest(table);
IndexFileHandler indexFileHandler =
table.store().newIndexFileHandler();
Filter<IndexManifestEntry> indexFileFilter =
entry -> {
@@ -128,7 +146,19 @@ public class DataEvolutionFullTextScan implements
FullTextScan {
}
}
- return () -> splits;
+ @Nullable Snapshot planSnapshot = snapshot;
+ return new Plan() {
+ @Override
+ public List<FullTextSearchSplit> splits() {
+ return splits;
+ }
+
+ @Override
+ @Nullable
+ public Snapshot snapshot() {
+ return planSnapshot;
+ }
+ };
}
/**
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextScan.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextScan.java
index 563ffe1c9d..28e76f9fee 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextScan.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/FullTextScan.java
@@ -18,6 +18,10 @@
package org.apache.paimon.table.source;
+import org.apache.paimon.Snapshot;
+
+import javax.annotation.Nullable;
+
import java.util.List;
/** Full-text scan to scan index files. */
@@ -28,5 +32,11 @@ public interface FullTextScan {
/** Plan of full-text scan. */
interface Plan {
List<FullTextSearchSplit> splits();
+
+ /** Snapshot the plan was built against; the read pins live-row
filtering to it. */
+ @Nullable
+ default Snapshot snapshot() {
+ return null;
+ }
}
}
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 873e73b2f4..dccd3d7da2 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
@@ -78,7 +78,10 @@ public class FullTextSearchBuilderImpl implements
FullTextSearchBuilder {
? new PrimaryKeyFullTextScan(
table, definition.get(), partitionFilter,
pinnedSnapshot)
: new DataEvolutionFullTextScan(
- table, partitionFilter,
Collections.singletonList(textColumn));
+ table,
+ partitionFilter,
+ Collections.singletonList(textColumn),
+ pinnedSnapshot);
}
@Override
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
index ca142430c3..86c6d2a1d2 100644
---
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
@@ -18,6 +18,8 @@
package org.apache.paimon.table.source;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.PositionOutputStream;
@@ -64,6 +66,7 @@ import static
org.apache.paimon.utils.Preconditions.checkNotNull;
class RawFullTextReadImpl {
private final FileStoreTable table;
+ @Nullable private final Snapshot planSnapshot;
@Nullable private final PartitionPredicate partitionFilter;
private final int limit;
private final DataField textColumn;
@@ -71,11 +74,13 @@ class RawFullTextReadImpl {
RawFullTextReadImpl(
FileStoreTable table,
+ @Nullable Snapshot planSnapshot,
@Nullable PartitionPredicate partitionFilter,
int limit,
DataField textColumn,
IndexSearch indexSearch) {
this.table = table;
+ this.planSnapshot = planSnapshot;
this.partitionFilter = partitionFilter;
this.limit = limit;
this.textColumn = textColumn;
@@ -243,13 +248,30 @@ class RawFullTextReadImpl {
}
private ReadBuilder rawReadBuilder(RowType readType) {
- ReadBuilder readBuilder =
table.newReadBuilder().withReadType(readType);
+ ReadBuilder readBuilder =
rawReadTable().newReadBuilder().withReadType(readType);
if (partitionFilter != null) {
readBuilder.withPartitionFilter(partitionFilter);
}
return readBuilder;
}
+ private FileStoreTable rawReadTable() {
+ if (planSnapshot == null) {
+ return table;
+ }
+
+ Map<String, String> pinOptions = new HashMap<>();
+ pinOptions.put(
+ CoreOptions.SCAN_MODE.key(),
CoreOptions.StartupMode.FROM_SNAPSHOT.toString());
+ pinOptions.put(CoreOptions.SCAN_SNAPSHOT_ID.key(),
String.valueOf(planSnapshot.id()));
+ pinOptions.put(CoreOptions.SCAN_VERSION.key(), null);
+ pinOptions.put(CoreOptions.SCAN_TAG_NAME.key(), null);
+ pinOptions.put(CoreOptions.SCAN_WATERMARK.key(), null);
+ pinOptions.put(CoreOptions.SCAN_TIMESTAMP.key(), null);
+ pinOptions.put(CoreOptions.SCAN_TIMESTAMP_MILLIS.key(), null);
+ return table.copyWithoutTimeTravel(pinOptions);
+ }
+
private Options rawSearchOptions() {
Options options = new
Options(table.coreOptions().toConfiguration().toMap());
options.setString("full-text.searcher-pool.max-size", "0");
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 16c418dc64..0cc025e040 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/source/FullTextSearchBuilderTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/source/FullTextSearchBuilderTest.java
@@ -19,6 +19,10 @@
package org.apache.paimon.table.source;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.append.dataevolution.DataEvolutionCompactCoordinator;
+import org.apache.paimon.append.dataevolution.DataEvolutionCompactTask;
+import
org.apache.paimon.append.dataevolution.DataEvolutionCompactionCommitPreparation;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryString;
@@ -65,7 +69,9 @@ import java.io.ObjectOutputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import static
org.apache.paimon.table.source.DeletionVectorTestUtils.commitDeletionVectors;
import static org.assertj.core.api.Assertions.assertThat;
@@ -162,6 +168,101 @@ public class FullTextSearchBuilderTest extends
TableTestBase {
assertThat(readIds(table, result)).containsExactlyInAnyOrder(2, 3);
}
+ @Test
+ public void testFullTextSearchPinsLiveRowFilterToPlanSnapshot() throws
Exception {
+ Identifier identifier = identifier("full_text_pinned_live_rows");
+ Schema schema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column(TEXT_FIELD_NAME, DataTypes.STRING())
+ .option(CoreOptions.BUCKET.key(), "-1")
+ .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(),
"true")
+ .option(CoreOptions.DELETION_VECTORS_ENABLED.key(),
"true")
+ .build();
+ catalog.createTable(identifier, schema, false);
+ FileStoreTable table = getTable(identifier);
+
+ String[] documents = {
+ "paimon keyword", "paimon keyword", "paimon keyword", "paimon
keyword"
+ };
+ writeDocuments(table, documents);
+ buildAndCommitIndex(table, documents);
+
+ FullTextSearchBuilder builder =
+ table.newFullTextSearchBuilder()
+ .withQuery(TEXT_FIELD_NAME, matchQuery("keyword"))
+ .withLimit(4);
+ FullTextScan.Plan plan = builder.newFullTextScan().scan();
+ assertThat(plan.snapshot()).isNotNull();
+
+ // Row 0 was live when the index plan was created, so a later DV must
not affect this read.
+ commitDeletionVectors(table, 0L);
+
+ GlobalIndexResult result = builder.newFullTextRead().read(plan);
+ assertThat(result.results()).containsExactlyInAnyOrder(0L, 1L, 2L, 3L);
+ }
+
+ @Test
+ public void testFullTextRawFallbackPinsDataReadToPlanSnapshot() throws
Exception {
+ Identifier identifier = identifier("full_text_pinned_raw_fallback");
+ Schema schema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column(TEXT_FIELD_NAME, DataTypes.STRING())
+ .option(CoreOptions.BUCKET.key(), "-1")
+ .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true")
+ .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(),
"true")
+ .option(CoreOptions.DELETION_VECTORS_ENABLED.key(),
"true")
+ .option(CoreOptions.FULL_TEXT_INDEX_SEARCH_MODE.key(),
"full")
+ .build();
+ catalog.createTable(identifier, schema, false);
+ FileStoreTable table = getTable(identifier);
+
+ String[] documents = {"indexed keyword", "raw keyword"};
+ writeDocuments(table, documents);
+ buildAndCommitIndexRange(
+ table,
+ new String[] {documents[0]},
+
Collections.singletonList(table.rowType().getField(TEXT_FIELD_NAME)),
+ 0);
+
+ FullTextSearchBuilder builder =
+ table.newFullTextSearchBuilder()
+ .withQuery(TEXT_FIELD_NAME, matchQuery("keyword"))
+ .withLimit(2);
+ FullTextScan.Plan plan = builder.newFullTextScan().scan();
+
assertThat(plan.splits()).anyMatch(RawFullTextSearchSplit.class::isInstance);
+
+ // Materialization rewrites the surviving row ids after planning. Both
the indexed and raw
+ // sides must still be evaluated against the pre-compaction snapshot
carried by the plan.
+ commitDeletionVectors(table, 0L);
+ Map<String, String> compactOptions = new HashMap<>();
+
compactOptions.put(CoreOptions.DATA_EVOLUTION_COMPACTION_REWRITE_ROW_IDS.key(),
"true");
+ FileStoreTable compactTable = table.copy(compactOptions);
+ Snapshot compactSnapshot = compactTable.latestSnapshot().get();
+ DataEvolutionCompactCoordinator coordinator =
+ new DataEvolutionCompactCoordinator(compactTable, false,
false, compactSnapshot);
+ List<DataEvolutionCompactTask> tasks = coordinator.plan();
+ assertThat(tasks)
+ .singleElement()
+ .extracting(DataEvolutionCompactTask::type)
+
.isEqualTo(DataEvolutionCompactTask.TaskType.MATERIALIZE_DELETION);
+ List<CommitMessage> messages = new ArrayList<>();
+ for (DataEvolutionCompactTask task : tasks) {
+ messages.add(task.doCompact(compactTable,
"test-full-text-snapshot-pin"));
+ }
+ messages.addAll(
+ new DataEvolutionCompactionCommitPreparation(compactTable,
compactSnapshot)
+ .prepare(messages));
+ try (BatchTableCommit commit =
compactTable.newBatchWriteBuilder().newCommit()) {
+ commit.commit(messages);
+ }
+
+ GlobalIndexResult result = builder.newFullTextRead().read(plan);
+ assertThat(result.results()).containsExactlyInAnyOrder(0L, 1L);
+ }
+
@Test
public void testFullTextSearchNonFastModesScanUnindexedData() throws
Exception {
createTableDefault();
