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 9edc73ed6dc6..a7dfcdeb0f12 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 @@ 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 index a1ae650b433a..d7e195418b32 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.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 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 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 index 92909a618da4..799aa3ef66de 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 List scan( return result; } + public List scanSourceIndexes( + Snapshot snapshot, BinaryRow partition, int bucket) { + List 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, List> scan( long snapshot, String indexType, Set 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 f566aca1c831..6976d214b364 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 extends ChangelogMergeTreeRewrite private final LookupLevels lookupLevels; private final MergeFunctionWrapperFactory wrapperFactory; private final boolean noSequenceField; + private final boolean forceRewriteLevelZeroForVectorIndex; @Nullable private final BucketedDvMaintainer dvMaintainer; private final IntFunction 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 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 index fb1db4e841a3..1014ddd11816 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.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 implements FileStoreWrite { 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 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> checkpoint() { writerContainer.writer.maxSequenceNumber(), writerContainer.dynamicBucketMaintainer, writerContainer.deletionVectorsMaintainer, + writerContainer.vectorIndexMaintainer, increment)); } } @@ -407,6 +441,7 @@ public void restore(List> 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 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 restoreFiles = restored.dataFiles(); if (restoreFiles == null) { @@ -497,6 +540,7 @@ public WriterContainer 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 { 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 index 574143192e19..0e4eeb9e4e23 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 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 index 24cb9c128703..5875840053ff 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.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 { 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 index e7faf2a24569..46e174cc21ab 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 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 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 1bbbfd2c306c..df2faf2dc800 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.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 index 06263d4ec66a..2a178d83cd04 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 @@ 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); 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 8005bda2911d..0887149819c4 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 dataFiles; private final @Nullable IndexFileMeta dynamicBucketIndex; private final @Nullable List deleteVectorsIndex; + private final @Nullable List vectorIndexPayloads; public RestoreFiles( @Nullable Snapshot snapshot, @Nullable Integer totalBuckets, @Nullable List dataFiles, @Nullable IndexFileMeta dynamicBucketIndex, - @Nullable List deleteVectorsIndex) { + @Nullable List deleteVectorsIndex, + @Nullable List 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 List deleteVectorsIndex() { return deleteVectorsIndex; } + @Nullable + public List 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 5d4e335571d1..284d22575a0f 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 @@ RestoreFiles restoreFiles( BinaryRow partition, int bucket, boolean scanDynamicBucketIndex, - boolean scanDeleteVectorsIndex); + boolean scanDeleteVectorsIndex, + boolean scanVectorIndexPayloads); @Nullable static Integer extractDataFiles(List entries, List 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 9fa8afcd674c..43c51d7ee7d7 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 PostponeBucketFileStoreWrite( 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 000000000000..5c85b37df33f --- /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 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 000000000000..401439be919e --- /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 000000000000..d2feef4c256d --- /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 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 { + + private TestingFileStoreWrite() { + super( + mock(SnapshotManager.class), + mock(FileStoreScan.class), + null, + null, + null, + "test-table", + new CoreOptions(new HashMap<>()), + RowType.of()); + } + + private void install( + RecordWriter writer, BucketedVectorIndexMaintainer vectorMaintainer) { + writers.computeIfAbsent(EMPTY_ROW, ignored -> new HashMap<>()) + .put(0, new WriterContainer<>(writer, 1, null, null, vectorMaintainer, null)); + } + + @Override + protected Function, Boolean> createWriterCleanChecker() { + return writer -> false; + } + + @Override + protected RecordWriter createWriter( + BinaryRow partition, + int bucket, + List 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 31438cfbde12..a9b3c3c9a2a7 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 RestoreFiles restoreFiles( 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 RestoreFiles restoreFiles( 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 81eb6867534f..b37c00a0ff18 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 boolean scanDynamicBucketIndex() { 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 46e411e91939..7fbc9150ee6f 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 dataFiles; @Nullable private final byte[] dynamicBucketIndex; @Nullable private final List deleteVectorsIndex; + @Nullable private final List vectorIndexPayloads; public ScanCoordinationResponse( @Nullable Snapshot snapshot, @Nullable Integer totalBuckets, @Nullable List dataFiles, @Nullable IndexFileMeta dynamicBucketIndex, - @Nullable List deleteVectorsIndex) + @Nullable List deleteVectorsIndex, + @Nullable List vectorIndexPayloads) throws IOException { this.snapshot = snapshot; this.totalBuckets = totalBuckets; @@ -78,6 +80,15 @@ public ScanCoordinationResponse( } 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 List extractDeletionVectorsIndex() throws IOException { } return metas; } + + @Nullable + public List extractVectorIndexPayloads() throws IOException { + if (vectorIndexPayloads == null) { + return null; + } + IndexFileMetaSerializer serializer = new IndexFileMetaSerializer(); + List 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 1cff11eeffc1..ac7d5475ee80 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 synchronized PagedCoordinationResponse scan(PagedCoordinationRequest requ 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 synchronized PagedCoordinationResponse scan(PagedCoordinationRequest requ 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 synchronized ScanCoordinationResponse scan(ScanCoordinationRequest reques indexFileHandler.scan(snapshot, DELETION_VECTORS_INDEX, partition, bucket); } + List 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 16367204b6c3..790958087675 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.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 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 @@ public void testLatestIdentifierAndScan(boolean initSnapshot) throws Exception { // 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 @@ public void testPrefetchManifestsWarmsCache() throws Exception { // 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 @@ public void testPrefetchWarmsAllManifestsAfterScan() throws Exception { // 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 8114ac17eb38..21edabbb42ac 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 RecordWriter createMergeTreeWriter(BinaryRow partition, int buc null, // not used, we only create an empty writer null, null, + null, options, EXTRACTOR, tablePath.getName());