[core] Support LocalKvDb merge and TTL compaction

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<MemorySlice, byte[]> memTable;
@@ -134,6 +136,8 @@ private LocalKvDb(
             long maxSstFileSize,
             int level0FileNumCompactTrigger,
             int sizeRatio,
+            @Nullable Predicate<byte[]> 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<MemorySlice, byte[]> 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<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();
         } 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.
+     *
+     * <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
     // -------------------------------------------------------------------------
@@ -661,6 +690,8 @@ public static class Builder {
         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) {
@@ -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<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
@@ -783,6 +830,8 @@ public LocalKvDb build() {
                     maxSstFileSize,
                     level0FileNumCompactTrigger,
                     sizeRatio,
+                    expiredValuePredicate,
+                    mergeOperator,
                     compactionExecutor);
         }
     }

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<byte[]> 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;
+    }
+}