fix

JingsongLi · JingsongLi · commit 14e368353cd8 · 2026-03-11T20:23:14.000+08:00
diff --git a/paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/SimpleLsmKvDb.java b/paimon-common/src/main/java/org/apache/paimon/lookup/sort/db/SimpleLsmKvDb.java
@@ -40,8 +40,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.ConcurrentSkipListMap;
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.TreeMap;
 
 /**
  * A simple LSM-Tree based KV database built on top of {@link SortLookupStoreFactory}.
@@ -80,10 +79,10 @@ public class SimpleLsmKvDb implements Closeable {
     private final LsmCompactor compactor;
 
     /** Active MemTable: key -> value bytes (empty byte[] = tombstone). */
-    private ConcurrentSkipListMap<MemorySlice, byte[]> memTable;
+    private TreeMap<MemorySlice, byte[]> memTable;
 
     /** Estimated size of the current MemTable in bytes. */
-    private final AtomicLong memTableSize;
+    private long memTableSize;
 
     /**
      * Multi-level SST file storage. Each level contains a list of {@link SstFileMetadata} ordered
@@ -95,8 +94,8 @@ public class SimpleLsmKvDb implements Closeable {
     /** Cached readers for SST files, keyed by file path. Lazily populated on first lookup. */
     private final Map<File, SortLookupStoreReader> readerCache;
 
-    private final AtomicLong fileSequence;
-    private volatile boolean closed;
+    private long fileSequence;
+    private boolean closed;
 
     private SimpleLsmKvDb(
             File dataDirectory,
@@ -109,14 +108,14 @@ private SimpleLsmKvDb(
         this.storeFactory = storeFactory;
         this.keyComparator = keyComparator;
         this.memTableFlushThreshold = memTableFlushThreshold;
-        this.memTable = new ConcurrentSkipListMap<>(keyComparator);
-        this.memTableSize = new AtomicLong(0);
+        this.memTable = new TreeMap<>(keyComparator);
+        this.memTableSize = 0;
         this.levels = new ArrayList<>();
         for (int i = 0; i < MAX_LEVELS; i++) {
             this.levels.add(new ArrayList<SstFileMetadata>());
         }
         this.readerCache = new HashMap<>();
-        this.fileSequence = new AtomicLong(0);
+        this.fileSequence = 0;
         this.closed = false;
         this.compactor =
                 new LsmCompactor(
@@ -182,7 +181,7 @@ public void put(byte[] key, byte[] value) throws IOException {
         if (oldValue != null) {
             delta -= (key.length + oldValue.length);
         }
-        memTableSize.addAndGet(delta);
+        memTableSize += delta;
         maybeFlushMemTable();
     }
 
@@ -209,7 +208,7 @@ public void delete(byte[] key) throws IOException {
         if (oldValue != null) {
             delta -= (key.length + oldValue.length);
         }
-        memTableSize.addAndGet(delta);
+        memTableSize += delta;
         maybeFlushMemTable();
     }
 
@@ -246,32 +245,30 @@ public byte[] get(byte[] key) throws IOException {
         }
 
         // 2. Search each level from L0 to Lmax
-        synchronized (levels) {
-            for (int level = 0; level < MAX_LEVELS; level++) {
-                List<SstFileMetadata> levelFiles = levels.get(level);
-                if (levelFiles.isEmpty()) {
-                    continue;
-                }
+        for (int level = 0; level < MAX_LEVELS; level++) {
+            List<SstFileMetadata> levelFiles = levels.get(level);
+            if (levelFiles.isEmpty()) {
+                continue;
+            }
 
-                if (level == 0) {
-                    // L0: files may have overlapping keys, search newest-first
-                    for (SstFileMetadata meta : levelFiles) {
-                        if (!meta.mightContainKey(wrappedKey, keyComparator)) {
-                            continue;
-                        }
-                        byte[] value = lookupInFile(meta.getFile(), key);
-                        if (value != null) {
-                            return LsmCompactor.isTombstone(value) ? null : value;
-                        }
+            if (level == 0) {
+                // L0: files may have overlapping keys, search newest-first
+                for (SstFileMetadata meta : levelFiles) {
+                    if (!meta.mightContainKey(wrappedKey, keyComparator)) {
+                        continue;
                     }
-                } else {
-                    // L1+: files have non-overlapping key ranges, binary search
-                    SstFileMetadata target = findFileForKey(levelFiles, wrappedKey);
-                    if (target != null) {
-                        byte[] value = lookupInFile(target.getFile(), key);
-                        if (value != null) {
-                            return LsmCompactor.isTombstone(value) ? null : value;
-                        }
+                    byte[] value = lookupInFile(meta.getFile(), key);
+                    if (value != null) {
+                        return LsmCompactor.isTombstone(value) ? null : value;
+                    }
+                }
+            } else {
+                // L1+: files have non-overlapping key ranges, binary search
+                SstFileMetadata target = findFileForKey(levelFiles, wrappedKey);
+                if (target != null) {
+                    byte[] value = lookupInFile(target.getFile(), key);
+                    if (value != null) {
+                        return LsmCompactor.isTombstone(value) ? null : value;
                     }
                 }
             }
@@ -303,24 +300,20 @@ public void flush() throws IOException {
             return;
         }
 
-        ConcurrentSkipListMap<MemorySlice, byte[]> snapshot = memTable;
-        memTable = new ConcurrentSkipListMap<>(keyComparator);
-        memTableSize.set(0);
+        TreeMap<MemorySlice, byte[]> snapshot = memTable;
+        memTable = new TreeMap<>(keyComparator);
+        memTableSize = 0;
 
         SstFileMetadata metadata = writeMemTableToSst(snapshot);
 
-        synchronized (levels) {
-            levels.get(0).add(0, metadata);
-        }
+        levels.get(0).add(0, metadata);
 
         LOG.info(
                 "Flushed MemTable to L0 SST file: {}, entries: {}",
                 metadata.getFile().getName(),
                 snapshot.size());
 
-        synchronized (levels) {
-            compactor.maybeCompact(levels, MAX_LEVELS, this::newSstFile);
-        }
+        compactor.maybeCompact(levels, MAX_LEVELS, this::newSstFile);
     }
 
     /**
@@ -330,9 +323,7 @@ public void flush() throws IOException {
      */
     public void compact() throws IOException {
         ensureOpen();
-        synchronized (levels) {
-            compactor.fullCompact(levels, MAX_LEVELS, this::newSstFile);
-        }
+        compactor.fullCompact(levels, MAX_LEVELS, this::newSstFile);
     }
 
     // -------------------------------------------------------------------------
@@ -348,14 +339,12 @@ public void close() throws IOException {
 
         // Flush remaining MemTable data to L0
         if (!memTable.isEmpty()) {
-            ConcurrentSkipListMap<MemorySlice, byte[]> snapshot = memTable;
-            memTable = new ConcurrentSkipListMap<>(keyComparator);
-            memTableSize.set(0);
+            TreeMap<MemorySlice, byte[]> snapshot = memTable;
+            memTable = new TreeMap<>(keyComparator);
+            memTableSize = 0;
 
             SstFileMetadata metadata = writeMemTableToSst(snapshot);
-            synchronized (levels) {
-                levels.get(0).add(0, metadata);
-            }
+            levels.get(0).add(0, metadata);
         }
 
         // Close all cached readers
@@ -373,53 +362,47 @@ public void close() throws IOException {
 
     /** Return the total number of SST files across all levels. */
     public int getSstFileCount() {
-        synchronized (levels) {
-            int count = 0;
-            for (List<SstFileMetadata> levelFiles : levels) {
-                count += levelFiles.size();
-            }
-            return count;
+        int count = 0;
+        for (List<SstFileMetadata> levelFiles : levels) {
+            count += levelFiles.size();
         }
+        return count;
     }
 
     /** Return the number of SST files at a specific level. */
     public int getLevelFileCount(int level) {
-        synchronized (levels) {
-            if (level < 0 || level >= MAX_LEVELS) {
-                return 0;
-            }
-            return levels.get(level).size();
+        if (level < 0 || level >= MAX_LEVELS) {
+            return 0;
         }
+        return levels.get(level).size();
     }
 
     /** Return the estimated MemTable size in bytes. */
     public long getMemTableSize() {
-        return memTableSize.get();
+        return memTableSize;
     }
 
     /** Return a human-readable summary of file counts per level. */
     public String getLevelStats() {
-        synchronized (levels) {
-            StringBuilder sb = new StringBuilder();
-            for (int i = 0; i < MAX_LEVELS; i++) {
-                int count = levels.get(i).size();
-                if (count > 0) {
-                    if (sb.length() > 0) {
-                        sb.append(", ");
-                    }
-                    sb.append("L").append(i).append("=").append(count);
+        StringBuilder sb = new StringBuilder();
+        for (int i = 0; i < MAX_LEVELS; i++) {
+            int count = levels.get(i).size();
+            if (count > 0) {
+                if (sb.length() > 0) {
+                    sb.append(", ");
                 }
+                sb.append("L").append(i).append("=").append(count);
             }
-            return sb.length() == 0 ? "empty" : sb.toString();
         }
+        return sb.length() == 0 ? "empty" : sb.toString();
     }
 
     // -------------------------------------------------------------------------
     //  Internal Helpers
     // -------------------------------------------------------------------------
 
     private void maybeFlushMemTable() throws IOException {
-        if (memTableSize.get() >= memTableFlushThreshold) {
+        if (memTableSize >= memTableFlushThreshold) {
             flush();
         }
     }
@@ -452,7 +435,7 @@ private SstFileMetadata findFileForKey(List<SstFileMetadata> sortedFiles, Memory
         return null;
     }
 
-    private SstFileMetadata writeMemTableToSst(ConcurrentSkipListMap<MemorySlice, byte[]> data)
+    private SstFileMetadata writeMemTableToSst(TreeMap<MemorySlice, byte[]> data)
             throws IOException {
         File sstFile = newSstFile();
         SortLookupStoreWriter writer = storeFactory.createWriter(sstFile, null);
@@ -477,7 +460,7 @@ private SstFileMetadata writeMemTableToSst(ConcurrentSkipListMap<MemorySlice, by
     }
 
     private File newSstFile() {
-        long sequence = fileSequence.getAndIncrement();
+        long sequence = fileSequence++;
         return new File(dataDirectory, String.format("sst-%06d.db", sequence));
     }
 
