From 11afeb0c5abf50a752efb33c22706c49ffbe87a0 Mon Sep 17 00:00:00 2001
From: dev-donghwan <ycdoit@naver.com>
Date: Wed, 30 Sep 2026 16:46:07 +0900
Subject: [PATCH] [core] Treat a snapshot rename that throws but succeeded as
 committed

RenamingSnapshotCommit already treats a rename that returns false as
committed when the target snapshot file exists with the expected
content, and then writes the LATEST hint. When the rename throws
instead, for example because the response of a successful object store
copy cannot be parsed, the commit fails and the retry in
FileStoreCommitImpl finds the snapshot already committed and returns
success without writing the hint. The LATEST hint then stays behind.

Apply the same check when the rename throws, and rethrow only when the
target snapshot file does not exist.
---
 .../catalog/RenamingSnapshotCommit.java       | 12 +++-
 .../catalog/RenamingSnapshotCommitTest.java   | 70 +++++++++++++++++++
 2 files changed, 81 insertions(+), 1 deletion(-)

diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/RenamingSnapshotCommit.java b/paimon-core/src/main/java/org/apache/paimon/catalog/RenamingSnapshotCommit.java
index 8a655f6076f6..b880d5c4a26d 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/RenamingSnapshotCommit.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/RenamingSnapshotCommit.java
@@ -63,7 +63,17 @@ public boolean commit(
 
         Callable<Boolean> callable =
                 () -> {
-                    boolean committed = fileIO.tryToWriteAtomic(newSnapshotPath, snapshot.toJson());
+                    boolean committed;
+                    try {
+                        committed = fileIO.tryToWriteAtomic(newSnapshotPath, snapshot.toJson());
+                    } catch (IOException e) {
+                        // the rename may have succeeded even if the file system reports an
+                        // error, check the snapshot file below
+                        if (!fileIO.exists(newSnapshotPath)) {
+                            throw e;
+                        }
+                        committed = false;
+                    }
                     if (!committed) {
                         if (!fileIO.exists(newSnapshotPath)) {
                             throw new IOException(
diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/RenamingSnapshotCommitTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/RenamingSnapshotCommitTest.java
index 4e7fa27e7b7f..d7213ff08811 100644
--- a/paimon-core/src/test/java/org/apache/paimon/catalog/RenamingSnapshotCommitTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/catalog/RenamingSnapshotCommitTest.java
@@ -101,6 +101,51 @@ public void testCommitTargetSnapshotMissing(@TempDir java.nio.file.Path tmp) thr
         assertThat(fileIO.exists(snapshotManager.snapshotPath(snapshot.id()))).isFalse();
     }
 
+    /**
+     * Simulate the object storage scenario where rename actually succeeds but throws, for example
+     * when the response of a successful copy cannot be parsed. The commit should succeed and write
+     * the LATEST hint.
+     */
+    @Test
+    public void testCommitSucceedsAndRenameThrows(@TempDir java.nio.file.Path tmp)
+            throws Exception {
+        FileIO fileIO = new ThrowingRenameLocalFileIO(true);
+        Path tablePath = new Path(tmp.toUri());
+        SnapshotManager snapshotManager = new SnapshotManager(fileIO, tablePath, null, null, null);
+
+        RenamingSnapshotCommit commit = new RenamingSnapshotCommit(snapshotManager, Lock.empty());
+
+        Snapshot snapshot = createSnapshot(3L);
+
+        boolean committed = commit.commit(null, snapshot, "main", Collections.emptyList());
+
+        assertThat(committed).isTrue();
+        Path latestHint = new Path(snapshotManager.snapshotDirectory(), HintFileUtils.LATEST);
+        assertThat(fileIO.readOverwrittenFileUtf8(latestHint).orElse(null))
+                .isEqualTo(String.valueOf(snapshot.id()));
+    }
+
+    /** When rename throws and the target snapshot file is not created, the error is rethrown. */
+    @Test
+    public void testCommitRenameThrowsAndTargetMissing(@TempDir java.nio.file.Path tmp)
+            throws Exception {
+        FileIO fileIO = new ThrowingRenameLocalFileIO(false);
+        Path tablePath = new Path(tmp.toUri());
+        SnapshotManager snapshotManager = new SnapshotManager(fileIO, tablePath, null, null, null);
+
+        RenamingSnapshotCommit commit = new RenamingSnapshotCommit(snapshotManager, Lock.empty());
+
+        Snapshot snapshot = createSnapshot(4L);
+
+        IOException ex =
+                assertThrows(
+                        IOException.class,
+                        () -> commit.commit(null, snapshot, "main", Collections.emptyList()));
+
+        assertThat(ex).hasMessageContaining(ThrowingRenameLocalFileIO.MESSAGE);
+        assertThat(fileIO.exists(snapshotManager.snapshotPath(snapshot.id()))).isFalse();
+    }
+
     private static Snapshot createSnapshot(long id) throws IOException {
         long schemaId = 1L;
         String baseManifestList = "manifest-list-base";
@@ -171,4 +216,29 @@ public boolean rename(Path src, Path dst) throws IOException {
             }
         }
     }
+
+    /**
+     * A FileIO based on LocalFileIO that always throws from rename().
+     *
+     * <p>When {@code actuallyMove} is true, it performs the move then throws; otherwise it throws
+     * without moving.
+     */
+    private static class ThrowingRenameLocalFileIO extends LocalFileIO {
+
+        private static final String MESSAGE = "Failed to parse the rename response";
+
+        private final boolean actuallyMove;
+
+        ThrowingRenameLocalFileIO(boolean actuallyMove) {
+            this.actuallyMove = actuallyMove;
+        }
+
+        @Override
+        public boolean rename(Path src, Path dst) throws IOException {
+            if (actuallyMove) {
+                super.rename(src, dst);
+            }
+            throw new IOException(MESSAGE);
+        }
+    }
 }