[core] Introduce deleteTags and deleteBranches to Table

askwang · Jul 5, 2024 · db8bcd7 · db8bcd7
1 parent 4088bc7
commit db8bcd7
Show file tree

Hide file tree

Showing 9 changed files with 25 additions and 33 deletions.
diff --git a/docs/content/flink/procedures.md b/docs/content/flink/procedures.md
@@ -127,7 +127,7 @@ All available procedures are listed below.
       <td>
          To delete a tag. Arguments:
             <li>identifier: the target table identifier. Cannot be empty.</li>
-            <li>tagName: name of the tag to be deleted.If you specify multiple tags, delimiter is ','.</li>
+            <li>tagName: name of the tag to be deleted. If you specify multiple tags, delimiter is ','.</li>
       </td>
       <td>
          CALL sys.delete_tag('default.T', 'my_tag')
@@ -331,7 +331,7 @@ All available procedures are listed below.
       <td>
          To delete a branch. Arguments:
             <li>identifier: the target table identifier. Cannot be empty.</li>
-            <li>branchName: name of the branch to be deleted.If you specify multiple branches, delimiter is ','.</li>
+            <li>branchName: name of the branch to be deleted. If you specify multiple branches, delimiter is ','.</li>
       </td>
       <td>
          CALL sys.delete_branch('default.T', 'branch1')

diff --git a/docs/content/spark/procedures.md b/docs/content/spark/procedures.md
@@ -169,7 +169,7 @@ This section introduce all available spark procedures about paimon.
       <td>
          To merge a branch to main branch. Arguments:
             <li>table: the target table identifier. Cannot be empty.</li>
-            <li>branch: name of the branch to be merged.If you specify multiple branches, delimiter is ','.</li>
+            <li>branch: name of the branch to be merged. If you specify multiple branches, delimiter is ','.</li>
       </td>
       <td>
           CALL sys.delete_branch(table => 'test_db.T', branch => 'test_branch')

diff --git a/paimon-core/src/main/java/org/apache/paimon/table/Table.java b/paimon-core/src/main/java/org/apache/paimon/table/Table.java
@@ -95,6 +95,14 @@ default String fullName() {
     @Experimental
     void deleteTag(String tagName);
 
+    /** Delete tags, tags are separated by commas. */
+    @Experimental
+    default void deleteTags(String tagNames) {
+        for (String tagName : tagNames.split(",")) {
+            deleteTag(tagName);
+        }
+    }
+
     /** Rollback table's state to a specific tag. */
     @Experimental
     void rollbackTo(String tagName);
@@ -115,6 +123,14 @@ default String fullName() {
     @Experimental
     void deleteBranch(String branchName);
 
+    /** Delete branches, branches are separated by commas. */
+    @Experimental
+    default void deleteBranches(String branchNames) {
+        for (String branch : branchNames.split(",")) {
+            deleteBranch(branch);
+        }
+    }
+
     /** Merge a branch to main branch. */
     @Experimental
     void fastForward(String branchName);

diff --git a/.../paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DeleteBranchAction.java b/.../paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DeleteBranchAction.java
@@ -37,9 +37,6 @@ public DeleteBranchAction(
 
     @Override
     public void run() throws Exception {
-        String[] branches = branchNames.split(",");
-        for (String branch : branches) {
-            table.deleteBranch(branch);
-        }
+        table.deleteBranches(branchNames);
     }
 }
diff --git a/...ink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DeleteTagAction.java b/...ink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DeleteTagAction.java
@@ -37,9 +37,6 @@ public DeleteTagAction(
 
     @Override
     public void run() throws Exception {
-        String[] tagNames = tagNameStr.split(",");
-        for (String tagName : tagNames) {
-            table.deleteTag(tagName);
-        }
+        table.deleteTags(tagNameStr);
     }
 }
diff --git a/...n-flink-common/src/main/java/org/apache/paimon/flink/procedure/DeleteBranchProcedure.java b/...n-flink-common/src/main/java/org/apache/paimon/flink/procedure/DeleteBranchProcedure.java
@@ -20,7 +20,6 @@
 
 import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.catalog.Identifier;
-import org.apache.paimon.table.Table;
 
 import org.apache.flink.table.procedure.ProcedureContext;
 
@@ -42,12 +41,7 @@ public String identifier() {
 
     public String[] call(ProcedureContext procedureContext, String tableId, String branchStr)
             throws Catalog.TableNotExistException {
-        String[] branchs = branchStr.split(",");
-        for (String branch : branchs) {
-            Table table = catalog.getTable(Identifier.fromString(tableId));
-            table.deleteBranch(branch);
-        }
-
+        catalog.getTable(Identifier.fromString(tableId)).deleteBranches(branchStr);
         return new String[] {"Success"};
     }
 }
diff --git a/...imon-flink-common/src/main/java/org/apache/paimon/flink/procedure/DeleteTagProcedure.java b/...imon-flink-common/src/main/java/org/apache/paimon/flink/procedure/DeleteTagProcedure.java
@@ -38,11 +38,7 @@ public class DeleteTagProcedure extends ProcedureBase {
     public String[] call(ProcedureContext procedureContext, String tableId, String tagNameStr)
             throws Catalog.TableNotExistException {
         Table table = catalog.getTable(Identifier.fromString(tableId));
-        String[] tagNames = tagNameStr.split(",");
-        for (String tagName : tagNames) {
-            table.deleteTag(tagName);
-        }
-
+        table.deleteTags(tagNameStr);
         return new String[] {"Success"};
     }
 

diff --git a/...n-spark-common/src/main/java/org/apache/paimon/spark/procedure/DeleteBranchProcedure.java b/...n-spark-common/src/main/java/org/apache/paimon/spark/procedure/DeleteBranchProcedure.java
@@ -61,14 +61,10 @@ public StructType outputType() {
     public InternalRow[] call(InternalRow args) {
         Identifier tableIdent = toIdentifier(args.getString(0), PARAMETERS[0].name());
         String branchStr = args.getString(1);
-        String[] branches = branchStr.split(",");
-
         return modifyPaimonTable(
                 tableIdent,
                 table -> {
-                    for (String branch : branches) {
-                        table.deleteBranch(branch);
-                    }
+                    table.deleteBranches(branchStr);
                     InternalRow outputRow = newInternalRow(true);
                     return new InternalRow[] {outputRow};
                 });

diff --git a/...imon-spark-common/src/main/java/org/apache/paimon/spark/procedure/DeleteTagProcedure.java b/...imon-spark-common/src/main/java/org/apache/paimon/spark/procedure/DeleteTagProcedure.java
@@ -61,14 +61,10 @@ public StructType outputType() {
     public InternalRow[] call(InternalRow args) {
         Identifier tableIdent = toIdentifier(args.getString(0), PARAMETERS[0].name());
         String tagStr = args.getString(1);
-        String[] tags = tagStr.split(",");
-
         return modifyPaimonTable(
                 tableIdent,
                 table -> {
-                    for (String tag : tags) {
-                        table.deleteTag(tag);
-                    }
+                    table.deleteTags(tagStr);
                     InternalRow outputRow = newInternalRow(true);
                     return new InternalRow[] {outputRow};
                 });