[core] Introduce expire_tags action and procedure with olderThan user custom time

docs/content/flink/procedures.md

@@ -182,6 +182,20 @@ All available procedures are listed below.
          CALL sys.delete_tag('default.T', 'my_tag')
       </td>
    </tr>
+   <tr>
+      <td>expire_tags</td>
+      <td>
+         CALL [catalog.]sys.expire_tags('identifier', 'expiration_time')
+      </td>
+      <td>
+         To expire tags by time. Arguments:
+            <li>identifier: the target table identifier. Cannot be empty.</li>
+            <li>expiration_time: tagCreateTime before which tags will be removed.</li>
+      </td>
+      <td>
+         CALL sys.expire_tags(table => 'default.T', expiration_time => '2024-09-06 11:00:00')
+      </td>
+   </tr>
    <tr>
       <td>merge_into</td>
       <td>

docs/content/spark/procedures.md

@@ -118,6 +118,17 @@ This section introduce all available spark procedures about paimon.
       </td>
       <td>CALL sys.delete_tag(table => 'default.T', tag => 'my_tag')</td>
     </tr>
+    <tr>
+      <td>expire_tags</td>
+      <td>
+         To expire tags by time. Arguments:
+            <li>table: the target table identifier. Cannot be empty.</li>
+            <li>expiration_time: tagCreateTime before which tags will be removed.</li>
+      </td>
+      <td>
+         CALL sys.expire_tags(table => 'default.T', expiration_time => '2024-09-06 11:00:00')
+      </td>
+    </tr>
     <tr>
       <td>rollback</td>
       <td>

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

@@ -37,6 +37,7 @@
 import org.apache.paimon.table.source.InnerTableRead;
 import org.apache.paimon.table.source.StreamDataTableScan;
 import org.apache.paimon.table.source.snapshot.SnapshotReader;
+import org.apache.paimon.tag.TagTimeExpire;
 import org.apache.paimon.utils.BranchManager;
 import org.apache.paimon.utils.SimpleFileReader;
 import org.apache.paimon.utils.TagManager;
@@ -194,6 +195,12 @@ public ExpireSnapshots newExpireChangelog() {
         return wrapped.newExpireChangelog();
     }
 
+    @Override
+    public TagTimeExpire newExpireTags() {
+        privilegeChecker.assertCanInsert(identifier);
+        return wrapped.newExpireTags();
+    }
+
     @Override
     public FileStoreTable copyWithoutTimeTravel(Map<String, String> dynamicOptions) {
         return new PrivilegedFileStoreTable(

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

@@ -59,6 +59,7 @@
 import org.apache.paimon.table.source.snapshot.StaticFromTimestampStartingScanner;
 import org.apache.paimon.table.source.snapshot.StaticFromWatermarkStartingScanner;
 import org.apache.paimon.tag.TagPreview;
+import org.apache.paimon.tag.TagTimeExpire;
 import org.apache.paimon.utils.BranchManager;
 import org.apache.paimon.utils.Preconditions;
 import org.apache.paimon.utils.SegmentsCache;
@@ -372,6 +373,15 @@ public ExpireSnapshots newExpireChangelog() {
                 snapshotManager(), tagManager(), store().newChangelogDeletion());
     }
 
+    @Override
+    public TagTimeExpire newExpireTags() {
+        return TagTimeExpire.create(
+                snapshotManager(),
+                tagManager(),
+                store().newTagDeletion(),
+                store().createTagCallbacks());
+    }
+
     @Override
     public TableCommitImpl newCommit(String commitUser) {
         CoreOptions options = coreOptions();

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

@@ -35,6 +35,7 @@
 import org.apache.paimon.table.source.InnerTableRead;
 import org.apache.paimon.table.source.StreamDataTableScan;
 import org.apache.paimon.table.source.snapshot.SnapshotReader;
+import org.apache.paimon.tag.TagTimeExpire;
 import org.apache.paimon.utils.BranchManager;
 import org.apache.paimon.utils.SegmentsCache;
 import org.apache.paimon.utils.SnapshotManager;
@@ -199,6 +200,11 @@ public ExpireSnapshots newExpireChangelog() {
         return wrapped.newExpireChangelog();
     }
 
+    @Override
+    public TagTimeExpire newExpireTags() {
+        return wrapped.newExpireTags();
+    }
+
     @Override
     public DataTableScan newScan() {
         return wrapped.newScan();

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

@@ -28,6 +28,7 @@
 import org.apache.paimon.table.sink.BatchWriteBuilder;
 import org.apache.paimon.table.sink.StreamWriteBuilder;
 import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.tag.TagTimeExpire;
 import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.SimpleFileReader;
 
@@ -306,6 +307,11 @@ default ExpireSnapshots newExpireChangelog() {
         throw new UnsupportedOperationException();
     }
 
+    @Override
+    default TagTimeExpire newExpireTags() {
+        throw new UnsupportedOperationException();
+    }
+
     @Override
     default ReadBuilder newReadBuilder() {
         throw new UnsupportedOperationException();

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

@@ -29,6 +29,7 @@
 import org.apache.paimon.table.sink.StreamWriteBuilder;
 import org.apache.paimon.table.sink.WriteSelector;
 import org.apache.paimon.table.source.StreamDataTableScan;
+import org.apache.paimon.tag.TagTimeExpire;
 import org.apache.paimon.utils.SimpleFileReader;
 
 import java.time.Duration;
@@ -252,4 +253,12 @@ default ExpireSnapshots newExpireChangelog() {
                         "Readonly Table %s does not support expireChangelog.",
                         this.getClass().getSimpleName()));
     }
+
+    @Override
+    default TagTimeExpire newExpireTags() {
+        throw new UnsupportedOperationException(
+                String.format(
+                        "Readonly Table %s does not support expireTags.",
+                        this.getClass().getSimpleName()));
+    }
 }

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

@@ -28,6 +28,7 @@
 import org.apache.paimon.table.sink.BatchWriteBuilder;
 import org.apache.paimon.table.sink.StreamWriteBuilder;
 import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.tag.TagTimeExpire;
 import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.SimpleFileReader;
 
@@ -164,6 +165,9 @@ default void deleteBranches(String branchNames) {
     @Experimental
     ExpireSnapshots newExpireChangelog();
 
+    /** Expire tags. */
+    TagTimeExpire newExpireTags();
+
     // =============== Read & Write Operations ==================
 
     /** Returns a new read builder. */

paimon-core/src/main/java/org/apache/paimon/tag/TagAutoManager.java

@@ -42,7 +42,7 @@ public void run() {
             tagAutoCreation.run();
         }
         if (tagTimeExpire != null) {
-            tagTimeExpire.run();
+            tagTimeExpire.expire();
         }
     }
 

paimon-core/src/main/java/org/apache/paimon/tag/TagTimeExpire.java

@@ -29,6 +29,7 @@
 
 import java.time.Duration;
 import java.time.LocalDateTime;
+import java.util.ArrayList;
 import java.util.List;
 
 /** A manager to expire tags by time. */
@@ -41,6 +42,8 @@ public class TagTimeExpire {
     private final TagDeletion tagDeletion;
     private final List<TagCallback> callbacks;
 
+    private LocalDateTime expirationTime;
+
     private TagTimeExpire(
             SnapshotManager snapshotManager,
             TagManager tagManager,
@@ -52,8 +55,9 @@ private TagTimeExpire(
         this.callbacks = callbacks;
     }
 
-    public void run() {
+    public List<String> expire() {
         List<Pair<Tag, String>> tags = tagManager.tagObjects();
+        List<String> expired = new ArrayList<>();
         for (Pair<Tag, String> pair : tags) {
             Tag tag = pair.getLeft();
             String tagName = pair.getRight();
@@ -62,14 +66,23 @@ public void run() {
             if (createTime == null || timeRetained == null) {
                 continue;
             }
-            if (LocalDateTime.now().isAfter(createTime.plus(timeRetained))) {
+            if ((expirationTime == null
+                            && LocalDateTime.now().isAfter(createTime.plus(timeRetained)))
+                    || (expirationTime != null && expirationTime.isAfter(createTime))) {
                 LOG.info(
                         "Delete tag {}, because its existence time has reached its timeRetained of {}.",
                         tagName,
                         timeRetained);
                 tagManager.deleteTag(tagName, tagDeletion, snapshotManager, callbacks);
+                expired.add(tagName);
             }
         }
+        return expired;
+    }
+
+    public TagTimeExpire withExpirationTime(LocalDateTime expirationTime) {
+        this.expirationTime = expirationTime;
+        return this;
     }
 
     public static TagTimeExpire create(

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

@@ -0,0 +1,49 @@
+/*
+ * 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.flink.procedure.ExpireTagsProcedure;
+
+import org.apache.flink.table.procedure.DefaultProcedureContext;
+
+import java.util.Map;
+
+/** Expire tags action for Flink. */
+public class ExpireTagsAction extends ActionBase {
+
+    private final String table;
+    private final String expirationTime;
+
+    public ExpireTagsAction(
+            String warehouse,
+            String table,
+            String expirationTime,
+            Map<String, String> catalogConfig) {
+        super(warehouse, catalogConfig);
+        this.table = table;
+        this.expirationTime = expirationTime;
+    }
+
+    @Override
+    public void run() throws Exception {
+        ExpireTagsProcedure expireTagsProcedure = new ExpireTagsProcedure();
+        expireTagsProcedure.withCatalog(catalog);
+        expireTagsProcedure.call(new DefaultProcedureContext(env), table, expirationTime);
+    }
+}

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/ExpireTagsActionFactory.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.action;
+
+import java.util.Map;
+import java.util.Optional;
+
+/** Factory to create {@link ExpireTagsAction}. */
+public class ExpireTagsActionFactory implements ActionFactory {
+
+    public static final String IDENTIFIER = "expire_tags";
+
+    private static final String EXPIRATION_TIME = "expiration_time";
+
+    @Override
+    public String identifier() {
+        return IDENTIFIER;
+    }
+
+    @Override
+    public Optional<Action> create(MultipleParameterToolAdapter params) {
+        String warehouse = params.get(WAREHOUSE);
+        String table = params.get(TABLE);
+        String expirationTime = params.get(EXPIRATION_TIME);
+        Map<String, String> catalogConfig = optionalConfigMap(params, CATALOG_CONF);
+
+        ExpireTagsAction expireTagsAction =
+                new ExpireTagsAction(warehouse, table, expirationTime, catalogConfig);
+        return Optional.of(expireTagsAction);
+    }
+
+    @Override
+    public void printHelp() {
+        System.out.println("Action \"expire_tags\" expire tags by time.");
+        System.out.println();
+
+        System.out.println("Syntax:");
+        System.out.println(
+                "  expire_tags --warehouse <warehouse_path> "
+                        + "--table <database.table_name> "
+                        + "[--expiration_time <expiration_time>]");
+    }
+}