diff --git a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/BucketedDvMaintainer.java b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/BucketedDvMaintainer.java index 788ea1cce85d..14a6ec8475de 100644 --- a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/BucketedDvMaintainer.java +++ b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/BucketedDvMaintainer.java @@ -87,7 +87,7 @@ public void notifyNewDeletion(String fileName, DeletionVector deletionVector) { public void mergeNewDeletion(String fileName, DeletionVector deletionVector) { DeletionVector old = deletionVectors.get(fileName); if (old != null) { - deletionVector.merge(old); + deletionVector = DeletionVector.mergeVectors(deletionVector, old); } deletionVectors.put(fileName, deletionVector); modified = true; diff --git a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/DeletionVector.java b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/DeletionVector.java index d3f7eed1cede..e8c9cba9d42a 100644 --- a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/DeletionVector.java +++ b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/DeletionVector.java @@ -197,6 +197,27 @@ static DeletionVector deserializeFromBytes(byte[] bytes) { } } + /** + * Merges {@code stored} into {@code fresh} and returns the vector holding the union. + * + *

{@code deletion-vectors.bitmap64} can be flipped on an existing table, so a freshly + * created vector and a previously stored one may not share the same bitmap width. When they + * differ, the {@link BitmapDeletionVector} side is promoted to {@link Bitmap64DeletionVector} + * so the merge runs at the wider format instead of {@link #merge} throwing on the type + * mismatch. A bitmap64 result stays readable under a bitmap32 table because vectors are + * dispatched by magic number on read. + */ + static DeletionVector mergeVectors(DeletionVector fresh, DeletionVector stored) { + if (fresh instanceof Bitmap64DeletionVector && stored instanceof BitmapDeletionVector) { + stored = Bitmap64DeletionVector.fromBitmapDeletionVector((BitmapDeletionVector) stored); + } else if (fresh instanceof BitmapDeletionVector + && stored instanceof Bitmap64DeletionVector) { + fresh = Bitmap64DeletionVector.fromBitmapDeletionVector((BitmapDeletionVector) fresh); + } + fresh.merge(stored); + return fresh; + } + /** Interface to create {@link DeletionVector}. */ interface Factory { Optional create(String fileName) throws IOException; diff --git a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/append/AppendDeleteFileMaintainer.java b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/append/AppendDeleteFileMaintainer.java index 93ea8a7676a8..496bbc846d0e 100644 --- a/paimon-core/src/main/java/org/apache/paimon/deletionvectors/append/AppendDeleteFileMaintainer.java +++ b/paimon-core/src/main/java/org/apache/paimon/deletionvectors/append/AppendDeleteFileMaintainer.java @@ -131,7 +131,8 @@ public DeletionFile notifyRemovedDeletionVector(String dataFile) { public void notifyNewDeletionVector(String dataFile, DeletionVector deletionVector) { DeletionFile previous = notifyRemovedDeletionVector(dataFile); if (previous != null) { - deletionVector.merge(dvIndexFile.readDeletionVector(previous)); + DeletionVector stored = dvIndexFile.readDeletionVector(previous); + deletionVector = DeletionVector.mergeVectors(deletionVector, stored); } deletionVectors.put(dataFile, deletionVector); } diff --git a/paimon-core/src/test/java/org/apache/paimon/deletionvectors/BucketedDvMaintainerTest.java b/paimon-core/src/test/java/org/apache/paimon/deletionvectors/BucketedDvMaintainerTest.java index 018d55e93304..2dd3c44199a9 100644 --- a/paimon-core/src/test/java/org/apache/paimon/deletionvectors/BucketedDvMaintainerTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/deletionvectors/BucketedDvMaintainerTest.java @@ -283,6 +283,52 @@ public void testReadAndWriteMixedDv(boolean bitmap64) { assertThat(dvs.get("f3").getCardinality()).isEqualTo(2); } + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testMergeNewDeletionAcrossBitmap64Flip(boolean initialBitmap64) { + // write a stored dv under one bitmap64 setting, then flip the option and merge a fresh + // vector of the opposite type into it: the merge must not crash on the type mismatch + initIndexHandler(initialBitmap64); + BucketedDvMaintainer.Factory factory1 = BucketedDvMaintainer.factory(fileHandler); + BucketedDvMaintainer dvMaintainer1 = factory1.create(partition, 0, new HashMap<>()); + dvMaintainer1.notifyNewDeletion("f1", 1); + dvMaintainer1.notifyNewDeletion("f1", 3); + commitIndexFile(dvMaintainer1.writeDeletionVectorsIndex().get()); + + initIndexHandler(!initialBitmap64); + BucketedDvMaintainer.Factory factory2 = BucketedDvMaintainer.factory(fileHandler); + List indexFiles = + fileHandler.scan( + table.latestSnapshot().get(), DELETION_VECTORS_INDEX, partition, 0); + BucketedDvMaintainer dvMaintainer2 = factory2.create(partition, 0, indexFiles); + + DeletionVector fresh = createDeletionVector(!initialBitmap64); + fresh.delete(5); + dvMaintainer2.mergeNewDeletion("f1", fresh); + + DeletionVector merged = dvMaintainer2.deletionVectorOf("f1").get(); + assertThat(merged.isDeleted(1)).isTrue(); + assertThat(merged.isDeleted(3)).isTrue(); + assertThat(merged.isDeleted(5)).isTrue(); + assertThat(merged.getCardinality()).isEqualTo(3); + } + + private void commitIndexFile(IndexFileMeta file) { + CommitMessage commitMessage = + new CommitMessageImpl( + partition, + 0, + 1, + DataIncrement.emptyIncrement(), + new CompactIncrement( + Collections.emptyList(), + Collections.emptyList(), + Collections.emptyList(), + Collections.singletonList(file), + Collections.emptyList())); + table.newBatchWriteBuilder().newCommit().commit(Collections.singletonList(commitMessage)); + } + private DeletionVector createDeletionVector(boolean bitmap64) { return bitmap64 ? new Bitmap64DeletionVector() : new BitmapDeletionVector(); } diff --git a/paimon-core/src/test/java/org/apache/paimon/deletionvectors/append/AppendDeletionFileMaintainerTest.java b/paimon-core/src/test/java/org/apache/paimon/deletionvectors/append/AppendDeletionFileMaintainerTest.java index 3a03985c885d..78fe7a2098b8 100644 --- a/paimon-core/src/test/java/org/apache/paimon/deletionvectors/append/AppendDeletionFileMaintainerTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/deletionvectors/append/AppendDeletionFileMaintainerTest.java @@ -21,6 +21,8 @@ import org.apache.paimon.CoreOptions; import org.apache.paimon.TestAppendFileStore; import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.deletionvectors.Bitmap64DeletionVector; +import org.apache.paimon.deletionvectors.BitmapDeletionVector; import org.apache.paimon.deletionvectors.DeletionVector; import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.local.LocalFileIO; @@ -32,6 +34,7 @@ import org.apache.paimon.table.sink.CommitMessageImpl; import org.apache.paimon.table.source.DeletionFile; +import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; @@ -48,6 +51,88 @@ class AppendDeletionFileMaintainerTest { @TempDir java.nio.file.Path tempDir; + @Test + public void testMergeStoredBitmap32IntoBitmap64() throws Exception { + // write DVs as bitmap32, then flip deletion-vectors.bitmap64 on: the next + // notification must merge the stored vector instead of crashing on the type + Map options = new HashMap<>(); + options.put(CoreOptions.DELETION_VECTOR_BITMAP64.key(), "false"); + TestAppendFileStore store = TestAppendFileStore.createAppendStore(tempDir, options); + + CommitMessageImpl commitMessage = + store.writeDVIndexFiles( + BinaryRow.EMPTY_ROW, + 0, + Collections.singletonMap("f1", Arrays.asList(1, 3, 5))); + store.commit(commitMessage); + + IndexPathFactory indexPathFactory = + store.pathFactory().indexFileFactory(BinaryRow.EMPTY_ROW, 0); + Map dataFileToDeletionFiles = + createDeletionFileMapFromIndexFileMetas( + indexPathFactory, commitMessage.newFilesIncrement().newIndexFiles()); + + Map flipped = new HashMap<>(); + flipped.put(CoreOptions.DELETION_VECTOR_BITMAP64.key(), "true"); + TestAppendFileStore flippedStore = TestAppendFileStore.createAppendStore(tempDir, flipped); + AppendDeleteFileMaintainer dvIFMaintainer = + flippedStore.createDVIFMaintainer(BinaryRow.EMPTY_ROW, dataFileToDeletionFiles); + + Bitmap64DeletionVector fresh = new Bitmap64DeletionVector(); + fresh.delete(7); + dvIFMaintainer.notifyNewDeletionVector("f1", fresh); + + List res = dvIFMaintainer.persist(); + assertThat(res).hasSize(2); + // the old index file is replaced by one holding the merged vector: stored 3 + // deletions plus the fresh one + assertThat(res).anyMatch(entry -> entry.kind() == FileKind.DELETE); + IndexManifestEntry added = + res.stream().filter(entry -> entry.kind() == FileKind.ADD).findAny().get(); + assertThat(added.indexFile().dvRanges()).containsKey("f1"); + assertThat(added.indexFile().dvRanges().get("f1").cardinality()).isEqualTo(4); + } + + @Test + public void testMergeStoredBitmap64IntoBitmap32() throws Exception { + // reverse of the case above: write DVs as bitmap64, then flip deletion-vectors.bitmap64 + // off. A fresh bitmap32 vector must still merge the stored bitmap64 one instead of crashing + Map options = new HashMap<>(); + options.put(CoreOptions.DELETION_VECTOR_BITMAP64.key(), "true"); + TestAppendFileStore store = TestAppendFileStore.createAppendStore(tempDir, options); + + CommitMessageImpl commitMessage = + store.writeDVIndexFiles( + BinaryRow.EMPTY_ROW, + 0, + Collections.singletonMap("f1", Arrays.asList(1, 3, 5))); + store.commit(commitMessage); + + IndexPathFactory indexPathFactory = + store.pathFactory().indexFileFactory(BinaryRow.EMPTY_ROW, 0); + Map dataFileToDeletionFiles = + createDeletionFileMapFromIndexFileMetas( + indexPathFactory, commitMessage.newFilesIncrement().newIndexFiles()); + + Map flipped = new HashMap<>(); + flipped.put(CoreOptions.DELETION_VECTOR_BITMAP64.key(), "false"); + TestAppendFileStore flippedStore = TestAppendFileStore.createAppendStore(tempDir, flipped); + AppendDeleteFileMaintainer dvIFMaintainer = + flippedStore.createDVIFMaintainer(BinaryRow.EMPTY_ROW, dataFileToDeletionFiles); + + BitmapDeletionVector fresh = new BitmapDeletionVector(); + fresh.delete(7); + dvIFMaintainer.notifyNewDeletionVector("f1", fresh); + + List res = dvIFMaintainer.persist(); + assertThat(res).hasSize(2); + assertThat(res).anyMatch(entry -> entry.kind() == FileKind.DELETE); + IndexManifestEntry added = + res.stream().filter(entry -> entry.kind() == FileKind.ADD).findAny().get(); + assertThat(added.indexFile().dvRanges()).containsKey("f1"); + assertThat(added.indexFile().dvRanges().get("f1").cardinality()).isEqualTo(4); + } + @ParameterizedTest @ValueSource(booleans = {true, false}) public void test(boolean bitmap64) throws Exception {