[core] Add custom tag expiration strategy

docs/layouts/shortcodes/generated/core_configuration.html

@@ -1038,6 +1038,12 @@
             <td>String</td>
             <td>Use customized name when creating tags in Batch mode.</td>
         </tr>
+        <tr>
+            <td><h5>tag.expiration-strategy</h5></td>
+            <td style="word-wrap: break-word;">hybrid</td>
+            <td><p>Enum</p></td>
+            <td>The strategy determines how to expire tags.<br /><br />Possible values:<ul><li>"retain-time": Expire tags by retain time only.</li><li>"retain-number": Expire tags by retain number only.</li><li>"hybrid": Expire tags by retain time and retain number.</li></ul></td>
+        </tr>
         <tr>
             <td><h5>tag.callback.#.param</h5></td>
             <td style="word-wrap: break-word;">(none)</td>

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

@@ -1430,6 +1430,12 @@ public class CoreOptions implements Serializable {
                     .noDefaultValue()
                     .withDescription("Use customized name when creating tags in Batch mode.");
 
+    public static final ConfigOption<TagExpireStrategy> TAG_EXPIRATION_STRATEGY =
+            key("tag.expiration-strategy")
+                    .enumType(TagExpireStrategy.class)
+                    .defaultValue(TagExpireStrategy.HYBRID)
+                    .withDescription("The strategy determines how to expire tags.");
+
     public static final ConfigOption<Duration> SNAPSHOT_WATERMARK_IDLE_TIMEOUT =
             key("snapshot.watermark-idle-timeout")
                     .durationType()
@@ -2545,6 +2551,10 @@ public String tagBatchCustomizedName() {
         return options.get(TAG_BATCH_CUSTOMIZED_NAME);
     }
 
+    public TagExpireStrategy tagExpireStrategy() {
+        return options.get(TAG_EXPIRATION_STRATEGY);
+    }
+
     public Duration snapshotWatermarkIdleTimeout() {
         return options.get(SNAPSHOT_WATERMARK_IDLE_TIMEOUT);
     }
@@ -3305,6 +3315,33 @@ public InlineElement getDescription() {
         }
     }
 
+    /** The tag expiration strategy. */
+    public enum TagExpireStrategy implements DescribedEnum {
+        RETAIN_TIME("retain-time", "Expire tags by retain time only"),
+
+        RETAIN_NUMBER("retain-number", "Expire tags by retain number only."),
+
+        HYBRID("hybrid", "Expire tags by retain time and retain number.");
+
+        private final String value;
+
+        private final String description;
+
+        TagExpireStrategy(String value, String description) {
+            this.value = value;
+            this.description = description;
+        }
+
+        public String toString() {
+            return value;
+        }
+
+        @Override
+        public InlineElement getDescription() {
+            return text(description);
+        }
+    }
+
     /** Specifies the strategy for selecting external storage paths. */
     public enum ExternalPathStrategy implements DescribedEnum {
         NONE(

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

@@ -30,19 +30,19 @@
 public class TagAutoManager {
 
     private final TagAutoCreation tagAutoCreation;
-    private final TagTimeExpire tagTimeExpire;
+    private final TagExpire tagExpire;
 
-    private TagAutoManager(TagAutoCreation tagAutoCreation, TagTimeExpire tagTimeExpire) {
+    private TagAutoManager(TagAutoCreation tagAutoCreation, TagExpire tagExpire) {
         this.tagAutoCreation = tagAutoCreation;
-        this.tagTimeExpire = tagTimeExpire;
+        this.tagExpire = tagExpire;
     }
 
     public void run() {
         if (tagAutoCreation != null) {
             tagAutoCreation.run();
         }
-        if (tagTimeExpire != null) {
-            tagTimeExpire.expire();
+        if (tagExpire != null) {
+            tagExpire.expire();
         }
     }
 
@@ -60,14 +60,17 @@ public static TagAutoManager create(
                         ? null
                         : TagAutoCreation.create(
                                 options, snapshotManager, tagManager, tagDeletion, callbacks),
-                TagTimeExpire.create(snapshotManager, tagManager, tagDeletion, callbacks));
+                options.tagCreationMode() == CoreOptions.TagCreationMode.BATCH
+                        ? null
+                        : TagExpire.createTagExpireStrategy(
+                                options, snapshotManager, tagManager, tagDeletion, callbacks));
     }
 
     public TagAutoCreation getTagAutoCreation() {
         return tagAutoCreation;
     }
 
-    public TagTimeExpire getTagTimeExpire() {
-        return tagTimeExpire;
+    public TagExpire getTagExpire() {
+        return tagExpire;
     }
 }

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

@@ -22,6 +22,7 @@
 import org.apache.paimon.Snapshot;
 import org.apache.paimon.operation.TagDeletion;
 import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.TagCallback;
 import org.apache.paimon.utils.SnapshotManager;
 import org.apache.paimon.utils.TagManager;
 
@@ -46,12 +47,19 @@ public class TagBatchCreation {
     private final SnapshotManager snapshotManager;
     private final TagDeletion tagDeletion;
 
+    private final TagExpire tagExpire;
+    private final List<TagCallback> callbacks;
+
     public TagBatchCreation(FileStoreTable table) {
         this.table = table;
         this.snapshotManager = table.snapshotManager();
         this.tagManager = table.tagManager();
         this.tagDeletion = table.store().newTagDeletion();
         this.options = table.coreOptions();
+        this.callbacks = table.store().createTagCallbacks(table);
+        this.tagExpire =
+                TagExpire.createTagExpireStrategy(
+                        options, snapshotManager, tagManager, tagDeletion, callbacks);
     }
 
     public void createTag() {
@@ -92,40 +100,6 @@ public void createTag() {
             }
         }
         // Expire the tag
-        expireTag();
-    }
-
-    private void expireTag() {
-        Integer tagNumRetainedMax = options.tagNumRetainedMax();
-        if (tagNumRetainedMax != null) {
-            if (snapshotManager.latestSnapshot() == null) {
-                return;
-            }
-            long tagCount = tagManager.tagCount();
-
-            while (tagCount > tagNumRetainedMax) {
-                for (List<String> tagNames : tagManager.tags().values()) {
-                    if (tagCount - tagNames.size() > tagNumRetainedMax) {
-                        tagManager.deleteAllTagsOfOneSnapshot(
-                                tagNames, tagDeletion, snapshotManager);
-                        tagCount = tagCount - tagNames.size();
-                    } else {
-                        List<String> sortedTagNames = tagManager.sortTagsOfOneSnapshot(tagNames);
-                        for (String toBeDeleted : sortedTagNames) {
-                            tagManager.deleteTag(
-                                    toBeDeleted,
-                                    tagDeletion,
-                                    snapshotManager,
-                                    table.store().createTagCallbacks(table));
-                            tagCount--;
-                            if (tagCount == tagNumRetainedMax) {
-                                break;
-                            }
-                        }
-                        break;
-                    }
-                }
-            }
-        }
+        tagExpire.expire();
     }
 }

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

@@ -0,0 +1,79 @@
+/*
+ * 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.tag;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.operation.TagDeletion;
+import org.apache.paimon.table.sink.TagCallback;
+import org.apache.paimon.utils.SnapshotManager;
+import org.apache.paimon.utils.TagManager;
+
+import java.time.LocalDateTime;
+import java.util.List;
+
+/** Strategy for tag expiration. */
+public abstract class TagExpire {
+
+    protected final CoreOptions options;
+
+    protected final SnapshotManager snapshotManager;
+
+    protected final TagManager tagManager;
+
+    protected final TagDeletion tagDeletion;
+
+    protected final List<TagCallback> callbacks;
+
+    protected TagExpire(
+            CoreOptions options,
+            SnapshotManager snapshotManager,
+            TagManager tagManager,
+            TagDeletion tagDeletion,
+            List<TagCallback> callbacks) {
+        this.options = options;
+        this.snapshotManager = snapshotManager;
+        this.tagManager = tagManager;
+        this.tagDeletion = tagDeletion;
+        this.callbacks = callbacks;
+    }
+
+    public abstract List<String> expire();
+
+    public abstract void withOlderThanTime(LocalDateTime olderThanTime);
+
+    public static TagExpire createTagExpireStrategy(
+            CoreOptions options,
+            SnapshotManager snapshotManager,
+            TagManager tagManager,
+            TagDeletion tagDeletion,
+            List<TagCallback> callbacks) {
+        switch (options.tagExpireStrategy()) {
+            case RETAIN_TIME:
+                return TagTimeExpire.create(
+                        options, snapshotManager, tagManager, tagDeletion, callbacks);
+            case RETAIN_NUMBER:
+                return TagNumberExpire.create(
+                        options, snapshotManager, tagManager, tagDeletion, callbacks);
+            case HYBRID:
+            default:
+                return TagHybridExpire.create(
+                        options, snapshotManager, tagManager, tagDeletion, callbacks);
+        }
+    }
+}

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

@@ -0,0 +1,73 @@
+/*
+ * 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.tag;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.operation.TagDeletion;
+import org.apache.paimon.table.sink.TagCallback;
+import org.apache.paimon.utils.SnapshotManager;
+import org.apache.paimon.utils.TagManager;
+
+import java.time.LocalDateTime;
+import java.util.List;
+
+/** A strategy to expire tags by retain time and retain number. */
+public class TagHybridExpire extends TagExpire {
+
+    private final TagTimeExpire tagTimeExpire;
+
+    private final TagNumberExpire tagNumberExpire;
+
+    public TagHybridExpire(
+            CoreOptions options,
+            SnapshotManager snapshotManager,
+            TagManager tagManager,
+            TagDeletion tagDeletion,
+            List<TagCallback> callbacks) {
+        super(options, snapshotManager, tagManager, tagDeletion, callbacks);
+        this.tagTimeExpire =
+                TagTimeExpire.create(options, snapshotManager, tagManager, tagDeletion, callbacks);
+        this.tagNumberExpire =
+                TagNumberExpire.create(
+                        options, snapshotManager, tagManager, tagDeletion, callbacks);
+    }
+
+    @Override
+    public List<String> expire() {
+        List<String> expired = tagTimeExpire.expire();
+        expired.addAll(tagNumberExpire.expire());
+        return expired;
+    }
+
+    @Override
+    public void withOlderThanTime(LocalDateTime olderThanTime) {
+        if (tagTimeExpire != null) {
+            tagTimeExpire.withOlderThanTime(olderThanTime);
+        }
+    }
+
+    public static TagHybridExpire create(
+            CoreOptions options,
+            SnapshotManager snapshotManager,
+            TagManager tagManager,
+            TagDeletion tagDeletion,
+            List<TagCallback> callbacks) {
+        return new TagHybridExpire(options, snapshotManager, tagManager, tagDeletion, callbacks);
+    }
+}