[core] Rollback 'COMPACT' commit for row-level operations (#6968)

Author: JingsongLi

paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java

@@ -21,6 +21,7 @@
 import org.apache.paimon.CoreOptions.ExternalPathStrategy;
 import org.apache.paimon.catalog.RenamingSnapshotCommit;
 import org.apache.paimon.catalog.SnapshotCommit;
+import org.apache.paimon.catalog.TableRollback;
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.format.FileFormat;
 import org.apache.paimon.fs.FileIO;
@@ -43,6 +44,7 @@
 import org.apache.paimon.operation.PartitionExpire;
 import org.apache.paimon.operation.SnapshotDeletion;
 import org.apache.paimon.operation.TagDeletion;
+import org.apache.paimon.operation.commit.CommitRollback;
 import org.apache.paimon.operation.commit.ConflictDetection;
 import org.apache.paimon.operation.commit.StrictModeChecker;
 import org.apache.paimon.partition.PartitionExpireStrategy;
@@ -288,6 +290,11 @@ public FileStoreCommitImpl newCommit(String commitUser, FileStoreTable table) {
                         commitUser,
                         this::newScan,
                         options.commitStrictModeLastSafeSnapshot().orElse(null));
+        CommitRollback rollback = null;
+        TableRollback tableRollback = catalogEnvironment.catalogTableRollback();
+        if (tableRollback != null) {
+            rollback = new CommitRollback(tableRollback);
+        }
         return new FileStoreCommitImpl(
                 snapshotCommit,
                 fileIO,
@@ -320,7 +327,8 @@ public FileStoreCommitImpl newCommit(String commitUser, FileStoreTable table) {
                 options.rowTrackingEnabled(),
                 options.commitDiscardDuplicateFiles(),
                 conflictDetection,
-                strictModeChecker);
+                strictModeChecker,
+                rollback);
     }
 
     @Override

paimon-core/src/main/java/org/apache/paimon/catalog/TableRollback.java

@@ -0,0 +1,29 @@
+/*
+ * 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.catalog;
+
+import org.apache.paimon.table.Instant;
+
+import javax.annotation.Nullable;
+
+/** Rollback table to instant from snapshot. */
+public interface TableRollback {
+
+    void rollbackTo(Instant instant, @Nullable Long fromSnapshot);
+}

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

@@ -43,10 +43,12 @@
 import org.apache.paimon.operation.commit.CommitChangesProvider;
 import org.apache.paimon.operation.commit.CommitCleaner;
 import org.apache.paimon.operation.commit.CommitResult;
+import org.apache.paimon.operation.commit.CommitRollback;
 import org.apache.paimon.operation.commit.CommitScanner;
 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;
@@ -138,6 +140,7 @@ public class FileStoreCommitImpl implements FileStoreCommit {
     private final ManifestFile manifestFile;
     private final ManifestList manifestList;
     private final IndexManifestFile indexManifestFile;
+    @Nullable private final CommitRollback rollback;
     private final CommitScanner scanner;
     private final int numBucket;
     private final MemorySize manifestTargetSize;
@@ -195,7 +198,8 @@ public FileStoreCommitImpl(
             boolean rowTrackingEnabled,
             boolean discardDuplicateFiles,
             ConflictDetection conflictDetection,
-            @Nullable StrictModeChecker strictModeChecker) {
+            @Nullable StrictModeChecker strictModeChecker,
+            @Nullable CommitRollback rollback) {
         this.snapshotCommit = snapshotCommit;
         this.fileIO = fileIO;
         this.schemaManager = schemaManager;
@@ -209,6 +213,7 @@ public FileStoreCommitImpl(
         this.manifestFile = manifestFileFactory.create();
         this.manifestList = manifestListFactory.create();
         this.indexManifestFile = indexManifestFileFactory.create();
+        this.rollback = rollback;
         this.scanner = new CommitScanner(scan, indexManifestFile, options);
         this.numBucket = numBucket;
         this.manifestTargetSize = manifestTargetSize;
@@ -313,10 +318,13 @@ public int commit(ManifestCommittable committable, boolean checkAppendFiles) {
                 if (appendCommitCheckConflict) {
                     checkAppendFiles = true;
                 }
+
+                boolean allowRollback = false;
                 if (containsFileDeletionOrDeletionVectors(
                         appendSimpleEntries, changes.appendIndexFiles)) {
                     commitKind = CommitKind.OVERWRITE;
                     checkAppendFiles = true;
+                    allowRollback = true;
                 }
 
                 attempts +=
@@ -329,6 +337,7 @@ public int commit(ManifestCommittable committable, boolean checkAppendFiles) {
                                 committable.watermark(),
                                 committable.properties(),
                                 commitKind,
+                                allowRollback,
                                 checkAppendFiles,
                                 null);
                 generatedSnapshot += 1;
@@ -347,6 +356,7 @@ public int commit(ManifestCommittable committable, boolean checkAppendFiles) {
                                 committable.watermark(),
                                 committable.properties(),
                                 CommitKind.COMPACT,
+                                false,
                                 true,
                                 null);
                 generatedSnapshot += 1;
@@ -512,6 +522,7 @@ public int overwritePartition(
                                 committable.watermark(),
                                 committable.properties(),
                                 CommitKind.COMPACT,
+                                false,
                                 true,
                                 null);
                 generatedSnapshot += 1;
@@ -652,6 +663,7 @@ public void commitStatistics(Statistics stats, long commitIdentifier) {
                 Collections.emptyMap(),
                 CommitKind.ANALYZE,
                 false,
+                false,
                 statsFileName);
     }
 
@@ -678,6 +690,7 @@ private int tryCommit(
             @Nullable Long watermark,
             Map<String, String> properties,
             CommitKind commitKind,
+            boolean allowRollback,
             boolean detectConflicts,
             @Nullable String statsFileName) {
         int retryCount = 0;
@@ -696,6 +709,7 @@ private int tryCommit(
                             watermark,
                             properties,
                             commitKind,
+                            allowRollback,
                             latestSnapshot,
                             detectConflicts,
                             statsFileName);
@@ -742,6 +756,7 @@ private int tryOverwritePartition(
                 watermark,
                 properties,
                 CommitKind.OVERWRITE,
+                false,
                 true,
                 null);
     }
@@ -756,20 +771,23 @@ CommitResult tryCommitOnce(
             @Nullable Long watermark,
             Map<String, String> properties,
             CommitKind commitKind,
+            boolean allowRollback,
             @Nullable Snapshot latestSnapshot,
             boolean detectConflicts,
             @Nullable String newStatsFileName) {
         long startMillis = System.currentTimeMillis();
 
         // 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);
@@ -813,11 +831,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) {
-                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));
@@ -837,12 +861,21 @@ CommitResult tryCommitOnce(
                                 .filter(entry -> !baseIdentifiers.contains(entry.identifier()))
                                 .collect(Collectors.toList());
             }
-            conflictDetection.checkNoConflictsOrFail(
-                    latestSnapshot,
-                    baseDataFiles,
-                    SimpleFileEntry.from(deltaFiles),
-                    indexFiles,
-                    commitKind);
+            Optional<RuntimeException> exception =
+                    conflictDetection.checkConflicts(
+                            latestSnapshot,
+                            baseDataFiles,
+                            SimpleFileEntry.from(deltaFiles),
+                            indexFiles,
+                            commitKind);
+            if (exception.isPresent()) {
+                if (allowRollback && rollback != null) {
+                    if (rollback.tryToRollback(latestSnapshot)) {
+                        return RetryCommitResult.forRollback(exception.get());
+                    }
+                }
+                throw exception.get();
+            }
         }
 
         Snapshot newSnapshot;
@@ -971,7 +1004,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 new RetryCommitResult(latestSnapshot, baseDataFiles, e);
+            return RetryCommitResult.forCommitFail(latestSnapshot, baseDataFiles, e);
         }
 
         if (!success) {
@@ -988,7 +1021,7 @@ CommitResult tryCommitOnce(
                     commitTime);
             commitCleaner.cleanUpNoReuseTmpManifests(
                     baseManifestList, mergeBeforeManifests, mergeAfterManifests);
-            return new RetryCommitResult(latestSnapshot, baseDataFiles, null);
+            return RetryCommitResult.forCommitFail(latestSnapshot, baseDataFiles, null);
         }
 
         LOG.info(

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

@@ -0,0 +1,45 @@
+/*
+ * 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.operation.commit;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.catalog.TableRollback;
+import org.apache.paimon.table.Instant;
+
+/** Commit rollback to rollback 'COMPACT' commits for resolving conflicts. */
+public class CommitRollback {
+
+    private final TableRollback tableRollback;
+
+    public CommitRollback(TableRollback tableRollback) {
+        this.tableRollback = tableRollback;
+    }
+
+    public boolean tryToRollback(Snapshot latestSnapshot) {
+        if (latestSnapshot.commitKind() == Snapshot.CommitKind.COMPACT) {
+            long latest = latestSnapshot.id();
+            try {
+                tableRollback.rollbackTo(Instant.snapshot(latest - 1), latest);
+                return true;
+            } catch (Exception ignored) {
+            }
+        }
+        return false;
+    }
+}