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

askwang · 2024-09-06T04:37:59Z

Purpose

Introduce expire_tags procedure, support expire tags by time.

-- expire tags by tagCreateTime and tagTimeRetained
CALL sys.expire_tags(table => 'default.T')
-- expire tags that tagCreateTime less than older_than
CALL sys.expire_tags(table => 'default.T', older_than => '2024-09-06 11:00:00')

Tests

API and Format

Documentation

JingsongLi · 2024-09-19T02:03:46Z

docs/content/flink/procedures.md

+      <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>


JingsongLi · 2024-09-24T01:54:20Z

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

+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT cases for {@link ExpireTagsAction}. */
+public class ExpireTagsActionITTest extends ActionITCaseBase {


ExpireTagsActionTest

JingsongLi · 2024-09-24T01:55:13Z

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

@@ -62,14 +66,23 @@ public void run() {
            if (createTime == null || timeRetained == null) {
                continue;
            }
-            if (LocalDateTime.now().isAfter(createTime.plus(timeRetained))) {
+            if ((olderThanTime == null


if ((LocalDateTime.now().isAfter(createTime.plus(timeRetained)))
|| (olderThanTime != null && olderThanTime.isAfter(createTime)))

For tag without createTime, can we just check its file creation time?

Add check file creation time for tag without createTime.
This if locial makes some adjustments, help me review it again

JingsongLi · 2024-09-24T01:55:43Z

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

@@ -62,14 +66,23 @@ public void run() {
            if (createTime == null || timeRetained == null) {
                continue;


For tag without createTime, can we just check its file creation time?

JingsongLi · 2024-09-24T01:56:17Z

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

@@ -164,6 +165,9 @@ default void deleteBranches(String branchNames) {
    @Experimental
    ExpireSnapshots newExpireChangelog();

+    /** Expire tags. */
+    TagTimeExpire newExpireTags();


Dont introduce this method to Table, it is not so useful.

addressed. Removed it, just depends on TagAutoManager.

askwang · 2024-09-25T09:37:33Z

@JingsongLi Thanks for review, updated

JingsongLi · 2024-09-26T08:09:45Z

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

+            if (createTime == null) {
+                FileStatus tagFileStatus;
+                try {
+                    tagFileStatus = fileIO.getFileStatus(tagManager.tagPath(tagName));


if (timeRetained != null && olderThanTime != null), in this case, we get file status for this tag.

maybe if (createTime == null && olderThanTime != null) ?
if timeRetained != null, the tag has own createTime, no need to get file status.

wwj6591812 · 2024-09-27T02:35:37Z

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

            }
-            if (LocalDateTime.now().isAfter(createTime.plus(timeRetained))) {
+            Duration timeRetained = tag.getTagTimeRetained();


1、Method invocation 'plus' may produce 'NullPointerException'，because timeRetained in Tag is nullable. So you should check and handle it.

2、Argument 'createTime' might be null，because createTime in Tag is nullable. So you should check and handle it.

3、Change line 81 if statement to :
boolean xxx = timeRetained != null && LocalDateTime.now().isAfter(createTime.plus(timeRetained));
boolean yyy = olderThanTime != null && olderThanTime.isAfter(createTime);
if (xxx || yyy) {
...
}

wwj6591812 · 2024-09-27T02:35:49Z

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

-            if (LocalDateTime.now().isAfter(createTime.plus(timeRetained))) {
+            Duration timeRetained = tag.getTagTimeRetained();
+            if ((timeRetained != null && LocalDateTime.now().isAfter(createTime.plus(timeRetained)))
+                    || (olderThanTime != null && olderThanTime.isAfter(createTime))) {
                LOG.info(
                        "Delete tag {}, because its existence time has reached its timeRetained of {}.",


LOG.info(
"Delete tag {}, because its existence time has reached its timeRetained of {} or" +
" its createTime {} is olderThan olderThanTime {}.",
tagName,
timeRetained,
createTime,
olderThanTime);

wwj6591812 · 2024-09-27T02:37:32Z

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

+            ProcedureContext procedureContext, String tableId, @Nullable String olderThanStr)
+            throws Catalog.TableNotExistException {
+        FileStoreTable fileStoreTable = (FileStoreTable) table(tableId);
+        FileStore fileStore = fileStoreTable.store();


inline line 56 to line 57

wwj6591812 · 2024-09-27T02:37:39Z

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

+/** A procedure to expire tags by time. */
+public class ExpireTagsProcedure extends ProcedureBase {
+
+    public static final String IDENTIFIER = "expire_tags";


wwj6591812 · 2024-09-27T02:38:02Z

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

+        List<String> expired = tagTimeExpire.expire();
+        return expired.isEmpty()
+                ? new Row[] {Row.of("No expired tags.")}
+                : expired.stream().map(x -> Row.of(x)).toArray(Row[]::new);


Change to : expired.stream().map(Row::of).toArray(Row[]::new);

wwj6591812 · 2024-09-27T02:38:26Z

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

+/** Factory to create {@link ExpireTagsAction}. */
+public class ExpireTagsActionFactory implements ActionFactory {
+
+    public static final String IDENTIFIER = "expire_tags";


wwj6591812 · 2024-09-27T02:39:32Z

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

        List<Pair<Tag, String>> tags = tagManager.tagObjects();
+        FileIO fileIO = snapshotManager.fileIO();


inline this to line 73:
tagFileStatus = snapshotManager.fileIO().getFileStatus(tagManager.tagPath(tagName));

wwj6591812 · 2024-09-27T02:40:22Z

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

+                tableIdent,
+                table -> {
+                    FileStoreTable fileStoreTable = (FileStoreTable) table;
+                    FileStore fileStore = fileStoreTable.store();


inline line 77 to line 79

wwj6591812 · 2024-09-27T02:42:40Z

Suggest change title to "Introduce expire_tags action and procedure with olderThan user custom time".

askwang · 2024-09-27T06:28:03Z

@wwj6591812 thank you, updated

askwang added 2 commits September 6, 2024 11:52

expire tags procedure

9ba6a4e

support flink

be3cbe0

askwang changed the title ~~[WIP][core] Introduce expire_tags procedure~~ [core] Introduce expire_tags procedure Sep 6, 2024

askwang changed the title ~~[core] Introduce expire_tags procedure~~ [WIP][core] Introduce expire_tags procedure Sep 10, 2024

add flink action

43e512a

askwang changed the title ~~[WIP][core] Introduce expire_tags procedure~~ [core] Introduce expire_tags procedure Sep 10, 2024

JingsongLi reviewed Sep 19, 2024

View reviewed changes

askwang added 3 commits September 19, 2024 17:31

rename

0d97581

1

19f4895

fix

559f322

JingsongLi reviewed Sep 24, 2024

View reviewed changes

fix comments

b48f58b

JingsongLi reviewed Sep 26, 2024

View reviewed changes

fix

64772ed

wwj6591812 reviewed Sep 27, 2024

View reviewed changes

fix comments

b5362c7

askwang changed the title ~~[core] Introduce expire_tags procedure~~ [core] Introduce expire_tags action and procedure with olderThan user custom time Sep 27, 2024

support flink 1.18

d6b616b

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

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

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

askwang commented Sep 6, 2024 •

edited

Loading

JingsongLi Sep 19, 2024

askwang Sep 19, 2024

JingsongLi Sep 24, 2024

JingsongLi Sep 24, 2024

askwang Sep 25, 2024

JingsongLi Sep 24, 2024

JingsongLi Sep 24, 2024

askwang Sep 25, 2024

askwang commented Sep 25, 2024

JingsongLi Sep 26, 2024

askwang Sep 26, 2024

wwj6591812 Sep 27, 2024

wwj6591812 Sep 27, 2024

wwj6591812 Sep 27, 2024

wwj6591812 Sep 27, 2024

wwj6591812 Sep 27, 2024

wwj6591812 Sep 27, 2024

wwj6591812 Sep 27, 2024

wwj6591812 Sep 27, 2024

wwj6591812 commented Sep 27, 2024

askwang commented Sep 27, 2024

		@@ -62,14 +66,23 @@ public void run() {
		if (createTime == null \|\| timeRetained == null) {
		continue;

		List<Pair<Tag, String>> tags = tagManager.tagObjects();
		FileIO fileIO = snapshotManager.fileIO();

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

Are you sure you want to change the base?

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

Conversation

askwang commented Sep 6, 2024 • edited Loading

Purpose

Tests

API and Format

Documentation

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

askwang commented Sep 25, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wwj6591812 commented Sep 27, 2024

askwang commented Sep 27, 2024

askwang commented Sep 6, 2024 •

edited

Loading