diff --git a/paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/LocalKvDb.java b/paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/LocalKvDb.java index 6875fc2c31a2..2fde5118b798 100644 --- a/paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/LocalKvDb.java +++ b/paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/LocalKvDb.java @@ -47,6 +47,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.atomic.AtomicLong; import java.util.function.LongFunction; +import java.util.function.Predicate; import static org.apache.paimon.utils.Preconditions.checkArgument; @@ -112,6 +113,7 @@ public class LocalKvDb implements Closeable { private final long maxSstFileSize; private final LsmLevels levels; private final LsmCompactor compaction; + @Nullable private final MergeOperator mergeOperator; /** Active MemTable: key -> value bytes (empty byte[] = tombstone). */ private TreeMap memTable; @@ -134,6 +136,8 @@ private LocalKvDb( long maxSstFileSize, int level0FileNumCompactTrigger, int sizeRatio, + @Nullable Predicate expiredValuePredicate, + @Nullable MergeOperator mergeOperator, @Nullable ExecutorService compactionExecutor) { this.dataDirectory = dataDirectory; this.uuid = UUID.randomUUID().toString(); @@ -148,6 +152,7 @@ private LocalKvDb( this.readerCache = new HashMap<>(); this.fileSequence = new AtomicLong(); this.closed = false; + this.mergeOperator = mergeOperator; LsmCompactor.CompactorFactory compactorFactory = fileDeleter -> new UniversalCompactor( @@ -157,6 +162,8 @@ private LocalKvDb( maxSstFileSize, level0FileNumCompactTrigger, sizeRatio, + expiredValuePredicate, + mergeOperator, fileDeleter); this.compaction = compactionExecutor == null @@ -611,25 +618,34 @@ private SstFileMetadata writeMemTableToSst(TreeMap data) throws IOException { File sstFile = newSstFile(); SortLookupStoreWriter writer = - storeFactory.createWriter(sstFile, bloomFilterBuilderFactory.apply(data.size())); - MemorySlice minKey = null; - MemorySlice maxKey = null; - long tombstoneCount = 0; + storeFactory.createWriter( + sstFile, + bloomFilterBuilderFactory.apply( + mergeOperator == null ? data.size() : UNKNOWN_NUM_ENTRIES)); + MemorySlice[] boundaryKeys = new MemorySlice[2]; + long[] tombstoneCount = new long[1]; + RecordCombiningWriter combiningWriter = + new RecordCombiningWriter( + mergeOperator, + (key, value) -> { + writer.put(key.copyBytes(), value); + if (boundaryKeys[0] == null) { + boundaryKeys[0] = key; + } + boundaryKeys[1] = key; + if (isTombstone(value)) { + tombstoneCount[0]++; + } + }); try { for (Map.Entry entry : data.entrySet()) { - writer.put(entry.getKey().copyBytes(), entry.getValue()); - if (minKey == null) { - minKey = entry.getKey(); - } - maxKey = entry.getKey(); - if (isTombstone(entry.getValue())) { - tombstoneCount++; - } + combiningWriter.put(entry.getKey(), entry.getValue()); } + combiningWriter.finish(); } finally { writer.close(); } - return new SstFileMetadata(sstFile, minKey, maxKey, tombstoneCount, 0); + return new SstFileMetadata(sstFile, boundaryKeys[0], boundaryKeys[1], tombstoneCount[0], 0); } private File newSstFile() { @@ -643,6 +659,19 @@ private void ensureOpen() { } } + /** + * Operator for combining adjacent logical records while flushing and compacting SST files. + * + *

MemTable writes remain independent so merge-heavy workloads do not pay repeated + * read-modify-write costs. The first record's key is retained for the combined value. + */ + public interface MergeOperator { + + boolean canMerge(MemorySlice firstKey, MemorySlice nextKey); + + byte[] merge(List values) throws IOException; + } + // ------------------------------------------------------------------------- // Builder // ------------------------------------------------------------------------- @@ -661,6 +690,8 @@ public static class Builder { private Comparator keyComparator = MemorySlice::compareTo; private boolean bloomFilterEnabled = true; private double bloomFilterFpp = 0.1; + @Nullable private Predicate expiredValuePredicate; + @Nullable private MergeOperator mergeOperator; @Nullable private ExecutorService compactionExecutor; Builder(File dataDirectory) { @@ -729,6 +760,22 @@ public Builder bloomFilterFpp(double bloomFilterFpp) { return this; } + /** + * Set a predicate which identifies expired stored values during compaction. Partial + * compaction converts matching values into tombstones to avoid resurrecting older values; + * full compaction drops them. + */ + public Builder expiredValuePredicate(@Nullable Predicate expiredValuePredicate) { + this.expiredValuePredicate = expiredValuePredicate; + return this; + } + + /** Set an operator for combining adjacent logical records in generated SST files. */ + public Builder mergeOperator(@Nullable MergeOperator mergeOperator) { + this.mergeOperator = mergeOperator; + return this; + } + /** * Set the executor for asynchronous compaction. Compaction runs synchronously when no * executor is configured. The executor remains owned by the caller and is not shut down @@ -783,6 +830,8 @@ public LocalKvDb build() { maxSstFileSize, level0FileNumCompactTrigger, sizeRatio, + expiredValuePredicate, + mergeOperator, compactionExecutor); } } diff --git a/paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/RecordCombiningWriter.java b/paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/RecordCombiningWriter.java new file mode 100644 index 000000000000..b9c488112125 --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/RecordCombiningWriter.java @@ -0,0 +1,97 @@ +/* + * 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.lookup.sort.db; + +import org.apache.paimon.memory.MemorySlice; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; + +import static org.apache.paimon.lookup.sort.db.LocalKvDb.isTombstone; + +/** Combines adjacent records before forwarding them to an SST writer. */ +final class RecordCombiningWriter { + + @Nullable private final LocalKvDb.MergeOperator mergeOperator; + private final RecordConsumer consumer; + + @Nullable private MemorySlice pendingKey; + private final List pendingValues; + + RecordCombiningWriter( + @Nullable LocalKvDb.MergeOperator mergeOperator, RecordConsumer consumer) { + this.mergeOperator = mergeOperator; + this.consumer = consumer; + this.pendingValues = new ArrayList<>(); + } + + void put(MemorySlice key, byte[] value) throws IOException { + if (mergeOperator == null) { + consumer.accept(key, value); + return; + } + + if (isTombstone(value)) { + flushPending(); + consumer.accept(key, value); + return; + } + + if (pendingKey == null) { + startGroup(key, value); + } else if (mergeOperator.canMerge(pendingKey, key)) { + pendingValues.add(value); + } else { + flushPending(); + startGroup(key, value); + } + } + + void finish() throws IOException { + flushPending(); + } + + private void startGroup(MemorySlice key, byte[] value) { + pendingKey = MemorySlice.wrap(key.copyBytes()); + pendingValues.add(value); + } + + private void flushPending() throws IOException { + if (pendingKey == null) { + return; + } + + byte[] value = + pendingValues.size() == 1 + ? pendingValues.get(0) + : mergeOperator.merge(pendingValues); + consumer.accept(pendingKey, value); + pendingKey = null; + pendingValues.clear(); + } + + /** Callback receiving one combined record. */ + interface RecordConsumer { + + void accept(MemorySlice key, byte[] value) throws IOException; + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/UniversalCompactor.java b/paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/UniversalCompactor.java index fc6c8b8328ea..594e61076ae4 100644 --- a/paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/UniversalCompactor.java +++ b/paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/UniversalCompactor.java @@ -29,6 +29,8 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import javax.annotation.Nullable; + import java.io.File; import java.io.IOException; import java.util.ArrayList; @@ -40,6 +42,7 @@ import java.util.PriorityQueue; import java.util.Set; import java.util.function.LongFunction; +import java.util.function.Predicate; import static org.apache.paimon.lookup.sort.db.LocalKvDb.UNKNOWN_NUM_ENTRIES; import static org.apache.paimon.lookup.sort.db.LocalKvDb.isTombstone; @@ -69,6 +72,7 @@ public class UniversalCompactor { private static final Logger LOG = LoggerFactory.getLogger(UniversalCompactor.class); + private static final byte[] TOMBSTONE = new byte[0]; private final Comparator keyComparator; private final SortLookupStoreFactory storeFactory; @@ -76,6 +80,8 @@ public class UniversalCompactor { private final long maxOutputFileSize; private final int level0FileNumCompactTrigger; private final int sizeRatioPercent; + @Nullable private final Predicate expiredValuePredicate; + @Nullable private final LocalKvDb.MergeOperator mergeOperator; private final FileDeleter fileDeleter; public UniversalCompactor( @@ -85,6 +91,8 @@ public UniversalCompactor( long maxOutputFileSize, int level0FileNumCompactTrigger, int sizeRatioPercent, + @Nullable Predicate expiredValuePredicate, + @Nullable LocalKvDb.MergeOperator mergeOperator, FileDeleter fileDeleter) { this.keyComparator = keyComparator; this.storeFactory = storeFactory; @@ -92,6 +100,8 @@ public UniversalCompactor( this.maxOutputFileSize = maxOutputFileSize; this.level0FileNumCompactTrigger = level0FileNumCompactTrigger; this.sizeRatioPercent = sizeRatioPercent; + this.expiredValuePredicate = expiredValuePredicate; + this.mergeOperator = mergeOperator; this.fileDeleter = fileDeleter; } @@ -178,7 +188,7 @@ public void fullCompact( List> levels, int maxLevels, FileSupplier fileSupplier) throws IOException { List sortedRuns = collectSortedRuns(levels, maxLevels); - if (sortedRuns.size() <= 1) { + if (sortedRuns.isEmpty() || (sortedRuns.size() == 1 && expiredValuePredicate == null)) { return; } @@ -317,7 +327,9 @@ private MergeResult mergeSortedRuns( for (List group : mergedGroups) { if (group.size() == 1) { SstFileMetadata singleFile = group.get(0); - boolean canSkip = !dropTombstones || !singleFile.hasTombstones(); + boolean canSkip = + expiredValuePredicate == null + && (!dropTombstones || !singleFile.hasTombstones()); if (canSkip) { SstFileMetadata promoted = singleFile.withLevel(outputLevel); outputFiles.add(promoted); @@ -483,7 +495,9 @@ private List mergeFileGroup( return Integer.compare(b.sequence, a.sequence); }); - SortLookupStoreWriter currentWriter = null; + CompactionSstOutput output = + new CompactionSstOutput(result, fileSupplier, outputLevel, dropTombstones); + Throwable failure = null; try { for (int seq = 0; seq < orderedFiles.size(); seq++) { SortLookupStoreReader reader = @@ -495,11 +509,6 @@ private List mergeFileGroup( minHeap.add(source.currentEntry()); } } - File currentSstFile = null; - MemorySlice currentFileMinKey = null; - MemorySlice currentFileMaxKey = null; - long currentBatchSize = 0; - long currentTombstoneCount = 0; MemorySlice previousKey = null; while (!minHeap.isEmpty()) { @@ -518,67 +527,32 @@ private List mergeFileGroup( minHeap.add(entry.source.currentEntry()); } - if (dropTombstones && isTombstone(entry.value)) { - continue; - } - - if (currentWriter == null) { - currentSstFile = fileSupplier.newSstFile(); - currentWriter = - storeFactory.createWriter( - currentSstFile, - bloomFilterBuilderFactory.apply(UNKNOWN_NUM_ENTRIES)); - currentFileMinKey = entry.key; - currentBatchSize = 0; - currentTombstoneCount = 0; - } - - currentWriter.put(entry.key.copyBytes(), entry.value); - currentFileMaxKey = entry.key; - currentBatchSize += entry.key.length() + entry.value.length; - if (isTombstone(entry.value)) { - currentTombstoneCount++; - } - - if (currentBatchSize >= maxOutputFileSize) { - currentWriter.close(); - result.add( - new SstFileMetadata( - currentSstFile, - currentFileMinKey, - currentFileMaxKey, - currentTombstoneCount, - outputLevel)); - currentWriter = null; - currentSstFile = null; - currentFileMinKey = null; - currentFileMaxKey = null; - } + output.put(entry.key, entry.value); } - if (currentWriter != null) { - currentWriter.close(); - result.add( - new SstFileMetadata( - currentSstFile, - currentFileMinKey, - currentFileMaxKey, - currentTombstoneCount, - outputLevel)); - } - } catch (IOException | RuntimeException e) { - // Close the in-progress writer on failure to avoid resource leak - if (currentWriter != null) { - try { - currentWriter.close(); - } catch (IOException suppressed) { - e.addSuppressed(suppressed); - } - } + output.finish(); + } catch (IOException | RuntimeException | Error e) { + failure = e; + output.abort(e); throw e; } finally { + IOException closeFailure = null; for (SortLookupStoreReader reader : openReaders) { - reader.close(); + try { + reader.close(); + } catch (IOException e) { + if (closeFailure == null) { + closeFailure = e; + } else { + closeFailure.addSuppressed(e); + } + } + } + if (closeFailure != null) { + if (failure == null) { + throw closeFailure; + } + failure.addSuppressed(closeFailure); } } @@ -625,6 +599,105 @@ private static void clearLevelsOfRuns( } } + /** Writes compacted records, combining adjacent values before enforcing output file sizes. */ + private final class CompactionSstOutput { + + private final List result; + private final FileSupplier fileSupplier; + private final int outputLevel; + private final boolean dropTombstones; + private final RecordCombiningWriter combiningWriter; + + @Nullable private SortLookupStoreWriter currentWriter; + @Nullable private File currentSstFile; + @Nullable private MemorySlice currentFileMinKey; + @Nullable private MemorySlice currentFileMaxKey; + private long currentBatchSize; + private long currentTombstoneCount; + + private CompactionSstOutput( + List result, + FileSupplier fileSupplier, + int outputLevel, + boolean dropTombstones) { + this.result = result; + this.fileSupplier = fileSupplier; + this.outputLevel = outputLevel; + this.dropTombstones = dropTombstones; + this.combiningWriter = + new RecordCombiningWriter(mergeOperator, this::writeCombinedRecord); + } + + private void put(MemorySlice key, byte[] value) throws IOException { + combiningWriter.put(key, value); + } + + private void finish() throws IOException { + combiningWriter.finish(); + closeCurrentWriter(); + } + + private void abort(Throwable failure) { + if (currentWriter != null) { + try { + currentWriter.close(); + } catch (IOException suppressed) { + failure.addSuppressed(suppressed); + } + currentWriter = null; + } + } + + private void writeCombinedRecord(MemorySlice key, byte[] value) throws IOException { + // Evaluate expiration after combining so a TTL-aware merge can refresh the result. + boolean expired = expiredValuePredicate != null && expiredValuePredicate.test(value); + if (dropTombstones && (isTombstone(value) || expired)) { + return; + } + byte[] outputValue = expired ? TOMBSTONE : value; + + if (currentWriter == null) { + currentSstFile = fileSupplier.newSstFile(); + currentWriter = + storeFactory.createWriter( + currentSstFile, + bloomFilterBuilderFactory.apply(UNKNOWN_NUM_ENTRIES)); + currentFileMinKey = key; + currentBatchSize = 0; + currentTombstoneCount = 0; + } + + currentWriter.put(key.copyBytes(), outputValue); + currentFileMaxKey = key; + currentBatchSize += key.length() + outputValue.length; + if (isTombstone(outputValue)) { + currentTombstoneCount++; + } + + if (currentBatchSize >= maxOutputFileSize) { + closeCurrentWriter(); + } + } + + private void closeCurrentWriter() throws IOException { + if (currentWriter == null) { + return; + } + currentWriter.close(); + result.add( + new SstFileMetadata( + currentSstFile, + currentFileMinKey, + currentFileMaxKey, + currentTombstoneCount, + outputLevel)); + currentWriter = null; + currentSstFile = null; + currentFileMinKey = null; + currentFileMaxKey = null; + } + } + /** Delete old SST files from the merged sorted runs, skipping files that were preserved. */ private void deleteOldFiles(List oldRuns, Set skippedFiles) { for (SortedRun run : oldRuns) { diff --git a/paimon-common/src/test/java/org/apache/paimon/lookup/sort/db/LocalKvDbTest.java b/paimon-common/src/test/java/org/apache/paimon/lookup/sort/db/LocalKvDbTest.java index 751a209d314e..345cb47566be 100644 --- a/paimon-common/src/test/java/org/apache/paimon/lookup/sort/db/LocalKvDbTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/lookup/sort/db/LocalKvDbTest.java @@ -32,6 +32,7 @@ import java.io.RandomAccessFile; import java.util.AbstractMap; import java.util.ArrayList; +import java.util.Collections; import java.util.Comparator; import java.util.List; import java.util.Map; @@ -313,6 +314,80 @@ public void testCompactionRemovesTombstones() throws IOException { } } + @Test + public void testPartialCompactionDoesNotResurrectValueFilteredByExpiration() + throws IOException { + File directory = new File(tempDir.toFile(), "expiration-compaction-db"); + try (LocalKvDb db = + LocalKvDb.builder(directory) + .memTableFlushThreshold(1024) + .maxSstFileSize(1024) + .blockSize(128) + .level0FileNumCompactTrigger(2) + .compressOptions(new CompressOptions("none", 1)) + .expiredValuePredicate(value -> "expired".equals(new String(value, UTF_8))) + .build()) { + db.bulkLoad(Collections.singletonList(entry("key", "old-value")).iterator(), 1); + + putString(db, "key", "expired"); + db.flush(); + putString(db, "other-key", "other-value"); + db.flush(); + + Assertions.assertNull(getString(db, "key")); + Assertions.assertEquals("other-value", getString(db, "other-key")); + Assertions.assertEquals(1, db.getLevelFileCount(LocalKvDb.MAX_LEVELS - 1)); + + db.compact(); + Assertions.assertNull(getString(db, "key")); + Assertions.assertEquals("other-value", getString(db, "other-key")); + } + } + + @Test + public void testCompactionMergesBeforeFilteringExpiredValues() throws IOException { + File directory = new File(tempDir.toFile(), "expiration-merge-db"); + LocalKvDb.MergeOperator mergeOperator = + new LocalKvDb.MergeOperator() { + @Override + public boolean canMerge(MemorySlice firstKey, MemorySlice nextKey) { + return firstKey.readByte(0) == nextKey.readByte(0); + } + + @Override + public byte[] merge(List values) { + StringBuilder merged = new StringBuilder(); + for (byte[] value : values) { + if (merged.length() > 0) { + merged.append('+'); + } + merged.append(new String(value, UTF_8)); + } + return merged.toString().getBytes(UTF_8); + } + }; + try (LocalKvDb db = + LocalKvDb.builder(directory) + .memTableFlushThreshold(1024) + .maxSstFileSize(1024) + .blockSize(128) + .level0FileNumCompactTrigger(100) + .compressOptions(new CompressOptions("none", 1)) + .expiredValuePredicate(value -> "expired".equals(new String(value, UTF_8))) + .mergeOperator(mergeOperator) + .build()) { + putString(db, "a-0", "expired"); + db.flush(); + putString(db, "a-1", "live"); + db.flush(); + + db.compact(); + + Assertions.assertEquals("expired+live", getString(db, "a-0")); + Assertions.assertNull(getString(db, "a-1")); + } + } + @Test public void testManualCompaction() throws IOException { try (LocalKvDb db = createDb()) {