[flink] Add flink actions for branch (#3622)

docs/content/maintenance/manage-branches.md

@@ -54,6 +54,24 @@ CALL sys.create_branch('default.T', 'branch1');
 ```
 {{< /tab >}}
 
+{{< tab "Flink Action Jar" >}}
+
+Run the following command:
+
+```bash
+<FLINK_HOME>/bin/flink run \
+    /path/to/paimon-flink-action-{{< version >}}.jar \
+    create_branch \
+    --warehouse <warehouse-path> \
+    --database <database-name> \ 
+    --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 >}}
+
 {{< /tabs >}}
 
 ## Delete Branches
@@ -71,6 +89,22 @@ CALL sys.delete_branch('default.T', 'branch1');
 ```
 {{< /tab >}}
 
+{{< tab "Flink Action Jar" >}}
+
+Run the following command:
+
+```bash
+<FLINK_HOME>/bin/flink run \
+    /path/to/paimon-flink-action-{{< version >}}.jar \
+    delete_branch \
+    --warehouse <warehouse-path> \
+    --database <database-name> \ 
+    --table <table-name> \
+    --branch_name <branch-name> \
+    [--catalog_conf <paimon-catalog-conf> [--catalog_conf <paimon-catalog-conf> ...]]
+```
+{{< /tab >}}
+
 {{< /tabs >}}
 
 ## Read / Write With Branch
@@ -107,4 +141,20 @@ CALL sys.merge_branch('default.T', 'branch1');
 
 {{< /tab >}}
 
+{{< tab "Flink Action Jar" >}}
+
+Run the following command:
+
+```bash
+<FLINK_HOME>/bin/flink run \
+    /path/to/paimon-flink-action-{{< version >}}.jar \
+    merge_branch \
+    --warehouse <warehouse-path> \
+    --database <database-name> \ 
+    --table <table-name> \
+    --branch_name <branch-name> \
+    [--catalog_conf <paimon-catalog-conf> [--catalog_conf <paimon-catalog-conf> ...]]
+```
+{{< /tab >}}
+
 {{< /tabs >}}

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

@@ -0,0 +1,56 @@
+/*
+ * 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.flink.action;
+
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.Map;
+
+/** Create branch action for Flink. */
+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) {
+        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

@@ -0,0 +1,82 @@
+/*
+ * 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.flink.action;
+
+import org.apache.flink.api.java.tuple.Tuple3;
+
+import java.util.Map;
+import java.util.Optional;
+
+/** Factory to create {@link CreateBranchAction}. */
+public class CreateBranchActionFactory implements ActionFactory {
+
+    public static final String IDENTIFIER = "create_branch";
+
+    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() {
+        return IDENTIFIER;
+    }
+
+    @Override
+    public Optional<Action> create(MultipleParameterToolAdapter params) {
+        checkRequiredArgument(params, BRANCH_NAME);
+
+        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);
+        }
+
+        String branchName = params.get(BRANCH_NAME);
+
+        CreateBranchAction action =
+                new CreateBranchAction(
+                        tablePath.f0,
+                        tablePath.f1,
+                        tablePath.f2,
+                        catalogConfig,
+                        branchName,
+                        tagName,
+                        snapshot);
+        return Optional.of(action);
+    }
+
+    @Override
+    public void printHelp() {
+        System.out.println("Action \"create_branch\" create a branch from given tag.");
+        System.out.println();
+
+        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>]");
+        System.out.println();
+    }
+}

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

@@ -0,0 +1,42 @@
+/*
+ * 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.flink.action;
+
+import java.util.Map;
+
+/** Delete branch action for Flink. */
+public class DeleteBranchAction extends TableActionBase {
+
+    private final String branchName;
+
+    public DeleteBranchAction(
+            String warehouse,
+            String databaseName,
+            String tableName,
+            Map<String, String> catalogConfig,
+            String branchName) {
+        super(warehouse, databaseName, tableName, catalogConfig);
+        this.branchName = branchName;
+    }
+
+    @Override
+    public void run() throws Exception {
+        table.deleteBranch(branchName);
+    }
+}

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

@@ -0,0 +1,63 @@
+/*
+ * 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.flink.action;
+
+import org.apache.flink.api.java.tuple.Tuple3;
+
+import java.util.Map;
+import java.util.Optional;
+
+/** Factory to create {@link DeleteBranchAction}. */
+public class DeleteBranchActionFactory implements ActionFactory {
+
+    public static final String IDENTIFIER = "delete_branch";
+
+    private static final String BRANCH_NAME = "branch_name";
+
+    @Override
+    public String identifier() {
+        return IDENTIFIER;
+    }
+
+    @Override
+    public Optional<Action> create(MultipleParameterToolAdapter params) {
+        checkRequiredArgument(params, BRANCH_NAME);
+
+        Tuple3<String, String, String> tablePath = getTablePath(params);
+        Map<String, String> catalogConfig = optionalConfigMap(params, CATALOG_CONF);
+        String branchName = params.get(BRANCH_NAME);
+
+        DeleteBranchAction action =
+                new DeleteBranchAction(
+                        tablePath.f0, tablePath.f1, tablePath.f2, catalogConfig, branchName);
+        return Optional.of(action);
+    }
+
+    @Override
+    public void printHelp() {
+        System.out.println("Action \"delete_branch\" delete a branch by name.");
+        System.out.println();
+
+        System.out.println("Syntax:");
+        System.out.println(
+                "  delete_branch --warehouse <warehouse_path> --database <database_name> "
+                        + "--table <table_name> --branch_name <branch_name>");
+        System.out.println();
+    }
+}

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

@@ -0,0 +1,41 @@
+/*
+ * 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.flink.action;
+
+import java.util.Map;
+
+/** Merge branch action for Flink. */
+public class MergeBranchAction extends TableActionBase {
+    private final String branchName;
+
+    public MergeBranchAction(
+            String warehouse,
+            String databaseName,
+            String tableName,
+            Map<String, String> catalogConfig,
+            String branchName) {
+        super(warehouse, databaseName, tableName, catalogConfig);
+        this.branchName = branchName;
+    }
+
+    @Override
+    public void run() throws Exception {
+        table.mergeBranch(branchName);
+    }
+}