fix

Author: JingsongLi

paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java

@@ -49,6 +49,7 @@
 import org.apache.paimon.operation.commit.ConflictDetection;
 import org.apache.paimon.operation.commit.ManifestEntryChanges;
 import org.apache.paimon.operation.commit.RetryCommitResult;
+import org.apache.paimon.operation.commit.RetryCommitResult.CommitFailRetryResult;
 import org.apache.paimon.operation.commit.RowTrackingCommitUtils.RowTrackingAssigned;
 import org.apache.paimon.operation.commit.StrictModeChecker;
 import org.apache.paimon.operation.commit.SuccessCommitResult;
@@ -786,13 +787,15 @@ CommitResult tryCommitOnce(
 
         // Check if the commit has been completed. At this point, there will be no more repeated
         // commits and just return success
-        if (retryResult != null && latestSnapshot != null) {
+        if (retryResult instanceof CommitFailRetryResult && latestSnapshot != null) {
+            CommitFailRetryResult commitFailRetry = (CommitFailRetryResult) retryResult;
             Map<Long, Snapshot> snapshotCache = new HashMap<>();
             snapshotCache.put(latestSnapshot.id(), latestSnapshot);
             long startCheckSnapshot = Snapshot.FIRST_SNAPSHOT_ID;
-            if (retryResult.latestSnapshot != null) {
-                snapshotCache.put(retryResult.latestSnapshot.id(), retryResult.latestSnapshot);
-                startCheckSnapshot = retryResult.latestSnapshot.id() + 1;
+            if (commitFailRetry.latestSnapshot != null) {
+                snapshotCache.put(
+                        commitFailRetry.latestSnapshot.id(), commitFailRetry.latestSnapshot);
+                startCheckSnapshot = commitFailRetry.latestSnapshot.id() + 1;
             }
             for (long i = startCheckSnapshot; i <= latestSnapshot.id(); i++) {
                 Snapshot snapshot = snapshotCache.computeIfAbsent(i, snapshotManager::snapshot);
@@ -836,13 +839,17 @@ CommitResult tryCommitOnce(
             // latestSnapshotId is different from the snapshot id we've checked for conflicts,
             // so we have to check again
             List<BinaryRow> changedPartitions = changedPartitions(deltaFiles, indexFiles);
-            if (retryResult != null
-                    && retryResult.latestSnapshot != null
-                    && retryResult.baseDataFiles != null) {
-                baseDataFiles = new ArrayList<>(retryResult.baseDataFiles);
+            CommitFailRetryResult commitFailRetry =
+                    retryResult instanceof CommitFailRetryResult
+                            ? (CommitFailRetryResult) retryResult
+                            : null;
+            if (commitFailRetry != null
+                    && commitFailRetry.latestSnapshot != null
+                    && commitFailRetry.baseDataFiles != null) {
+                baseDataFiles = new ArrayList<>(commitFailRetry.baseDataFiles);
                 List<SimpleFileEntry> incremental =
                         scanner.readIncrementalChanges(
-                                retryResult.latestSnapshot, latestSnapshot, changedPartitions);
+                                commitFailRetry.latestSnapshot, latestSnapshot, changedPartitions);
                 if (!incremental.isEmpty()) {
                     baseDataFiles.addAll(incremental);
                     baseDataFiles = new ArrayList<>(FileEntry.mergeEntries(baseDataFiles));
@@ -872,7 +879,7 @@ CommitResult tryCommitOnce(
             if (exception.isPresent()) {
                 if (allowRollback && rollback != null) {
                     if (rollback.tryToRollback(latestSnapshot)) {
-                        return RetryCommitResult.ofEmpty(exception.get());
+                        return RetryCommitResult.forRollback(exception.get());
                     }
                 }
                 throw exception.get();
@@ -1005,7 +1012,7 @@ CommitResult tryCommitOnce(
         } catch (Exception e) {
             // commit exception, not sure about the situation and should not clean up the files
             LOG.warn("Retry commit for exception.", e);
-            return RetryCommitResult.ofContext(latestSnapshot, baseDataFiles, e);
+            return RetryCommitResult.forCommitFail(latestSnapshot, baseDataFiles, e);
         }
 
         if (!success) {
@@ -1022,7 +1029,7 @@ CommitResult tryCommitOnce(
                     commitTime);
             commitCleaner.cleanUpNoReuseTmpManifests(
                     baseManifestList, mergeBeforeManifests, mergeAfterManifests);
-            return RetryCommitResult.ofContext(latestSnapshot, baseDataFiles, null);
+            return RetryCommitResult.forCommitFail(latestSnapshot, baseDataFiles, null);
         }
 
         LOG.info(

paimon-core/src/main/java/org/apache/paimon/operation/commit/RetryCommitResult.java

@@ -26,32 +26,49 @@
 import java.util.List;
 
 /** Need to retry commit of {@link CommitResult}. */
-public class RetryCommitResult implements CommitResult {
+public abstract class RetryCommitResult implements CommitResult {
 
-    public final @Nullable Snapshot latestSnapshot;
-    public final @Nullable List<SimpleFileEntry> baseDataFiles;
     public final Exception exception;
 
-    private RetryCommitResult(
-            @Nullable Snapshot latestSnapshot,
-            @Nullable List<SimpleFileEntry> baseDataFiles,
-            Exception exception) {
-        this.latestSnapshot = latestSnapshot;
-        this.baseDataFiles = baseDataFiles;
+    private RetryCommitResult(Exception exception) {
         this.exception = exception;
     }
 
-    public static RetryCommitResult ofContext(
+    public static RetryCommitResult forCommitFail(
             Snapshot snapshot, List<SimpleFileEntry> baseDataFiles, Exception exception) {
-        return new RetryCommitResult(snapshot, baseDataFiles, exception);
+        return new CommitFailRetryResult(snapshot, baseDataFiles, exception);
     }
 
-    public static RetryCommitResult ofEmpty(Exception exception) {
-        return new RetryCommitResult(null, null, exception);
+    public static RetryCommitResult forRollback(Exception exception) {
+        return new RollbackRetryResult(exception);
     }
 
     @Override
     public boolean isSuccess() {
         return false;
     }
+
+    /** Retry result for commit failing. */
+    public static class CommitFailRetryResult extends RetryCommitResult {
+
+        public final @Nullable Snapshot latestSnapshot;
+        public final @Nullable List<SimpleFileEntry> baseDataFiles;
+
+        private CommitFailRetryResult(
+                @Nullable Snapshot latestSnapshot,
+                @Nullable List<SimpleFileEntry> baseDataFiles,
+                Exception exception) {
+            super(exception);
+            this.latestSnapshot = latestSnapshot;
+            this.baseDataFiles = baseDataFiles;
+        }
+    }
+
+    /** Retry result for rollback. */
+    public static class RollbackRetryResult extends RetryCommitResult {
+
+        private RollbackRetryResult(Exception exception) {
+            super(exception);
+        }
+    }
 }

paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java

@@ -1025,7 +1025,7 @@ public void testCommitTwiceWithDifferentKind() throws Exception {
                     null);
             // Compact
             commit.tryCommitOnce(
-                    RetryCommitResult.ofContext(firstLatest, Collections.emptyList(), null),
+                    RetryCommitResult.forCommitFail(firstLatest, Collections.emptyList(), null),
                     Collections.emptyList(),
                     Collections.emptyList(),
                     Collections.emptyList(),

paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java

@@ -3056,6 +3056,10 @@ private void doTestConflictRollback(boolean insertMiddle) throws Exception {
         }
 
         // do delete commit after
+        // expire snapshots first
+        SnapshotManager snapshotManager = ((FileStoreTable) table).snapshotManager();
+        snapshotManager.deleteSnapshot(1);
+        snapshotManager.deleteSnapshot(2);
         try (BatchTableCommit commit = writeBuilder.newCommit()) {
             List<CommitMessage> messages = singletonList(deleteCommitMessage);
             if (insertMiddle) {