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 9107d0a6a1 [core] Generalize the primary-key index write lifecycle
(#8598)
9107d0a6a1 is described below
commit 9107d0a6a161b8604197fb00506b89f5133a17fe
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Jul 13 18:52:38 2026 +0800
[core] Generalize the primary-key index write lifecycle (#8598)
Prepare the primary-key index write lifecycle for additional
source-backed index families, such as BTree and Bitmap, while preserving
the existing vector-index behavior and Postpone Bucket compaction
support.
---
.../java/org/apache/paimon/KeyValueFileStore.java | 23 ++--
.../pk/BucketedPrimaryKeyIndexMaintainer.java | 139 +++++++++++++++++++++
.../paimon/operation/AbstractFileStoreWrite.java | 127 +++++++------------
.../apache/paimon/operation/FileStoreWrite.java | 10 +-
.../paimon/operation/FileSystemWriteRestore.java | 10 +-
.../paimon/operation/KeyValueFileStoreWrite.java | 6 +-
.../paimon/operation/MemoryFileStoreWrite.java | 6 +-
.../org/apache/paimon/operation/RestoreFiles.java | 10 +-
.../org/apache/paimon/operation/WriteRestore.java | 2 +-
.../operation/FileSystemWriteRestoreTest.java | 8 +-
.../operation/PrimaryKeyVectorIndexWriteTest.java | 11 +-
11 files changed, 234 insertions(+), 118 deletions(-)
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 d7e195418b..8e0d4da7e2 100644
--- a/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
+++ b/paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java
@@ -24,6 +24,7 @@ import org.apache.paimon.deletionvectors.BucketedDvMaintainer;
import org.apache.paimon.format.FileFormatDiscover;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.index.DynamicBucketIndexMaintainer;
+import org.apache.paimon.index.pk.BucketedPrimaryKeyIndexMaintainer;
import org.apache.paimon.index.pkvector.BucketedVectorIndexMaintainer;
import org.apache.paimon.io.KeyValueFileReaderFactory;
import org.apache.paimon.mergetree.compact.MergeFunctionFactory;
@@ -171,18 +172,20 @@ public class KeyValueFileStore extends
AbstractFileStore<KeyValue> {
if (options.deletionVectorsEnabled()) {
dvMaintainerFactory =
BucketedDvMaintainer.factory(newIndexFileHandler());
}
- BucketedVectorIndexMaintainer.Factory vectorIndexMaintainerFactory =
null;
+ BucketedPrimaryKeyIndexMaintainer.Factory
primaryKeyIndexMaintainerFactory = null;
if (options.primaryKeyVectorIndexEnabled()) {
String vectorColumn = options.primaryKeyVectorIndexColumn();
DataField vectorField = schema.nameToFieldMap().get(vectorColumn);
- vectorIndexMaintainerFactory =
- new BucketedVectorIndexMaintainer.Factory(
- newIndexFileHandler(),
- newReaderFactoryBuilder(),
- vectorField,
-
options.primaryKeyVectorDistanceMetric(vectorColumn),
-
options.primaryKeyVectorIndexType(vectorColumn))
-
.withIndexOptions(options.primaryKeyVectorIndexOptions(vectorColumn));
+ primaryKeyIndexMaintainerFactory =
+ BucketedPrimaryKeyIndexMaintainer.Factory.ofVector(
+ new BucketedVectorIndexMaintainer.Factory(
+ newIndexFileHandler(),
+ newReaderFactoryBuilder(),
+ vectorField,
+
options.primaryKeyVectorDistanceMetric(vectorColumn),
+
options.primaryKeyVectorIndexType(vectorColumn))
+ .withIndexOptions(
+
options.primaryKeyVectorIndexOptions(vectorColumn)));
}
return new KeyValueFileStoreWrite(
fileIO,
@@ -202,7 +205,7 @@ public class KeyValueFileStore extends
AbstractFileStore<KeyValue> {
newScan(),
indexFactory,
dvMaintainerFactory,
- vectorIndexMaintainerFactory,
+ primaryKeyIndexMaintainerFactory,
options,
keyValueFieldsExtractor,
tableName);
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
new file mode 100644
index 0000000000..702a1dc9cf
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/index/pk/BucketedPrimaryKeyIndexMaintainer.java
@@ -0,0 +1,139 @@
+/*
+ * 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.pk;
+
+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.pkvector.BucketedVectorIndexMaintainer;
+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.Collections;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+
+/** Coordinates source-backed primary-key indexes for one bucket. */
+public final class BucketedPrimaryKeyIndexMaintainer {
+
+ private final BucketedVectorIndexMaintainer vectorMaintainer;
+
+ private BucketedPrimaryKeyIndexMaintainer(BucketedVectorIndexMaintainer
vectorMaintainer) {
+ this.vectorMaintainer = vectorMaintainer;
+ }
+
+ public static BucketedPrimaryKeyIndexMaintainer ofVector(
+ BucketedVectorIndexMaintainer vectorMaintainer) {
+ return new BucketedPrimaryKeyIndexMaintainer(vectorMaintainer);
+ }
+
+ public void prepareCommit(
+ DataIncrement appendIncrement,
+ CompactIncrement compactIncrement,
+ boolean waitCompaction)
+ throws Exception {
+ BucketedVectorIndexMaintainer.VectorIndexCommit vectorCommit =
+ vectorMaintainer.prepareCommit(appendIncrement,
compactIncrement, waitCompaction);
+ vectorCommit
+ .appendIncrement()
+ .ifPresent(
+ increment ->
+ applyIndexIncrement(
+ appendIncrement.newIndexFiles(),
+ appendIncrement.deletedIndexFiles(),
+ increment.newIndexFiles(),
+ increment.deletedIndexFiles()));
+ vectorCommit
+ .compactIncrement()
+ .ifPresent(
+ increment ->
+ applyIndexIncrement(
+ compactIncrement.newIndexFiles(),
+ compactIncrement.deletedIndexFiles(),
+ increment.newIndexFiles(),
+ increment.deletedIndexFiles()));
+ }
+
+ private static void applyIndexIncrement(
+ List<IndexFileMeta> targetNew,
+ List<IndexFileMeta> targetDeleted,
+ List<IndexFileMeta> sourceNew,
+ List<IndexFileMeta> sourceDeleted) {
+ targetNew.addAll(sourceNew);
+ targetDeleted.addAll(sourceDeleted);
+ }
+
+ public boolean buildNotCompleted() {
+ return vectorMaintainer.buildNotCompleted();
+ }
+
+ public void withExecutor(ExecutorService executor) {
+ vectorMaintainer.withExecutor(executor);
+ }
+
+ public void close() {
+ vectorMaintainer.close();
+ }
+
+ /** Factory to restore configured source-backed primary-key indexes for a
bucket. */
+ public static final class Factory {
+
+ private final BucketedVectorIndexMaintainer.Factory vectorFactory;
+
+ private Factory(BucketedVectorIndexMaintainer.Factory vectorFactory) {
+ this.vectorFactory = vectorFactory;
+ }
+
+ public static Factory ofVector(BucketedVectorIndexMaintainer.Factory
vectorFactory) {
+ return new Factory(vectorFactory);
+ }
+
+ public IndexFileHandler indexFileHandler() {
+ return vectorFactory.indexFileHandler();
+ }
+
+ public BucketedPrimaryKeyIndexMaintainer create(
+ BinaryRow partition,
+ int bucket,
+ @Nullable List<DataFileMeta> restoredDataFiles,
+ @Nullable List<IndexFileMeta> restoredPayloads,
+ ExecutorService executor) {
+ List<DataFileMeta> dataFiles =
+ restoredDataFiles == null ? Collections.emptyList() :
restoredDataFiles;
+ List<IndexFileMeta> payloads =
+ restoredPayloads == null ? Collections.emptyList() :
restoredPayloads;
+ return BucketedPrimaryKeyIndexMaintainer.ofVector(
+ vectorFactory.create(partition, bucket, dataFiles,
payloads, executor));
+ }
+
+ public BucketedPrimaryKeyIndexMaintainer create(
+ BinaryRow partition,
+ int bucket,
+ @Nullable List<DataFileMeta> restoredDataFiles,
+ @Nullable List<IndexFileMeta> restoredPayloads,
+ ExecutorService executor,
+ @Nullable IOManager ioManager) {
+ return create(partition, bucket, restoredDataFiles,
restoredPayloads, executor);
+ }
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java
b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java
index 0409f9492a..a393613b35 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java
@@ -29,7 +29,7 @@ import org.apache.paimon.deletionvectors.BucketedDvMaintainer;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.index.DynamicBucketIndexMaintainer;
import org.apache.paimon.index.IndexFileHandler;
-import org.apache.paimon.index.pkvector.BucketedVectorIndexMaintainer;
+import org.apache.paimon.index.pk.BucketedPrimaryKeyIndexMaintainer;
import org.apache.paimon.io.CompactIncrement;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.io.DataIncrement;
@@ -84,7 +84,10 @@ public abstract class AbstractFileStoreWrite<T> implements
FileStoreWrite<T> {
private final int writerNumberMax;
@Nullable private final DynamicBucketIndexMaintainer.Factory
dbMaintainerFactory;
@Nullable private final BucketedDvMaintainer.Factory dvMaintainerFactory;
- @Nullable private final BucketedVectorIndexMaintainer.Factory
vectorIndexMaintainerFactory;
+
+ @Nullable
+ private final BucketedPrimaryKeyIndexMaintainer.Factory
primaryKeyIndexMaintainerFactory;
+
private final int numBuckets;
private final RowType partitionType;
@@ -96,7 +99,7 @@ public abstract class AbstractFileStoreWrite<T> implements
FileStoreWrite<T> {
protected WriteRestore restore;
private ExecutorService lazyCompactExecutor;
- private ExecutorService lazyVectorIndexExecutor;
+ private ExecutorService lazyPrimaryKeyIndexExecutor;
private boolean closeCompactExecutorWhenLeaving = true;
private boolean ignorePreviousFiles = false;
private boolean ignoreNumBucketCheck = false;
@@ -112,7 +115,7 @@ public abstract class AbstractFileStoreWrite<T> implements
FileStoreWrite<T> {
FileStoreScan scan,
@Nullable DynamicBucketIndexMaintainer.Factory dbMaintainerFactory,
@Nullable BucketedDvMaintainer.Factory dvMaintainerFactory,
- @Nullable BucketedVectorIndexMaintainer.Factory
vectorIndexMaintainerFactory,
+ @Nullable BucketedPrimaryKeyIndexMaintainer.Factory
primaryKeyIndexMaintainerFactory,
String tableName,
CoreOptions options,
RowType partitionType) {
@@ -121,14 +124,14 @@ public abstract class AbstractFileStoreWrite<T>
implements FileStoreWrite<T> {
indexFileHandler = dbMaintainerFactory.indexFileHandler();
} else if (dvMaintainerFactory != null) {
indexFileHandler = dvMaintainerFactory.indexFileHandler();
- } else if (vectorIndexMaintainerFactory != null) {
- indexFileHandler = vectorIndexMaintainerFactory.indexFileHandler();
+ } else if (primaryKeyIndexMaintainerFactory != null) {
+ indexFileHandler =
primaryKeyIndexMaintainerFactory.indexFileHandler();
}
this.snapshotManager = snapshotManager;
this.restore = new FileSystemWriteRestore(options, snapshotManager,
scan, indexFileHandler);
this.dbMaintainerFactory = dbMaintainerFactory;
this.dvMaintainerFactory = dvMaintainerFactory;
- this.vectorIndexMaintainerFactory = vectorIndexMaintainerFactory;
+ this.primaryKeyIndexMaintainerFactory =
primaryKeyIndexMaintainerFactory;
this.numBuckets = options.bucket();
this.partitionType = partitionType;
this.writers = new HashMap<>();
@@ -249,11 +252,10 @@ public abstract class AbstractFileStoreWrite<T>
implements FileStoreWrite<T> {
.newIndexFiles()
.addAll(writerContainer.dynamicBucketMaintainer.prepareCommit());
}
- applyVectorIndexCommit(
- writerContainer.vectorIndexMaintainer,
- newFilesIncrement,
- compactIncrement,
- waitCompaction);
+ if (writerContainer.primaryKeyIndexMaintainer != null) {
+ writerContainer.primaryKeyIndexMaintainer.prepareCommit(
+ newFilesIncrement, compactIncrement,
waitCompaction);
+ }
CompactDeletionFile compactDeletionFile =
increment.compactDeletionFile();
if (compactDeletionFile != null) {
compactDeletionFile
@@ -271,8 +273,8 @@ public abstract class AbstractFileStoreWrite<T> implements
FileStoreWrite<T> {
if (committable.isEmpty()) {
if (writerCleanChecker.apply(writerContainer)
- && (writerContainer.vectorIndexMaintainer == null
- || !writerContainer.vectorIndexMaintainer
+ && (writerContainer.primaryKeyIndexMaintainer ==
null
+ ||
!writerContainer.primaryKeyIndexMaintainer
.buildNotCompleted())) {
// Clear writer if no update, and if its latest
modification has committed.
//
@@ -289,8 +291,8 @@ public abstract class AbstractFileStoreWrite<T> implements
FileStoreWrite<T> {
commitIdentifier);
}
writerContainer.writer.close();
- if (writerContainer.vectorIndexMaintainer != null) {
- writerContainer.vectorIndexMaintainer.close();
+ if (writerContainer.primaryKeyIndexMaintainer != null)
{
+ writerContainer.primaryKeyIndexMaintainer.close();
}
bucketIter.remove();
}
@@ -307,43 +309,6 @@ public abstract class AbstractFileStoreWrite<T> implements
FileStoreWrite<T> {
return result;
}
- private void applyVectorIndexCommit(
- @Nullable BucketedVectorIndexMaintainer vectorIndexMaintainer,
- DataIncrement newFilesIncrement,
- CompactIncrement compactIncrement,
- boolean waitCompaction)
- throws Exception {
- if (vectorIndexMaintainer == null) {
- return;
- }
-
- BucketedVectorIndexMaintainer.VectorIndexCommit vectorCommit =
- vectorIndexMaintainer.prepareCommit(
- newFilesIncrement, compactIncrement, waitCompaction);
- vectorCommit
- .appendIncrement()
- .ifPresent(
- vectorIncrement -> {
- newFilesIncrement
- .newIndexFiles()
- .addAll(vectorIncrement.newIndexFiles());
- newFilesIncrement
- .deletedIndexFiles()
-
.addAll(vectorIncrement.deletedIndexFiles());
- });
- vectorCommit
- .compactIncrement()
- .ifPresent(
- vectorIncrement -> {
- compactIncrement
- .newIndexFiles()
- .addAll(vectorIncrement.newIndexFiles());
- compactIncrement
- .deletedIndexFiles()
-
.addAll(vectorIncrement.deletedIndexFiles());
- });
- }
-
// This abstract function returns a whole function (instead of just a
boolean value),
// because we do not want to introduce `commitUser` into this base class.
//
@@ -382,8 +347,8 @@ public abstract class AbstractFileStoreWrite<T> implements
FileStoreWrite<T> {
for (Map<Integer, WriterContainer<T>> bucketWriters :
writers.values()) {
for (WriterContainer<T> writerContainer : bucketWriters.values()) {
writerContainer.writer.close();
- if (writerContainer.vectorIndexMaintainer != null) {
- writerContainer.vectorIndexMaintainer.close();
+ if (writerContainer.primaryKeyIndexMaintainer != null) {
+ writerContainer.primaryKeyIndexMaintainer.close();
}
}
}
@@ -391,8 +356,8 @@ public abstract class AbstractFileStoreWrite<T> implements
FileStoreWrite<T> {
if (lazyCompactExecutor != null && closeCompactExecutorWhenLeaving) {
lazyCompactExecutor.shutdownNow();
}
- if (lazyVectorIndexExecutor != null) {
- lazyVectorIndexExecutor.shutdownNow();
+ if (lazyPrimaryKeyIndexExecutor != null) {
+ lazyPrimaryKeyIndexExecutor.shutdownNow();
}
if (compactionMetrics != null) {
compactionMetrics.close();
@@ -414,11 +379,10 @@ public abstract class AbstractFileStoreWrite<T>
implements FileStoreWrite<T> {
CommitIncrement increment;
try {
increment = writerContainer.writer.prepareCommit(false);
- applyVectorIndexCommit(
- writerContainer.vectorIndexMaintainer,
- increment.newFilesIncrement(),
- increment.compactIncrement(),
- true);
+ if (writerContainer.primaryKeyIndexMaintainer != null) {
+
writerContainer.primaryKeyIndexMaintainer.prepareCommit(
+ increment.newFilesIncrement(),
increment.compactIncrement(), true);
+ }
} catch (Exception e) {
throw new RuntimeException(
"Failed to extract state from writer of partition "
@@ -441,7 +405,7 @@ public abstract class AbstractFileStoreWrite<T> implements
FileStoreWrite<T> {
writerContainer.writer.maxSequenceNumber(),
writerContainer.dynamicBucketMaintainer,
writerContainer.deletionVectorsMaintainer,
- writerContainer.vectorIndexMaintainer,
+ writerContainer.primaryKeyIndexMaintainer,
increment));
}
}
@@ -468,8 +432,8 @@ public abstract class AbstractFileStoreWrite<T> implements
FileStoreWrite<T> {
// not ignore them.
false);
notifyNewWriter(writer);
- if (state.vectorIndexMaintainer != null) {
-
state.vectorIndexMaintainer.withExecutor(vectorIndexExecutor());
+ if (state.primaryKeyIndexMaintainer != null) {
+
state.primaryKeyIndexMaintainer.withExecutor(primaryKeyIndexExecutor());
}
WriterContainer<T> writerContainer =
new WriterContainer<>(
@@ -477,7 +441,7 @@ public abstract class AbstractFileStoreWrite<T> implements
FileStoreWrite<T> {
state.totalBuckets,
state.indexMaintainer,
state.deletionVectorsMaintainer,
- state.vectorIndexMaintainer,
+ state.primaryKeyIndexMaintainer,
state.baseSnapshotId);
writerContainer.lastModifiedCommitIdentifier =
state.lastModifiedCommitIdentifier;
writers.computeIfAbsent(state.partition, k -> new HashMap<>())
@@ -544,15 +508,16 @@ public abstract class AbstractFileStoreWrite<T>
implements FileStoreWrite<T> {
? null
: dvMaintainerFactory.create(
partition, bucket,
restored.deleteVectorsIndex());
- BucketedVectorIndexMaintainer vectorIndexMaintainer =
- vectorIndexMaintainerFactory == null
+ BucketedPrimaryKeyIndexMaintainer primaryKeyIndexMaintainer =
+ primaryKeyIndexMaintainerFactory == null
? null
- : vectorIndexMaintainerFactory.create(
+ : primaryKeyIndexMaintainerFactory.create(
partition,
bucket,
restored.dataFiles(),
- restored.vectorIndexPayloads(),
- vectorIndexExecutor());
+ restored.sourceIndexPayloads(),
+ primaryKeyIndexExecutor(),
+ ioManager);
List<DataFileMeta> restoreFiles = restored.dataFiles();
if (restoreFiles == null) {
@@ -577,7 +542,7 @@ public abstract class AbstractFileStoreWrite<T> implements
FileStoreWrite<T> {
firstNonNull(restored.totalBuckets(), numBuckets),
indexMaintainer,
dvMaintainer,
- vectorIndexMaintainer,
+ primaryKeyIndexMaintainer,
previousSnapshot == null ? null : previousSnapshot.id());
}
@@ -640,7 +605,7 @@ public abstract class AbstractFileStoreWrite<T> implements
FileStoreWrite<T> {
bucket,
dbMaintainerFactory != null,
dvMaintainerFactory != null,
- vectorIndexMaintainerFactory != null);
+ primaryKeyIndexMaintainerFactory != null);
} catch (RuntimeException e) {
throw new RuntimeException(
String.format(
@@ -673,14 +638,14 @@ public abstract class AbstractFileStoreWrite<T>
implements FileStoreWrite<T> {
return lazyCompactExecutor;
}
- private ExecutorService vectorIndexExecutor() {
- if (lazyVectorIndexExecutor == null) {
- lazyVectorIndexExecutor =
+ private ExecutorService primaryKeyIndexExecutor() {
+ if (lazyPrimaryKeyIndexExecutor == null) {
+ lazyPrimaryKeyIndexExecutor =
Executors.newSingleThreadExecutor(
new ExecutorThreadFactory(
- Thread.currentThread().getName() +
"-vector-index"));
+ Thread.currentThread().getName() +
"-primary-key-index"));
}
- return lazyVectorIndexExecutor;
+ return lazyPrimaryKeyIndexExecutor;
}
@VisibleForTesting
@@ -713,7 +678,7 @@ public abstract class AbstractFileStoreWrite<T> implements
FileStoreWrite<T> {
public final int totalBuckets;
@Nullable public final DynamicBucketIndexMaintainer
dynamicBucketMaintainer;
@Nullable public final BucketedDvMaintainer deletionVectorsMaintainer;
- @Nullable public final BucketedVectorIndexMaintainer
vectorIndexMaintainer;
+ @Nullable public final BucketedPrimaryKeyIndexMaintainer
primaryKeyIndexMaintainer;
protected final long baseSnapshotId;
protected long lastModifiedCommitIdentifier;
@@ -722,13 +687,13 @@ public abstract class AbstractFileStoreWrite<T>
implements FileStoreWrite<T> {
int totalBuckets,
@Nullable DynamicBucketIndexMaintainer dynamicBucketMaintainer,
@Nullable BucketedDvMaintainer deletionVectorsMaintainer,
- @Nullable BucketedVectorIndexMaintainer vectorIndexMaintainer,
+ @Nullable BucketedPrimaryKeyIndexMaintainer
primaryKeyIndexMaintainer,
Long baseSnapshotId) {
this.writer = writer;
this.totalBuckets = totalBuckets;
this.dynamicBucketMaintainer = dynamicBucketMaintainer;
this.deletionVectorsMaintainer = deletionVectorsMaintainer;
- this.vectorIndexMaintainer = vectorIndexMaintainer;
+ this.primaryKeyIndexMaintainer = primaryKeyIndexMaintainer;
this.baseSnapshotId =
baseSnapshotId == null ? Snapshot.FIRST_SNAPSHOT_ID - 1 :
baseSnapshotId;
this.lastModifiedCommitIdentifier = Long.MIN_VALUE;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreWrite.java
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreWrite.java
index 5875840053..b94bd04fc4 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreWrite.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreWrite.java
@@ -24,7 +24,7 @@ import org.apache.paimon.data.BlobConsumer;
import org.apache.paimon.deletionvectors.BucketedDvMaintainer;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.index.DynamicBucketIndexMaintainer;
-import org.apache.paimon.index.pkvector.BucketedVectorIndexMaintainer;
+import org.apache.paimon.index.pk.BucketedPrimaryKeyIndexMaintainer;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.memory.MemoryPoolFactory;
import org.apache.paimon.metrics.MetricRegistry;
@@ -151,7 +151,7 @@ public interface FileStoreWrite<T> extends
Restorable<List<FileStoreWrite.State<
protected final long maxSequenceNumber;
@Nullable protected final DynamicBucketIndexMaintainer indexMaintainer;
@Nullable protected final BucketedDvMaintainer
deletionVectorsMaintainer;
- @Nullable protected final BucketedVectorIndexMaintainer
vectorIndexMaintainer;
+ @Nullable protected final BucketedPrimaryKeyIndexMaintainer
primaryKeyIndexMaintainer;
protected final CommitIncrement commitIncrement;
protected State(
@@ -164,7 +164,7 @@ public interface FileStoreWrite<T> extends
Restorable<List<FileStoreWrite.State<
long maxSequenceNumber,
@Nullable DynamicBucketIndexMaintainer indexMaintainer,
@Nullable BucketedDvMaintainer deletionVectorsMaintainer,
- @Nullable BucketedVectorIndexMaintainer vectorIndexMaintainer,
+ @Nullable BucketedPrimaryKeyIndexMaintainer
primaryKeyIndexMaintainer,
CommitIncrement commitIncrement) {
this.partition = partition;
this.bucket = bucket;
@@ -175,7 +175,7 @@ public interface FileStoreWrite<T> extends
Restorable<List<FileStoreWrite.State<
this.maxSequenceNumber = maxSequenceNumber;
this.indexMaintainer = indexMaintainer;
this.deletionVectorsMaintainer = deletionVectorsMaintainer;
- this.vectorIndexMaintainer = vectorIndexMaintainer;
+ this.primaryKeyIndexMaintainer = primaryKeyIndexMaintainer;
this.commitIncrement = commitIncrement;
}
@@ -192,7 +192,7 @@ public interface FileStoreWrite<T> extends
Restorable<List<FileStoreWrite.State<
maxSequenceNumber,
indexMaintainer,
deletionVectorsMaintainer,
- vectorIndexMaintainer,
+ primaryKeyIndexMaintainer,
commitIncrement);
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/FileSystemWriteRestore.java
b/paimon-core/src/main/java/org/apache/paimon/operation/FileSystemWriteRestore.java
index 3a2017ef56..740e7399f0 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/FileSystemWriteRestore.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/FileSystemWriteRestore.java
@@ -90,7 +90,7 @@ public class FileSystemWriteRestore implements WriteRestore {
int bucket,
boolean scanDynamicBucketIndex,
boolean scanDeleteVectorsIndex,
- boolean scanVectorIndexPayloads) {
+ boolean scanSourceIndexPayloads) {
// NOTE: don't use snapshotManager.latestSnapshot() here,
// because we don't want to flood the catalog with high concurrency
Snapshot snapshot =
@@ -118,9 +118,9 @@ public class FileSystemWriteRestore implements WriteRestore
{
indexFileHandler.scan(snapshot, DELETION_VECTORS_INDEX,
partition, bucket);
}
- List<IndexFileMeta> vectorIndexPayloads = null;
- if (scanVectorIndexPayloads) {
- vectorIndexPayloads = indexFileHandler.scanSourceIndexes(snapshot,
partition, bucket);
+ List<IndexFileMeta> sourceIndexPayloads = null;
+ if (scanSourceIndexPayloads) {
+ sourceIndexPayloads = indexFileHandler.scanSourceIndexes(snapshot,
partition, bucket);
}
return new RestoreFiles(
@@ -129,6 +129,6 @@ public class FileSystemWriteRestore implements WriteRestore
{
restoreFiles,
dynamicBucketIndex,
deleteVectorsIndex,
- vectorIndexPayloads);
+ sourceIndexPayloads);
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
b/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
index df2faf2dc8..c672853684 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
@@ -31,7 +31,7 @@ import org.apache.paimon.disk.IOManager;
import org.apache.paimon.format.FileFormatDiscover;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.index.DynamicBucketIndexMaintainer;
-import org.apache.paimon.index.pkvector.BucketedVectorIndexMaintainer;
+import org.apache.paimon.index.pk.BucketedPrimaryKeyIndexMaintainer;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.io.KeyValueFileReaderFactory;
import org.apache.paimon.io.KeyValueFileWriterFactory;
@@ -97,7 +97,7 @@ public class KeyValueFileStoreWrite extends
MemoryFileStoreWrite<KeyValue> {
FileStoreScan scan,
@Nullable DynamicBucketIndexMaintainer.Factory dbMaintainerFactory,
@Nullable BucketedDvMaintainer.Factory dvMaintainerFactory,
- @Nullable BucketedVectorIndexMaintainer.Factory
vectorIndexMaintainerFactory,
+ @Nullable BucketedPrimaryKeyIndexMaintainer.Factory
primaryKeyIndexMaintainerFactory,
CoreOptions options,
KeyValueFieldsExtractor extractor,
String tableName) {
@@ -108,7 +108,7 @@ public class KeyValueFileStoreWrite extends
MemoryFileStoreWrite<KeyValue> {
partitionType,
dbMaintainerFactory,
dvMaintainerFactory,
- vectorIndexMaintainerFactory,
+ primaryKeyIndexMaintainerFactory,
tableName);
this.valueType = valueType;
this.commitUser = commitUser;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/MemoryFileStoreWrite.java
b/paimon-core/src/main/java/org/apache/paimon/operation/MemoryFileStoreWrite.java
index 2a178d83cd..96a6d6b483 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/MemoryFileStoreWrite.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/MemoryFileStoreWrite.java
@@ -21,7 +21,7 @@ package org.apache.paimon.operation;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.deletionvectors.BucketedDvMaintainer;
import org.apache.paimon.index.DynamicBucketIndexMaintainer;
-import org.apache.paimon.index.pkvector.BucketedVectorIndexMaintainer;
+import org.apache.paimon.index.pk.BucketedPrimaryKeyIndexMaintainer;
import org.apache.paimon.io.cache.CacheManager;
import org.apache.paimon.memory.HeapMemorySegmentPool;
import org.apache.paimon.memory.MemoryOwner;
@@ -67,14 +67,14 @@ public abstract class MemoryFileStoreWrite<T> extends
AbstractFileStoreWrite<T>
RowType partitionType,
@Nullable DynamicBucketIndexMaintainer.Factory dbMaintainerFactory,
@Nullable BucketedDvMaintainer.Factory dvMaintainerFactory,
- @Nullable BucketedVectorIndexMaintainer.Factory
vectorIndexMaintainerFactory,
+ @Nullable BucketedPrimaryKeyIndexMaintainer.Factory
primaryKeyIndexMaintainerFactory,
String tableName) {
super(
snapshotManager,
scan,
dbMaintainerFactory,
dvMaintainerFactory,
- vectorIndexMaintainerFactory,
+ primaryKeyIndexMaintainerFactory,
tableName,
options,
partitionType);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/RestoreFiles.java
b/paimon-core/src/main/java/org/apache/paimon/operation/RestoreFiles.java
index 0887149819..3284aa811a 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/RestoreFiles.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/RestoreFiles.java
@@ -34,7 +34,7 @@ public class RestoreFiles {
private final @Nullable List<DataFileMeta> dataFiles;
private final @Nullable IndexFileMeta dynamicBucketIndex;
private final @Nullable List<IndexFileMeta> deleteVectorsIndex;
- private final @Nullable List<IndexFileMeta> vectorIndexPayloads;
+ private final @Nullable List<IndexFileMeta> sourceIndexPayloads;
public RestoreFiles(
@Nullable Snapshot snapshot,
@@ -42,13 +42,13 @@ public class RestoreFiles {
@Nullable List<DataFileMeta> dataFiles,
@Nullable IndexFileMeta dynamicBucketIndex,
@Nullable List<IndexFileMeta> deleteVectorsIndex,
- @Nullable List<IndexFileMeta> vectorIndexPayloads) {
+ @Nullable List<IndexFileMeta> sourceIndexPayloads) {
this.snapshot = snapshot;
this.totalBuckets = totalBuckets;
this.dataFiles = dataFiles;
this.dynamicBucketIndex = dynamicBucketIndex;
this.deleteVectorsIndex = deleteVectorsIndex;
- this.vectorIndexPayloads = vectorIndexPayloads;
+ this.sourceIndexPayloads = sourceIndexPayloads;
}
@Nullable
@@ -77,8 +77,8 @@ public class RestoreFiles {
}
@Nullable
- public List<IndexFileMeta> vectorIndexPayloads() {
- return vectorIndexPayloads;
+ public List<IndexFileMeta> sourceIndexPayloads() {
+ return sourceIndexPayloads;
}
public static RestoreFiles empty() {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/WriteRestore.java
b/paimon-core/src/main/java/org/apache/paimon/operation/WriteRestore.java
index 284d22575a..f57d9ab055 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/WriteRestore.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/WriteRestore.java
@@ -36,7 +36,7 @@ public interface WriteRestore {
int bucket,
boolean scanDynamicBucketIndex,
boolean scanDeleteVectorsIndex,
- boolean scanVectorIndexPayloads);
+ boolean scanSourceIndexPayloads);
@Nullable
static Integer extractDataFiles(List<ManifestEntry> entries,
List<DataFileMeta> dataFiles) {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/FileSystemWriteRestoreTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/FileSystemWriteRestoreTest.java
index 830fa5ded0..69d41094c4 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/FileSystemWriteRestoreTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/FileSystemWriteRestoreTest.java
@@ -40,7 +40,7 @@ import static org.mockito.Mockito.when;
class FileSystemWriteRestoreTest {
@Test
- void testRestoreFromPinnedSnapshot() {
+ void testRestoreFromPinnedSnapshotForPostponeBucket() {
Snapshot pinned = mock(Snapshot.class);
Snapshot latest = mock(Snapshot.class);
SnapshotManager snapshotManager = mock(SnapshotManager.class);
@@ -70,14 +70,14 @@ class FileSystemWriteRestoreTest {
RestoreFiles restored = restore.restoreFiles(EMPTY_ROW, 0, false,
false, true);
assertThat(restored.snapshot()).isSameAs(pinned);
- assertThat(restored.vectorIndexPayloads()).containsExactly(ann);
+ assertThat(restored.sourceIndexPayloads()).containsExactly(ann);
verify(scan).withSnapshot(pinned);
verify(indexFileHandler).scanSourceIndexes(pinned, EMPTY_ROW, 0);
verify(snapshotManager, never()).latestSnapshotFromFileSystem();
}
@Test
- void testRestoreVectorIndexPayloadsWithoutDirectory() {
+ void testRestoreSourceIndexPayloadsWithoutDirectory() {
Snapshot snapshot = mock(Snapshot.class);
SnapshotManager snapshotManager = mock(SnapshotManager.class);
when(snapshotManager.latestSnapshotFromFileSystem()).thenReturn(snapshot);
@@ -100,6 +100,6 @@ class FileSystemWriteRestoreTest {
RestoreFiles restored = restore.restoreFiles(EMPTY_ROW, 0, false,
false, true);
- assertThat(restored.vectorIndexPayloads()).containsExactly(ann);
+ assertThat(restored.sourceIndexPayloads()).containsExactly(ann);
}
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyVectorIndexWriteTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyVectorIndexWriteTest.java
index 994c51445f..9b2aa3dcb1 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyVectorIndexWriteTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyVectorIndexWriteTest.java
@@ -22,6 +22,7 @@ import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.deletionvectors.BucketedDvMaintainer;
import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.pk.BucketedPrimaryKeyIndexMaintainer;
import org.apache.paimon.index.pkvector.BucketedVectorIndexMaintainer;
import org.apache.paimon.io.CompactIncrement;
import org.apache.paimon.io.DataFileMeta;
@@ -161,7 +162,15 @@ class PrimaryKeyVectorIndexWriteTest {
private void install(
RecordWriter<String> writer, BucketedVectorIndexMaintainer
vectorMaintainer) {
writers.computeIfAbsent(EMPTY_ROW, ignored -> new HashMap<>())
- .put(0, new WriterContainer<>(writer, 1, null, null,
vectorMaintainer, null));
+ .put(
+ 0,
+ new WriterContainer<>(
+ writer,
+ 1,
+ null,
+ null,
+
BucketedPrimaryKeyIndexMaintainer.ofVector(vectorMaintainer),
+ null));
}
@Override
