[core] Support enable and disable tag time expiration

Author: yuzelin

docs/layouts/shortcodes/generated/core_configuration.html

@@ -1344,6 +1344,12 @@
             <td><p>Enum</p></td>
             <td>The date format for tag periods.<br /><br />Possible values:<ul><li>"with_dashes": Dates and hours with dashes, e.g., 'yyyy-MM-dd HH'</li><li>"without_dashes": Dates and hours without dashes, e.g., 'yyyyMMdd HH'</li><li>"without_dashes_and_spaces": Dates and hours without dashes and spaces, e.g., 'yyyyMMddHH'</li></ul></td>
         </tr>
+        <tr>
+            <td><h5>tag.time-expire-enabled</h5></td>
+            <td style="word-wrap: break-word;">true</td>
+            <td>Boolean</td>
+            <td>Whether to enable tag expiration by retained time.</td>
+        </tr>
         <tr>
             <td><h5>target-file-size</h5></td>
             <td style="word-wrap: break-word;">(none)</td>

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -1612,6 +1612,12 @@ public InlineElement getDescription() {
                             "The default maximum time retained for newly created tags. "
                                     + "It affects both auto-created tags and manually created (by procedure) tags.");
 
+    public static final ConfigOption<Boolean> TAG_TIME_EXPIRE_ENABLED =
+            key("tag.time-expire-enabled")
+                    .booleanType()
+                    .defaultValue(true)
+                    .withDescription("Whether to enable tag expiration by retained time.");
+
     public static final ConfigOption<Boolean> TAG_AUTOMATIC_COMPLETION =
             key("tag.automatic-completion")
                     .booleanType()
@@ -3029,6 +3035,10 @@ public Duration tagDefaultTimeRetained() {
         return options.get(TAG_DEFAULT_TIME_RETAINED);
     }
 
+    public boolean tagTimeExpireEnabled() {
+        return options.get(TAG_TIME_EXPIRE_ENABLED);
+    }
+
     public boolean tagAutomaticCompletion() {
         return options.get(TAG_AUTOMATIC_COMPLETION);
     }

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

@@ -24,15 +24,18 @@
 import org.apache.paimon.utils.SnapshotManager;
 import org.apache.paimon.utils.TagManager;
 
+import javax.annotation.Nullable;
+
 import java.util.List;
 
 /** A manager to create and expire tags. */
 public class TagAutoManager {
 
-    private final TagAutoCreation tagAutoCreation;
-    private final TagTimeExpire tagTimeExpire;
+    @Nullable private final TagAutoCreation tagAutoCreation;
+    @Nullable private final TagTimeExpire tagTimeExpire;
 
-    private TagAutoManager(TagAutoCreation tagAutoCreation, TagTimeExpire tagTimeExpire) {
+    private TagAutoManager(
+            @Nullable TagAutoCreation tagAutoCreation, @Nullable TagTimeExpire tagTimeExpire) {
         this.tagAutoCreation = tagAutoCreation;
         this.tagTimeExpire = tagTimeExpire;
     }
@@ -60,13 +63,17 @@ public static TagAutoManager create(
                         ? null
                         : TagAutoCreation.create(
                                 options, snapshotManager, tagManager, tagDeletion, callbacks),
-                TagTimeExpire.create(snapshotManager, tagManager, tagDeletion, callbacks));
+                options.tagTimeExpireEnabled()
+                        ? TagTimeExpire.create(snapshotManager, tagManager, tagDeletion, callbacks)
+                        : null);
     }
 
+    @Nullable
     public TagAutoCreation getTagAutoCreation() {
         return tagAutoCreation;
     }
 
+    @Nullable
     public TagTimeExpire getTagTimeExpire() {
         return tagTimeExpire;
     }

paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/ExpireTagsProcedure.java

@@ -18,6 +18,7 @@
 
 package org.apache.paimon.flink.procedure;
 
+import org.apache.paimon.CoreOptions;
 import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.tag.TagTimeExpire;
@@ -26,6 +27,7 @@
 import org.apache.flink.table.procedure.ProcedureContext;
 
 import java.time.LocalDateTime;
+import java.util.Collections;
 import java.util.List;
 import java.util.TimeZone;
 
@@ -42,6 +44,10 @@ public String[] call(ProcedureContext procedureContext, String tableId)
     public String[] call(ProcedureContext procedureContext, String tableId, String olderThanStr)
             throws Catalog.TableNotExistException {
         FileStoreTable fileStoreTable = (FileStoreTable) table(tableId);
+        fileStoreTable =
+                fileStoreTable.copy(
+                        Collections.singletonMap(
+                                CoreOptions.TAG_TIME_EXPIRE_ENABLED.key(), "true"));
         TagTimeExpire tagTimeExpire =
                 fileStoreTable.store().newTagAutoManager(fileStoreTable).getTagTimeExpire();
         if (olderThanStr != null) {

paimon-flink/paimon-flink-1.18/src/test/java/org/apache/paimon/flink/procedure/ProcedurePositionalArgumentsITCase.java

@@ -32,6 +32,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatCode;
@@ -499,14 +500,18 @@ public void testRewriteFileIndex() {
 
     @Test
     public void testExpireTags() throws Exception {
+        boolean tagTimeExpireEnabled = ThreadLocalRandom.current().nextBoolean();
         sql(
                 "CREATE TABLE T ("
                         + " k STRING,"
                         + " dt STRING,"
                         + " PRIMARY KEY (k, dt) NOT ENFORCED"
                         + ") PARTITIONED BY (dt) WITH ("
-                        + " 'bucket' = '1'"
-                        + ")");
+                        + " 'bucket' = '1',"
+                        + " 'tag.time-expire-enabled' = '%s'"
+                        + ")",
+                tagTimeExpireEnabled);
+
         FileStoreTable table = paimonTable("T");
         for (int i = 1; i <= 3; i++) {
             sql("INSERT INTO T VALUES ('" + i + "', '" + i + "')");

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/ExpireTagsProcedure.java

@@ -18,6 +18,7 @@
 
 package org.apache.paimon.flink.procedure;
 
+import org.apache.paimon.CoreOptions;
 import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.tag.TagTimeExpire;
@@ -32,6 +33,7 @@
 import javax.annotation.Nullable;
 
 import java.time.LocalDateTime;
+import java.util.Collections;
 import java.util.List;
 import java.util.TimeZone;
 
@@ -52,6 +54,10 @@ public class ExpireTagsProcedure extends ProcedureBase {
             ProcedureContext procedureContext, String tableId, @Nullable String olderThanStr)
             throws Catalog.TableNotExistException {
         FileStoreTable fileStoreTable = (FileStoreTable) table(tableId);
+        fileStoreTable =
+                fileStoreTable.copy(
+                        Collections.singletonMap(
+                                CoreOptions.TAG_TIME_EXPIRE_ENABLED.key(), "true"));
         TagTimeExpire tagTimeExpire =
                 fileStoreTable.store().newTagAutoManager(fileStoreTable).getTagTimeExpire();
         if (olderThanStr != null) {

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/ExpireTagsActionTest.java

@@ -29,6 +29,7 @@
 
 import java.nio.file.Path;
 import java.time.LocalDateTime;
+import java.util.concurrent.ThreadLocalRandom;
 
 import static org.apache.paimon.flink.util.ReadWriteTableTestUtil.bEnv;
 import static org.apache.paimon.flink.util.ReadWriteTableTestUtil.init;
@@ -47,10 +48,13 @@ public void setUp() {
     @ParameterizedTest
     @ValueSource(booleans = {false, true})
     public void testExpireTags(boolean startFlinkJob) throws Exception {
+        boolean tagTimeExpireEnabled = ThreadLocalRandom.current().nextBoolean();
         bEnv.executeSql(
                 "CREATE TABLE T (id STRING, name STRING,"
                         + " PRIMARY KEY (id) NOT ENFORCED)"
-                        + " WITH ('bucket'='1', 'write-only'='true')");
+                        + " WITH ('bucket'='1', 'write-only'='true', 'tag.time-expire-enabled' = '"
+                        + tagTimeExpireEnabled
+                        + "')");
 
         expireTags(startFlinkJob);
     }

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/ExpireTagsProcedureITCase.java

@@ -29,6 +29,7 @@
 import java.io.IOException;
 import java.time.LocalDateTime;
 import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -37,10 +38,12 @@ public class ExpireTagsProcedureITCase extends CatalogITCaseBase {
 
     @Test
     public void testExpireTagsByTagCreateTimeAndTagTimeRetained() throws Exception {
+        boolean tagTimeExpireEnabled = ThreadLocalRandom.current().nextBoolean();
         sql(
                 "CREATE TABLE T (id STRING, name STRING,"
                         + " PRIMARY KEY (id) NOT ENFORCED)"
-                        + " WITH ('bucket'='1', 'write-only'='true')");
+                        + " WITH ('bucket'='1', 'write-only'='true', 'tag.time-expire-enabled'='%s')",
+                tagTimeExpireEnabled);
 
         FileStoreTable table = paimonTable("T");
         SnapshotManager snapshotManager = table.snapshotManager();

paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ExpireTagsProcedure.java

@@ -18,6 +18,7 @@
 
 package org.apache.paimon.spark.procedure;
 
+import org.apache.paimon.CoreOptions;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.tag.TagTimeExpire;
 import org.apache.paimon.utils.DateTimeUtils;
@@ -31,6 +32,7 @@
 import org.apache.spark.unsafe.types.UTF8String;
 
 import java.time.LocalDateTime;
+import java.util.Collections;
 import java.util.List;
 import java.util.TimeZone;
 
@@ -73,6 +75,10 @@ public InternalRow[] call(InternalRow args) {
                 tableIdent,
                 table -> {
                     FileStoreTable fileStoreTable = (FileStoreTable) table;
+                    fileStoreTable =
+                            fileStoreTable.copy(
+                                    Collections.singletonMap(
+                                            CoreOptions.TAG_TIME_EXPIRE_ENABLED.key(), "true"));
                     TagTimeExpire tagTimeExpire =
                             fileStoreTable
                                     .store()

paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/ExpireTagsProcedureTest.scala

@@ -28,9 +28,11 @@ import org.assertj.core.api.Assertions.assertThat
 class ExpireTagsProcedureTest extends PaimonSparkTestBase {
 
   test("Paimon procedure: expire tags that reached its timeRetained") {
+    val tagTimeExpireEnabled = scala.util.Random.nextBoolean()
     spark.sql(s"""
                  |CREATE TABLE T (id STRING, name STRING)
                  |USING PAIMON
+                 |TBLPROPERTIES ('tag.time-expire-enabled' = '$tagTimeExpireEnabled')
                  |""".stripMargin)
 
     val table = loadTable("T")