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 e599991d83 [core] Support LocalKvDb merge and TTL compaction (#8872)
e599991d83 is described below
commit e599991d8341fec89618d3168342505ad15b8f3c
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Jul 28 10:24:54 2026 +0800
[core] Support LocalKvDb merge and TTL compaction (#8872)
---
.../apache/paimon/lookup/sort/db/LocalKvDb.java | 88 ++++++-
.../lookup/sort/db/RecordCombiningWriter.java | 107 +++++++++
.../paimon/lookup/sort/db/UniversalCompactor.java | 264 +++++++++++++++------
.../paimon/lookup/sort/db/LocalKvDbTest.java | 169 +++++++++++++
4 files changed, 537 insertions(+), 91 deletions(-)
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 1bf4aa8dfa..f319ee9463 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
@@ -53,6 +53,7 @@ import java.util.UUID;
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;
@@ -118,6 +119,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<MemorySlice, byte[]> memTable;
@@ -142,6 +144,8 @@ public class LocalKvDb implements Closeable {
long maxSstFileSize,
int level0FileNumCompactTrigger,
int sizeRatio,
+ @Nullable Predicate<byte[]> expiredValuePredicate,
+ @Nullable MergeOperator mergeOperator,
@Nullable ExecutorService compactionExecutor) {
this.dataDirectory = dataDirectory;
this.uuid = UUID.randomUUID().toString();
@@ -158,6 +162,7 @@ public class LocalKvDb implements Closeable {
this.activeBulkLoadWriter = null;
this.openRangeIterators = 0;
this.closed = false;
+ this.mergeOperator = mergeOperator;
LsmCompactor.CompactorFactory compactorFactory =
fileDeleter ->
new UniversalCompactor(
@@ -167,6 +172,8 @@ public class LocalKvDb implements Closeable {
maxSstFileSize,
level0FileNumCompactTrigger,
sizeRatio,
+ expiredValuePredicate,
+ mergeOperator,
fileDeleter);
this.compaction =
compactionExecutor == null
@@ -807,26 +814,23 @@ public class LocalKvDb implements Closeable {
throws IOException {
File sstFile = newSstFile();
SortLookupStoreWriter writer = null;
- MemorySlice minKey = null;
- MemorySlice maxKey = null;
- long tombstoneCount = 0;
try {
writer =
storeFactory.createWriter(
- sstFile,
bloomFilterBuilderFactory.apply(data.size()));
+ sstFile,
+ bloomFilterBuilderFactory.apply(
+ mergeOperator == null ? data.size() :
UNKNOWN_NUM_ENTRIES));
+ SstMetadataWriter output = new SstMetadataWriter(writer);
+ RecordCombiningWriter combiningWriter =
+ new RecordCombiningWriter(mergeOperator, output);
for (Map.Entry<MemorySlice, byte[]> 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();
writer.close();
writer = null;
- return new SstFileMetadata(sstFile, minKey, maxKey,
tombstoneCount, 0);
+ return new SstFileMetadata(
+ sstFile, output.minKey, output.maxKey,
output.tombstoneCount, 0);
} catch (IOException | RuntimeException e) {
if (writer != null) {
try {
@@ -840,6 +844,31 @@ public class LocalKvDb implements Closeable {
}
}
+ private static final class SstMetadataWriter implements
RecordCombiningWriter.RecordConsumer {
+
+ private final SortLookupStoreWriter writer;
+
+ @Nullable private MemorySlice minKey;
+ @Nullable private MemorySlice maxKey;
+ private long tombstoneCount;
+
+ private SstMetadataWriter(SortLookupStoreWriter writer) {
+ this.writer = writer;
+ }
+
+ @Override
+ public void accept(MemorySlice key, byte[] value) throws IOException {
+ writer.put(key.copyBytes(), value);
+ if (minKey == null) {
+ minKey = key;
+ }
+ maxKey = key;
+ if (isTombstone(value)) {
+ tombstoneCount++;
+ }
+ }
+ }
+
private File newSstFile() {
long sequence = fileSequence.getAndIncrement();
return new File(dataDirectory, String.format("sst-%s-%06d.db", uuid,
sequence));
@@ -1119,6 +1148,19 @@ public class LocalKvDb implements Closeable {
void accept(MemorySlice key, MemorySlice value) throws IOException;
}
+ /**
+ * Operator for combining adjacent logical records while flushing and
compacting SST files.
+ *
+ * <p>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<byte[]> values) throws IOException;
+ }
+
//
-------------------------------------------------------------------------
// Builder
//
-------------------------------------------------------------------------
@@ -1137,6 +1179,8 @@ public class LocalKvDb implements Closeable {
private Comparator<MemorySlice> keyComparator = MemorySlice::compareTo;
private boolean bloomFilterEnabled = true;
private double bloomFilterFpp = 0.1;
+ @Nullable private Predicate<byte[]> expiredValuePredicate;
+ @Nullable private MergeOperator mergeOperator;
@Nullable private ExecutorService compactionExecutor;
Builder(File dataDirectory) {
@@ -1205,6 +1249,22 @@ public class LocalKvDb implements Closeable {
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<byte[]>
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
@@ -1259,6 +1319,8 @@ public class LocalKvDb implements Closeable {
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 0000000000..edf5eedd0d
--- /dev/null
+++
b/paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/RecordCombiningWriter.java
@@ -0,0 +1,107 @@
+/*
+ * 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 {
+
+ private static final byte[] TOMBSTONE = new byte[0];
+
+ @Nullable private final LocalKvDb.MergeOperator mergeOperator;
+ private final RecordConsumer consumer;
+
+ @Nullable private MemorySlice pendingKey;
+ private final List<MemorySlice> pendingKeys;
+ private final List<byte[]> pendingValues;
+
+ RecordCombiningWriter(
+ @Nullable LocalKvDb.MergeOperator mergeOperator, RecordConsumer
consumer) {
+ this.mergeOperator = mergeOperator;
+ this.consumer = consumer;
+ this.pendingKeys = new ArrayList<>();
+ 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)) {
+ pendingKeys.add(MemorySlice.wrap(key.copyBytes()));
+ 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());
+ pendingKeys.add(pendingKey);
+ 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);
+ for (int i = 1; i < pendingKeys.size(); i++) {
+ consumer.accept(pendingKeys.get(i), TOMBSTONE);
+ }
+ pendingKey = null;
+ pendingKeys.clear();
+ 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 fc6c8b8328..b099b3eb0e 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.apache.paimon.utils.BloomFilter;
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.Map;
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 @@ import static
org.apache.paimon.lookup.sort.db.LocalKvDb.isTombstone;
public class UniversalCompactor {
private static final Logger LOG =
LoggerFactory.getLogger(UniversalCompactor.class);
+ private static final byte[] TOMBSTONE = new byte[0];
private final Comparator<MemorySlice> 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<byte[]> expiredValuePredicate;
+ @Nullable private final LocalKvDb.MergeOperator mergeOperator;
private final FileDeleter fileDeleter;
public UniversalCompactor(
@@ -85,6 +91,8 @@ public class UniversalCompactor {
long maxOutputFileSize,
int level0FileNumCompactTrigger,
int sizeRatioPercent,
+ @Nullable Predicate<byte[]> expiredValuePredicate,
+ @Nullable LocalKvDb.MergeOperator mergeOperator,
FileDeleter fileDeleter) {
this.keyComparator = keyComparator;
this.storeFactory = storeFactory;
@@ -92,6 +100,8 @@ public class UniversalCompactor {
this.maxOutputFileSize = maxOutputFileSize;
this.level0FileNumCompactTrigger = level0FileNumCompactTrigger;
this.sizeRatioPercent = sizeRatioPercent;
+ this.expiredValuePredicate = expiredValuePredicate;
+ this.mergeOperator = mergeOperator;
this.fileDeleter = fileDeleter;
}
@@ -178,7 +188,7 @@ public class UniversalCompactor {
List<List<SstFileMetadata>> levels, int maxLevels, FileSupplier
fileSupplier)
throws IOException {
List<SortedRun> sortedRuns = collectSortedRuns(levels, maxLevels);
- if (sortedRuns.size() <= 1) {
+ if (sortedRuns.isEmpty() || (sortedRuns.size() == 1 &&
expiredValuePredicate == null)) {
return;
}
@@ -314,24 +324,51 @@ public class UniversalCompactor {
int skippedGroupCount = 0;
int mergedGroupCount = 0;
- for (List<SstFileMetadata> group : mergedGroups) {
- if (group.size() == 1) {
- SstFileMetadata singleFile = group.get(0);
- boolean canSkip = !dropTombstones ||
!singleFile.hasTombstones();
- if (canSkip) {
- SstFileMetadata promoted =
singleFile.withLevel(outputLevel);
- outputFiles.add(promoted);
- skippedFileSet.add(promoted.getFile());
- skippedGroupCount++;
- continue;
+ if (mergeOperator == null) {
+ for (List<SstFileMetadata> group : mergedGroups) {
+ if (group.size() == 1) {
+ SstFileMetadata singleFile = group.get(0);
+ boolean canSkip =
+ expiredValuePredicate == null
+ && (!dropTombstones ||
!singleFile.hasTombstones());
+ if (canSkip) {
+ SstFileMetadata promoted =
singleFile.withLevel(outputLevel);
+ outputFiles.add(promoted);
+ skippedFileSet.add(promoted.getFile());
+ skippedGroupCount++;
+ continue;
+ }
}
- }
- mergedGroupCount++;
- List<SstFileMetadata> groupMerged =
+ mergedGroupCount++;
+ outputFiles.addAll(
+ mergeFileGroup(
+ group,
+ dropTombstones,
+ fileSupplier,
+ fileToRunSequence,
+ outputLevel,
+ null));
+ }
+ } else {
+ CompactionSstOutput output =
+ new CompactionSstOutput(outputFiles, fileSupplier,
outputLevel, dropTombstones);
+ try {
+ for (List<SstFileMetadata> group : mergedGroups) {
+ mergedGroupCount++;
mergeFileGroup(
- group, dropTombstones, fileSupplier,
fileToRunSequence, outputLevel);
- outputFiles.addAll(groupMerged);
+ group,
+ dropTombstones,
+ fileSupplier,
+ fileToRunSequence,
+ outputLevel,
+ output);
+ }
+ output.finish();
+ } catch (IOException | RuntimeException | Error e) {
+ output.abort(e);
+ throw e;
+ }
}
outputFiles.sort((a, b) -> keyComparator.compare(a.getMinKey(),
b.getMinKey()));
@@ -452,6 +489,7 @@ public class UniversalCompactor {
* @param fileSupplier supplier for new SST file paths
* @param fileToRunSequence maps each file to its run sequence number for
dedup ordering
* @param outputLevel the level to assign to output files
+ * @param sharedOutput optional output shared across file groups to
preserve merge state
* @return the list of merged output files
*/
private List<SstFileMetadata> mergeFileGroup(
@@ -459,7 +497,8 @@ public class UniversalCompactor {
boolean dropTombstones,
FileSupplier fileSupplier,
Map<File, Integer> fileToRunSequence,
- int outputLevel)
+ int outputLevel,
+ @Nullable CompactionSstOutput sharedOutput)
throws IOException {
// Sort files by run sequence (older first) for correct dedup ordering
@@ -483,7 +522,11 @@ public class UniversalCompactor {
return Integer.compare(b.sequence, a.sequence);
});
- SortLookupStoreWriter currentWriter = null;
+ CompactionSstOutput output =
+ sharedOutput == null
+ ? new CompactionSstOutput(result, fileSupplier,
outputLevel, dropTombstones)
+ : sharedOutput;
+ Throwable failure = null;
try {
for (int seq = 0; seq < orderedFiles.size(); seq++) {
SortLookupStoreReader reader =
@@ -495,11 +538,6 @@ public class UniversalCompactor {
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 +556,34 @@ public class UniversalCompactor {
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);
- }
+ if (sharedOutput == null) {
+ 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 +630,109 @@ public class UniversalCompactor {
}
}
+ /** Writes compacted records, combining adjacent values before enforcing
output file sizes. */
+ private final class CompactionSstOutput {
+
+ private final List<SstFileMetadata> 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<SstFileMetadata> 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 tombstone = isTombstone(value);
+ boolean expired =
+ !tombstone
+ && expiredValuePredicate != null
+ && expiredValuePredicate.test(value);
+ if (dropTombstones && (tombstone || 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<SortedRun> oldRuns, Set<File>
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 0f49fba43b..a022cb2265 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
@@ -489,6 +489,175 @@ public class LocalKvDbTest {
}
}
+ @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<byte[]> 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
testCompactionMergesAcrossFileGroupsBeforeFilteringExpiration() throws
IOException {
+ File directory = new File(tempDir.toFile(),
"cross-group-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<byte[]> values) {
+ return "live".getBytes(UTF_8);
+ }
+ };
+ try (LocalKvDb db =
+ LocalKvDb.builder(directory)
+ .maxSstFileSize(1)
+ .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", "expired");
+ db.flush();
+
+ db.compact();
+
+ Assertions.assertEquals("live", getString(db, "a-0"));
+ Assertions.assertNull(getString(db, "a-1"));
+ }
+ }
+
+ @Test
+ public void testExpirationPredicateDoesNotReceiveTombstones() throws
IOException {
+ File directory = new File(tempDir.toFile(), "tombstone-expiration-db");
+ try (LocalKvDb db =
+ LocalKvDb.builder(directory)
+ .level0FileNumCompactTrigger(100)
+ .compressOptions(new CompressOptions("none", 1))
+ .expiredValuePredicate(value -> value[0] == 1)
+ .build()) {
+ db.put("key".getBytes(UTF_8), new byte[] {2});
+ db.flush();
+ db.delete("key".getBytes(UTF_8));
+ db.flush();
+
+ db.compact();
+
+ Assertions.assertNull(db.get("key".getBytes(UTF_8)));
+ }
+ }
+
+ @Test
+ public void testFlushMergeShadowsConsumedKeysInOlderRuns() throws
IOException {
+ File directory = new File(tempDir.toFile(), "flush-merge-shadow-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<byte[]> values) {
+ return (new String(values.get(0), UTF_8)
+ + "+"
+ + new String(values.get(1), UTF_8))
+ .getBytes(UTF_8);
+ }
+ };
+ try (LocalKvDb db =
+ LocalKvDb.builder(directory)
+ .level0FileNumCompactTrigger(100)
+ .compressOptions(new CompressOptions("none", 1))
+ .mergeOperator(mergeOperator)
+ .build()) {
+ List<Map.Entry<byte[], byte[]>> oldValues = new ArrayList<>();
+ oldValues.add(entry("a-1", "old-1"));
+ oldValues.add(entry("a-2", "old-2"));
+ db.bulkLoad(oldValues.iterator(), oldValues.size());
+
+ putString(db, "a-1", "new-1");
+ putString(db, "a-2", "new-2");
+ Assertions.assertEquals("new-2", getString(db, "a-2"));
+
+ db.flush();
+
+ Assertions.assertEquals("new-1+new-2", getString(db, "a-1"));
+ Assertions.assertNull(getString(db, "a-2"));
+ }
+ }
+
@Test
public void testManualCompaction() throws IOException {
try (LocalKvDb db = createDb()) {
