(paimon) branch master updated: [core] Make LocalKvDb flush failure-safe (#8869)

[lzljs3620320]

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 2e8f69a154 [core] Make LocalKvDb flush failure-safe (#8869)
2e8f69a154 is described below

commit 2e8f69a15401fb8747b21eb52ec8cece3aee36fc
Author: Jingsong Lee <jingsongl...@gmail.com>
AuthorDate: Tue Jul 28 08:19:14 2026 +0800

    [core] Make LocalKvDb flush failure-safe (#8869)
---
 .../apache/paimon/lookup/sort/db/LocalKvDb.java    | 26 +++++++++++++++-------
 .../paimon/lookup/sort/db/LocalKvDbTest.java       | 26 ++++++++++++++++++++++
 2 files changed, 44 insertions(+), 8 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 6875fc2c31..f1a0f43960 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
@@ -471,12 +471,10 @@ public class LocalKvDb implements Closeable {
 
     private void flushMemTable() throws IOException {
         TreeMap<MemorySlice, byte[]> snapshot = memTable;
-        memTable = new TreeMap<>(keyComparator);
-        memTableSize = 0;
-
         SstFileMetadata metadata = writeMemTableToSst(snapshot);
-
         levels.addLevelZeroFile(metadata);
+        memTable = new TreeMap<>(keyComparator);
+        memTableSize = 0;
 
         LOG.info(
                 "Flushed MemTable to L0 SST file: {}, entries: {}",
@@ -610,12 +608,14 @@ public class LocalKvDb implements Closeable {
     private SstFileMetadata writeMemTableToSst(TreeMap<MemorySlice, byte[]> 
data)
             throws IOException {
         File sstFile = newSstFile();
-        SortLookupStoreWriter writer =
-                storeFactory.createWriter(sstFile, 
bloomFilterBuilderFactory.apply(data.size()));
+        SortLookupStoreWriter writer = null;
         MemorySlice minKey = null;
         MemorySlice maxKey = null;
         long tombstoneCount = 0;
         try {
+            writer =
+                    storeFactory.createWriter(
+                            sstFile, 
bloomFilterBuilderFactory.apply(data.size()));
             for (Map.Entry<MemorySlice, byte[]> entry : data.entrySet()) {
                 writer.put(entry.getKey().copyBytes(), entry.getValue());
                 if (minKey == null) {
@@ -626,10 +626,20 @@ public class LocalKvDb implements Closeable {
                     tombstoneCount++;
                 }
             }
-        } finally {
             writer.close();
+            writer = null;
+            return new SstFileMetadata(sstFile, minKey, maxKey, 
tombstoneCount, 0);
+        } catch (IOException | RuntimeException e) {
+            if (writer != null) {
+                try {
+                    writer.close();
+                } catch (IOException suppressed) {
+                    e.addSuppressed(suppressed);
+                }
+            }
+            deleteFileQuietly(sstFile);
+            throw e;
         }
-        return new SstFileMetadata(sstFile, minKey, maxKey, tombstoneCount, 0);
     }
 
     private File newSstFile() {
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 751a209d31..f3be0c8047 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
@@ -30,6 +30,7 @@ import org.junit.jupiter.api.io.TempDir;
 import java.io.File;
 import java.io.IOException;
 import java.io.RandomAccessFile;
+import java.nio.file.Files;
 import java.util.AbstractMap;
 import java.util.ArrayList;
 import java.util.Comparator;
@@ -391,6 +392,31 @@ public class LocalKvDbTest {
         Assertions.assertEquals(1, db.getSstFileCount());
     }
 
+    @Test
+    public void testFlushFailureKeepsMemTableForRetry() throws IOException {
+        File dbDir = new File(tempDir.toFile(), "flush-failure-db");
+        try (LocalKvDb db =
+                LocalKvDb.builder(dbDir)
+                        .memTableFlushThreshold(1024 * 1024)
+                        .blockSize(256)
+                        .compressOptions(new CompressOptions("none", 1))
+                        .build()) {
+            putString(db, "key", "value");
+
+            Files.delete(dbDir.toPath());
+            Files.createFile(dbDir.toPath());
+            Assertions.assertThrows(IOException.class, db::flush);
+            Assertions.assertEquals("value", getString(db, "key"));
+            Assertions.assertTrue(db.getMemTableSize() > 0);
+
+            Files.delete(dbDir.toPath());
+            Files.createDirectories(dbDir.toPath());
+            db.flush();
+            Assertions.assertEquals("value", getString(db, "key"));
+            Assertions.assertEquals(0, db.getMemTableSize());
+        }
+    }
+
     @Test
     public void testClosedDbThrowsException() throws IOException {
         LocalKvDb db = createDb();