format code

sunxiaojian · Feb 26, 2024 · caac79f · caac79f
1 parent 170e65b
commit caac79f
Show file tree

Hide file tree

Showing 3 changed files with 49 additions and 10 deletions.
diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/BranchManager.java b/paimon-core/src/main/java/org/apache/paimon/utils/BranchManager.java
@@ -33,9 +33,11 @@
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.SortedMap;
 import java.util.stream.Collectors;
 
@@ -244,40 +246,54 @@ public void replaceBranch(String branchName) {
 
     /** Calculate copy main branch to target branch. */
     private void calculateCopyMainBranchToTargetBranch(String branchName) throws IOException {
-        TableBranch referenceBranch =
+        TableBranch fromBranch =
                 this.branches().stream()
                         .filter(branch -> branch.getBranchName().equals(branchName))
                         .findFirst()
                         .orElse(null);
-        if (referenceBranch == null) {
+        if (fromBranch == null) {
             throw new RuntimeException(String.format("No branches found %s", branchName));
         }
-        long createBranchTime = referenceBranch.getCreateTime();
-        // delete tags
+        long createBranchTime = fromBranch.getCreateTime();
+        Snapshot fromSnapshot = snapshotManager.snapshot(fromBranch.getCreatedFromSnapshot());
+        // Copy tags
         List<TableTag> tags = tagManager.tableTags();
         for (TableTag tag : tags) {
+            if (tagManager.branchTagExists(branchName, tag.getTagName())) {
+                // If it already exists, skip it directly
+                continue;
+            }
             if (tag.getCreateTime() < createBranchTime) {
                 fileIO.copyFileUtf8(
                         tagManager.tagPath(tag.getTagName()),
                         tagManager.branchTagPath(branchName, tag.getTagName()));
             }
         }
 
-        // delete snapshots
-        Iterator<Snapshot> snapshotIterator = snapshotManager.snapshots();
-        while (snapshotIterator.hasNext()) {
-            Snapshot snapshot = snapshotIterator.next();
-            if (snapshot.timeMillis() < createBranchTime) {
+        // Copy snapshots
+        Iterator<Snapshot> snapshots = snapshotManager.snapshots();
+        while (snapshots.hasNext()) {
+            Snapshot snapshot = snapshots.next();
+            if (snapshotManager.branchSnapshotExists(branchName, snapshot.id())) {
+                // If it already exists, skip it directly
+                continue;
+            }
+            if (snapshot.timeMillis() < fromSnapshot.timeMillis()) {
                 fileIO.copyFileUtf8(
                         snapshotManager.snapshotPath(snapshot.id()),
                         snapshotManager.branchSnapshotPath(branchName, snapshot.id()));
             }
         }
 
-        // Delete schemas
+        // Copy schemas
         List<Long> schemaIds = schemaManager.listAllIds();
+        Set<Long> existsSchemas = new HashSet<>(schemaManager.listAllIdsWithBranch(branchName));
         for (Long schemaId : schemaIds) {
             TableSchema tableSchema = schemaManager.schema(schemaId);
+            if (existsSchemas.contains(schemaId)) {
+                // If it already exists, skip it directly
+                continue;
+            }
             if (tableSchema.timeMillis() < createBranchTime) {
                 fileIO.copyFileUtf8(
                         schemaManager.toSchemaPath(schemaId),

diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java b/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java
@@ -115,6 +115,17 @@ public Snapshot snapshot(String branchName, long snapshotId) {
         return Snapshot.fromPath(fileIO, snapshotPath);
     }
 
+    public boolean branchSnapshotExists(String branchName, long snapshotId) {
+        Path path = snapshotPathByBranch(branchName, snapshotId);
+        try {
+            return fileIO.exists(path);
+        } catch (IOException e) {
+            throw new RuntimeException(
+                    "Failed to determine if snapshot #" + snapshotId + " exists in path " + path,
+                    e);
+        }
+    }
+
     public boolean snapshotExists(long snapshotId) {
         Path path = snapshotPath(snapshotId);
         try {

diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/TagManager.java b/paimon-core/src/main/java/org/apache/paimon/utils/TagManager.java
@@ -199,6 +199,18 @@ private void doClean(
                 taggedSnapshot, tagDeletion.manifestSkippingSet(skippedSnapshots));
     }
 
+    /** Check if a tag exists. */
+    public boolean branchTagExists(String branchName, String tagName) {
+        Path path = branchTagPath(branchName, tagName);
+        try {
+            return fileIO.exists(path);
+        } catch (IOException e) {
+            throw new RuntimeException(
+                    String.format(
+                            "Failed to determine if tag '%s' exists in path %s.", tagName, path),
+                    e);
+        }
+    }
     /** Check if a tag exists. */
     public boolean tagExists(String tagName) {
         Path path = tagPath(tagName);