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 803ed5c8ce [core] Maintain primary-key full-text index archives (#8651)
803ed5c8ce is described below
commit 803ed5c8ce0453c35c697110e03be15a6a5dd53a
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Jul 15 15:52:41 2026 +0800
[core] Maintain primary-key full-text index archives (#8651)
Build on #8649 by adding LSM-style incremental maintenance for
primary-key full-text index archives.
---
.../main/java/org/apache/paimon/CoreOptions.java | 57 +++
.../java/org/apache/paimon/KeyValueFileStore.java | 1 +
.../pk/BucketedPrimaryKeyIndexMaintainer.java | 148 +++++-
.../paimon/index/pk/PrimaryKeyIndexDefinition.java | 3 +-
.../index/pk/PrimaryKeyIndexDefinitions.java | 20 +-
.../BucketedFullTextIndexMaintainer.java | 557 +++++++++++++++++++++
.../org/apache/paimon/schema/SchemaManager.java | 3 +-
.../org/apache/paimon/schema/SchemaValidation.java | 60 ++-
.../pk/BucketedPrimaryKeyIndexMaintainerTest.java | 130 +++++
.../index/pk/PrimaryKeyIndexDefinitionsTest.java | 32 ++
.../BucketedFullTextIndexMaintainerTest.java | 343 +++++++++++++
.../paimon/operation/PrimaryKeyIndexWriteTest.java | 23 +
.../PrimaryKeyFullTextIndexValidationTest.java | 243 +++++++++
.../apache/paimon/schema/SchemaManagerTest.java | 36 ++
14 files changed, 1644 insertions(+), 12 deletions(-)
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index 2537fbb793..271e46346e 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -2755,6 +2755,14 @@ public class CoreOptions implements Serializable {
.withDescription(
"Comma-separated columns indexed by primary-key
Bitmap indexes.");
+ public static final ConfigOption<String> PK_FULL_TEXT_INDEX_COLUMNS =
+ key("pk-full-text.index.columns")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Comma-separated character columns indexed by
primary-key full-text indexes. "
+ + "The first release supports exactly one
column.");
+
@Immutable
public static final ConfigOption<Boolean> PK_CLUSTERING_OVERRIDE =
key("pk-clustering-override")
@@ -4287,6 +4295,10 @@ public class CoreOptions implements Serializable {
return options.getOptional(PK_VECTOR_INDEX_COLUMNS).isPresent();
}
+ public boolean primaryKeyFullTextIndexEnabled() {
+ return options.getOptional(PK_FULL_TEXT_INDEX_COLUMNS).isPresent();
+ }
+
public int primaryKeyIndexCompactionLevelFanout(String column) {
return
options.getInteger(primaryKeyIndexCompactionLevelFanoutKey(column), 5);
}
@@ -4315,6 +4327,10 @@ public class CoreOptions implements Serializable {
return primaryKeyIndexColumns(PK_BITMAP_INDEX_COLUMNS);
}
+ public List<String> primaryKeyFullTextIndexColumns() {
+ return primaryKeyIndexColumns(PK_FULL_TEXT_INDEX_COLUMNS);
+ }
+
private List<String> primaryKeyIndexColumns(ConfigOption<String> option) {
String columns = options.get(option);
if (columns == null) {
@@ -4331,6 +4347,47 @@ public class CoreOptions implements Serializable {
return primaryKeySortedIndexOptions(column, "pk-bitmap",
"bitmap-index.");
}
+ public Options primaryKeyFullTextIndexOptions(String column) {
+ String optionKey = "fields." + column + ".pk-full-text.index.options";
+ TreeMap<String, String> resolved = new TreeMap<>();
+ for (Map.Entry<String, String> entry :
toConfiguration().toMap().entrySet()) {
+ if (entry.getKey().startsWith("full-text.")) {
+ resolved.put(entry.getKey(), entry.getValue());
+ }
+ }
+
+ String serialized = options.get(optionKey);
+ if (serialized == null || serialized.trim().isEmpty()) {
+ return new Options(resolved);
+ }
+
+ LinkedHashMap<String, String> parsed;
+ try {
+ parsed = JsonSerdeUtil.parseJsonMap(serialized, String.class);
+ } catch (RuntimeException e) {
+ throw new IllegalArgumentException(
+ optionKey + " must be a JSON object of option key-value
pairs.", e);
+ }
+
+ for (Map.Entry<String, String> entry : parsed.entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+ checkArgument(
+ key != null && !key.trim().isEmpty(),
+ "%s contains an empty option key.",
+ optionKey);
+ checkArgument(value != null, "%s value for key %s must not be
null.", optionKey, key);
+ String qualifiedKey = key.startsWith("full-text.") ? key :
"full-text." + key;
+ String previous = resolved.put(qualifiedKey, value);
+ checkArgument(
+ previous == null || previous.equals(value),
+ "%s defines conflicting values for %s.",
+ optionKey,
+ qualifiedKey);
+ }
+ return new Options(resolved);
+ }
+
private Options primaryKeySortedIndexOptions(
String column, String optionFamily, String algorithmPrefix) {
Options resolved = new Options(toConfiguration().toMap());
diff --git a/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
b/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
index 7531f91a77..b3c06c98c1 100644
--- a/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
+++ b/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
@@ -172,6 +172,7 @@ public class KeyValueFileStore extends
AbstractFileStore<KeyValue> {
}
BucketedPrimaryKeyIndexMaintainer.Factory
primaryKeyIndexMaintainerFactory = null;
if (options.primaryKeyVectorIndexEnabled()
+ || options.primaryKeyFullTextIndexEnabled()
|| !options.primaryKeyBTreeIndexColumns().isEmpty()
|| !options.primaryKeyBitmapIndexColumns().isEmpty()) {
primaryKeyIndexMaintainerFactory =
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainer.java
b/paimon-core/src/main/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainer.java
index 938c3ed10e..dd3f244bb7 100644
---
a/paimon-core/src/main/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainer.java
+++
b/paimon-core/src/main/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainer.java
@@ -23,6 +23,10 @@ import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.index.IndexFileHandler;
import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.pkfulltext.BucketedFullTextIndexMaintainer;
+import org.apache.paimon.index.pkfulltext.PkFullTextDataFileReader;
+import org.apache.paimon.index.pkfulltext.PkFullTextIndexBuilder;
+import org.apache.paimon.index.pkfulltext.PkFullTextIndexFile;
import org.apache.paimon.index.pksorted.BucketedSortedIndexMaintainer;
import org.apache.paimon.index.pksorted.PkSortedDataFileReader;
import org.apache.paimon.index.pksorted.PkSortedIndexBuilder;
@@ -50,13 +54,16 @@ import static
org.apache.paimon.utils.Preconditions.checkArgument;
public final class BucketedPrimaryKeyIndexMaintainer {
@Nullable private final BucketedVectorIndexMaintainer vectorMaintainer;
+ @Nullable private final BucketedFullTextIndexMaintainer fullTextMaintainer;
private final List<BucketedSortedIndexMaintainer> sortedMaintainers;
private int nextSortedMaintainerIndex;
private BucketedPrimaryKeyIndexMaintainer(
@Nullable BucketedVectorIndexMaintainer vectorMaintainer,
+ @Nullable BucketedFullTextIndexMaintainer fullTextMaintainer,
List<BucketedSortedIndexMaintainer> sortedMaintainers) {
this.vectorMaintainer = vectorMaintainer;
+ this.fullTextMaintainer = fullTextMaintainer;
List<BucketedSortedIndexMaintainer> sorted = new
ArrayList<>(sortedMaintainers);
sorted.sort(Comparator.comparingInt(BucketedSortedIndexMaintainer::fieldId));
this.sortedMaintainers = Collections.unmodifiableList(sorted);
@@ -64,18 +71,33 @@ public final class BucketedPrimaryKeyIndexMaintainer {
public static BucketedPrimaryKeyIndexMaintainer ofVector(
BucketedVectorIndexMaintainer vectorMaintainer) {
- return new BucketedPrimaryKeyIndexMaintainer(vectorMaintainer,
Collections.emptyList());
+ return new BucketedPrimaryKeyIndexMaintainer(
+ vectorMaintainer, null, Collections.emptyList());
+ }
+
+ public static BucketedPrimaryKeyIndexMaintainer ofFullText(
+ BucketedFullTextIndexMaintainer fullTextMaintainer) {
+ return new BucketedPrimaryKeyIndexMaintainer(
+ null, fullTextMaintainer, Collections.emptyList());
}
public static BucketedPrimaryKeyIndexMaintainer of(
BucketedVectorIndexMaintainer vectorMaintainer,
List<BucketedSortedIndexMaintainer> sortedMaintainers) {
- return new BucketedPrimaryKeyIndexMaintainer(vectorMaintainer,
sortedMaintainers);
+ return new BucketedPrimaryKeyIndexMaintainer(vectorMaintainer, null,
sortedMaintainers);
+ }
+
+ public static BucketedPrimaryKeyIndexMaintainer of(
+ @Nullable BucketedVectorIndexMaintainer vectorMaintainer,
+ @Nullable BucketedFullTextIndexMaintainer fullTextMaintainer,
+ List<BucketedSortedIndexMaintainer> sortedMaintainers) {
+ return new BucketedPrimaryKeyIndexMaintainer(
+ vectorMaintainer, fullTextMaintainer, sortedMaintainers);
}
public static BucketedPrimaryKeyIndexMaintainer ofSorted(
List<BucketedSortedIndexMaintainer> sortedMaintainers) {
- return new BucketedPrimaryKeyIndexMaintainer(null, sortedMaintainers);
+ return new BucketedPrimaryKeyIndexMaintainer(null, null,
sortedMaintainers);
}
public synchronized void prepareCommit(
@@ -84,6 +106,7 @@ public final class BucketedPrimaryKeyIndexMaintainer {
boolean waitCompaction)
throws Exception {
BucketedVectorIndexMaintainer.VectorIndexCommit vectorCommit = null;
+ BucketedFullTextIndexMaintainer.FullTextIndexCommit fullTextCommit =
null;
List<BucketedSortedIndexMaintainer.SortedIndexCommit> sortedCommits =
new ArrayList<>();
try {
if (vectorMaintainer != null) {
@@ -91,6 +114,11 @@ public final class BucketedPrimaryKeyIndexMaintainer {
vectorMaintainer.prepareCommit(
appendIncrement, compactIncrement,
waitCompaction);
}
+ if (fullTextMaintainer != null) {
+ fullTextCommit =
+ fullTextMaintainer.prepareCommit(
+ appendIncrement, compactIncrement,
waitCompaction);
+ }
if (waitCompaction) {
prepareSortedBlocking(appendIncrement, compactIncrement,
sortedCommits);
@@ -101,6 +129,9 @@ public final class BucketedPrimaryKeyIndexMaintainer {
if (vectorCommit != null) {
mergeVectorCommit(appendIncrement, compactIncrement,
vectorCommit);
}
+ if (fullTextCommit != null) {
+ mergeFullTextCommit(appendIncrement, compactIncrement,
fullTextCommit);
+ }
for (BucketedSortedIndexMaintainer.SortedIndexCommit sortedCommit
: sortedCommits) {
mergeSortedCommit(appendIncrement, compactIncrement,
sortedCommit);
}
@@ -108,6 +139,9 @@ public final class BucketedPrimaryKeyIndexMaintainer {
for (int i = sortedCommits.size() - 1; i >= 0; i--) {
sortedCommits.get(i).abort(failure);
}
+ if (fullTextCommit != null) {
+ fullTextCommit.abort(failure);
+ }
if (vectorCommit != null) {
vectorCommit.abort(failure);
}
@@ -219,6 +253,28 @@ public final class BucketedPrimaryKeyIndexMaintainer {
increment.deletedIndexFiles()));
}
+ private static void mergeFullTextCommit(
+ DataIncrement appendIncrement,
+ CompactIncrement compactIncrement,
+ BucketedFullTextIndexMaintainer.FullTextIndexCommit commit) {
+ commit.appendIncrement()
+ .ifPresent(
+ increment ->
+ applyIndexIncrement(
+ appendIncrement.newIndexFiles(),
+ appendIncrement.deletedIndexFiles(),
+ increment.newIndexFiles(),
+ increment.deletedIndexFiles()));
+ commit.compactIncrement()
+ .ifPresent(
+ increment ->
+ applyIndexIncrement(
+ compactIncrement.newIndexFiles(),
+ compactIncrement.deletedIndexFiles(),
+ increment.newIndexFiles(),
+ increment.deletedIndexFiles()));
+ }
+
private static void applyIndexIncrement(
List<IndexFileMeta> targetNew,
List<IndexFileMeta> targetDeleted,
@@ -232,6 +288,9 @@ public final class BucketedPrimaryKeyIndexMaintainer {
if (vectorMaintainer != null && vectorMaintainer.buildNotCompleted()) {
return true;
}
+ if (fullTextMaintainer != null &&
fullTextMaintainer.buildNotCompleted()) {
+ return true;
+ }
return activeSortedMaintainer() != null;
}
@@ -239,6 +298,9 @@ public final class BucketedPrimaryKeyIndexMaintainer {
if (vectorMaintainer != null) {
vectorMaintainer.withExecutor(executor);
}
+ if (fullTextMaintainer != null) {
+ fullTextMaintainer.withExecutor(executor);
+ }
for (BucketedSortedIndexMaintainer maintainer : sortedMaintainers) {
maintainer.withExecutor(executor);
}
@@ -248,6 +310,9 @@ public final class BucketedPrimaryKeyIndexMaintainer {
if (vectorMaintainer != null) {
vectorMaintainer.close();
}
+ if (fullTextMaintainer != null) {
+ fullTextMaintainer.close();
+ }
for (BucketedSortedIndexMaintainer maintainer : sortedMaintainers) {
maintainer.close();
}
@@ -258,20 +323,23 @@ public final class BucketedPrimaryKeyIndexMaintainer {
private final IndexFileHandler handler;
@Nullable private final BucketedVectorIndexMaintainer.Factory
vectorFactory;
+ @Nullable private final FullTextDefinitionFactory fullTextFactory;
private final List<SortedDefinitionFactory> sortedFactories;
private Factory(
IndexFileHandler handler,
@Nullable BucketedVectorIndexMaintainer.Factory vectorFactory,
+ @Nullable FullTextDefinitionFactory fullTextFactory,
List<SortedDefinitionFactory> sortedFactories) {
this.handler = handler;
this.vectorFactory = vectorFactory;
+ this.fullTextFactory = fullTextFactory;
this.sortedFactories = Collections.unmodifiableList(new
ArrayList<>(sortedFactories));
}
public static Factory ofVector(BucketedVectorIndexMaintainer.Factory
vectorFactory) {
return new Factory(
- vectorFactory.indexFileHandler(), vectorFactory,
Collections.emptyList());
+ vectorFactory.indexFileHandler(), vectorFactory, null,
Collections.emptyList());
}
public static Factory create(
@@ -281,6 +349,7 @@ public final class BucketedPrimaryKeyIndexMaintainer {
CoreOptions coreOptions = new CoreOptions(schema.options());
Map<String, DataField> fields = schema.nameToFieldMap();
BucketedVectorIndexMaintainer.Factory vectorFactory = null;
+ FullTextDefinitionFactory fullTextFactory = null;
List<SortedDefinitionFactory> sortedFactories = new ArrayList<>();
for (PrimaryKeyIndexDefinition definition :
PrimaryKeyIndexDefinitions.create(schema).definitions()) {
@@ -315,15 +384,27 @@ public final class BucketedPrimaryKeyIndexMaintainer {
definition.compactionLevelFanout(),
definition.compactionStaleRatioThreshold()));
break;
+ case FULL_TEXT:
+ checkArgument(
+ fullTextFactory == null,
+ "Only one primary-key full-text index is
supported.");
+ fullTextFactory =
+ new FullTextDefinitionFactory(
+ readerFactoryBuilder,
+ field,
+ definition.options(),
+ definition.compactionLevelFanout(),
+
definition.compactionStaleRatioThreshold());
+ break;
default:
throw new IllegalArgumentException(
"Unsupported primary-key index family " +
definition.family());
}
}
checkArgument(
- vectorFactory != null || !sortedFactories.isEmpty(),
+ vectorFactory != null || fullTextFactory != null ||
!sortedFactories.isEmpty(),
"No primary-key index definition is configured.");
- return new Factory(handler, vectorFactory, sortedFactories);
+ return new Factory(handler, vectorFactory, fullTextFactory,
sortedFactories);
}
public IndexFileHandler indexFileHandler() {
@@ -355,6 +436,11 @@ public final class BucketedPrimaryKeyIndexMaintainer {
? null
: vectorFactory.create(
partition, bucket, dataFiles, payloads,
executor);
+ BucketedFullTextIndexMaintainer fullText =
+ fullTextFactory == null
+ ? null
+ : fullTextFactory.create(
+ handler, partition, bucket, dataFiles,
payloads, executor);
List<BucketedSortedIndexMaintainer> sorted = new ArrayList<>();
for (SortedDefinitionFactory factory : sortedFactories) {
sorted.add(
@@ -362,7 +448,55 @@ public final class BucketedPrimaryKeyIndexMaintainer {
handler, partition, bucket, dataFiles,
payloads, executor,
ioManager));
}
- return new BucketedPrimaryKeyIndexMaintainer(vector, sorted);
+ return new BucketedPrimaryKeyIndexMaintainer(vector, fullText,
sorted);
+ }
+
+ private static final class FullTextDefinitionFactory {
+
+ private final KeyValueFileReaderFactory.Builder
readerFactoryBuilder;
+ private final DataField field;
+ private final org.apache.paimon.options.Options options;
+ private final int compactionLevelFanout;
+ private final double compactionStaleRatioThreshold;
+
+ private FullTextDefinitionFactory(
+ KeyValueFileReaderFactory.Builder readerFactoryBuilder,
+ DataField field,
+ org.apache.paimon.options.Options options,
+ int compactionLevelFanout,
+ double compactionStaleRatioThreshold) {
+ this.readerFactoryBuilder = readerFactoryBuilder;
+ this.field = field;
+ this.options = options;
+ this.compactionLevelFanout = compactionLevelFanout;
+ this.compactionStaleRatioThreshold =
compactionStaleRatioThreshold;
+ }
+
+ private BucketedFullTextIndexMaintainer create(
+ IndexFileHandler handler,
+ BinaryRow partition,
+ int bucket,
+ List<DataFileMeta> restoredDataFiles,
+ List<IndexFileMeta> restoredPayloads,
+ ExecutorService executor) {
+ PkFullTextIndexFile indexFile =
handler.pkFullTextIndex(partition, bucket);
+ PkFullTextIndexBuilder builder =
+ new PkFullTextIndexBuilder(
+ indexFile,
+ new PkFullTextDataFileReader.Factory(
+ readerFactoryBuilder, partition,
bucket, field),
+ field,
+ options);
+ return new BucketedFullTextIndexMaintainer(
+ field.id(),
+ indexFile,
+ builder,
+ compactionLevelFanout,
+ compactionStaleRatioThreshold,
+ restoredDataFiles,
+ restoredPayloads,
+ executor);
+ }
}
private static final class SortedDefinitionFactory {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinition.java
b/paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinition.java
index 266f529051..f5d47f8b60 100644
---
a/paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinition.java
+++
b/paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinition.java
@@ -27,7 +27,8 @@ public class PrimaryKeyIndexDefinition {
public enum Family {
VECTOR,
BTREE,
- BITMAP
+ BITMAP,
+ FULL_TEXT
}
private final String column;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinitions.java
b/paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinitions.java
index 9d56db1e67..0ac0353e04 100644
---
a/paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinitions.java
+++
b/paimon-core/src/main/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinitions.java
@@ -46,10 +46,12 @@ public class PrimaryKeyIndexDefinitions {
List<String> vectorColumns = options.primaryKeyVectorIndexColumns();
List<String> btreeColumns = options.primaryKeyBTreeIndexColumns();
List<String> bitmapColumns = options.primaryKeyBitmapIndexColumns();
+ List<String> fullTextColumns =
options.primaryKeyFullTextIndexColumns();
validateNoDuplicates(vectorColumns,
CoreOptions.PK_VECTOR_INDEX_COLUMNS.key());
validateNoDuplicates(btreeColumns,
CoreOptions.PK_BTREE_INDEX_COLUMNS.key());
validateNoDuplicates(bitmapColumns,
CoreOptions.PK_BITMAP_INDEX_COLUMNS.key());
- validateOneIndexPerColumn(vectorColumns, btreeColumns, bitmapColumns);
+ validateNoDuplicates(fullTextColumns,
CoreOptions.PK_FULL_TEXT_INDEX_COLUMNS.key());
+ validateOneIndexPerColumn(vectorColumns, btreeColumns, bitmapColumns,
fullTextColumns);
List<PrimaryKeyIndexDefinition> definitions = new ArrayList<>();
for (DataField field : schema.fields()) {
@@ -84,6 +86,16 @@ public class PrimaryKeyIndexDefinitions {
PrimaryKeyIndexDefinition.Family.VECTOR,
options.primaryKeyIndexCompactionLevelFanout(column),
options.primaryKeyIndexCompactionStaleRatioThreshold(column)));
+ } else if (fullTextColumns.contains(column)) {
+ definitions.add(
+ new PrimaryKeyIndexDefinition(
+ column,
+ field.id(),
+ "full-text",
+ options.primaryKeyFullTextIndexOptions(column),
+ PrimaryKeyIndexDefinition.Family.FULL_TEXT,
+
options.primaryKeyIndexCompactionLevelFanout(column),
+
options.primaryKeyIndexCompactionStaleRatioThreshold(column)));
}
}
@@ -102,11 +114,15 @@ public class PrimaryKeyIndexDefinitions {
}
private static void validateOneIndexPerColumn(
- List<String> vectorColumns, List<String> btreeColumns,
List<String> bitmapColumns) {
+ List<String> vectorColumns,
+ List<String> btreeColumns,
+ List<String> bitmapColumns,
+ List<String> fullTextColumns) {
Set<String> indexedColumns = new HashSet<>();
validateUniqueColumns(indexedColumns, vectorColumns);
validateUniqueColumns(indexedColumns, btreeColumns);
validateUniqueColumns(indexedColumns, bitmapColumns);
+ validateUniqueColumns(indexedColumns, fullTextColumns);
}
private static void validateUniqueColumns(Set<String> indexedColumns,
List<String> columns) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/pkfulltext/BucketedFullTextIndexMaintainer.java
b/paimon-core/src/main/java/org/apache/paimon/index/pkfulltext/BucketedFullTextIndexMaintainer.java
new file mode 100644
index 0000000000..9231c78979
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/index/pkfulltext/BucketedFullTextIndexMaintainer.java
@@ -0,0 +1,557 @@
+/*
+ * 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.index.IndexFileMeta;
+import org.apache.paimon.index.pk.PrimaryKeyIndexLevels;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceFile;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourceMeta;
+import org.apache.paimon.index.pk.PrimaryKeyIndexSourcePolicy;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataIncrement;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** Maintains bucket-local full-text archives with source-backed LSM
consolidation. */
+public class BucketedFullTextIndexMaintainer {
+
+ private static final int DEFAULT_LEVEL_FANOUT = 5;
+ private static final double DEFAULT_STALE_RATIO_THRESHOLD = 0.2;
+
+ private final int textFieldId;
+ private final PkFullTextIndexFile indexFile;
+ private final PkFullTextIndexBuilder indexBuilder;
+ private final PrimaryKeyIndexLevels<IndexFileMeta> levels;
+ private final List<IndexFileMeta> currentPayloads = new ArrayList<>();
+ private final List<IndexFileMeta> retiredPayloads = new ArrayList<>();
+ private final Map<String, DataFileMeta> activeSourceFiles = new
LinkedHashMap<>();
+ private ExecutorService executor;
+ @Nullable private PendingBuild pendingBuild;
+
+ public BucketedFullTextIndexMaintainer(
+ int textFieldId,
+ PkFullTextIndexFile indexFile,
+ PkFullTextIndexBuilder indexBuilder,
+ List<DataFileMeta> restoredDataFiles,
+ List<IndexFileMeta> restoredPayloads,
+ ExecutorService executor) {
+ this(
+ textFieldId,
+ indexFile,
+ indexBuilder,
+ DEFAULT_LEVEL_FANOUT,
+ DEFAULT_STALE_RATIO_THRESHOLD,
+ restoredDataFiles,
+ restoredPayloads,
+ executor);
+ }
+
+ public BucketedFullTextIndexMaintainer(
+ int textFieldId,
+ PkFullTextIndexFile indexFile,
+ PkFullTextIndexBuilder indexBuilder,
+ int levelFanout,
+ double staleRatioThreshold,
+ List<DataFileMeta> restoredDataFiles,
+ List<IndexFileMeta> restoredPayloads,
+ ExecutorService executor) {
+ this.textFieldId = textFieldId;
+ this.indexFile = indexFile;
+ this.indexBuilder = indexBuilder;
+ this.levels =
+ new PrimaryKeyIndexLevels<>(
+ levelFanout,
+ staleRatioThreshold,
+ IndexFileMeta::fileName,
+ payload -> sourceMeta(payload).sourceFiles());
+ this.executor = executor;
+ for (DataFileMeta file : restoredDataFiles) {
+ if (PrimaryKeyIndexSourcePolicy.shouldRead(file)) {
+ activeSourceFiles.put(file.fileName(), file);
+ }
+ }
+
+ PkFullTextBucketIndexState restoredState =
+ PkFullTextBucketIndexState.fromActivePayloads(textFieldId,
restoredPayloads);
+ currentPayloads.addAll(restoredState.currentPayloads());
+ retiredPayloads.addAll(restoredState.stalePayloads());
+ validateActiveSourceRows();
+ }
+
+ public synchronized FullTextIndexCommit prepareCommit(
+ DataIncrement appendIncrement,
+ CompactIncrement compactIncrement,
+ boolean waitCompaction)
+ throws Exception {
+ checkArgument(
+ eligibleFiles(appendIncrement.newFiles()).isEmpty(),
+ "Append files must not be primary-key full-text index
sources.");
+
+ List<IndexFileMeta> originalPayloads = new
ArrayList<>(currentPayloads);
+ List<IndexFileMeta> originalRetired = new ArrayList<>(retiredPayloads);
+ Map<String, DataFileMeta> originalSources = new
LinkedHashMap<>(activeSourceFiles);
+ List<IndexFileMeta> generated = new ArrayList<>();
+ try {
+ applyDataTransition(compactIncrement);
+ List<IndexFileMeta> created = new ArrayList<>();
+ List<IndexFileMeta> removed = new ArrayList<>(retiredPayloads);
+ retiredPayloads.clear();
+
+ while (true) {
+ Optional<CompletedBuild> completed =
finishPendingBuild(waitCompaction);
+ if (completed.isPresent()) {
+ CompletedBuild build = completed.get();
+ generated.add(build.payload);
+ if (canAccept(build)) {
+ replacePayloads(
+ build.inputPayloads, build.payload, created,
removed, generated);
+ } else {
+ deleteGenerated(build.payload, generated);
+ }
+ }
+
+ if (pendingBuild == null) {
+ List<DataFileMeta> uncovered = uncoveredFiles();
+ if (!uncovered.isEmpty()) {
+ startBuild(uncovered, Collections.emptyList());
+ } else {
+ Optional<PrimaryKeyIndexLevels.Plan<IndexFileMeta>>
plan =
+ levels.pick(currentPayloads,
activeSourceFiles);
+ if (plan.isPresent()) {
+ if (plan.get().sourceFiles().isEmpty()) {
+ removePayloads(
+ plan.get().inputUnits(), created,
removed, generated);
+ continue;
+ }
+ startBuild(plan.get().sourceFiles(),
plan.get().inputUnits());
+ }
+ }
+ }
+ if (!waitCompaction || pendingBuild == null) {
+ break;
+ }
+ }
+
+ boolean hasCompactDataTransition =
+ !compactIncrement.compactBefore().isEmpty()
+ || !compactIncrement.compactAfter().isEmpty();
+ boolean changed = !created.isEmpty() || !removed.isEmpty();
+ Optional<FullTextIndexIncrement> appendChange =
+ changed && !hasCompactDataTransition
+ ? Optional.of(new FullTextIndexIncrement(created,
removed))
+ : Optional.empty();
+ Optional<FullTextIndexIncrement> compactChange =
+ changed && hasCompactDataTransition
+ ? Optional.of(new FullTextIndexIncrement(created,
removed))
+ : Optional.empty();
+ return new FullTextIndexCommit(
+ appendChange,
+ compactChange,
+ failure ->
+ rollback(
+ originalPayloads,
+ originalRetired,
+ originalSources,
+ generated,
+ failure));
+ } catch (Throwable failure) {
+ rollback(originalPayloads, originalRetired, originalSources,
generated, failure);
+ if (failure instanceof Exception) {
+ throw (Exception) failure;
+ }
+ if (failure instanceof Error) {
+ throw (Error) failure;
+ }
+ throw new RuntimeException(failure);
+ }
+ }
+
+ private void applyDataTransition(CompactIncrement compactIncrement) {
+ for (DataFileMeta file : compactIncrement.compactBefore()) {
+ if (!containsFile(compactIncrement.compactAfter(),
file.fileName())) {
+ activeSourceFiles.remove(file.fileName());
+ }
+ }
+ for (DataFileMeta file : compactIncrement.compactAfter()) {
+ if (PrimaryKeyIndexSourcePolicy.shouldRead(file)) {
+ activeSourceFiles.put(file.fileName(), file);
+ }
+ }
+ }
+
+ private List<DataFileMeta> uncoveredFiles() {
+ Set<String> covered = coveredSources(currentPayloads);
+ List<DataFileMeta> uncovered = new ArrayList<>();
+ for (DataFileMeta source : activeSourceFiles.values()) {
+ if (!covered.contains(source.fileName())) {
+ uncovered.add(source);
+ }
+ }
+ uncovered.sort(Comparator.comparing(DataFileMeta::fileName));
+ return uncovered;
+ }
+
+ private void startBuild(List<DataFileMeta> sourceFiles,
List<IndexFileMeta> inputPayloads) {
+ PendingBuild build = new PendingBuild(sourceFiles, inputPayloads);
+ build.start();
+ pendingBuild = build;
+ }
+
+ private Optional<CompletedBuild> finishPendingBuild(boolean blocking)
throws Exception {
+ if (pendingBuild == null || (!blocking && !pendingBuild.isDone())) {
+ return Optional.empty();
+ }
+ PendingBuild completed = pendingBuild;
+ try {
+ IndexFileMeta payload = completed.get();
+ pendingBuild = null;
+ return Optional.of(
+ new CompletedBuild(completed.sourceFiles,
completed.inputPayloads, payload));
+ } catch (CancellationException e) {
+ pendingBuild = null;
+ return Optional.empty();
+ } catch (ExecutionException e) {
+ pendingBuild = null;
+ Throwable cause = e.getCause();
+ if (cause instanceof Exception) {
+ throw (Exception) cause;
+ }
+ if (cause instanceof Error) {
+ throw (Error) cause;
+ }
+ throw new RuntimeException(cause);
+ }
+ }
+
+ private boolean canAccept(CompletedBuild build) {
+ if (!currentPayloads.containsAll(build.inputPayloads)) {
+ return false;
+ }
+ PkFullTextBucketIndexState outputState =
+ PkFullTextBucketIndexState.fromActivePayloads(
+ textFieldId, Collections.singletonList(build.payload));
+ if (outputState.currentPayloads().size() != 1) {
+ return false;
+ }
+ List<PrimaryKeyIndexSourceFile> actualSources =
sourceMeta(build.payload).sourceFiles();
+ if (actualSources.size() != build.sourceFiles.size()) {
+ return false;
+ }
+
+ List<IndexFileMeta> retained = new ArrayList<>(currentPayloads);
+ retained.removeAll(build.inputPayloads);
+ Set<String> retainedSources = coveredSources(retained);
+ for (int i = 0; i < actualSources.size(); i++) {
+ PrimaryKeyIndexSourceFile actual = actualSources.get(i);
+ DataFileMeta expected = build.sourceFiles.get(i);
+ DataFileMeta active = activeSourceFiles.get(actual.fileName());
+ if (!actual.fileName().equals(expected.fileName())
+ || actual.rowCount() != expected.rowCount()
+ || active == null
+ || active.rowCount() != actual.rowCount()
+ || retainedSources.contains(actual.fileName())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private void replacePayloads(
+ List<IndexFileMeta> inputs,
+ IndexFileMeta output,
+ List<IndexFileMeta> created,
+ List<IndexFileMeta> removed,
+ List<IndexFileMeta> generated) {
+ removePayloads(inputs, created, removed, generated);
+ currentPayloads.add(output);
+ created.add(output);
+ }
+
+ private void removePayloads(
+ List<IndexFileMeta> inputs,
+ List<IndexFileMeta> created,
+ List<IndexFileMeta> removed,
+ List<IndexFileMeta> generated) {
+ for (IndexFileMeta input : inputs) {
+ checkArgument(
+ currentPayloads.remove(input), "Full-text rebuild input is
no longer active.");
+ if (created.remove(input)) {
+ deleteGenerated(input, generated);
+ } else {
+ removed.add(input);
+ }
+ }
+ }
+
+ private void deleteGenerated(IndexFileMeta payload, List<IndexFileMeta>
generated) {
+ indexFile.delete(payload);
+ generated.remove(payload);
+ }
+
+ private synchronized void rollback(
+ List<IndexFileMeta> originalPayloads,
+ List<IndexFileMeta> originalRetired,
+ Map<String, DataFileMeta> originalSources,
+ List<IndexFileMeta> generated,
+ Throwable failure) {
+ currentPayloads.clear();
+ currentPayloads.addAll(originalPayloads);
+ retiredPayloads.clear();
+ retiredPayloads.addAll(originalRetired);
+ activeSourceFiles.clear();
+ activeSourceFiles.putAll(originalSources);
+
+ PendingBuild build = pendingBuild;
+ pendingBuild = null;
+ if (build != null) {
+ try {
+ build.cancel();
+ } catch (Throwable cleanupFailure) {
+ failure.addSuppressed(cleanupFailure);
+ }
+ }
+ for (IndexFileMeta payload : generated) {
+ try {
+ indexFile.delete(payload);
+ } catch (Throwable cleanupFailure) {
+ failure.addSuppressed(cleanupFailure);
+ }
+ }
+ }
+
+ public synchronized boolean buildNotCompleted() {
+ return pendingBuild != null;
+ }
+
+ public synchronized void withExecutor(ExecutorService executor) {
+ checkArgument(pendingBuild == null, "Cannot replace executor during a
full-text build.");
+ this.executor = executor;
+ }
+
+ public synchronized void close() {
+ PendingBuild build = pendingBuild;
+ pendingBuild = null;
+ if (build != null) {
+ build.cancel();
+ }
+ }
+
+ public synchronized PkFullTextBucketIndexState state() {
+ return PkFullTextBucketIndexState.fromActivePayloads(textFieldId,
currentPayloads);
+ }
+
+ public synchronized List<IndexFileMeta> payloads() {
+ return Collections.unmodifiableList(new ArrayList<>(currentPayloads));
+ }
+
+ private void validateActiveSourceRows() {
+ for (IndexFileMeta payload : currentPayloads) {
+ for (PrimaryKeyIndexSourceFile source :
sourceMeta(payload).sourceFiles()) {
+ DataFileMeta active = activeSourceFiles.get(source.fileName());
+ if (active != null) {
+ checkArgument(
+ active.rowCount() == source.rowCount(),
+ "Full-text source %s row count does not match its
active data file.",
+ source.fileName());
+ }
+ }
+ }
+ }
+
+ private class PendingBuild {
+
+ private final List<DataFileMeta> sourceFiles;
+ private final List<IndexFileMeta> inputPayloads;
+ @Nullable private IndexFileMeta result;
+ @Nullable private Future<IndexFileMeta> future;
+ private boolean cancelled;
+
+ private PendingBuild(List<DataFileMeta> sourceFiles,
List<IndexFileMeta> inputPayloads) {
+ this.sourceFiles = Collections.unmodifiableList(new
ArrayList<>(sourceFiles));
+ this.inputPayloads = Collections.unmodifiableList(new
ArrayList<>(inputPayloads));
+ }
+
+ private void start() {
+ future =
+ executor.submit(
+ () -> {
+ IndexFileMeta payload =
+ sourceFiles.size() == 1
+ ?
indexBuilder.build(sourceFiles.get(0))
+ :
indexBuilder.build(sourceFiles);
+ synchronized (PendingBuild.this) {
+ if (!cancelled) {
+ result = payload;
+ return payload;
+ }
+ }
+ indexFile.delete(payload);
+ throw new CancellationException();
+ });
+ }
+
+ private boolean isDone() {
+ return future.isDone();
+ }
+
+ private IndexFileMeta get() throws InterruptedException,
ExecutionException {
+ return future.get();
+ }
+
+ private void cancel() {
+ Future<IndexFileMeta> buildFuture;
+ IndexFileMeta builtPayload;
+ synchronized (this) {
+ cancelled = true;
+ buildFuture = future;
+ builtPayload = result;
+ result = null;
+ }
+ if (buildFuture != null) {
+ buildFuture.cancel(true);
+ }
+ if (builtPayload != null) {
+ indexFile.delete(builtPayload);
+ }
+ }
+ }
+
+ private static final class CompletedBuild {
+
+ private final List<DataFileMeta> sourceFiles;
+ private final List<IndexFileMeta> inputPayloads;
+ private final IndexFileMeta payload;
+
+ private CompletedBuild(
+ List<DataFileMeta> sourceFiles,
+ List<IndexFileMeta> inputPayloads,
+ IndexFileMeta payload) {
+ this.sourceFiles = sourceFiles;
+ this.inputPayloads = inputPayloads;
+ this.payload = payload;
+ }
+ }
+
+ private static Set<String> coveredSources(List<IndexFileMeta> payloads) {
+ Set<String> covered = new HashSet<>();
+ for (IndexFileMeta payload : payloads) {
+ for (PrimaryKeyIndexSourceFile source :
sourceMeta(payload).sourceFiles()) {
+ covered.add(source.fileName());
+ }
+ }
+ return covered;
+ }
+
+ private static List<DataFileMeta> eligibleFiles(List<DataFileMeta> files) {
+ List<DataFileMeta> eligible = new ArrayList<>();
+ for (DataFileMeta file : files) {
+ if (PrimaryKeyIndexSourcePolicy.shouldRead(file)) {
+ eligible.add(file);
+ }
+ }
+ return eligible;
+ }
+
+ private static boolean containsFile(List<DataFileMeta> files, String
fileName) {
+ for (DataFileMeta file : files) {
+ if (file.fileName().equals(fileName)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static PrimaryKeyIndexSourceMeta sourceMeta(IndexFileMeta payload)
{
+ return PrimaryKeyIndexSourceMeta.fromIndexFile(payload);
+ }
+
+ /** Full-text changes prepared for Paimon's append snapshot or compact
snapshot. */
+ public static class FullTextIndexCommit {
+
+ private final Optional<FullTextIndexIncrement> appendIncrement;
+ private final Optional<FullTextIndexIncrement> compactIncrement;
+ private final AbortAction abortAction;
+
+ private FullTextIndexCommit(
+ Optional<FullTextIndexIncrement> appendIncrement,
+ Optional<FullTextIndexIncrement> compactIncrement,
+ AbortAction abortAction) {
+ this.appendIncrement = appendIncrement;
+ this.compactIncrement = compactIncrement;
+ this.abortAction = abortAction;
+ }
+
+ public Optional<FullTextIndexIncrement> appendIncrement() {
+ return appendIncrement;
+ }
+
+ public Optional<FullTextIndexIncrement> compactIncrement() {
+ return compactIncrement;
+ }
+
+ public void abort(Throwable failure) {
+ abortAction.abort(failure);
+ }
+ }
+
+ @FunctionalInterface
+ private interface AbortAction {
+
+ void abort(Throwable failure);
+ }
+
+ /** Index-file additions and deletions emitted by one bucket state update.
*/
+ public static class FullTextIndexIncrement {
+
+ private final List<IndexFileMeta> newIndexFiles;
+ private final List<IndexFileMeta> deletedIndexFiles;
+
+ private FullTextIndexIncrement(
+ List<IndexFileMeta> newIndexFiles, List<IndexFileMeta>
deletedIndexFiles) {
+ this.newIndexFiles = Collections.unmodifiableList(new
ArrayList<>(newIndexFiles));
+ this.deletedIndexFiles =
+ Collections.unmodifiableList(new
ArrayList<>(deletedIndexFiles));
+ }
+
+ public List<IndexFileMeta> newIndexFiles() {
+ return newIndexFiles;
+ }
+
+ public List<IndexFileMeta> deletedIndexFiles() {
+ return deletedIndexFiles;
+ }
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java
b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java
index b8e27fb956..c9c1a6b5f5 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java
@@ -997,7 +997,8 @@ public class SchemaManager implements Serializable {
CoreOptions options = new CoreOptions(schema.options());
if (options.primaryKeyVectorIndexColumns().contains(fieldName)
|| options.primaryKeyBTreeIndexColumns().contains(fieldName)
- || options.primaryKeyBitmapIndexColumns().contains(fieldName))
{
+ || options.primaryKeyBitmapIndexColumns().contains(fieldName)
+ ||
options.primaryKeyFullTextIndexColumns().contains(fieldName)) {
throw new UnsupportedOperationException(
String.format(
"Cannot %s primary-key index column: [%s]",
operation, fieldName));
diff --git
a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
index fb0eb2e6f2..9f74310736 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
@@ -358,6 +358,7 @@ public class SchemaValidation {
validatePrimaryKeyIndexColumns(options);
validatePrimaryKeySortedIndexes(schema, options);
validatePrimaryKeyVectorIndex(schema, options);
+ validatePrimaryKeyFullTextIndex(schema, options);
validateMergeFunctionFactory(schema);
@@ -957,22 +958,79 @@ public class SchemaValidation {
options.primaryKeyVectorDistanceMetric(indexColumn));
}
+ private static void validatePrimaryKeyFullTextIndex(TableSchema schema,
CoreOptions options) {
+ if (!options.primaryKeyFullTextIndexEnabled()) {
+ return;
+ }
+
+ List<String> indexColumns = options.primaryKeyFullTextIndexColumns();
+ checkArgument(
+ indexColumns.size() == 1,
+ "%s must contain exactly one column in the first release, but
is %s.",
+ CoreOptions.PK_FULL_TEXT_INDEX_COLUMNS.key(),
+ indexColumns);
+ String indexColumn = indexColumns.get(0);
+ checkArgument(
+ !StringUtils.isNullOrWhitespaceOnly(indexColumn),
+ "%s must contain a non-empty column.",
+ CoreOptions.PK_FULL_TEXT_INDEX_COLUMNS.key());
+ checkArgument(
+ !schema.primaryKeys().isEmpty(),
+ "Primary-key full-text index requires a primary-key table.");
+ checkArgument(
+ options.mergeEngine() == MergeEngine.FIRST_ROW ||
options.deletionVectorsEnabled(),
+ "Primary-key full-text index requires deletion-vectors.enabled
= true.");
+ checkArgument(
+ !options.deletionVectorsMergeOnRead(),
+ "Primary-key full-text index requires
deletion-vectors.merge-on-read = false.");
+ checkArgument(
+ options.bucket() > 0 || options.bucket() ==
BucketMode.POSTPONE_BUCKET,
+ "Primary-key full-text index requires fixed or postpone bucket
mode "
+ + "(bucket > 0 or bucket = -2), but bucket is %s.",
+ options.bucket());
+ checkArgument(
+ !options.pkClusteringOverride(),
+ "Primary-key full-text index does not support
pk-clustering-override.");
+ checkArgument(
+ schema.nameToFieldMap().containsKey(indexColumn),
+ "%s entry '%s' must reference an existing column.",
+ CoreOptions.PK_FULL_TEXT_INDEX_COLUMNS.key(),
+ indexColumn);
+ DataTypeRoot typeRoot =
schema.nameToFieldMap().get(indexColumn).type().getTypeRoot();
+ checkArgument(
+ typeRoot == DataTypeRoot.CHAR || typeRoot ==
DataTypeRoot.VARCHAR,
+ "%s entry '%s' must reference a CHAR/VARCHAR/STRING column.",
+ CoreOptions.PK_FULL_TEXT_INDEX_COLUMNS.key(),
+ indexColumn);
+ options.primaryKeyFullTextIndexOptions(indexColumn);
+ }
+
private static void validatePrimaryKeyIndexColumns(CoreOptions options) {
List<String> vectorColumns = options.primaryKeyVectorIndexColumns();
List<String> btreeColumns = options.primaryKeyBTreeIndexColumns();
List<String> bitmapColumns = options.primaryKeyBitmapIndexColumns();
+ List<String> fullTextColumns =
options.primaryKeyFullTextIndexColumns();
validateNoDuplicatePrimaryKeyIndexColumns(
vectorColumns, CoreOptions.PK_VECTOR_INDEX_COLUMNS.key());
validateNoDuplicatePrimaryKeyIndexColumns(
btreeColumns, CoreOptions.PK_BTREE_INDEX_COLUMNS.key());
validateNoDuplicatePrimaryKeyIndexColumns(
bitmapColumns, CoreOptions.PK_BITMAP_INDEX_COLUMNS.key());
+ validateNoDuplicatePrimaryKeyIndexColumns(
+ fullTextColumns, CoreOptions.PK_FULL_TEXT_INDEX_COLUMNS.key());
Set<String> indexedColumns = new HashSet<>();
validateUniquePrimaryKeyIndexColumns(indexedColumns, vectorColumns);
validateUniquePrimaryKeyIndexColumns(indexedColumns, btreeColumns);
validateUniquePrimaryKeyIndexColumns(indexedColumns, bitmapColumns);
- for (String column : indexedColumns) {
+ validateUniquePrimaryKeyIndexColumns(indexedColumns, fullTextColumns);
+
+ Set<String> compactedIndexColumns = new HashSet<>();
+ compactedIndexColumns.addAll(vectorColumns);
+ compactedIndexColumns.addAll(btreeColumns);
+ compactedIndexColumns.addAll(bitmapColumns);
+ compactedIndexColumns.addAll(fullTextColumns);
+ for (String column : compactedIndexColumns) {
String fanoutKey =
CoreOptions.primaryKeyIndexCompactionLevelFanoutKey(column);
checkArgument(
options.primaryKeyIndexCompactionLevelFanout(column) > 1,
diff --git
a/paimon-core/src/test/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainerTest.java
b/paimon-core/src/test/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainerTest.java
index b5d7e8d6ea..cb0955237a 100644
---
a/paimon-core/src/test/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainerTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainerTest.java
@@ -18,19 +18,28 @@
package org.apache.paimon.index.pk;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.index.IndexFileHandler;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.index.IndexPathFactory;
+import org.apache.paimon.index.pkfulltext.BucketedFullTextIndexMaintainer;
+import org.apache.paimon.index.pkfulltext.PkFullTextIndexFile;
import org.apache.paimon.index.pksorted.BucketedSortedIndexMaintainer;
import org.apache.paimon.index.pksorted.PkSortedIndexFile;
import org.apache.paimon.index.pkvector.BucketedVectorIndexMaintainer;
import org.apache.paimon.io.CompactIncrement;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.io.KeyValueFileReaderFactory;
import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.stats.SimpleStats;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
@@ -39,7 +48,9 @@ import org.junit.jupiter.api.io.TempDir;
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 java.util.Optional;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
@@ -51,6 +62,7 @@ import java.util.concurrent.atomic.AtomicReference;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
+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;
@@ -84,6 +96,104 @@ class BucketedPrimaryKeyIndexMaintainerTest {
verify(vector).close();
}
+ @Test
+ void testDelegatesFullTextLifecycleAndMergesCommit() throws Exception {
+ IndexFileMeta payload =
+ new IndexFileMeta("full-text", "payload", 1, 1,
(GlobalIndexMeta) null, null);
+ BucketedFullTextIndexMaintainer fullText =
mock(BucketedFullTextIndexMaintainer.class);
+ BucketedFullTextIndexMaintainer.FullTextIndexCommit commit =
+
mock(BucketedFullTextIndexMaintainer.FullTextIndexCommit.class);
+ BucketedFullTextIndexMaintainer.FullTextIndexIncrement increment =
+
mock(BucketedFullTextIndexMaintainer.FullTextIndexIncrement.class);
+ when(fullText.prepareCommit(any(), any(),
eq(true))).thenReturn(commit);
+ when(commit.appendIncrement()).thenReturn(Optional.empty());
+ when(commit.compactIncrement()).thenReturn(Optional.of(increment));
+
when(increment.newIndexFiles()).thenReturn(Collections.singletonList(payload));
+
when(increment.deletedIndexFiles()).thenReturn(Collections.emptyList());
+ when(fullText.buildNotCompleted()).thenReturn(true);
+ BucketedPrimaryKeyIndexMaintainer maintainer =
+ BucketedPrimaryKeyIndexMaintainer.ofFullText(fullText);
+ CompactIncrement compactIncrement = CompactIncrement.emptyIncrement();
+
+ maintainer.prepareCommit(DataIncrement.emptyIncrement(),
compactIncrement, true);
+
+ assertThat(compactIncrement.newIndexFiles()).containsExactly(payload);
+ assertThat(maintainer.buildNotCompleted()).isTrue();
+ maintainer.withExecutor(buildExecutor);
+ maintainer.close();
+ verify(fullText).withExecutor(buildExecutor);
+ verify(fullText).close();
+ }
+
+ @Test
+ void testFactoryCreatesConfiguredFullTextMaintainer() {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.PK_FULL_TEXT_INDEX_COLUMNS.key(), "content");
+ options.put("fields.content.pk-index.compaction.level-fanout", "2");
+
options.put("fields.content.pk-index.compaction.stale-ratio-threshold", "1.0");
+ TableSchema schema =
+ new TableSchema(
+ 0,
+ Arrays.asList(
+ new DataField(0, "id",
DataTypes.INT().notNull()),
+ new DataField(1, "content",
DataTypes.STRING())),
+ 0,
+ Collections.emptyList(),
+ Collections.singletonList("id"),
+ options,
+ "");
+ IndexFileHandler handler = mock(IndexFileHandler.class);
+ when(handler.pkFullTextIndex(any(),
eq(0))).thenReturn(mock(PkFullTextIndexFile.class));
+ KeyValueFileReaderFactory.Builder readerBuilder =
+ mock(KeyValueFileReaderFactory.Builder.class, RETURNS_SELF);
+ DataFileMeta first = dataFile("data-1", 1);
+ DataFileMeta second = dataFile("data-2", 1);
+ BucketedPrimaryKeyIndexMaintainer.Factory factory =
+ BucketedPrimaryKeyIndexMaintainer.Factory.create(handler,
readerBuilder, schema);
+ BucketedPrimaryKeyIndexMaintainer maintainer =
+ factory.create(
+ BinaryRow.EMPTY_ROW,
+ 0,
+ Arrays.asList(first, second),
+ Arrays.asList(
+ fullTextPayload("payload-1", first),
+ fullTextPayload("payload-2", second)),
+ buildExecutor);
+
+ assertThat(factory.indexFileHandler()).isSameAs(handler);
+ assertThat(maintainer.buildNotCompleted()).isFalse();
+ }
+
+ @Test
+ void testLaterDefinitionFailureAbortsPreparedFullTextCommit() throws
Exception {
+ DataFileMeta source = dataFile("data-1", 3);
+ BucketedFullTextIndexMaintainer fullText =
mock(BucketedFullTextIndexMaintainer.class);
+ BucketedFullTextIndexMaintainer.FullTextIndexCommit fullTextCommit =
+
mock(BucketedFullTextIndexMaintainer.FullTextIndexCommit.class);
+ when(fullText.prepareCommit(any(), any(),
eq(true))).thenReturn(fullTextCommit);
+ when(fullTextCommit.appendIncrement()).thenReturn(Optional.empty());
+ when(fullTextCommit.compactIncrement()).thenReturn(Optional.empty());
+ BucketedSortedIndexMaintainer failing =
+ sortedMaintainer(
+ 8,
+ "bitmap",
+ source,
+ dataFile -> {
+ throw new IllegalStateException("expected sorted
failure");
+ });
+ BucketedPrimaryKeyIndexMaintainer maintainer =
+ BucketedPrimaryKeyIndexMaintainer.of(
+ null, fullText, Collections.singletonList(failing));
+
+ assertThatThrownBy(
+ () ->
+ maintainer.prepareCommit(
+ DataIncrement.emptyIncrement(),
compactAfter(source), true))
+ .hasMessage("expected sorted failure");
+
+ verify(fullTextCommit).abort(any());
+ }
+
@Test
void testScalarFailureAbortsOtherDefinitionsAndVector() throws Exception {
DataFileMeta source = dataFile("data-1", 3);
@@ -369,6 +479,26 @@ class BucketedPrimaryKeyIndexMaintainerTest {
null);
}
+ private static IndexFileMeta fullTextPayload(String fileName, DataFileMeta
sourceFile) {
+ PrimaryKeyIndexSourceMeta sourceMeta =
+ new PrimaryKeyIndexSourceMeta(
+ new PrimaryKeyIndexSourceFile(
+ sourceFile.fileName(), sourceFile.rowCount()));
+ return new IndexFileMeta(
+ "full-text",
+ fileName,
+ 1,
+ sourceFile.rowCount(),
+ new GlobalIndexMeta(
+ 0,
+ sourceFile.rowCount() - 1,
+ 1,
+ null,
+ new byte[] {1},
+ sourceMeta.serialize()),
+ null);
+ }
+
private static DataFileMeta dataFile(String fileName, long rowCount) {
return DataFileMeta.forAppend(
fileName,
diff --git
a/paimon-core/src/test/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinitionsTest.java
b/paimon-core/src/test/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinitionsTest.java
index ab8ca6d2bf..378c1ad274 100644
---
a/paimon-core/src/test/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinitionsTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/index/pk/PrimaryKeyIndexDefinitionsTest.java
@@ -85,6 +85,38 @@ class PrimaryKeyIndexDefinitionsTest {
assertThat(definition.compactionStaleRatioThreshold()).isEqualTo(0.4);
}
+ @Test
+ void testCreatesFullTextDefinitionWithoutIndexType() {
+ Map<String, String> options = new HashMap<>();
+ options.put("pk-full-text.index.columns", "name");
+
+ List<PrimaryKeyIndexDefinition> definitions =
+
PrimaryKeyIndexDefinitions.create(schema(options)).definitions();
+
+ assertThat(definitions).hasSize(1);
+ assertThat(definitions.get(0).column()).isEqualTo("name");
+ assertThat(definitions.get(0).indexType()).isEqualTo("full-text");
+ assertThat(definitions.get(0).family().name()).isEqualTo("FULL_TEXT");
+ }
+
+ @Test
+ void testResolvesFullTextIndexOptions() {
+ Map<String, String> options = new HashMap<>();
+ options.put("pk-full-text.index.columns", "name");
+ options.put(
+ "fields.name.pk-full-text.index.options",
+
"{\"full-text.tokenizer\":\"jieba\",\"ngram.min-gram\":\"2\"}");
+
+ PrimaryKeyIndexDefinition definition =
+
PrimaryKeyIndexDefinitions.create(schema(options)).definitions().get(0);
+
+
assertThat(definition.options().get("full-text.tokenizer")).isEqualTo("jieba");
+
assertThat(definition.options().get("full-text.ngram.min-gram")).isEqualTo("2");
+ assertThat(definition.options().toMap())
+ .doesNotContainKey("pk-full-text.index.columns")
+ .doesNotContainKey("fields.name.pk-full-text.index.options");
+ }
+
@Test
void testLegacyVectorCompactionOptionsAreIgnored() {
Map<String, String> options = new HashMap<>();
diff --git
a/paimon-core/src/test/java/org/apache/paimon/index/pkfulltext/BucketedFullTextIndexMaintainerTest.java
b/paimon-core/src/test/java/org/apache/paimon/index/pkfulltext/BucketedFullTextIndexMaintainerTest.java
new file mode 100644
index 0000000000..0036da2d87
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/index/pkfulltext/BucketedFullTextIndexMaintainerTest.java
@@ -0,0 +1,343 @@
+/*
+ * 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.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.DataFileMeta;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.stats.SimpleStats;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+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.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/** Tests file-aligned maintenance of primary-key full-text archives. */
+class BucketedFullTextIndexMaintainerTest {
+
+ private final ExecutorService executor =
Executors.newSingleThreadExecutor();
+
+ @AfterEach
+ void shutdownExecutor() {
+ executor.shutdownNow();
+ }
+
+ @Test
+ void testReplacesArchiveWithCompactSourceWithoutMergingArchives() throws
Exception {
+ DataFileMeta oldData = dataFile("old-data");
+ DataFileMeta newData = dataFile("new-data");
+ IndexFileMeta oldPayload = payload("old-payload", oldData);
+ IndexFileMeta newPayload = payload("new-payload", newData);
+ PkFullTextIndexBuilder builder = mock(PkFullTextIndexBuilder.class);
+ when(builder.build(newData)).thenReturn(newPayload);
+ BucketedFullTextIndexMaintainer maintainer =
+ new BucketedFullTextIndexMaintainer(
+ 7,
+ mock(PkFullTextIndexFile.class),
+ builder,
+ Collections.singletonList(oldData),
+ Collections.singletonList(oldPayload),
+ executor);
+
+ BucketedFullTextIndexMaintainer.FullTextIndexCommit commit =
+ maintainer.prepareCommit(
+ DataIncrement.emptyIncrement(),
+ new CompactIncrement(
+ Collections.singletonList(oldData),
+ Collections.singletonList(newData),
+ Collections.emptyList()),
+ true);
+
+ assertThat(commit.compactIncrement()).isPresent();
+
assertThat(commit.compactIncrement().get().newIndexFiles()).containsExactly(newPayload);
+
assertThat(commit.compactIncrement().get().deletedIndexFiles()).containsExactly(oldPayload);
+
assertThat(maintainer.state().payloadBySourceFile()).containsOnlyKeys("new-data");
+ verify(builder).build(newData);
+ }
+
+ @Test
+ void testMergesArchivesAtConfiguredFanout() throws Exception {
+ DataFileMeta first = dataFile("data-1");
+ DataFileMeta second = dataFile("data-2");
+ IndexFileMeta firstPayload = payload("payload-1", first);
+ IndexFileMeta secondPayload = payload("payload-2", second);
+ IndexFileMeta merged = payload("merged", Arrays.asList(first, second));
+ PkFullTextIndexBuilder builder = mock(PkFullTextIndexBuilder.class);
+ when(builder.build(Arrays.asList(first, second))).thenReturn(merged);
+ BucketedFullTextIndexMaintainer maintainer =
+ new BucketedFullTextIndexMaintainer(
+ 7,
+ mock(PkFullTextIndexFile.class),
+ builder,
+ 2,
+ 0.5,
+ Arrays.asList(first, second),
+ Arrays.asList(firstPayload, secondPayload),
+ executor);
+
+ BucketedFullTextIndexMaintainer.FullTextIndexCommit commit =
+ maintainer.prepareCommit(
+ DataIncrement.emptyIncrement(),
CompactIncrement.emptyIncrement(), true);
+
+ assertThat(commit.appendIncrement()).isPresent();
+
assertThat(commit.appendIncrement().get().newIndexFiles()).containsExactly(merged);
+ assertThat(commit.appendIncrement().get().deletedIndexFiles())
+ .containsExactly(firstPayload, secondPayload);
+
assertThat(maintainer.state().payloadBySourceFile()).containsOnlyKeys("data-1",
"data-2");
+ assertThat(maintainer.payloads()).containsExactly(merged);
+ verify(builder).build(Arrays.asList(first, second));
+ }
+
+ @Test
+ void testRebuildsArchiveAtConfiguredStaleRatio() throws Exception {
+ DataFileMeta active = dataFile("active");
+ DataFileMeta removed = dataFile("removed");
+ IndexFileMeta oldPayload = payload("old-payload",
Arrays.asList(active, removed));
+ IndexFileMeta rebuilt = payload("rebuilt", active);
+ PkFullTextIndexBuilder builder = mock(PkFullTextIndexBuilder.class);
+ when(builder.build(active)).thenReturn(rebuilt);
+ BucketedFullTextIndexMaintainer maintainer =
+ new BucketedFullTextIndexMaintainer(
+ 7,
+ mock(PkFullTextIndexFile.class),
+ builder,
+ 5,
+ 0.5,
+ Collections.singletonList(active),
+ Collections.singletonList(oldPayload),
+ executor);
+
+ BucketedFullTextIndexMaintainer.FullTextIndexCommit commit =
+ maintainer.prepareCommit(
+ DataIncrement.emptyIncrement(),
CompactIncrement.emptyIncrement(), true);
+
+ assertThat(commit.appendIncrement()).isPresent();
+
assertThat(commit.appendIncrement().get().newIndexFiles()).containsExactly(rebuilt);
+
assertThat(commit.appendIncrement().get().deletedIndexFiles()).containsExactly(oldPayload);
+ assertThat(maintainer.payloads()).containsExactly(rebuilt);
+ verify(builder).build(active);
+ }
+
+ @Test
+ void testRetriesBuildAfterFailure() throws Exception {
+ DataFileMeta data = dataFile("data");
+ IndexFileMeta newPayload = payload("new-payload", data);
+ PkFullTextIndexBuilder builder = mock(PkFullTextIndexBuilder.class);
+ when(builder.build(data))
+ .thenThrow(new IOException("first failure"))
+ .thenReturn(newPayload);
+ BucketedFullTextIndexMaintainer maintainer =
+ new BucketedFullTextIndexMaintainer(
+ 7,
+ mock(PkFullTextIndexFile.class),
+ builder,
+ Collections.singletonList(data),
+ Collections.emptyList(),
+ executor);
+
+ assertThatThrownBy(
+ () ->
+ maintainer.prepareCommit(
+ DataIncrement.emptyIncrement(),
+ CompactIncrement.emptyIncrement(),
+ true))
+ .hasMessageContaining("first failure");
+
+ BucketedFullTextIndexMaintainer.FullTextIndexCommit commit =
+ maintainer.prepareCommit(
+ DataIncrement.emptyIncrement(),
CompactIncrement.emptyIncrement(), true);
+
+ assertThat(commit.appendIncrement()).isPresent();
+
assertThat(commit.appendIncrement().get().newIndexFiles()).containsExactly(newPayload);
+
assertThat(commit.appendIncrement().get().deletedIndexFiles()).isEmpty();
+ }
+
+ @Test
+ void testNonBlockingBuildPublishesOnALaterAppendIncrement() throws
Exception {
+ DataFileMeta data = dataFile("data");
+ IndexFileMeta payload = payload("payload", data);
+ CountDownLatch buildStarted = new CountDownLatch(1);
+ CountDownLatch allowBuild = new CountDownLatch(1);
+ PkFullTextIndexBuilder builder = mock(PkFullTextIndexBuilder.class);
+ when(builder.build(data))
+ .thenAnswer(
+ ignored -> {
+ buildStarted.countDown();
+ assertThat(allowBuild.await(30,
TimeUnit.SECONDS)).isTrue();
+ return payload;
+ });
+ BucketedFullTextIndexMaintainer maintainer =
+ new BucketedFullTextIndexMaintainer(
+ 7,
+ mock(PkFullTextIndexFile.class),
+ builder,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ executor);
+ try {
+ BucketedFullTextIndexMaintainer.FullTextIndexCommit first =
+ maintainer.prepareCommit(
+ DataIncrement.emptyIncrement(),
+ new CompactIncrement(
+ Collections.emptyList(),
+ Collections.singletonList(data),
+ Collections.emptyList()),
+ false);
+ assertThat(first.compactIncrement()).isEmpty();
+ assertThat(buildStarted.await(30, TimeUnit.SECONDS)).isTrue();
+ assertThat(maintainer.buildNotCompleted()).isTrue();
+ allowBuild.countDown();
+
+ BucketedFullTextIndexMaintainer.FullTextIndexCommit second =
+ maintainer.prepareCommit(
+ DataIncrement.emptyIncrement(),
+ CompactIncrement.emptyIncrement(),
+ true);
+ assertThat(second.appendIncrement()).isPresent();
+
assertThat(second.appendIncrement().get().newIndexFiles()).containsExactly(payload);
+ } finally {
+ allowBuild.countDown();
+ }
+ }
+
+ @Test
+ void testCoordinatorAbortRestoresStateAndDeletesGeneratedArchive() throws
Exception {
+ DataFileMeta data = dataFile("data");
+ IndexFileMeta payload = payload("payload", data);
+ PkFullTextIndexFile indexFile = mock(PkFullTextIndexFile.class);
+ PkFullTextIndexBuilder builder = mock(PkFullTextIndexBuilder.class);
+ when(builder.build(data)).thenReturn(payload);
+ BucketedFullTextIndexMaintainer maintainer =
+ new BucketedFullTextIndexMaintainer(
+ 7,
+ indexFile,
+ builder,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ executor);
+
+ BucketedFullTextIndexMaintainer.FullTextIndexCommit commit =
+ maintainer.prepareCommit(
+ DataIncrement.emptyIncrement(),
+ new CompactIncrement(
+ Collections.emptyList(),
+ Collections.singletonList(data),
+ Collections.emptyList()),
+ true);
+ assertThat(maintainer.payloads()).containsExactly(payload);
+
+ commit.abort(new IllegalStateException("later definition failed"));
+
+ assertThat(maintainer.payloads()).isEmpty();
+ verify(indexFile).delete(payload);
+ }
+
+ @Test
+ void testAbortDoesNotDeleteRejectedArchiveTwice() throws Exception {
+ DataFileMeta data = dataFile("data");
+ IndexFileMeta rejected = payload("rejected", dataFile("other"));
+ IndexFileMeta accepted = payload("accepted", data);
+ PkFullTextIndexFile indexFile = mock(PkFullTextIndexFile.class);
+ PkFullTextIndexBuilder builder = mock(PkFullTextIndexBuilder.class);
+ when(builder.build(data)).thenReturn(rejected, accepted);
+ BucketedFullTextIndexMaintainer maintainer =
+ new BucketedFullTextIndexMaintainer(
+ 7,
+ indexFile,
+ builder,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ executor);
+
+ BucketedFullTextIndexMaintainer.FullTextIndexCommit commit =
+ maintainer.prepareCommit(
+ DataIncrement.emptyIncrement(),
+ new CompactIncrement(
+ Collections.emptyList(),
+ Collections.singletonList(data),
+ Collections.emptyList()),
+ true);
+
+ verify(indexFile).delete(rejected);
+ commit.abort(new IllegalStateException("later definition failed"));
+
+ verify(indexFile, times(1)).delete(rejected);
+ verify(indexFile).delete(accepted);
+ }
+
+ private static DataFileMeta dataFile(String fileName) {
+ return DataFileMeta.forAppend(
+ fileName,
+ 100,
+ 1,
+ SimpleStats.EMPTY_STATS,
+ 0,
+ 1,
+ 1,
+ Collections.emptyList(),
+ null,
+ FileSource.COMPACT,
+ null,
+ null,
+ null,
+ null)
+ .upgrade(1);
+ }
+
+ private static IndexFileMeta payload(String payloadName, DataFileMeta
source) {
+ return payload(payloadName, Collections.singletonList(source));
+ }
+
+ private static IndexFileMeta payload(String payloadName,
List<DataFileMeta> sources) {
+ List<PrimaryKeyIndexSourceFile> sourceFiles = new ArrayList<>();
+ long rowCount = 0;
+ for (DataFileMeta source : sources) {
+ sourceFiles.add(new PrimaryKeyIndexSourceFile(source.fileName(),
source.rowCount()));
+ rowCount += source.rowCount();
+ }
+ PrimaryKeyIndexSourceMeta sourceMeta = new
PrimaryKeyIndexSourceMeta(sourceFiles);
+ return new IndexFileMeta(
+ "full-text",
+ payloadName,
+ 100,
+ rowCount,
+ new GlobalIndexMeta(0, rowCount - 1, 7, null, null,
sourceMeta.serialize()),
+ null);
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyIndexWriteTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyIndexWriteTest.java
index afd56c6207..ed16b21e01 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyIndexWriteTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyIndexWriteTest.java
@@ -128,6 +128,29 @@ class PrimaryKeyIndexWriteTest {
write.close();
}
+ @Test
+ void testCreatesCoordinatorForFullTextDefinition() throws Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.BUCKET.key(), "10");
+ options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true");
+ options.put(CoreOptions.PK_FULL_TEXT_INDEX_COLUMNS.key(), "comment");
+ TestFileStore store = createStore(options);
+ KeyValueFileStoreWrite write = (KeyValueFileStoreWrite)
store.newWrite();
+ write.withIOManager(ioManager);
+ TestKeyValueGenerator generator = new TestKeyValueGenerator();
+ KeyValue record = generator.next();
+
+ AbstractFileStoreWrite.WriterContainer<KeyValue> container =
+ write.createWriterContainer(generator.getPartition(record), 1);
+
+ assertThat(container.primaryKeyIndexMaintainer).isNotNull();
+ assertThat(readField(container.primaryKeyIndexMaintainer,
"fullTextMaintainer"))
+ .isNotNull();
+ assertThat((List<?>) readField(container.primaryKeyIndexMaintainer,
"sortedMaintainers"))
+ .isEmpty();
+ write.close();
+ }
+
@Test
void testPostponeBucketDefersCoordinatorUntilFixedBucketCompaction()
throws Exception {
Map<String, String> options = new HashMap<>();
diff --git
a/paimon-core/src/test/java/org/apache/paimon/schema/PrimaryKeyFullTextIndexValidationTest.java
b/paimon-core/src/test/java/org/apache/paimon/schema/PrimaryKeyFullTextIndexValidationTest.java
new file mode 100644
index 0000000000..f8a59512ac
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/schema/PrimaryKeyFullTextIndexValidationTest.java
@@ -0,0 +1,243 @@
+/*
+ * 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.schema;
+
+import org.apache.paimon.CoreOptions;
+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.apache.paimon.schema.SchemaValidation.validateTableSchema;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for primary-key full-text index option validation. */
+class PrimaryKeyFullTextIndexValidationTest {
+
+ @Test
+ void testValidConfiguration() {
+ assertThatCode(() -> validateTableSchema(schema(enabledOptions())))
+ .doesNotThrowAnyException();
+ }
+
+ @Test
+ void testRejectsMultipleColumnsForFirstRelease() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.PK_FULL_TEXT_INDEX_COLUMNS.key(),
"content,other_content");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+
.hasMessageContaining(CoreOptions.PK_FULL_TEXT_INDEX_COLUMNS.key())
+ .hasMessageContaining("exactly one column");
+ }
+
+ @Test
+ void testRejectsDuplicateColumns() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.PK_FULL_TEXT_INDEX_COLUMNS.key(),
"content,content");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+
.hasMessageContaining(CoreOptions.PK_FULL_TEXT_INDEX_COLUMNS.key())
+ .hasMessageContaining("duplicate");
+ }
+
+ @Test
+ void testRejectsEmptyColumn() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.PK_FULL_TEXT_INDEX_COLUMNS.key(), " ");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+
.hasMessageContaining(CoreOptions.PK_FULL_TEXT_INDEX_COLUMNS.key())
+ .hasMessageContaining("non-empty column");
+ }
+
+ @Test
+ void testRejectsUnknownColumn() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.PK_FULL_TEXT_INDEX_COLUMNS.key(), "unknown");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+ .hasMessageContaining("entry 'unknown'")
+ .hasMessageContaining("existing column");
+ }
+
+ @Test
+ void testRejectsNonCharacterColumn() {
+ TableSchema intContentSchema =
+ new TableSchema(
+ 0,
+ Arrays.asList(
+ new DataField(0, "id",
DataTypes.INT().notNull()),
+ new DataField(1, "content", DataTypes.INT())),
+ 0,
+ Collections.emptyList(),
+ Collections.singletonList("id"),
+ enabledOptions(),
+ "");
+
+ assertThatThrownBy(() -> validateTableSchema(intContentSchema))
+ .hasMessageContaining("entry 'content'")
+ .hasMessageContaining("CHAR/VARCHAR/STRING");
+ }
+
+ @Test
+ void testRequiresDeletionVectors() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "false");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+ .hasMessageContaining("requires deletion-vectors.enabled =
true");
+ }
+
+ @Test
+ void testSupportsFirstRowWithoutDeletionVectors() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.MERGE_ENGINE.key(), "first-row");
+ options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "false");
+
+ assertThatCode(() ->
validateTableSchema(schema(options))).doesNotThrowAnyException();
+ }
+
+ @Test
+ void testRejectsDeletionVectorMergeOnRead() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.DELETION_VECTORS_MERGE_ON_READ.key(), "true");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+ .hasMessageContaining("requires deletion-vectors.merge-on-read
= false");
+ }
+
+ @Test
+ void testRejectsDynamicBucket() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.BUCKET.key(), "-1");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+ .hasMessageContaining("requires fixed or postpone bucket
mode");
+ }
+
+ @Test
+ void testSupportsPostponeBucket() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.BUCKET.key(), "-2");
+
+ assertThatCode(() ->
validateTableSchema(schema(options))).doesNotThrowAnyException();
+ }
+
+ @Test
+ void testRejectsPkClusteringOverride() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.PK_CLUSTERING_OVERRIDE.key(), "true");
+ options.put(CoreOptions.CLUSTERING_COLUMNS.key(), "content");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+ .hasMessageContaining("does not support
pk-clustering-override");
+ }
+
+ @Test
+ void testRejectsColumnConfiguredForAnotherPrimaryKeyIndexFamily() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.PK_BTREE_INDEX_COLUMNS.key(), "content");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+ .hasMessageContaining("content")
+ .hasMessageContaining("at most one primary-key index");
+ }
+
+ @Test
+ void testRejectsInvalidLsmCompactionOptions() {
+ Map<String, String> options = enabledOptions();
+ options.put("fields.content.pk-index.compaction.level-fanout", "1");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+
.hasMessageContaining("fields.content.pk-index.compaction.level-fanout")
+ .hasMessageContaining("greater than 1");
+ }
+
+ @Test
+ void testRejectsMalformedIndexOptions() {
+ Map<String, String> options = enabledOptions();
+ options.put("fields.content.pk-full-text.index.options", "{not-json");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+ .hasMessageContaining(
+ "fields.content.pk-full-text.index.options must be a
JSON object");
+ }
+
+ @Test
+ void testRejectsConflictingGlobalAndFieldIndexOptions() {
+ Map<String, String> options = enabledOptions();
+ options.put("full-text.tokenizer", "standard");
+ options.put("fields.content.pk-full-text.index.options",
"{\"tokenizer\":\"jieba\"}");
+
+ assertThatThrownBy(() -> validateTableSchema(schema(options)))
+ .hasMessageContaining(
+ "fields.content.pk-full-text.index.options defines
conflicting values for full-text.tokenizer");
+ }
+
+ @Test
+ void testRequiresPrimaryKeyTable() {
+ Map<String, String> options = enabledOptions();
+ options.put(CoreOptions.BUCKET_KEY.key(), "id");
+
+ TableSchema appendTable =
+ new TableSchema(
+ 0,
+ fields(),
+ 0,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ options,
+ "");
+
+ assertThatThrownBy(() -> validateTableSchema(appendTable))
+ .hasMessageContaining("Primary-key full-text index requires a
primary-key table");
+ }
+
+ private static Map<String, String> enabledOptions() {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.BUCKET.key(), "1");
+ options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true");
+ options.put(CoreOptions.PK_FULL_TEXT_INDEX_COLUMNS.key(), "content");
+ return options;
+ }
+
+ private static java.util.List<DataField> fields() {
+ return Arrays.asList(
+ new DataField(0, "id", DataTypes.INT().notNull()),
+ new DataField(1, "content", DataTypes.STRING()),
+ new DataField(2, "other_content", DataTypes.STRING()));
+ }
+
+ private static TableSchema schema(Map<String, String> options) {
+ return new TableSchema(
+ 0,
+ fields(),
+ 0,
+ Collections.emptyList(),
+ Collections.singletonList("id"),
+ options,
+ "");
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerTest.java
b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerTest.java
index 9cb401c98e..1b3140a616 100644
--- a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerTest.java
@@ -226,6 +226,42 @@ public class SchemaManagerTest {
.hasMessage("Cannot rename primary-key index column: [name]");
}
+ @Test
+ public void testRejectDestructivePrimaryKeyFullTextIndexColumnChanges()
throws Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.BUCKET.key(), "1");
+ options.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true");
+ options.put(CoreOptions.PK_FULL_TEXT_INDEX_COLUMNS.key(), "content");
+ Schema schema =
+ new Schema(
+ Arrays.asList(
+ new DataField(0, "id",
DataTypes.INT().notNull()),
+ new DataField(1, "content",
DataTypes.STRING())),
+ Collections.emptyList(),
+ Collections.singletonList("id"),
+ options,
+ "");
+ SchemaManager manager = new SchemaManager(LocalFileIO.create(), path);
+ manager.createTable(schema);
+
+ assertThatThrownBy(
+ () ->
+ manager.commitChanges(
+ SchemaChange.renameColumn(
+ new String[] {"content"},
"renamed_content")))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessage("Cannot rename primary-key index column:
[content]");
+ assertThatThrownBy(() ->
manager.commitChanges(SchemaChange.dropColumn("content")))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessage("Cannot drop primary-key index column: [content]");
+ assertThatThrownBy(
+ () ->
+ manager.commitChanges(
+
SchemaChange.updateColumnType("content", DataTypes.INT())))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessage("Cannot update type of primary-key index column:
[content]");
+ }
+
@Test
public void testRejectDropPrimaryKeyBitmapIndexColumn() throws Exception {
Map<String, String> options = new HashMap<>();
