[core] Simplify rollback to only clean snapshots

Author: JingsongLi

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

@@ -87,6 +87,7 @@ public void cleanUnusedManifests(Changelog changelog, Set<String> skippingSet) {
         // the index and statics manifest list should handle by snapshot deletion.
     }
 
+    @Override
     public Set<String> manifestSkippingSet(List<Snapshot> skippingSnapshots) {
         Set<String> skippingSet = new HashSet<>();
 

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

@@ -37,7 +37,6 @@
 
 import java.io.IOException;
 import java.util.Collection;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -106,10 +105,6 @@ public void cleanUnusedManifests(Snapshot taggedSnapshot, Set<String> skippingSe
         cleanUnusedManifests(taggedSnapshot, skippingSet, true, false);
     }
 
-    public Predicate<ExpireFileEntry> dataFileSkipper(Snapshot fromSnapshot) throws Exception {
-        return dataFileSkipper(Collections.singletonList(fromSnapshot));
-    }
-
     public Predicate<ExpireFileEntry> dataFileSkipper(List<Snapshot> fromSnapshots)
             throws Exception {
         Map<BinaryRow, Map<Integer, Set<String>>> skipped = new HashMap<>();

paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java

@@ -68,8 +68,7 @@
 
 import javax.annotation.Nullable;
 
-import java.io.IOException;
-import java.io.UncheckedIOException;
+import java.io.FileNotFoundException;
 import java.time.Duration;
 import java.util.HashMap;
 import java.util.List;
@@ -80,7 +79,6 @@
 import java.util.function.BiConsumer;
 
 import static org.apache.paimon.CoreOptions.PATH;
-import static org.apache.paimon.utils.Preconditions.checkArgument;
 
 /** Abstract {@link FileStoreTable}. */
 abstract class AbstractFileStoreTable implements FileStoreTable {
@@ -504,11 +502,27 @@ public void rollbackTo(long snapshotId) {
         try {
             snapshotManager.rollback(Instant.snapshot(snapshotId));
         } catch (UnsupportedOperationException e) {
-            checkArgument(
-                    snapshotManager.snapshotExists(snapshotId),
-                    "Rollback snapshot '%s' doesn't exist.",
-                    snapshotId);
-            rollbackHelper().updateLatestAndCleanLargerThan(snapshotManager.snapshot(snapshotId));
+            Snapshot snapshot;
+            try {
+                snapshot = snapshotManager.tryGetSnapshot(snapshotId);
+            } catch (FileNotFoundException ex) {
+                throw new IllegalArgumentException(
+                        String.format("Rollback snapshot '%s' doesn't exist.", snapshotId), ex);
+            }
+            rollbackHelper().cleanLargerThan(snapshot);
+        }
+    }
+
+    @Override
+    public void rollbackTo(String tagName) {
+        SnapshotManager snapshotManager = snapshotManager();
+        try {
+            snapshotManager.rollback(Instant.tag(tagName));
+        } catch (UnsupportedOperationException e) {
+            Snapshot taggedSnapshot = tagManager().getOrThrow(tagName).trimToSnapshot();
+            RollbackHelper rollbackHelper = rollbackHelper();
+            rollbackHelper.cleanLargerThan(taggedSnapshot);
+            rollbackHelper.createSnapshotFileIfNeeded(taggedSnapshot);
         }
     }
 
@@ -644,38 +658,6 @@ public void fastForward(String branchName) {
         branchManager().fastForward(branchName);
     }
 
-    @Override
-    public void rollbackTo(String tagName) {
-        SnapshotManager snapshotManager = snapshotManager();
-        try {
-            snapshotManager.rollback(Instant.tag(tagName));
-            return;
-        } catch (UnsupportedOperationException ignore) {
-
-        }
-        TagManager tagManager = tagManager();
-        checkArgument(tagManager.tagExists(tagName), "Rollback tag '%s' doesn't exist.", tagName);
-
-        Snapshot taggedSnapshot = tagManager.getOrThrow(tagName).trimToSnapshot();
-        rollbackHelper().updateLatestAndCleanLargerThan(taggedSnapshot);
-
-        try {
-            // it is possible that the earliest snapshot is later than the rollback tag because of
-            // snapshot expiration, in this case the `cleanLargerThan` method will delete all
-            // snapshots, so we should write the tag file to snapshot directory and modify the
-            // earliest hint
-            if (!snapshotManager.snapshotExists(taggedSnapshot.id())) {
-                fileIO.writeFile(
-                        snapshotManager().snapshotPath(taggedSnapshot.id()),
-                        fileIO.readFileUtf8(tagManager.tagPath(tagName)),
-                        false);
-                snapshotManager.commitEarliestHint(taggedSnapshot.id());
-            }
-        } catch (IOException e) {
-            throw new UncheckedIOException(e);
-        }
-    }
-
     @Override
     public TagManager tagManager() {
         return new TagManager(fileIO, path, currentBranch());
@@ -713,14 +695,7 @@ public FileStoreTable switchToBranch(String branchName) {
     }
 
     private RollbackHelper rollbackHelper() {
-        return new RollbackHelper(
-                snapshotManager(),
-                changelogManager(),
-                tagManager(),
-                fileIO,
-                store().newSnapshotDeletion(),
-                store().newChangelogDeletion(),
-                store().newTagDeletion());
+        return new RollbackHelper(snapshotManager(), changelogManager(), tagManager(), fileIO);
     }
 
     protected RowKindGenerator rowKindGenerator() {

paimon-core/src/main/java/org/apache/paimon/table/RollbackHelper.java

@@ -21,135 +21,91 @@
 import org.apache.paimon.Changelog;
 import org.apache.paimon.Snapshot;
 import org.apache.paimon.fs.FileIO;
-import org.apache.paimon.manifest.ExpireFileEntry;
-import org.apache.paimon.operation.ChangelogDeletion;
-import org.apache.paimon.operation.SnapshotDeletion;
-import org.apache.paimon.operation.TagDeletion;
 import org.apache.paimon.utils.ChangelogManager;
 import org.apache.paimon.utils.SnapshotManager;
 import org.apache.paimon.utils.TagManager;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashSet;
 import java.util.List;
-import java.util.Set;
 import java.util.SortedMap;
-import java.util.function.Predicate;
 
 import static org.apache.paimon.utils.Preconditions.checkNotNull;
 
 /** Helper class for {@link Table#rollbackTo} including utils to clean snapshots. */
 public class RollbackHelper {
 
-    private static final Logger LOG = LoggerFactory.getLogger(RollbackHelper.class);
-
     private final SnapshotManager snapshotManager;
     private final ChangelogManager changelogManager;
     private final TagManager tagManager;
     private final FileIO fileIO;
-    private final SnapshotDeletion snapshotDeletion;
-    private final ChangelogDeletion changelogDeletion;
-    private final TagDeletion tagDeletion;
 
     public RollbackHelper(
             SnapshotManager snapshotManager,
             ChangelogManager changelogManager,
             TagManager tagManager,
-            FileIO fileIO,
-            SnapshotDeletion snapshotDeletion,
-            ChangelogDeletion changelogDeletion,
-            TagDeletion tagDeletion) {
+            FileIO fileIO) {
         this.snapshotManager = snapshotManager;
         this.changelogManager = changelogManager;
         this.tagManager = tagManager;
         this.fileIO = fileIO;
-        this.snapshotDeletion = snapshotDeletion;
-        this.changelogDeletion = changelogDeletion;
-        this.tagDeletion = tagDeletion;
     }
 
     /** Clean snapshots and tags whose id is larger than given snapshot's and update latest hit. */
-    public void updateLatestAndCleanLargerThan(Snapshot retainedSnapshot) {
-        // clean data files
-        List<Snapshot> cleanedSnapshots = updateLatestAndCleanSnapshotsDataFiles(retainedSnapshot);
-        List<Changelog> cleanedChangelogs = cleanLongLivedChangelogDataFiles(retainedSnapshot);
-        List<Snapshot> cleanedTags = cleanTagsDataFiles(retainedSnapshot);
-        Set<Long> cleanedIds = new HashSet<>();
-
-        // clean manifests
-        // this can be used for snapshots and tags manifests cleaning both
-        Set<String> manifestsSkippingSet = snapshotDeletion.manifestSkippingSet(retainedSnapshot);
-
-        for (Snapshot snapshot : cleanedSnapshots) {
-            snapshotDeletion.cleanUnusedManifests(snapshot, manifestsSkippingSet);
-            cleanedIds.add(snapshot.id());
-        }
-
-        for (Changelog changelog : cleanedChangelogs) {
-            changelogDeletion.cleanUnusedManifests(changelog, manifestsSkippingSet);
-            cleanedIds.add(changelog.id());
+    public void cleanLargerThan(Snapshot retainedSnapshot) {
+        try {
+            cleanSnapshots(retainedSnapshot);
+            cleanLongLivedChangelogs(retainedSnapshot);
+            cleanTags(retainedSnapshot);
+        } catch (IOException e) {
+            throw new RuntimeException(e);
         }
+    }
 
-        for (Snapshot snapshot : cleanedTags) {
-            if (cleanedIds.contains(snapshot.id())) {
-                continue;
+    public void createSnapshotFileIfNeeded(Snapshot taggedSnapshot) {
+        // it is possible that the earliest snapshot is later than the rollback tag because of
+        // snapshot expiration, in this case the `cleanLargerThan` method will delete all
+        // snapshots, so we should write the tag file to snapshot directory and modify the
+        // earliest hint
+        if (!snapshotManager.snapshotExists(taggedSnapshot.id())) {
+            try {
+                fileIO.writeFile(
+                        snapshotManager.snapshotPath(taggedSnapshot.id()),
+                        taggedSnapshot.toJson(),
+                        false);
+                snapshotManager.commitEarliestHint(taggedSnapshot.id());
+            } catch (IOException e) {
+                throw new UncheckedIOException(e);
             }
-            tagDeletion.cleanUnusedManifests(snapshot, manifestsSkippingSet);
         }
     }
 
-    private List<Snapshot> updateLatestAndCleanSnapshotsDataFiles(Snapshot retainedSnapshot) {
+    private void cleanSnapshots(Snapshot retainedSnapshot) throws IOException {
         long earliest =
                 checkNotNull(
                         snapshotManager.earliestSnapshotId(), "Cannot find earliest snapshot.");
         long latest =
                 checkNotNull(snapshotManager.latestSnapshotId(), "Cannot find latest snapshot.");
 
         // modify the latest hint
-        try {
-            snapshotManager.commitLatestHint(retainedSnapshot.id());
-        } catch (IOException e) {
-            throw new UncheckedIOException(e);
-        }
-        // delete snapshot files first, cannot be read now
+        snapshotManager.commitLatestHint(retainedSnapshot.id());
+
         // it is possible that some snapshots have been expired
-        List<Snapshot> toBeCleaned = new ArrayList<>();
         long to = Math.max(earliest, retainedSnapshot.id() + 1);
         for (long i = latest; i >= to; i--) {
             // Ignore the non-existent snapshots
             if (snapshotManager.snapshotExists(i)) {
-                toBeCleaned.add(snapshotManager.snapshot(i));
                 snapshotManager.deleteSnapshot(i);
             }
         }
-
-        // delete data files of snapshots
-        // don't worry about tag data files because file deletion methods won't throw exception
-        // when deleting non-existing data files
-        for (Snapshot snapshot : toBeCleaned) {
-            snapshotDeletion.deleteAddedDataFiles(snapshot.deltaManifestList());
-            if (snapshot.changelogManifestList() != null) {
-                snapshotDeletion.deleteAddedDataFiles(snapshot.changelogManifestList());
-            }
-        }
-
-        // delete directories
-        snapshotDeletion.cleanEmptyDirectories();
-
-        return toBeCleaned;
     }
 
-    private List<Changelog> cleanLongLivedChangelogDataFiles(Snapshot retainedSnapshot) {
+    private void cleanLongLivedChangelogs(Snapshot retainedSnapshot) throws IOException {
         Long earliest = changelogManager.earliestLongLivedChangelogId();
         Long latest = changelogManager.latestLongLivedChangelogId();
         if (earliest == null || latest == null) {
-            return Collections.emptyList();
+            return;
         }
 
         // it is possible that some snapshots have been expired
@@ -160,73 +116,37 @@ private List<Changelog> cleanLongLivedChangelogDataFiles(Snapshot retainedSnapsh
         }
 
         // modify the latest hint
-        try {
-            if (toBeCleaned.size() > 0) {
-                if (to == earliest) {
-                    // all changelog has been cleaned, so we do not know the actual latest id
-                    // set to -1
-                    changelogManager.commitLongLivedChangelogLatestHint(-1);
-                } else {
-                    changelogManager.commitLongLivedChangelogLatestHint(to - 1);
-                }
+        if (!toBeCleaned.isEmpty()) {
+            if (to == earliest) {
+                // all changelog has been cleaned, so we do not know the actual latest id
+                // set to -1
+                changelogManager.commitLongLivedChangelogLatestHint(-1);
+            } else {
+                changelogManager.commitLongLivedChangelogLatestHint(to - 1);
             }
-        } catch (IOException e) {
-            throw new UncheckedIOException(e);
         }
 
         // delete data files of changelog
         for (Changelog changelog : toBeCleaned) {
-            // delete changelog files first, cannot be read now
             fileIO.deleteQuietly(changelogManager.longLivedChangelogPath(changelog.id()));
-            // clean the deleted file
-            changelogDeletion.cleanUnusedDataFiles(changelog, manifestEntry -> false);
         }
-
-        // delete directories
-        snapshotDeletion.cleanEmptyDirectories();
-
-        return toBeCleaned;
     }
 
-    private List<Snapshot> cleanTagsDataFiles(Snapshot retainedSnapshot) {
+    private void cleanTags(Snapshot retainedSnapshot) {
         SortedMap<Snapshot, List<String>> tags = tagManager.tags();
         if (tags.isEmpty()) {
-            return Collections.emptyList();
+            return;
         }
 
         List<Snapshot> taggedSnapshots = new ArrayList<>(tags.keySet());
-        List<Snapshot> toBeCleaned = new ArrayList<>();
 
         // delete tag files
         for (int i = taggedSnapshots.size() - 1; i >= 0; i--) {
             Snapshot tag = taggedSnapshots.get(i);
             if (tag.id() <= retainedSnapshot.id()) {
                 break;
             }
-            toBeCleaned.add(tag);
             tags.get(tag).forEach(tagName -> fileIO.deleteQuietly(tagManager.tagPath(tagName)));
         }
-
-        // delete data files
-        Predicate<ExpireFileEntry> dataFileSkipper = null;
-        boolean success = true;
-        try {
-            dataFileSkipper = tagDeletion.dataFileSkipper(retainedSnapshot);
-        } catch (Exception e) {
-            LOG.info(
-                    "Skip cleaning data files for deleted tags due to failed to build skipping set.",
-                    e);
-            success = false;
-        }
-
-        if (success) {
-            for (Snapshot s : toBeCleaned) {
-                tagDeletion.cleanUnusedDataFiles(s, dataFileSkipper);
-            }
-            // delete directories
-            tagDeletion.cleanEmptyDirectories();
-        }
-
-        return toBeCleaned;
     }
 }