This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 531b653e5 [core] Support rename tag action and procedure (#4277)
531b653e5 is described below
commit 531b653e5ff7eb6b2985a1532e34bd8d4268a088
Author: xuzifu666 <[email protected]>
AuthorDate: Tue Oct 8 10:04:50 2024 +0800
[core] Support rename tag action and procedure (#4277)
---
docs/content/spark/procedures.md | 12 +++
.../paimon/privilege/PrivilegedFileStoreTable.java | 6 ++
.../paimon/table/AbstractFileStoreTable.java | 5 ++
.../paimon/table/DelegatedFileStoreTable.java | 5 ++
.../java/org/apache/paimon/table/FormatTable.java | 5 ++
.../org/apache/paimon/table/ReadonlyTable.java | 8 ++
.../main/java/org/apache/paimon/table/Table.java | 3 +
.../java/org/apache/paimon/utils/TagManager.java | 18 ++++
.../paimon/flink/action/RenameTagAction.java | 52 ++++++++++++
.../flink/action/RenameTagActionFactory.java | 71 ++++++++++++++++
.../paimon/flink/procedure/RenameTagProcedure.java | 59 ++++++++++++++
.../services/org.apache.paimon.factories.Factory | 2 +
.../paimon/flink/action/TagActionITCase.java | 95 ++++++++++++++++++++++
.../flink/procedure/CreateTagsProcedureITCase.java | 24 ++++++
.../sql/CreateAndDeleteTagProcedureTest.scala | 10 ++-
.../org/apache/paimon/spark/SparkProcedures.java | 2 +
.../paimon/spark/procedure/RenameTagProcedure.java | 95 ++++++++++++++++++++++
.../CreateAndDeleteTagProcedureTest.scala | 12 ++-
18 files changed, 482 insertions(+), 2 deletions(-)
diff --git a/docs/content/spark/procedures.md b/docs/content/spark/procedures.md
index 9f6e061c1..1557fd6a7 100644
--- a/docs/content/spark/procedures.md
+++ b/docs/content/spark/procedures.md
@@ -109,6 +109,18 @@ This section introduce all available spark procedures
about paimon.
CALL sys.create_tag_from_timestamp(`table` => 'default.T', `tag` =>
'my_tag', `timestamp` => 1724404318750, time_retained => '1 d')
</td>
</tr>
+ <tr>
+ <td>rename_tag</td>
+ <td>
+ Rename a tag with a new tag name. Arguments:
+ <li>table: the target table identifier. Cannot be empty.</li>
+ <li>tag_name: name of the tag. Cannot be empty.</li>
+ <li>target_tag_name: the new tag name to rename. Cannot be
empty.</li>
+ </td>
+ <td>
+ CALL sys.rename_tag(table => 'default.T', tag_name => 'tag1',
target_tag_name => 'tag2')
+ </td>
+ </tr>
<tr>
<td>delete_tag</td>
<td>
diff --git
a/paimon-core/src/main/java/org/apache/paimon/privilege/PrivilegedFileStoreTable.java
b/paimon-core/src/main/java/org/apache/paimon/privilege/PrivilegedFileStoreTable.java
index 6842f0142..4eb1a1a90 100644
---
a/paimon-core/src/main/java/org/apache/paimon/privilege/PrivilegedFileStoreTable.java
+++
b/paimon-core/src/main/java/org/apache/paimon/privilege/PrivilegedFileStoreTable.java
@@ -121,6 +121,12 @@ public class PrivilegedFileStoreTable extends
DelegatedFileStoreTable {
wrapped.createTag(tagName, timeRetained);
}
+ @Override
+ public void renameTag(String tagName, String targetTagName) {
+ privilegeChecker.assertCanInsert(identifier);
+ wrapped.renameTag(tagName, targetTagName);
+ }
+
@Override
public void createTag(String tagName, long fromSnapshotId, Duration
timeRetained) {
privilegeChecker.assertCanInsert(identifier);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
index a98f368c5..171b7325a 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
@@ -571,6 +571,11 @@ abstract class AbstractFileStoreTable implements
FileStoreTable {
tagManager().createTag(fromSnapshot, tagName, timeRetained,
store().createTagCallbacks());
}
+ @Override
+ public void renameTag(String tagName, String targetTagName) {
+ tagManager().renameTag(tagName, targetTagName);
+ }
+
@Override
public void deleteTag(String tagName) {
tagManager()
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/DelegatedFileStoreTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/DelegatedFileStoreTable.java
index dc758a1af..5d6331aa4 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/DelegatedFileStoreTable.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/DelegatedFileStoreTable.java
@@ -182,6 +182,11 @@ public abstract class DelegatedFileStoreTable implements
FileStoreTable {
wrapped.createTag(tagName, fromSnapshotId, timeRetained);
}
+ @Override
+ public void renameTag(String tagName, String targetTagName) {
+ wrapped.renameTag(tagName, targetTagName);
+ }
+
@Override
public void deleteTag(String tagName) {
wrapped.deleteTag(tagName);
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/FormatTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/FormatTable.java
index efd4a00d0..3224131d4 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/FormatTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/FormatTable.java
@@ -266,6 +266,11 @@ public interface FormatTable extends Table {
throw new UnsupportedOperationException();
}
+ @Override
+ default void renameTag(String tagName, String targetTagName) {
+ throw new UnsupportedOperationException();
+ }
+
@Override
default void deleteTag(String tagName) {
throw new UnsupportedOperationException();
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/ReadonlyTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/ReadonlyTable.java
index 82edaa667..4ae593b55 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/ReadonlyTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/ReadonlyTable.java
@@ -189,6 +189,14 @@ public interface ReadonlyTable extends InnerTable {
this.getClass().getSimpleName()));
}
+ @Override
+ default void renameTag(String tagName, String targetTagName) {
+ throw new UnsupportedOperationException(
+ String.format(
+ "Readonly Table %s does not support renameTag.",
+ this.getClass().getSimpleName()));
+ }
+
@Override
default void deleteTag(String tagName) {
throw new UnsupportedOperationException(
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 d542732e2..613dfca31 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
@@ -117,6 +117,9 @@ public interface Table extends Serializable {
@Experimental
void createTag(String tagName, Duration timeRetained);
+ @Experimental
+ void renameTag(String tagName, String targetTagName);
+
/** Delete a tag by name. */
@Experimental
void deleteTag(String tagName);
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
index f6e69fb28..8aecb1ab7 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/TagManager.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/TagManager.java
@@ -144,6 +144,24 @@ public class TagManager {
}
}
+ public void renameTag(String tagName, String targetTagName) {
+ try {
+ if (!tagExists(tagName)) {
+ throw new RuntimeException(
+ String.format("The specified tag name [%s] does not
exist.", tagName));
+ }
+ if (tagExists(targetTagName)) {
+ throw new RuntimeException(
+ String.format(
+ "The specified target tag name [%s] existed,
please set a non-existent tag name.",
+ targetTagName));
+ }
+ fileIO.rename(tagPath(tagName), tagPath(targetTagName));
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
/** Make sure the tagNames are ALL tags of one snapshot. */
public void deleteAllTagsOfOneSnapshot(
List<String> tagNames, TagDeletion tagDeletion, SnapshotManager
snapshotManager) {
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RenameTagAction.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RenameTagAction.java
new file mode 100644
index 000000000..3b12b4c11
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RenameTagAction.java
@@ -0,0 +1,52 @@
+/*
+ * 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.paimon.utils.StringUtils;
+
+import java.util.Map;
+
+/** Rename Tag action for Flink. */
+public class RenameTagAction extends TableActionBase {
+ private final String tagName;
+ private final String targetTagName;
+
+ public RenameTagAction(
+ String warehouse,
+ String databaseName,
+ String tableName,
+ Map<String, String> catalogConfig,
+ String tagName,
+ String targetTagName) {
+ super(warehouse, databaseName, tableName, catalogConfig);
+ this.tagName = tagName;
+ this.targetTagName = targetTagName;
+ }
+
+ @Override
+ public void run() throws Exception {
+ if (StringUtils.isEmpty(tagName) ||
StringUtils.isEmpty(targetTagName)) {
+ throw new RuntimeException(
+ String.format(
+ "The specified tag name [%s] or target tag name
[%s] cannot be empty.",
+ tagName, targetTagName));
+ }
+ table.renameTag(tagName, targetTagName);
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RenameTagActionFactory.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RenameTagActionFactory.java
new file mode 100644
index 000000000..84f174d39
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/RenameTagActionFactory.java
@@ -0,0 +1,71 @@
+/*
+ * 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 RenameTagActionFactory}. */
+public class RenameTagActionFactory implements ActionFactory {
+
+ private static final String IDENTIFIER = "rename_tag";
+
+ private static final String TAG_NAME = "tag_name";
+ private static final String TARGET_TAG_NAME = "target_tag_name";
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Optional<Action> create(MultipleParameterToolAdapter params) {
+ checkRequiredArgument(params, TAG_NAME);
+ checkRequiredArgument(params, TARGET_TAG_NAME);
+
+ Tuple3<String, String, String> tablePath = getTablePath(params);
+ Map<String, String> catalogConfig = optionalConfigMap(params,
CATALOG_CONF);
+ String tagName = params.get(TAG_NAME);
+ String targetTagName = params.get(TARGET_TAG_NAME);
+
+ RenameTagAction action =
+ new RenameTagAction(
+ tablePath.f0,
+ tablePath.f1,
+ tablePath.f2,
+ catalogConfig,
+ tagName,
+ targetTagName);
+ return Optional.of(action);
+ }
+
+ @Override
+ public void printHelp() {
+ System.out.println("Action \"rename_tag\" rename a tag with a new tag
name.");
+ System.out.println();
+
+ System.out.println("Syntax:");
+ System.out.println(
+ " rename_tag --warehouse <warehouse_path> --tag_name
<tag_name> "
+ + "--target_tag_name <target_tag_name>");
+ System.out.println();
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RenameTagProcedure.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RenameTagProcedure.java
new file mode 100644
index 000000000..5476e807b
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RenameTagProcedure.java
@@ -0,0 +1,59 @@
+/*
+ * 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.procedure;
+
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.table.Table;
+
+import org.apache.flink.table.annotation.ArgumentHint;
+import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.ProcedureHint;
+import org.apache.flink.table.procedure.ProcedureContext;
+
+/**
+ * Create tag procedure. Usage:
+ *
+ * <pre><code>
+ * CALL sys.rename_tag('tableId', 'tagName', 'targetTagName')
+ * </code></pre>
+ */
+public class RenameTagProcedure extends ProcedureBase {
+ private static final String IDENTIFIER = "rename_tag";
+
+ @ProcedureHint(
+ argument = {
+ @ArgumentHint(name = "table", type = @DataTypeHint("STRING")),
+ @ArgumentHint(name = "tagName", type =
@DataTypeHint("STRING")),
+ @ArgumentHint(name = "targetTagName", type =
@DataTypeHint("STRING"))
+ })
+ public String[] call(
+ ProcedureContext procedureContext, String tableId, String tagName,
String targetTagName)
+ throws Catalog.TableNotExistException {
+ Table table = catalog.getTable(Identifier.fromString(tableId));
+ table.renameTag(tagName, targetTagName);
+ String ret = String.format("Rename [%s] to [%s] successfully.",
tagName, targetTagName);
+ return new String[] {ret};
+ }
+
+ @Override
+ public String identifier() {
+ return IDENTIFIER;
+ }
+}
diff --git
a/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
b/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
index 9eb450d33..3ae35ade5 100644
---
a/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
+++
b/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory
@@ -36,6 +36,7 @@ org.apache.paimon.flink.action.MarkPartitionDoneActionFactory
org.apache.paimon.flink.action.CreateBranchActionFactory
org.apache.paimon.flink.action.DeleteBranchActionFactory
org.apache.paimon.flink.action.FastForwardActionFactory
+org.apache.paimon.flink.action.RenameTagActionFactory
org.apache.paimon.flink.action.RepairActionFactory
org.apache.paimon.flink.action.RewriteFileIndexActionFactory
org.apache.paimon.flink.action.ExpireSnapshotsActionFactory
@@ -67,6 +68,7 @@
org.apache.paimon.flink.procedure.privilege.DropPrivilegedUserProcedure
org.apache.paimon.flink.procedure.privilege.GrantPrivilegeToUserProcedure
org.apache.paimon.flink.procedure.privilege.RevokePrivilegeFromUserProcedure
org.apache.paimon.flink.procedure.RepairProcedure
+org.apache.paimon.flink.procedure.RenameTagProcedure
org.apache.paimon.flink.procedure.FastForwardProcedure
org.apache.paimon.flink.procedure.MarkPartitionDoneProcedure
org.apache.paimon.flink.procedure.CloneProcedure
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/TagActionITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/TagActionITCase.java
index aa98de387..7310a68df 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/TagActionITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/TagActionITCase.java
@@ -240,6 +240,101 @@ public class TagActionITCase extends ActionITCaseBase {
assertThat(tagManager.tagExists("tag3")).isFalse();
}
+ @ParameterizedTest(name = "{0}")
+ @ValueSource(strings = {"action", "procedure_indexed", "procedure_named"})
+ public void testRenameTag(String invoker) throws Exception {
+ init(warehouse);
+
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.BIGINT(),
DataTypes.STRING()},
+ new String[] {"k", "v"});
+ FileStoreTable table =
+ createFileStoreTable(
+ rowType,
+ Collections.emptyList(),
+ Collections.singletonList("k"),
+ Collections.emptyList(),
+ Collections.emptyMap());
+
+ StreamWriteBuilder writeBuilder =
table.newStreamWriteBuilder().withCommitUser(commitUser);
+ write = writeBuilder.newWrite();
+ commit = writeBuilder.newCommit();
+
+ // 3 snapshots
+ writeData(rowData(1L, BinaryString.fromString("Hi")));
+ writeData(rowData(2L, BinaryString.fromString("Hello")));
+ writeData(rowData(3L, BinaryString.fromString("Paimon")));
+
+ TagManager tagManager = new TagManager(table.fileIO(),
table.location());
+ switch (invoker) {
+ case "action":
+ createAction(
+ CreateTagAction.class,
+ "create_tag",
+ "--warehouse",
+ warehouse,
+ "--database",
+ database,
+ "--table",
+ tableName,
+ "--tag_name",
+ "tag2")
+ .run();
+ break;
+ case "procedure_indexed":
+ executeSQL(
+ String.format(
+ "CALL sys.create_tag('%s.%s', 'tag2', 2)",
database, tableName));
+ break;
+ case "procedure_named":
+ executeSQL(
+ String.format(
+ "CALL sys.create_tag(`table` => '%s.%s', tag
=> 'tag2', snapshot_id => cast(2 as bigint))",
+ database, tableName));
+ break;
+ default:
+ throw new UnsupportedOperationException(invoker);
+ }
+ assertThat(tagManager.tagExists("tag2")).isTrue();
+
+ switch (invoker) {
+ case "action":
+ createAction(
+ RenameTagAction.class,
+ "rename_tag",
+ "--warehouse",
+ warehouse,
+ "--database",
+ database,
+ "--table",
+ tableName,
+ "--tag_name",
+ "tag2",
+ "--target_tag_name",
+ "tag3")
+ .run();
+ break;
+ case "procedure_indexed":
+ executeSQL(
+ String.format(
+ "CALL sys.rename_tag('%s.%s', 'tag2', 'tag3')",
+ database, tableName));
+ break;
+ case "procedure_named":
+ executeSQL(
+ String.format(
+ "CALL sys.rename_tag(`table` => '%s.%s',
tagName => 'tag2', targetTagName => 'tag3')",
+ database, tableName));
+ break;
+ default:
+ throw new UnsupportedOperationException(invoker);
+ }
+
+ assertThat(tagManager.tagExists("tag2")).isFalse();
+ assertThat(tagManager.tagExists("tag3")).isTrue();
+ }
+
@ParameterizedTest(name = "{0}")
@ValueSource(strings = {"action", "procedure_indexed", "procedure_named"})
public void testCreateLatestTag(String invoker) throws Exception {
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CreateTagsProcedureITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CreateTagsProcedureITCase.java
index aef952b6c..77e2b74c5 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CreateTagsProcedureITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CreateTagsProcedureITCase.java
@@ -58,6 +58,30 @@ public class CreateTagsProcedureITCase extends
CatalogITCaseBase {
.containsExactlyInAnyOrder("+I[k, 2024-01-01]");
}
+ @Test
+ public void testRenameTag() {
+ sql(
+ "CREATE TABLE T ("
+ + " k STRING,"
+ + " dt STRING,"
+ + " PRIMARY KEY (k, dt) NOT ENFORCED"
+ + ") PARTITIONED BY (dt) WITH ("
+ + " 'bucket' = '1'"
+ + ")");
+ sql("insert into T values('k', '2024-01-01')");
+ sql("insert into T values('k2', '2024-01-02')");
+
+ sql("CALL sys.create_tag('default.T', 'tag1')");
+
+ assertThat(sql("select tag_name from
`T$tags`").stream().map(Row::toString))
+ .containsExactlyInAnyOrder("+I[tag1]");
+
+ sql("CALL sys.rename_tag('default.T', 'tag1', 'tag2')");
+
+ assertThat(sql("select tag_name from
`T$tags`").stream().map(Row::toString))
+ .containsExactlyInAnyOrder("+I[tag2]");
+ }
+
@Test
public void testThrowSnapshotNotExistException() {
sql(
diff --git
a/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/CreateAndDeleteTagProcedureTest.scala
b/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/CreateAndDeleteTagProcedureTest.scala
index c4b861b29..3f59e897e 100644
---
a/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/CreateAndDeleteTagProcedureTest.scala
+++
b/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/CreateAndDeleteTagProcedureTest.scala
@@ -75,8 +75,16 @@ class CreateAndDeleteTagProcedureTest extends
PaimonSparkTestBase with StreamTes
checkAnswer(
spark.sql("SELECT tag_name FROM paimon.test.`T$tags`"),
Row("test_tag") :: Nil)
+ // test rename_tag
checkAnswer(
- spark.sql("CALL paimon.sys.delete_tag(table => 'test.T', tag =>
'test_tag')"),
+ spark.sql(
+ "CALL paimon.sys.rename_tag(table => 'test.T', tag =>
'test_tag', target_tag => 'test_tag_1')"),
+ Row(true) :: Nil)
+ checkAnswer(
+ spark.sql("SELECT tag_name FROM paimon.test.`T$tags`"),
+ Row("test_tag_1") :: Nil)
+ checkAnswer(
+ spark.sql("CALL paimon.sys.delete_tag(table => 'test.T', tag =>
'test_tag_1')"),
Row(true) :: Nil)
checkAnswer(spark.sql("SELECT tag_name FROM
paimon.test.`T$tags`"), Nil)
checkAnswer(
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
index b8e43c19c..1d5bd9df2 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java
@@ -34,6 +34,7 @@ import
org.apache.paimon.spark.procedure.MigrateTableProcedure;
import org.apache.paimon.spark.procedure.Procedure;
import org.apache.paimon.spark.procedure.ProcedureBuilder;
import org.apache.paimon.spark.procedure.RemoveOrphanFilesProcedure;
+import org.apache.paimon.spark.procedure.RenameTagProcedure;
import org.apache.paimon.spark.procedure.RepairProcedure;
import org.apache.paimon.spark.procedure.ResetConsumerProcedure;
import org.apache.paimon.spark.procedure.RollbackProcedure;
@@ -61,6 +62,7 @@ public class SparkProcedures {
ImmutableMap.builder();
procedureBuilders.put("rollback", RollbackProcedure::builder);
procedureBuilders.put("create_tag", CreateTagProcedure::builder);
+ procedureBuilders.put("rename_tag", RenameTagProcedure::builder);
procedureBuilders.put(
"create_tag_from_timestamp",
CreateTagFromTimestampProcedure::builder);
procedureBuilders.put("delete_tag", DeleteTagProcedure::builder);
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RenameTagProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RenameTagProcedure.java
new file mode 100644
index 000000000..73c0dc353
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/RenameTagProcedure.java
@@ -0,0 +1,95 @@
+/*
+ * 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.spark.procedure;
+
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.Metadata;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+
+import static org.apache.spark.sql.types.DataTypes.StringType;
+
+/**
+ * Rename Tag Procedure. Usage:
+ *
+ * <pre><code>
+ * CALL sys.rename_tag(tag => 'tag0', target_tag => 'tag1')
+ * </code></pre>
+ */
+public class RenameTagProcedure extends BaseProcedure {
+
+ private static final ProcedureParameter[] PARAMETERS =
+ new ProcedureParameter[] {
+ ProcedureParameter.required("table", StringType),
+ ProcedureParameter.required("tag", StringType),
+ ProcedureParameter.required("target_tag", StringType)
+ };
+
+ private static final StructType OUTPUT_TYPE =
+ new StructType(
+ new StructField[] {
+ new StructField("result", DataTypes.BooleanType, true,
Metadata.empty())
+ });
+
+ protected RenameTagProcedure(TableCatalog tableCatalog) {
+ super(tableCatalog);
+ }
+
+ @Override
+ public ProcedureParameter[] parameters() {
+ return PARAMETERS;
+ }
+
+ @Override
+ public StructType outputType() {
+ return OUTPUT_TYPE;
+ }
+
+ @Override
+ public InternalRow[] call(InternalRow args) {
+ Identifier tableIdent = toIdentifier(args.getString(0),
PARAMETERS[0].name());
+ String tag = args.getString(1);
+ String targetTag = args.getString(2);
+
+ return modifyPaimonTable(
+ tableIdent,
+ table -> {
+ table.renameTag(tag, targetTag);
+ InternalRow outputRow = newInternalRow(true);
+ return new InternalRow[] {outputRow};
+ });
+ }
+
+ public static ProcedureBuilder builder() {
+ return new BaseProcedure.Builder<RenameTagProcedure>() {
+ @Override
+ public RenameTagProcedure doBuild() {
+ return new RenameTagProcedure(tableCatalog());
+ }
+ };
+ }
+
+ @Override
+ public String description() {
+ return "RenameTagProcedure";
+ }
+}
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 296680919..3dc90db00 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
@@ -110,10 +110,20 @@ class CreateAndDeleteTagProcedureTest extends
PaimonSparkTestBase with StreamTes
spark.sql("SELECT tag_name FROM paimon.test.`T$tags`"),
Row("test_tag_1") :: Row("test_tag_2") :: Nil)
+ // test rename_tag
+ checkAnswer(
+ spark.sql(
+ "CALL paimon.sys.rename_tag(table => 'test.T', tag =>
'test_tag_1', target_tag => 'test_tag_3')"),
+ Row(true) :: Nil
+ )
+ checkAnswer(
+ 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
checkAnswer(
spark.sql(
- "CALL paimon.sys.delete_tag(table => 'test.T', tag =>
'test_tag_1,test_tag_2')"),
+ "CALL paimon.sys.delete_tag(table => 'test.T', tag =>
'test_tag_2,test_tag_3')"),
Row(true) :: Nil)
checkAnswer(spark.sql("SELECT tag_name FROM
paimon.test.`T$tags`"), Nil)
