apache · JingsongLi · Jul 11, 2026 · Jul 11, 2026
diff --git a/...micro-benchmarks/src/test/java/org/apache/paimon/benchmark/WriteRestoreScanBenchmark.java b/...micro-benchmarks/src/test/java/org/apache/paimon/benchmark/WriteRestoreScanBenchmark.java
@@ -400,6 +400,7 @@ private static void runRestoreAcrossBuckets(
                                                     entry.partition(),
                                                     entry.bucket(),
                                                     false,
+                                                    false,
                                                     false)));
         }
         for (Future<?> f : futures) {

diff --git 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.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.io.KeyValueFileReaderFactory;
 import org.apache.paimon.mergetree.compact.MergeFunctionFactory;
 import org.apache.paimon.operation.AbstractFileStoreWrite;
@@ -38,6 +39,7 @@
 import org.apache.paimon.schema.TableSchema;
 import org.apache.paimon.table.BucketMode;
 import org.apache.paimon.table.CatalogEnvironment;
+import org.apache.paimon.types.DataField;
 import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.KeyComparatorSupplier;
 import org.apache.paimon.utils.UserDefinedSeqComparator;
@@ -169,6 +171,19 @@ public AbstractFileStoreWrite<KeyValue> newWrite(String commitUser, @Nullable In
         if (options.deletionVectorsEnabled()) {
             dvMaintainerFactory = BucketedDvMaintainer.factory(newIndexFileHandler());
         }
+        BucketedVectorIndexMaintainer.Factory vectorIndexMaintainerFactory = 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));
+        }
         return new KeyValueFileStoreWrite(
                 fileIO,
                 schemaManager,
@@ -187,6 +202,7 @@ public AbstractFileStoreWrite<KeyValue> newWrite(String commitUser, @Nullable In
                 newScan(),
                 indexFactory,
                 dvMaintainerFactory,
+                vectorIndexMaintainerFactory,
                 options,
                 keyValueFieldsExtractor,
                 tableName);

diff --git a/paimon-core/src/main/java/org/apache/paimon/index/IndexFileHandler.java b/paimon-core/src/main/java/org/apache/paimon/index/IndexFileHandler.java
@@ -145,6 +145,23 @@ public List<IndexFileMeta> scan(
         return result;
     }
 
+    public List<IndexFileMeta> scanSourceIndexes(
+            Snapshot snapshot, BinaryRow partition, int bucket) {
+        List<IndexFileMeta> result = new ArrayList<>();
+        for (IndexManifestEntry entry :
+                scan(
+                        snapshot,
+                        candidate ->
+                                candidate.partition().equals(partition)
+                                        && candidate.bucket() == bucket
+                                        && candidate.indexFile().globalIndexMeta() != null
+                                        && candidate.indexFile().globalIndexMeta().sourceMeta()
+                                                != null)) {
+            result.add(entry.indexFile());
+        }
+        return result;
+    }
+
     public Map<Pair<BinaryRow, Integer>, List<IndexFileMeta>> scan(
             long snapshot, String indexType, Set<BinaryRow> partitions) {
         return scan(snapshotManager.snapshot(snapshot), indexType, partitions);

diff --git a/...ore/src/main/java/org/apache/paimon/mergetree/compact/LookupMergeTreeCompactRewriter.java b/...ore/src/main/java/org/apache/paimon/mergetree/compact/LookupMergeTreeCompactRewriter.java
@@ -58,6 +58,7 @@ public class LookupMergeTreeCompactRewriter<T> extends ChangelogMergeTreeRewrite
     private final LookupLevels<T> lookupLevels;
     private final MergeFunctionWrapperFactory<T> wrapperFactory;
     private final boolean noSequenceField;
+    private final boolean forceRewriteLevelZeroForVectorIndex;
     @Nullable private final BucketedDvMaintainer dvMaintainer;
     private final IntFunction<String> level2FileFormat;
 
@@ -93,6 +94,7 @@ public LookupMergeTreeCompactRewriter(
         this.lookupLevels = lookupLevels;
         this.wrapperFactory = wrapperFactory;
         this.noSequenceField = options.sequenceField().isEmpty();
+        this.forceRewriteLevelZeroForVectorIndex = options.primaryKeyVectorIndexEnabled();
         String fileFormat = options.fileFormatString();
         Map<Integer, String> fileFormatPerLevel = options.fileFormatPerLevel();
         this.level2FileFormat = level -> fileFormatPerLevel.getOrDefault(level, fileFormat);
@@ -135,6 +137,10 @@ protected UpgradeStrategy upgradeStrategy(int outputLevel, DataFileMeta file) {
             return NO_CHANGELOG_NO_REWRITE;
         }
 
+        if (forceRewriteLevelZeroForVectorIndex) {
+            return CHANGELOG_WITH_REWRITE;
+        }
+
         // forcing rewriting when upgrading from level 0 to level x with different file formats
         if (!level2FileFormat.apply(file.level()).equals(level2FileFormat.apply(outputLevel))) {
             return CHANGELOG_WITH_REWRITE;

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
@@ -29,6 +29,7 @@
 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.io.CompactIncrement;
 import org.apache.paimon.io.DataFileMeta;
 import org.apache.paimon.io.DataIncrement;
@@ -83,6 +84,7 @@ 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;
     private final int numBuckets;
     private final RowType partitionType;
 
@@ -109,6 +111,7 @@ protected AbstractFileStoreWrite(
             FileStoreScan scan,
             @Nullable DynamicBucketIndexMaintainer.Factory dbMaintainerFactory,
             @Nullable BucketedDvMaintainer.Factory dvMaintainerFactory,
+            @Nullable BucketedVectorIndexMaintainer.Factory vectorIndexMaintainerFactory,
             String tableName,
             CoreOptions options,
             RowType partitionType) {
@@ -117,11 +120,14 @@ protected AbstractFileStoreWrite(
             indexFileHandler = dbMaintainerFactory.indexFileHandler();
         } else if (dvMaintainerFactory != null) {
             indexFileHandler = dvMaintainerFactory.indexFileHandler();
+        } else if (vectorIndexMaintainerFactory != null) {
+            indexFileHandler = vectorIndexMaintainerFactory.indexFileHandler();
         }
         this.snapshotManager = snapshotManager;
         this.restore = new FileSystemWriteRestore(options, snapshotManager, scan, indexFileHandler);
         this.dbMaintainerFactory = dbMaintainerFactory;
         this.dvMaintainerFactory = dvMaintainerFactory;
+        this.vectorIndexMaintainerFactory = vectorIndexMaintainerFactory;
         this.numBuckets = options.bucket();
         this.partitionType = partitionType;
         this.writers = new HashMap<>();
@@ -242,6 +248,33 @@ public List<CommitMessage> prepareCommit(boolean waitCompaction, long commitIden
                             .newIndexFiles()
                             .addAll(writerContainer.dynamicBucketMaintainer.prepareCommit());
                 }
+                if (writerContainer.vectorIndexMaintainer != null) {
+                    BucketedVectorIndexMaintainer.VectorIndexCommit vectorCommit =
+                            writerContainer.vectorIndexMaintainer.prepareCommit(
+                                    newFilesIncrement, compactIncrement);
+                    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());
+                                    });
+                }
                 CompactDeletionFile compactDeletionFile = increment.compactDeletionFile();
                 if (compactDeletionFile != null) {
                     compactDeletionFile
@@ -375,6 +408,7 @@ public List<State<T>> checkpoint() {
                                 writerContainer.writer.maxSequenceNumber(),
                                 writerContainer.dynamicBucketMaintainer,
                                 writerContainer.deletionVectorsMaintainer,
+                                writerContainer.vectorIndexMaintainer,
                                 increment));
             }
         }
@@ -407,6 +441,7 @@ public void restore(List<State<T>> states) {
                             state.totalBuckets,
                             state.indexMaintainer,
                             state.deletionVectorsMaintainer,
+                            state.vectorIndexMaintainer,
                             state.baseSnapshotId);
             writerContainer.lastModifiedCommitIdentifier = state.lastModifiedCommitIdentifier;
             writers.computeIfAbsent(state.partition, k -> new HashMap<>())
@@ -473,6 +508,14 @@ public WriterContainer<T> createWriterContainer(BinaryRow partition, int bucket)
                         ? null
                         : dvMaintainerFactory.create(
                                 partition, bucket, restored.deleteVectorsIndex());
+        BucketedVectorIndexMaintainer vectorIndexMaintainer =
+                vectorIndexMaintainerFactory == null
+                        ? null
+                        : vectorIndexMaintainerFactory.create(
+                                partition,
+                                bucket,
+                                restored.dataFiles(),
+                                restored.vectorIndexPayloads());
 
         List<DataFileMeta> restoreFiles = restored.dataFiles();
         if (restoreFiles == null) {
@@ -497,6 +540,7 @@ public WriterContainer<T> createWriterContainer(BinaryRow partition, int bucket)
                 firstNonNull(restored.totalBuckets(), numBuckets),
                 indexMaintainer,
                 dvMaintainer,
+                vectorIndexMaintainer,
                 previousSnapshot == null ? null : previousSnapshot.id());
     }
 
@@ -558,7 +602,8 @@ private RestoreFiles scanExistingFileMetas(BinaryRow partition, int bucket) {
                             partition,
                             bucket,
                             dbMaintainerFactory != null,
-                            dvMaintainerFactory != null);
+                            dvMaintainerFactory != null,
+                            vectorIndexMaintainerFactory != null);
         } catch (RuntimeException e) {
             throw new RuntimeException(
                     String.format(
@@ -621,6 +666,7 @@ public static class WriterContainer<T> {
         public final int totalBuckets;
         @Nullable public final DynamicBucketIndexMaintainer dynamicBucketMaintainer;
         @Nullable public final BucketedDvMaintainer deletionVectorsMaintainer;
+        @Nullable public final BucketedVectorIndexMaintainer vectorIndexMaintainer;
         protected final long baseSnapshotId;
         protected long lastModifiedCommitIdentifier;
 
@@ -629,11 +675,13 @@ protected WriterContainer(
                 int totalBuckets,
                 @Nullable DynamicBucketIndexMaintainer dynamicBucketMaintainer,
                 @Nullable BucketedDvMaintainer deletionVectorsMaintainer,
+                @Nullable BucketedVectorIndexMaintainer vectorIndexMaintainer,
                 Long baseSnapshotId) {
             this.writer = writer;
             this.totalBuckets = totalBuckets;
             this.dynamicBucketMaintainer = dynamicBucketMaintainer;
             this.deletionVectorsMaintainer = deletionVectorsMaintainer;
+            this.vectorIndexMaintainer = vectorIndexMaintainer;
             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/BaseAppendFileStoreWrite.java b/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java
@@ -96,7 +96,15 @@ public BaseAppendFileStoreWrite(
             CoreOptions options,
             @Nullable BucketedDvMaintainer.Factory dvMaintainerFactory,
             String tableName) {
-        super(snapshotManager, scan, options, partitionType, null, dvMaintainerFactory, tableName);
+        super(
+                snapshotManager,
+                scan,
+                options,
+                partitionType,
+                null,
+                dvMaintainerFactory,
+                null,
+                tableName);
         this.fileIO = fileIO;
         this.readForCompact = readForCompact;
         this.schemaId = schemaId;

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
@@ -24,6 +24,7 @@
 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.io.DataFileMeta;
 import org.apache.paimon.memory.MemoryPoolFactory;
 import org.apache.paimon.metrics.MetricRegistry;
@@ -150,6 +151,7 @@ class State<T> {
         protected final long maxSequenceNumber;
         @Nullable protected final DynamicBucketIndexMaintainer indexMaintainer;
         @Nullable protected final BucketedDvMaintainer deletionVectorsMaintainer;
+        @Nullable protected final BucketedVectorIndexMaintainer vectorIndexMaintainer;
         protected final CommitIncrement commitIncrement;
 
         protected State(
@@ -162,6 +164,7 @@ protected State(
                 long maxSequenceNumber,
                 @Nullable DynamicBucketIndexMaintainer indexMaintainer,
                 @Nullable BucketedDvMaintainer deletionVectorsMaintainer,
+                @Nullable BucketedVectorIndexMaintainer vectorIndexMaintainer,
                 CommitIncrement commitIncrement) {
             this.partition = partition;
             this.bucket = bucket;
@@ -172,13 +175,14 @@ protected State(
             this.maxSequenceNumber = maxSequenceNumber;
             this.indexMaintainer = indexMaintainer;
             this.deletionVectorsMaintainer = deletionVectorsMaintainer;
+            this.vectorIndexMaintainer = vectorIndexMaintainer;
             this.commitIncrement = commitIncrement;
         }
 
         @Override
         public String toString() {
             return String.format(
-                    "{%s, %d, %d, %d, %d, %s, %d, %s, %s, %s}",
+                    "{%s, %d, %d, %d, %d, %s, %d, %s, %s, %s, %s}",
                     partition,
                     bucket,
                     totalBuckets,
@@ -188,6 +192,7 @@ public String toString() {
                     maxSequenceNumber,
                     indexMaintainer,
                     deletionVectorsMaintainer,
+                    vectorIndexMaintainer,
                     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
@@ -67,7 +67,8 @@ public RestoreFiles restoreFiles(
             BinaryRow partition,
             int bucket,
             boolean scanDynamicBucketIndex,
-            boolean scanDeleteVectorsIndex) {
+            boolean scanDeleteVectorsIndex,
+            boolean scanVectorIndexPayloads) {
         // NOTE: don't use snapshotManager.latestSnapshot() here,
         // because we don't want to flood the catalog with high concurrency
         Snapshot snapshot = snapshotManager.latestSnapshotFromFileSystem();
@@ -92,7 +93,17 @@ public RestoreFiles restoreFiles(
                     indexFileHandler.scan(snapshot, DELETION_VECTORS_INDEX, partition, bucket);
         }
 
+        List<IndexFileMeta> vectorIndexPayloads = null;
+        if (scanVectorIndexPayloads) {
+            vectorIndexPayloads = indexFileHandler.scanSourceIndexes(snapshot, partition, bucket);
+        }
+
         return new RestoreFiles(
-                snapshot, totalBuckets, restoreFiles, dynamicBucketIndex, deleteVectorsIndex);
+                snapshot,
+                totalBuckets,
+                restoreFiles,
+                dynamicBucketIndex,
+                deleteVectorsIndex,
+                vectorIndexPayloads);
     }
 }
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
@@ -31,6 +31,7 @@
 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.io.DataFileMeta;
 import org.apache.paimon.io.KeyValueFileReaderFactory;
 import org.apache.paimon.io.KeyValueFileWriterFactory;
@@ -96,6 +97,7 @@ public KeyValueFileStoreWrite(
             FileStoreScan scan,
             @Nullable DynamicBucketIndexMaintainer.Factory dbMaintainerFactory,
             @Nullable BucketedDvMaintainer.Factory dvMaintainerFactory,
+            @Nullable BucketedVectorIndexMaintainer.Factory vectorIndexMaintainerFactory,
             CoreOptions options,
             KeyValueFieldsExtractor extractor,
             String tableName) {
@@ -106,6 +108,7 @@ public KeyValueFileStoreWrite(
                 partitionType,
                 dbMaintainerFactory,
                 dvMaintainerFactory,
+                vectorIndexMaintainerFactory,
                 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
@@ -21,6 +21,7 @@
 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.io.cache.CacheManager;
 import org.apache.paimon.memory.HeapMemorySegmentPool;
 import org.apache.paimon.memory.MemoryOwner;
@@ -66,12 +67,14 @@ public MemoryFileStoreWrite(
             RowType partitionType,
             @Nullable DynamicBucketIndexMaintainer.Factory dbMaintainerFactory,
             @Nullable BucketedDvMaintainer.Factory dvMaintainerFactory,
+            @Nullable BucketedVectorIndexMaintainer.Factory vectorIndexMaintainerFactory,
             String tableName) {
         super(
                 snapshotManager,
                 scan,
                 dbMaintainerFactory,
                 dvMaintainerFactory,
+                vectorIndexMaintainerFactory,
                 tableName,
                 options,
                 partitionType);