[incubator-paimon] branch master updated: [core] Fix that tag auto creation cannot work after restarting job (#1810)

[lzljs3620320]

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/incubator-paimon.git


The following commit(s) were added to refs/heads/master by this push:
     new 5d9a584a3 [core] Fix that tag auto creation cannot work after 
restarting job (#1810)
5d9a584a3 is described below

commit 5d9a584a348c927def6dfa4ca95f5d0c4822f58c
Author: yuzelin <33053040+yuze...@users.noreply.github.com>
AuthorDate: Mon Aug 14 18:14:20 2023 +0800

    [core] Fix that tag auto creation cannot work after restarting job (#1810)
---
 .../java/org/apache/paimon/tag/TagAutoCreation.java  |  8 +++++++-
 .../org/apache/paimon/tag/TagAutoCreationTest.java   | 20 +++++++++++++++++---
 2 files changed, 24 insertions(+), 4 deletions(-)

diff --git 
a/paimon-core/src/main/java/org/apache/paimon/tag/TagAutoCreation.java 
b/paimon-core/src/main/java/org/apache/paimon/tag/TagAutoCreation.java
index 4900d0c8f..2a08695b3 100644
--- a/paimon-core/src/main/java/org/apache/paimon/tag/TagAutoCreation.java
+++ b/paimon-core/src/main/java/org/apache/paimon/tag/TagAutoCreation.java
@@ -120,7 +120,13 @@ public class TagAutoCreation {
                 tryToTag(snapshotManager.snapshot(nextSnapshot));
                 nextSnapshot++;
             } else {
-                break;
+                // avoid snapshot has been expired
+                Long earliest = snapshotManager.earliestSnapshotId();
+                if (earliest != null && earliest > nextSnapshot) {
+                    nextSnapshot = earliest;
+                } else {
+                    break;
+                }
             }
         }
     }
diff --git 
a/paimon-core/src/test/java/org/apache/paimon/tag/TagAutoCreationTest.java 
b/paimon-core/src/test/java/org/apache/paimon/tag/TagAutoCreationTest.java
index 6c0ab0426..c98d2e169 100644
--- a/paimon-core/src/test/java/org/apache/paimon/tag/TagAutoCreationTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/tag/TagAutoCreationTest.java
@@ -35,6 +35,8 @@ import java.time.LocalDateTime;
 import java.time.ZoneId;
 
 import static org.apache.paimon.CoreOptions.SINK_WATERMARK_TIME_ZONE;
+import static org.apache.paimon.CoreOptions.SNAPSHOT_NUM_RETAINED_MAX;
+import static org.apache.paimon.CoreOptions.SNAPSHOT_NUM_RETAINED_MIN;
 import static org.apache.paimon.CoreOptions.TAG_AUTOMATIC_CREATION;
 import static org.apache.paimon.CoreOptions.TAG_CREATION_DELAY;
 import static org.apache.paimon.CoreOptions.TAG_CREATION_PERIOD;
@@ -72,9 +74,21 @@ public class TagAutoCreationTest extends 
PrimaryKeyTableTestBase {
         assertThat(tagManager.tags().values())
                 .containsOnly("2023-07-18 12", "2023-07-18 13", "2023-07-18 
14");
 
-        // test restore
-        commit = table.newCommit(commitUser).ignoreEmptyCommit(false);
-        commit.commit(new ManifestCommittable(5, 
utcMills("2023-07-18T16:00:00")));
+        // test restore with snapshot expiration
+        commit.commit(new ManifestCommittable(5, 
utcMills("2023-07-18T15:01:00")));
+        commit.commit(new ManifestCommittable(6, 
utcMills("2023-07-18T15:02:00")));
+
+        Options expireSetting = new Options();
+        expireSetting.set(SNAPSHOT_NUM_RETAINED_MIN, 1);
+        expireSetting.set(SNAPSHOT_NUM_RETAINED_MAX, 1);
+        commit = 
table.copy(expireSetting.toMap()).newCommit(commitUser).ignoreEmptyCommit(false);
+
+        // trigger snapshot expiration
+        commit.commit(new ManifestCommittable(7, 
utcMills("2023-07-18T15:03:00")));
+        commit.commit(new ManifestCommittable(8, 
utcMills("2023-07-18T15:04:00")));
+
+        // check tags
+        commit.commit(new ManifestCommittable(9, 
utcMills("2023-07-18T16:00:00")));
         assertThat(tagManager.tags().values())
                 .containsOnly("2023-07-18 13", "2023-07-18 14", "2023-07-18 
15");
     }