[core] Remove the feature that creates branches from snapshots. (#3934)

docs/content/flink/procedures.md

@@ -308,22 +308,18 @@ All available procedures are listed below.
    <tr>
       <td>create_branch</td>
       <td>
-         -- based on the specified snapshot <br/>
-         CALL [catalog.]sys.create_branch('identifier', 'branchName', snapshotId) <br/>
          -- based on the specified tag <br/>
          CALL [catalog.]sys.create_branch('identifier', 'branchName', 'tagName')
          -- create empty branch <br/>
          CALL [catalog.]sys.create_branch('identifier', 'branchName')
       </td>
       <td>
-         To create a branch based on given snapshot / tag, or just create empty branch. Arguments:
+         To create a branch based on given tag, or just create empty branch. Arguments:
             <li>identifier: the target table identifier. Cannot be empty.</li>
             <li>branchName: name of the new branch.</li>
-            <li>snapshotId (Long): id of the snapshot which the new branch is based on.</li>
             <li>tagName: name of the tag which the new branch is based on.</li>
       </td>
       <td>
-         CALL sys.create_branch('default.T', 'branch1', 10)<br/><br/>
          CALL sys.create_branch('default.T', 'branch1', 'tag1')<br/><br/>
          CALL sys.create_branch('default.T', 'branch1')<br/><br/>
       </td>

docs/content/maintenance/manage-branches.md

@@ -46,9 +46,6 @@ Run the following sql:
 -- create branch named 'branch1' from tag 'tag1'
 CALL sys.create_branch('default.T', 'branch1', 'tag1');
 
--- create branch named 'branch1' from snapshot 1
-CALL sys.create_branch('default.T', 'branch1', 1);
-
 -- create empty branch named 'branch1'
 CALL sys.create_branch('default.T', 'branch1');
 ```
@@ -67,7 +64,6 @@ Run the following command:
     --table <table-name> \
     --branch_name <branch-name> \
     [--tag_name <tag-name>] \
-    [--snapshot <snapshot_id>] \
     [--catalog_conf <paimon-catalog-conf> [--catalog_conf <paimon-catalog-conf> ...]]
 ```
 {{< /tab >}}

docs/content/spark/procedures.md

@@ -171,12 +171,10 @@ This section introduce all available spark procedures about paimon.
             <li>table: the target table identifier. Cannot be empty.</li>
             <li>branch: name of the branch to be merged.</li>
             <li>tag: name of the new tag. Cannot be empty.</li>
-            <li>snapshot(Long):  id of the snapshot which the new tag is based on.</li>
       </td>
       <td>
           CALL sys.create_branch(table => 'test_db.T', branch => 'test_branch')<br/><br/>
           CALL sys.create_branch(table => 'test_db.T', branch => 'test_branch', tag => 'my_tag')<br/><br/>
-          CALL sys.create_branch(table => 'test_db.T', branch => 'test_branch', snapshot => 10)
       </td>
     </tr>
     <tr>

paimon-core/src/main/java/org/apache/paimon/privilege/PrivilegedFileStoreTable.java

@@ -145,12 +145,6 @@ public void createBranch(String branchName) {
         wrapped.createBranch(branchName);
     }
 
-    @Override
-    public void createBranch(String branchName, long snapshotId) {
-        privilegeChecker.assertCanInsert(identifier);
-        wrapped.createBranch(branchName, snapshotId);
-    }
-
     @Override
     public void createBranch(String branchName, String tagName) {
         privilegeChecker.assertCanInsert(identifier);

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

@@ -569,11 +569,6 @@ public void createBranch(String branchName) {
         branchManager().createBranch(branchName);
     }
 
-    @Override
-    public void createBranch(String branchName, long snapshotId) {
-        branchManager().createBranch(branchName, snapshotId);
-    }
-
     @Override
     public void createBranch(String branchName, String tagName) {
         branchManager().createBranch(branchName, tagName);

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

@@ -168,11 +168,6 @@ public void createBranch(String branchName) {
         wrapped.createBranch(branchName);
     }
 
-    @Override
-    public void createBranch(String branchName, long snapshotId) {
-        wrapped.createBranch(branchName, snapshotId);
-    }
-
     @Override
     public void createBranch(String branchName, String tagName) {
         wrapped.createBranch(branchName, tagName);

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

@@ -176,14 +176,6 @@ default void createBranch(String branchName) {
                         this.getClass().getSimpleName()));
     }
 
-    @Override
-    default void createBranch(String branchName, long snapshotId) {
-        throw new UnsupportedOperationException(
-                String.format(
-                        "Readonly Table %s does not support createBranch with snapshotId.",
-                        this.getClass().getSimpleName()));
-    }
-
     @Override
     default void createBranch(String branchName, String tagName) {
         throw new UnsupportedOperationException(

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

@@ -116,10 +116,6 @@ default void deleteTags(String tagNames) {
     @Experimental
     void createBranch(String branchName);
 
-    /** Create a branch from given snapshot. */
-    @Experimental
-    void createBranch(String branchName, long snapshotId);
-
     /** Create a branch from given tag. */
     @Experimental
     void createBranch(String branchName, String tagName);

paimon-core/src/main/java/org/apache/paimon/utils/BranchManager.java

@@ -125,40 +125,6 @@ branchName, branchPath(tablePath, branchName)),
         }
     }
 
-    public void createBranch(String branchName, long snapshotId) {
-        checkArgument(
-                !isMainBranch(branchName),
-                String.format(
-                        "Branch name '%s' is the default branch and cannot be used.",
-                        DEFAULT_MAIN_BRANCH));
-        checkArgument(!StringUtils.isBlank(branchName), "Branch name '%s' is blank.", branchName);
-        checkArgument(!branchExists(branchName), "Branch name '%s' already exists.", branchName);
-        checkArgument(
-                !branchName.chars().allMatch(Character::isDigit),
-                "Branch name cannot be pure numeric string but is '%s'.",
-                branchName);
-
-        Snapshot snapshot = snapshotManager.snapshot(snapshotId);
-
-        try {
-            // Copy the corresponding snapshot and schema files into the branch directory
-            fileIO.copyFile(
-                    snapshotManager.snapshotPath(snapshotId),
-                    snapshotManager.copyWithBranch(branchName).snapshotPath(snapshot.id()),
-                    true);
-            fileIO.copyFile(
-                    schemaManager.toSchemaPath(snapshot.schemaId()),
-                    schemaManager.copyWithBranch(branchName).toSchemaPath(snapshot.schemaId()),
-                    true);
-        } catch (IOException e) {
-            throw new RuntimeException(
-                    String.format(
-                            "Exception occurs when create branch '%s' (directory in %s).",
-                            branchName, branchPath(tablePath, branchName)),
-                    e);
-        }
-    }
-
     public void createBranch(String branchName, String tagName) {
         checkArgument(
                 !isMainBranch(branchName),

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CreateBranchAction.java

@@ -27,28 +27,22 @@ public class CreateBranchAction extends TableActionBase {
     private final String branchName;
     private final String tagName;
 
-    private final Long snapshotId;
-
     public CreateBranchAction(
             String warehouse,
             String databaseName,
             String tableName,
             Map<String, String> catalogConfig,
             String branchName,
-            String tagName,
-            Long snapshotId) {
+            String tagName) {
         super(warehouse, databaseName, tableName, catalogConfig);
         this.branchName = branchName;
         this.tagName = tagName;
-        this.snapshotId = snapshotId;
     }
 
     @Override
     public void run() throws Exception {
         if (!StringUtils.isBlank(tagName)) {
             table.createBranch(branchName, tagName);
-        } else if (snapshotId != null) {
-            table.createBranch(branchName, snapshotId);
         } else {
             table.createBranch(branchName);
         }

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CreateBranchActionFactory.java

@@ -30,7 +30,6 @@ public class CreateBranchActionFactory implements ActionFactory {
 
     private static final String TAG_NAME = "tag_name";
     private static final String BRANCH_NAME = "branch_name";
-    private static final String SNAPSHOT = "snapshot";
 
     @Override
     public String identifier() {
@@ -44,11 +43,6 @@ public Optional<Action> create(MultipleParameterToolAdapter params) {
         Tuple3<String, String, String> tablePath = getTablePath(params);
         Map<String, String> catalogConfig = optionalConfigMap(params, CATALOG_CONF);
 
-        Long snapshot = null;
-        if (params.has(SNAPSHOT)) {
-            snapshot = Long.parseLong(params.get(SNAPSHOT));
-        }
-
         String tagName = null;
         if (params.has(TAG_NAME)) {
             tagName = params.get(TAG_NAME);
@@ -63,8 +57,7 @@ public Optional<Action> create(MultipleParameterToolAdapter params) {
                         tablePath.f2,
                         catalogConfig,
                         branchName,
-                        tagName,
-                        snapshot);
+                        tagName);
         return Optional.of(action);
     }
 
@@ -76,7 +69,7 @@ public void printHelp() {
         System.out.println("Syntax:");
         System.out.println(
                 "  create_branch --warehouse <warehouse_path> --database <database_name> "
-                        + "--table <table_name> --branch_name <branch_name> [--tag_name <tag_name>] [--snapshot <snapshot_id>]");
+                        + "--table <table_name> --branch_name <branch_name> [--tag_name <tag_name>]");
         System.out.println();
     }
 }

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CreateBranchProcedure.java

@@ -44,27 +44,19 @@ public String identifier() {
     public String[] call(
             ProcedureContext procedureContext, String tableId, String branchName, String tagName)
             throws Catalog.TableNotExistException {
-        return innerCall(tableId, branchName, tagName, 0);
+        return innerCall(tableId, branchName, tagName);
     }
 
     public String[] call(ProcedureContext procedureContext, String tableId, String branchName)
             throws Catalog.TableNotExistException {
-        return innerCall(tableId, branchName, null, 0);
+        return innerCall(tableId, branchName, null);
     }
 
-    public String[] call(
-            ProcedureContext procedureContext, String tableId, String branchName, long snapshotId)
-            throws Catalog.TableNotExistException {
-        return innerCall(tableId, branchName, null, snapshotId);
-    }
-
-    private String[] innerCall(String tableId, String branchName, String tagName, long snapshotId)
+    private String[] innerCall(String tableId, String branchName, String tagName)
             throws Catalog.TableNotExistException {
         Table table = catalog.getTable(Identifier.fromString(tableId));
         if (!StringUtils.isBlank(tagName)) {
             table.createBranch(branchName, tagName);
-        } else if (snapshotId > 0) {
-            table.createBranch(branchName, snapshotId);
         } else {
             table.createBranch(branchName);
         }

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BranchSqlITCase.java

@@ -54,7 +54,9 @@ public void testAlterBranchTable() throws Exception {
                         + " (2, 10, 'cat'),"
                         + " (2, 20, 'dog')");
 
-        sql("CALL sys.create_branch('default.T', 'test', 1)");
+        sql("CALL sys.create_tag('default.T', 'tag1', 1)");
+
+        sql("CALL sys.create_branch('default.T', 'test', 'tag1')");
 
         FileStoreTable branchTable = paimonTable("T$branch_test");
         assertThat(branchTable.schema().fields().size()).isEqualTo(3);
@@ -134,37 +136,6 @@ public void testCreateBranchFromTag() throws Exception {
                         "+I[2, 20, dog]");
     }
 
-    @Test
-    public void testCreateBranchFromSnapshot() throws Exception {
-        sql(
-                "CREATE TABLE T ("
-                        + " pt INT"
-                        + ", k INT"
-                        + ", v STRING"
-                        + ", PRIMARY KEY (pt, k) NOT ENFORCED"
-                        + " ) PARTITIONED BY (pt) WITH ("
-                        + " 'bucket' = '2'"
-                        + " )");
-
-        // snapshot 1.
-        sql("INSERT INTO T VALUES(1, 10, 'apple')");
-
-        // snapshot 2.
-        sql("INSERT INTO T VALUES(1, 20, 'dog')");
-
-        sql("CALL sys.create_branch('default.T', 'test', 1)");
-        sql("CALL sys.create_branch('default.T', 'test2', 2)");
-
-        assertThat(collectResult("SELECT created_from_snapshot FROM `T$branches`"))
-                .containsExactlyInAnyOrder("+I[1]", "+I[2]");
-
-        assertThat(paimonTable("T$branch_test").snapshotManager().snapshotExists(1))
-                .isEqualTo(true);
-
-        assertThat(paimonTable("T$branch_test2").snapshotManager().snapshotExists(2))
-                .isEqualTo(true);
-    }
-
     @Test
     public void testCreateEmptyBranch() throws Exception {
         sql(
@@ -213,8 +184,12 @@ public void testDeleteBranchTable() throws Exception {
         // snapshot 2.
         sql("INSERT INTO T VALUES(1, 20, 'dog')");
 
-        sql("CALL sys.create_branch('default.T', 'test', 1)");
-        sql("CALL sys.create_branch('default.T', 'test2', 2)");
+        sql("CALL sys.create_tag('default.T', 'tag1', 1)");
+
+        sql("CALL sys.create_tag('default.T', 'tag2', 2)");
+
+        sql("CALL sys.create_branch('default.T', 'test', 'tag1')");
+        sql("CALL sys.create_branch('default.T', 'test2', 'tag2')");
 
         assertThat(collectResult("SELECT branch_name, created_from_snapshot FROM `T$branches`"))
                 .containsExactlyInAnyOrder("+I[test, 1]", "+I[test2, 2]");
@@ -244,9 +219,13 @@ public void testBranchManagerGetBranchSnapshotsList() throws Exception {
         FileStoreTable table = paimonTable("T");
         checkSnapshots(table.snapshotManager(), 1, 3);
 
-        sql("CALL sys.create_branch('default.T', 'test1', 1)");
-        sql("CALL sys.create_branch('default.T', 'test2', 2)");
-        sql("CALL sys.create_branch('default.T', 'test3', 3)");
+        sql("CALL sys.create_tag('default.T', 'tag1', 1)");
+        sql("CALL sys.create_tag('default.T', 'tag2', 2)");
+        sql("CALL sys.create_tag('default.T', 'tag3', 3)");
+
+        sql("CALL sys.create_branch('default.T', 'test1', 'tag1')");
+        sql("CALL sys.create_branch('default.T', 'test2', 'tag2')");
+        sql("CALL sys.create_branch('default.T', 'test3', 'tag3')");
 
         assertThat(collectResult("SELECT created_from_snapshot FROM `T$branches`"))
                 .containsExactlyInAnyOrder("+I[1]", "+I[2]", "+I[3]");
@@ -277,7 +256,9 @@ public void testBranchFastForward() throws Exception {
                 .containsExactlyInAnyOrder(
                         "+I[1, 10, hunter]", "+I[1, 20, hunter]", "+I[1, 30, hunter]");
 
-        sql("CALL sys.create_branch('default.T', 'test', 1)");
+        sql("CALL sys.create_tag('default.T', 'tag1', 1)");
+
+        sql("CALL sys.create_branch('default.T', 'test', 'tag1')");
 
         sql("INSERT INTO `T$branch_test` VALUES (2, 10, 'hunterX')");