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 fb3484df6e [core][flink] Publish primary-key vector indexes on write
(#8563)
fb3484df6e is described below
commit fb3484df6e0053a94ed281f43e2dad09d04cf704
Author: Jingsong Lee <[email protected]>
AuthorDate: Sun Jul 12 07:37:36 2026 +0800
[core][flink] Publish primary-key vector indexes on write (#8563)
Activates primary-key vector index maintenance in the write path on top
of #8549 and #8562. Complete Level 1 compact files become ANN sources,
and bucket-local ANN additions and removals are published with the same
compact increment as their source data transition.
---
.../benchmark/WriteRestoreScanBenchmark.java | 1 +
.../java/org/apache/paimon/KeyValueFileStore.java | 16 +++
.../org/apache/paimon/index/IndexFileHandler.java | 17 +++
.../compact/LookupMergeTreeCompactRewriter.java | 6 +
.../paimon/operation/AbstractFileStoreWrite.java | 50 +++++++-
.../paimon/operation/BaseAppendFileStoreWrite.java | 10 +-
.../apache/paimon/operation/FileStoreWrite.java | 7 +-
.../paimon/operation/FileSystemWriteRestore.java | 15 ++-
.../paimon/operation/KeyValueFileStoreWrite.java | 3 +
.../paimon/operation/MemoryFileStoreWrite.java | 3 +
.../org/apache/paimon/operation/RestoreFiles.java | 12 +-
.../org/apache/paimon/operation/WriteRestore.java | 3 +-
.../postpone/PostponeBucketFileStoreWrite.java | 2 +-
.../LookupMergeTreeCompactRewriterTest.java | 90 +++++++++++++++
.../operation/FileSystemWriteRestoreTest.java | 66 +++++++++++
.../operation/PrimaryKeyVectorIndexWriteTest.java | 126 +++++++++++++++++++++
.../sink/coordinator/CoordinatedWriteRestore.java | 9 +-
.../sink/coordinator/ScanCoordinationRequest.java | 9 +-
.../sink/coordinator/ScanCoordinationResponse.java | 26 ++++-
.../sink/coordinator/TableWriteCoordinator.java | 17 ++-
.../coordinator/TableWriteCoordinatorTest.java | 55 ++++++++-
.../flink/source/TestChangelogDataReadWrite.java | 1 +
22 files changed, 524 insertions(+), 20 deletions(-)
diff --git
a/paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/WriteRestoreScanBenchmark.java
b/paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/WriteRestoreScanBenchmark.java
index 9edc73ed6d..a7dfcdeb0f 100644
---
a/paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/WriteRestoreScanBenchmark.java
+++
b/paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/WriteRestoreScanBenchmark.java
@@ -400,6 +400,7 @@ public class WriteRestoreScanBenchmark extends
TableBenchmark {
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
index a1ae650b43..d7e195418b 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.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.SchemaManager;
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 class KeyValueFileStore extends
AbstractFileStore<KeyValue> {
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 class KeyValueFileStore extends
AbstractFileStore<KeyValue> {
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
index 92909a618d..799aa3ef66 100644
--- 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 class IndexFileHandler {
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/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupMergeTreeCompactRewriter.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupMergeTreeCompactRewriter.java
index f566aca1c8..6976d214b3 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupMergeTreeCompactRewriter.java
+++
b/paimon-core/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 class LookupMergeTreeCompactRewriter<T> extends
ChangelogMergeTreeRewrite
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 @@ public class LookupMergeTreeCompactRewriter<T> extends
ChangelogMergeTreeRewrite
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
index fb1db4e841..1014ddd118 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,6 +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.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 @@ public abstract class AbstractFileStoreWrite<T> implements
FileStoreWrite<T> {
FileStoreScan scan,
@Nullable DynamicBucketIndexMaintainer.Factory dbMaintainerFactory,
@Nullable BucketedDvMaintainer.Factory dvMaintainerFactory,
+ @Nullable BucketedVectorIndexMaintainer.Factory
vectorIndexMaintainerFactory,
String tableName,
CoreOptions options,
RowType partitionType) {
@@ -117,11 +120,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();
}
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 abstract class AbstractFileStoreWrite<T> implements
FileStoreWrite<T> {
.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 abstract class AbstractFileStoreWrite<T> implements
FileStoreWrite<T> {
writerContainer.writer.maxSequenceNumber(),
writerContainer.dynamicBucketMaintainer,
writerContainer.deletionVectorsMaintainer,
+ writerContainer.vectorIndexMaintainer,
increment));
}
}
@@ -407,6 +441,7 @@ public abstract class AbstractFileStoreWrite<T> implements
FileStoreWrite<T> {
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 abstract class AbstractFileStoreWrite<T> implements
FileStoreWrite<T> {
? 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 abstract class AbstractFileStoreWrite<T> implements
FileStoreWrite<T> {
firstNonNull(restored.totalBuckets(), numBuckets),
indexMaintainer,
dvMaintainer,
+ vectorIndexMaintainer,
previousSnapshot == null ? null : previousSnapshot.id());
}
@@ -558,7 +602,8 @@ public abstract class AbstractFileStoreWrite<T> implements
FileStoreWrite<T> {
partition,
bucket,
dbMaintainerFactory != null,
- dvMaintainerFactory != null);
+ dvMaintainerFactory != null,
+ vectorIndexMaintainerFactory != null);
} catch (RuntimeException e) {
throw new RuntimeException(
String.format(
@@ -621,6 +666,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;
protected final long baseSnapshotId;
protected long lastModifiedCommitIdentifier;
@@ -629,11 +675,13 @@ public abstract class AbstractFileStoreWrite<T>
implements FileStoreWrite<T> {
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
index 574143192e..0e4eeb9e4e 100644
---
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 abstract class BaseAppendFileStoreWrite extends
MemoryFileStoreWrite<Inte
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
index 24cb9c1287..5875840053 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,6 +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.io.DataFileMeta;
import org.apache.paimon.memory.MemoryPoolFactory;
import org.apache.paimon.metrics.MetricRegistry;
@@ -150,6 +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;
protected final CommitIncrement commitIncrement;
protected State(
@@ -162,6 +164,7 @@ public interface FileStoreWrite<T> extends
Restorable<List<FileStoreWrite.State<
long maxSequenceNumber,
@Nullable DynamicBucketIndexMaintainer indexMaintainer,
@Nullable BucketedDvMaintainer deletionVectorsMaintainer,
+ @Nullable BucketedVectorIndexMaintainer vectorIndexMaintainer,
CommitIncrement commitIncrement) {
this.partition = partition;
this.bucket = bucket;
@@ -172,13 +175,14 @@ public interface FileStoreWrite<T> extends
Restorable<List<FileStoreWrite.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 interface FileStoreWrite<T> extends
Restorable<List<FileStoreWrite.State<
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
index e7faf2a245..46e174cc21 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
@@ -67,7 +67,8 @@ public class FileSystemWriteRestore implements WriteRestore {
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 class FileSystemWriteRestore implements WriteRestore {
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
index 1bbbfd2c30..df2faf2dc8 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,6 +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.io.DataFileMeta;
import org.apache.paimon.io.KeyValueFileReaderFactory;
import org.apache.paimon.io.KeyValueFileWriterFactory;
@@ -96,6 +97,7 @@ public class KeyValueFileStoreWrite extends
MemoryFileStoreWrite<KeyValue> {
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 class KeyValueFileStoreWrite extends
MemoryFileStoreWrite<KeyValue> {
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
index 06263d4ec6..2a178d83cd 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,6 +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.io.cache.CacheManager;
import org.apache.paimon.memory.HeapMemorySegmentPool;
import org.apache.paimon.memory.MemoryOwner;
@@ -66,12 +67,14 @@ public abstract class MemoryFileStoreWrite<T> extends
AbstractFileStoreWrite<T>
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);
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 8005bda291..0887149819 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,18 +34,21 @@ 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;
public RestoreFiles(
@Nullable Snapshot snapshot,
@Nullable Integer totalBuckets,
@Nullable List<DataFileMeta> dataFiles,
@Nullable IndexFileMeta dynamicBucketIndex,
- @Nullable List<IndexFileMeta> deleteVectorsIndex) {
+ @Nullable List<IndexFileMeta> deleteVectorsIndex,
+ @Nullable List<IndexFileMeta> vectorIndexPayloads) {
this.snapshot = snapshot;
this.totalBuckets = totalBuckets;
this.dataFiles = dataFiles;
this.dynamicBucketIndex = dynamicBucketIndex;
this.deleteVectorsIndex = deleteVectorsIndex;
+ this.vectorIndexPayloads = vectorIndexPayloads;
}
@Nullable
@@ -73,7 +76,12 @@ public class RestoreFiles {
return deleteVectorsIndex;
}
+ @Nullable
+ public List<IndexFileMeta> vectorIndexPayloads() {
+ return vectorIndexPayloads;
+ }
+
public static RestoreFiles empty() {
- return new RestoreFiles(null, null, null, null, null);
+ return new RestoreFiles(null, null, null, null, null, null);
}
}
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 5d4e335571..284d22575a 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
@@ -35,7 +35,8 @@ public interface WriteRestore {
BinaryRow partition,
int bucket,
boolean scanDynamicBucketIndex,
- boolean scanDeleteVectorsIndex);
+ boolean scanDeleteVectorsIndex,
+ boolean scanVectorIndexPayloads);
@Nullable
static Integer extractDataFiles(List<ManifestEntry> entries,
List<DataFileMeta> dataFiles) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/postpone/PostponeBucketFileStoreWrite.java
b/paimon-core/src/main/java/org/apache/paimon/postpone/PostponeBucketFileStoreWrite.java
index 9fa8afcd67..43c51d7ee7 100644
---
a/paimon-core/src/main/java/org/apache/paimon/postpone/PostponeBucketFileStoreWrite.java
+++
b/paimon-core/src/main/java/org/apache/paimon/postpone/PostponeBucketFileStoreWrite.java
@@ -95,7 +95,7 @@ public class PostponeBucketFileStoreWrite extends
MemoryFileStoreWrite<KeyValue>
CoreOptions options,
String tableName,
@Nullable Integer writeId) {
- super(snapshotManager, scan, options, partitionType, null, null,
tableName);
+ super(snapshotManager, scan, options, partitionType, null, null, null,
tableName);
this.fileIO = fileIO;
this.pathFactory = pathFactory;
this.mfFactory = mfFactory;
diff --git
a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/LookupMergeTreeCompactRewriterTest.java
b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/LookupMergeTreeCompactRewriterTest.java
new file mode 100644
index 0000000000..5c85b37df3
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/LookupMergeTreeCompactRewriterTest.java
@@ -0,0 +1,90 @@
+/*
+ * 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.mergetree.compact;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.FileReaderFactory;
+import org.apache.paimon.io.KeyValueFileWriterFactory;
+import org.apache.paimon.manifest.FileSource;
+import org.apache.paimon.mergetree.LookupLevels;
+import org.apache.paimon.mergetree.MergeSorter;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.stats.SimpleStats;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+import java.util.Collections;
+import java.util.Comparator;
+
+import static
org.apache.paimon.mergetree.compact.ChangelogMergeTreeRewriter.UpgradeStrategy.CHANGELOG_WITH_REWRITE;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+
+/** Tests for {@link LookupMergeTreeCompactRewriter}. */
+class LookupMergeTreeCompactRewriterTest {
+
+ @ParameterizedTest
+ @EnumSource(
+ value = CoreOptions.MergeEngine.class,
+ names = {"DEDUPLICATE", "PARTIAL_UPDATE"})
+ void testVectorIndexForcesLevelZeroRewrite(CoreOptions.MergeEngine
mergeEngine) {
+ Options rawOptions = new Options();
+ rawOptions.set(CoreOptions.MERGE_ENGINE, mergeEngine);
+ rawOptions.set(CoreOptions.PK_VECTOR_INDEX_COLUMNS, "embedding");
+ CoreOptions options = new CoreOptions(rawOptions);
+ LookupMergeTreeCompactRewriter<Object> rewriter =
+ new LookupMergeTreeCompactRewriter<>(
+ 2,
+ mergeEngine,
+ mock(LookupLevels.class),
+ mock(FileReaderFactory.class),
+ mock(KeyValueFileWriterFactory.class),
+ Comparator.comparingInt(row -> row.getInt(0)),
+ null,
+ mock(MergeFunctionFactory.class),
+ mock(MergeSorter.class),
+
mock(LookupMergeTreeCompactRewriter.MergeFunctionWrapperFactory.class),
+ false,
+ null,
+ options,
+ null);
+
+ assertThat(rewriter.upgradeStrategy(2,
levelZeroFile())).isEqualTo(CHANGELOG_WITH_REWRITE);
+ }
+
+ private static DataFileMeta levelZeroFile() {
+ return DataFileMeta.forAppend(
+ "data-1",
+ 100,
+ 10,
+ SimpleStats.EMPTY_STATS,
+ 0,
+ 0,
+ 1,
+ Collections.emptyList(),
+ null,
+ FileSource.APPEND,
+ null,
+ null,
+ null,
+ null);
+ }
+}
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
new file mode 100644
index 0000000000..401439be91
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/FileSystemWriteRestoreTest.java
@@ -0,0 +1,66 @@
+/*
+ * 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.operation;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.index.IndexFileHandler;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+
+import static org.apache.paimon.data.BinaryRow.EMPTY_ROW;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/** Tests for {@link FileSystemWriteRestore}. */
+class FileSystemWriteRestoreTest {
+
+ @Test
+ void testRestoreVectorIndexPayloadsWithoutDirectory() {
+ Snapshot snapshot = mock(Snapshot.class);
+ SnapshotManager snapshotManager = mock(SnapshotManager.class);
+
when(snapshotManager.latestSnapshotFromFileSystem()).thenReturn(snapshot);
+
+ FileStoreScan scan = mock(FileStoreScan.class);
+ FileStoreScan.Plan plan = mock(FileStoreScan.Plan.class);
+ when(scan.withSnapshot(snapshot)).thenReturn(scan);
+ when(scan.withPartitionBucket(EMPTY_ROW, 0)).thenReturn(scan);
+ when(scan.plan()).thenReturn(plan);
+ when(plan.files()).thenReturn(Collections.emptyList());
+
+ IndexFileMeta ann = new IndexFileMeta("test-vector-ann", "ann", 1, 1,
null, null, null);
+ IndexFileHandler indexFileHandler = mock(IndexFileHandler.class);
+ when(indexFileHandler.scanSourceIndexes(snapshot, EMPTY_ROW, 0))
+ .thenReturn(Collections.singletonList(ann));
+
+ FileSystemWriteRestore restore =
+ new FileSystemWriteRestore(
+ new CoreOptions(new HashMap<>()), snapshotManager,
scan, indexFileHandler);
+
+ RestoreFiles restored = restore.restoreFiles(EMPTY_ROW, 0, false,
false, true);
+
+ assertThat(restored.vectorIndexPayloads()).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
new file mode 100644
index 0000000000..d2feef4c25
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/PrimaryKeyVectorIndexWriteTest.java
@@ -0,0 +1,126 @@
+/*
+ * 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.operation;
+
+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.pkvector.BucketedVectorIndexMaintainer;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.CommitIncrement;
+import org.apache.paimon.utils.RecordWriter;
+import org.apache.paimon.utils.SnapshotManager;
+
+import org.junit.jupiter.api.Test;
+
+import javax.annotation.Nullable;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+import java.util.function.Function;
+
+import static org.apache.paimon.data.BinaryRow.EMPTY_ROW;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.same;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/** Tests primary-key vector index publication through {@link
AbstractFileStoreWrite}. */
+class PrimaryKeyVectorIndexWriteTest {
+
+ @Test
+ void testPublishesVectorChangesWithCompactDataTransition() throws
Exception {
+ DataIncrement dataIncrement = DataIncrement.emptyIncrement();
+ CompactIncrement compactIncrement = CompactIncrement.emptyIncrement();
+ RecordWriter<String> writer = mock(RecordWriter.class);
+ when(writer.prepareCommit(false))
+ .thenReturn(new CommitIncrement(dataIncrement,
compactIncrement, null));
+
+ IndexFileMeta added = mock(IndexFileMeta.class);
+ IndexFileMeta deleted = mock(IndexFileMeta.class);
+ BucketedVectorIndexMaintainer.VectorIndexIncrement vectorIncrement =
+ mock(BucketedVectorIndexMaintainer.VectorIndexIncrement.class);
+
when(vectorIncrement.newIndexFiles()).thenReturn(Collections.singletonList(added));
+
when(vectorIncrement.deletedIndexFiles()).thenReturn(Collections.singletonList(deleted));
+ BucketedVectorIndexMaintainer.VectorIndexCommit vectorCommit =
+ mock(BucketedVectorIndexMaintainer.VectorIndexCommit.class);
+ when(vectorCommit.appendIncrement()).thenReturn(Optional.empty());
+
when(vectorCommit.compactIncrement()).thenReturn(Optional.of(vectorIncrement));
+ BucketedVectorIndexMaintainer maintainer =
mock(BucketedVectorIndexMaintainer.class);
+ when(maintainer.prepareCommit(same(dataIncrement),
same(compactIncrement)))
+ .thenReturn(vectorCommit);
+
+ TestingFileStoreWrite write = new TestingFileStoreWrite();
+ write.install(writer, maintainer);
+
+ CommitMessageImpl message = (CommitMessageImpl)
write.prepareCommit(false, 1).get(0);
+
+
assertThat(message.compactIncrement().newIndexFiles()).containsExactly(added);
+
assertThat(message.compactIncrement().deletedIndexFiles()).containsExactly(deleted);
+ verify(maintainer).prepareCommit(same(dataIncrement),
same(compactIncrement));
+ }
+
+ private static class TestingFileStoreWrite extends
AbstractFileStoreWrite<String> {
+
+ private TestingFileStoreWrite() {
+ super(
+ mock(SnapshotManager.class),
+ mock(FileStoreScan.class),
+ null,
+ null,
+ null,
+ "test-table",
+ new CoreOptions(new HashMap<>()),
+ RowType.of());
+ }
+
+ 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));
+ }
+
+ @Override
+ protected Function<WriterContainer<String>, Boolean>
createWriterCleanChecker() {
+ return writer -> false;
+ }
+
+ @Override
+ protected RecordWriter<String> createWriter(
+ BinaryRow partition,
+ int bucket,
+ List<DataFileMeta> restoreFiles,
+ long restoredMaxSeqNumber,
+ @Nullable CommitIncrement restoreIncrement,
+ ExecutorService compactExecutor,
+ @Nullable BucketedDvMaintainer deletionVectorsMaintainer,
+ boolean ignorePreviousFiles) {
+ throw new UnsupportedOperationException();
+ }
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CoordinatedWriteRestore.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CoordinatedWriteRestore.java
index 31438cfbde..a9b3c3c9a2 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CoordinatedWriteRestore.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/CoordinatedWriteRestore.java
@@ -71,13 +71,15 @@ public class CoordinatedWriteRestore implements
WriteRestore {
BinaryRow partition,
int bucket,
boolean scanDynamicBucketIndex,
- boolean scanDeleteVectorsIndex) {
+ boolean scanDeleteVectorsIndex,
+ boolean scanVectorIndexPayloads) {
ScanCoordinationRequest coordinationRequest =
new ScanCoordinationRequest(
serializeBinaryRow(partition),
bucket,
scanDynamicBucketIndex,
- scanDeleteVectorsIndex);
+ scanDeleteVectorsIndex,
+ scanVectorIndexPayloads);
try {
byte[] requestContent = serializeObject(coordinationRequest);
Integer nextPageToken = null;
@@ -106,7 +108,8 @@ public class CoordinatedWriteRestore implements
WriteRestore {
response.totalBuckets(),
response.extractDataFiles(),
response.extractDynamicBucketIndex(),
- response.extractDeletionVectorsIndex());
+ response.extractDeletionVectorsIndex(),
+ response.extractVectorIndexPayloads());
} catch (IOException
| ExecutionException
| InterruptedException
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/ScanCoordinationRequest.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/ScanCoordinationRequest.java
index 81eb686753..b37c00a0ff 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/ScanCoordinationRequest.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/ScanCoordinationRequest.java
@@ -29,16 +29,19 @@ public class ScanCoordinationRequest implements
CoordinationRequest {
private final int bucket;
private final boolean scanDynamicBucketIndex;
private final boolean scanDeleteVectorsIndex;
+ private final boolean scanVectorIndexPayloads;
public ScanCoordinationRequest(
byte[] partition,
int bucket,
boolean scanDynamicBucketIndex,
- boolean scanDeleteVectorsIndex) {
+ boolean scanDeleteVectorsIndex,
+ boolean scanVectorIndexPayloads) {
this.partition = partition;
this.bucket = bucket;
this.scanDynamicBucketIndex = scanDynamicBucketIndex;
this.scanDeleteVectorsIndex = scanDeleteVectorsIndex;
+ this.scanVectorIndexPayloads = scanVectorIndexPayloads;
}
public byte[] partition() {
@@ -56,4 +59,8 @@ public class ScanCoordinationRequest implements
CoordinationRequest {
public boolean scanDeleteVectorsIndex() {
return scanDeleteVectorsIndex;
}
+
+ public boolean scanVectorIndexPayloads() {
+ return scanVectorIndexPayloads;
+ }
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/ScanCoordinationResponse.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/ScanCoordinationResponse.java
index 46e411e919..7fbc9150ee 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/ScanCoordinationResponse.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/ScanCoordinationResponse.java
@@ -42,13 +42,15 @@ public class ScanCoordinationResponse implements
CoordinationResponse {
@Nullable private final List<byte[]> dataFiles;
@Nullable private final byte[] dynamicBucketIndex;
@Nullable private final List<byte[]> deleteVectorsIndex;
+ @Nullable private final List<byte[]> vectorIndexPayloads;
public ScanCoordinationResponse(
@Nullable Snapshot snapshot,
@Nullable Integer totalBuckets,
@Nullable List<DataFileMeta> dataFiles,
@Nullable IndexFileMeta dynamicBucketIndex,
- @Nullable List<IndexFileMeta> deleteVectorsIndex)
+ @Nullable List<IndexFileMeta> deleteVectorsIndex,
+ @Nullable List<IndexFileMeta> vectorIndexPayloads)
throws IOException {
this.snapshot = snapshot;
this.totalBuckets = totalBuckets;
@@ -78,6 +80,15 @@ public class ScanCoordinationResponse implements
CoordinationResponse {
} else {
this.deleteVectorsIndex = null;
}
+
+ if (vectorIndexPayloads != null) {
+ this.vectorIndexPayloads = new
ArrayList<>(vectorIndexPayloads.size());
+ for (IndexFileMeta indexFile : vectorIndexPayloads) {
+
this.vectorIndexPayloads.add(indexSerializer.serializeToBytes(indexFile));
+ }
+ } else {
+ this.vectorIndexPayloads = null;
+ }
}
@Nullable
@@ -123,4 +134,17 @@ public class ScanCoordinationResponse implements
CoordinationResponse {
}
return metas;
}
+
+ @Nullable
+ public List<IndexFileMeta> extractVectorIndexPayloads() throws IOException
{
+ if (vectorIndexPayloads == null) {
+ return null;
+ }
+ IndexFileMetaSerializer serializer = new IndexFileMetaSerializer();
+ List<IndexFileMeta> metas = new
ArrayList<>(vectorIndexPayloads.size());
+ for (byte[] file : vectorIndexPayloads) {
+ metas.add(serializer.deserializeFromBytes(file));
+ }
+ return metas;
+ }
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/TableWriteCoordinator.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/TableWriteCoordinator.java
index 1cff11eeff..ac7d5475ee 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/TableWriteCoordinator.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/TableWriteCoordinator.java
@@ -118,7 +118,8 @@ public class TableWriteCoordinator {
throws IOException {
if (snapshot == null) {
return new PagedCoordinationResponse(
- serializeObject(new ScanCoordinationResponse(null, null,
null, null, null)),
+ serializeObject(
+ new ScanCoordinationResponse(null, null, null,
null, null, null)),
null);
}
@@ -161,7 +162,7 @@ public class TableWriteCoordinator {
public synchronized ScanCoordinationResponse scan(ScanCoordinationRequest
request)
throws IOException {
if (snapshot == null) {
- return new ScanCoordinationResponse(null, null, null, null, null);
+ return new ScanCoordinationResponse(null, null, null, null, null,
null);
}
BinaryRow partition = deserializeBinaryRow(request.partition());
@@ -183,8 +184,18 @@ public class TableWriteCoordinator {
indexFileHandler.scan(snapshot, DELETION_VECTORS_INDEX,
partition, bucket);
}
+ List<IndexFileMeta> vectorIndexPayloads = null;
+ if (request.scanVectorIndexPayloads()) {
+ vectorIndexPayloads = indexFileHandler.scanSourceIndexes(snapshot,
partition, bucket);
+ }
+
return new ScanCoordinationResponse(
- snapshot, totalBuckets, restoreFiles, dynamicBucketIndex,
deleteVectorsIndex);
+ snapshot,
+ totalBuckets,
+ restoreFiles,
+ dynamicBucketIndex,
+ deleteVectorsIndex,
+ vectorIndexPayloads);
}
public synchronized long latestCommittedIdentifier(String user) {
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/TableWriteCoordinatorTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/TableWriteCoordinatorTest.java
index 16367204b6..7909580876 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/TableWriteCoordinatorTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/TableWriteCoordinatorTest.java
@@ -24,12 +24,20 @@ import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.flink.FlinkConnectorOptions;
import org.apache.paimon.fs.Path;
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.index.pkvector.PkVectorSourceFile;
+import org.apache.paimon.index.pkvector.PkVectorSourceMeta;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.DataIncrement;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.options.MemorySize;
import org.apache.paimon.options.Options;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.TableTestBase;
+import org.apache.paimon.table.sink.CommitMessageImpl;
+import org.apache.paimon.table.sink.TableCommitImpl;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.utils.SegmentsCache;
@@ -39,6 +47,7 @@ import org.junit.jupiter.params.provider.ValueSource;
import java.lang.reflect.Field;
import java.time.Duration;
+import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;
@@ -75,12 +84,52 @@ class TableWriteCoordinatorTest extends TableTestBase {
// scan should scan snapshot 2
ScanCoordinationRequest request =
- new ScanCoordinationRequest(serializeBinaryRow(EMPTY_ROW), 0,
false, false);
+ new ScanCoordinationRequest(serializeBinaryRow(EMPTY_ROW), 0,
false, false, false);
ScanCoordinationResponse scan = coordinator.scan(request);
assertThat(scan.snapshot().id()).isEqualTo(latest.id());
assertThat(scan.extractDataFiles().size()).isEqualTo(initSnapshot ? 2
: 1);
}
+ @Test
+ public void testScanVectorIndexPayloads() throws Exception {
+ Identifier identifier = new Identifier("db", "table");
+ Schema schema = Schema.newBuilder().column("f0",
DataTypes.INT()).build();
+ catalog.createDatabase("db", false);
+ catalog.createTable(identifier, schema, false);
+ FileStoreTable table = getTable(identifier);
+ write(table, GenericRow.of(1));
+
+ PkVectorSourceMeta sourceMeta =
+ new PkVectorSourceMeta(
+ Collections.singletonList(new
PkVectorSourceFile("data", 1)));
+ IndexFileMeta ann =
+ new IndexFileMeta(
+ "test-vector-ann",
+ "ann",
+ 1,
+ 1,
+ new GlobalIndexMeta(0, 0, 0, null, new byte[0],
sourceMeta.serialize()),
+ null);
+ try (TableCommitImpl commit = table.newCommit("vector-index")) {
+ commit.commit(
+ 100,
+ Collections.singletonList(
+ new CommitMessageImpl(
+ EMPTY_ROW,
+ 0,
+ 1,
+
DataIncrement.indexIncrement(Collections.singletonList(ann)),
+ CompactIncrement.emptyIncrement())));
+ }
+
+ TableWriteCoordinator coordinator = new TableWriteCoordinator(table);
+ ScanCoordinationRequest request =
+ new ScanCoordinationRequest(serializeBinaryRow(EMPTY_ROW), 0,
false, false, true);
+ ScanCoordinationResponse scan = coordinator.scan(request);
+
+ assertThat(scan.extractVectorIndexPayloads()).containsExactly(ann);
+ }
+
@Test
public void testPrefetchManifestsWarmsCache() throws Exception {
Identifier identifier = new Identifier("db", "table");
@@ -120,7 +169,7 @@ class TableWriteCoordinatorTest extends TableTestBase {
// scan results remain correct after warming
ScanCoordinationRequest request =
- new ScanCoordinationRequest(serializeBinaryRow(EMPTY_ROW), 0,
false, false);
+ new ScanCoordinationRequest(serializeBinaryRow(EMPTY_ROW), 0,
false, false, false);
ScanCoordinationResponse scan = coordinator.scan(request);
assertThat(scan.snapshot().id()).isEqualTo(table.latestSnapshot().get().id());
assertThat(scan.extractDataFiles().size()).isEqualTo(2);
@@ -170,7 +219,7 @@ class TableWriteCoordinatorTest extends TableTestBase {
// manifest, proving the bucket filter is active (and leaving the
stale bucket state on the
// shared scan)
ScanCoordinationRequest request =
- new ScanCoordinationRequest(serializeBinaryRow(EMPTY_ROW), 0,
false, false);
+ new ScanCoordinationRequest(serializeBinaryRow(EMPTY_ROW), 0,
false, false, false);
coordinator.scan(request);
long filteredCacheBytes = table.getManifestCache().totalCacheBytes();
assertThat(filteredCacheBytes).isGreaterThan(0);
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/TestChangelogDataReadWrite.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/TestChangelogDataReadWrite.java
index 8114ac17eb..21edabbb42 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/TestChangelogDataReadWrite.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/TestChangelogDataReadWrite.java
@@ -195,6 +195,7 @@ public class TestChangelogDataReadWrite {
null, // not used, we only create an empty writer
null,
null,
+ null,
options,
EXTRACTOR,
tablePath.getName());
