From 1be1ba19088cd17763d834cf1be3b8572bb48a72 Mon Sep 17 00:00:00 2001
From: Askwang <135721692+Askwang@users.noreply.github.com>
Date: Wed, 30 Oct 2024 23:02:27 +0800
Subject: [PATCH] 1

---
 .../java/org/apache/paimon/table/Table.java   |  7 +++--
 .../CreateAndDeleteTagProcedureTest.scala     | 28 ++++++++++++++++++-
 2 files changed, 32 insertions(+), 3 deletions(-)

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
index 613dfca3158a..11dc135a6253 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/Table.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/Table.java
@@ -33,6 +33,7 @@
 
 import java.io.Serializable;
 import java.time.Duration;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -126,8 +127,10 @@ default String fullName() {
 
     /** Delete tags, tags are separated by commas. */
     @Experimental
-    default void deleteTags(String tagNames) {
-        for (String tagName : tagNames.split(",")) {
+    default void deleteTags(String tagStr) {
+        String[] tagNames =
+                Arrays.stream(tagStr.split(",")).map(String::trim).toArray(String[]::new);
+        for (String tagName : tagNames) {
             deleteTag(tagName);
         }
     }
diff --git a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/CreateAndDeleteTagProcedureTest.scala b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/CreateAndDeleteTagProcedureTest.scala
index 3621d44b8395..6400cb88c02f 100644
--- a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/CreateAndDeleteTagProcedureTest.scala
+++ b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/procedure/CreateAndDeleteTagProcedureTest.scala
@@ -120,7 +120,7 @@ class CreateAndDeleteTagProcedureTest extends PaimonSparkTestBase with StreamTes
               spark.sql("SELECT tag_name FROM paimon.test.`T$tags`"),
               Row("test_tag_2") :: Row("test_tag_3") :: Nil)
 
-            // delete test_tag_1 and test_tag_2
+            // delete test_tag_2 and test_tag_3
             checkAnswer(
               spark.sql(
                 "CALL paimon.sys.delete_tag(table => 'test.T', tag => 'test_tag_2,test_tag_3')"),
@@ -199,4 +199,30 @@ class CreateAndDeleteTagProcedureTest extends PaimonSparkTestBase with StreamTes
       spark.sql("CALL paimon.sys.delete_tag(table => 'test.T', tag => 'test_tag')"),
       Row(true) :: Nil)
   }
+
+  test("Paimon Procedure: delete multiple tags") {
+    spark.sql("CREATE TABLE T (id INT, name STRING) USING PAIMON")
+    spark.sql("insert into T values (1, 'a')")
+
+    // create four tags
+    spark.sql("CALL paimon.sys.create_tag(table => 'test.T', tag => 'tag-1')")
+    spark.sql("CALL paimon.sys.create_tag(table => 'test.T', tag => 'tag-2')")
+    spark.sql("CALL paimon.sys.create_tag(table => 'test.T', tag => 'tag-3')")
+    spark.sql("CALL paimon.sys.create_tag(table => 'test.T', tag => 'tag-4')")
+    checkAnswer(spark.sql("SELECT count(*) FROM paimon.test.`T$tags`"), Row(4) :: Nil)
+
+    // multiple tags with no space
+    checkAnswer(
+      spark.sql("CALL paimon.sys.delete_tag(table => 'test.T', tag => 'tag-1,tag-2')"),
+      Row(true) :: Nil)
+    checkAnswer(
+      spark.sql("SELECT tag_name FROM paimon.test.`T$tags`"),
+      Row("tag-3") :: Row("tag-4") :: Nil)
+
+    // multiple tags with space
+    checkAnswer(
+      spark.sql("CALL paimon.sys.delete_tag(table => 'test.T', tag => 'tag-3, tag-4')"),
+      Row(true) :: Nil)
+    checkAnswer(spark.sql("SELECT tag_name FROM paimon.test.`T$tags`"), Nil)
+  }
 }