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 43ddab54c [core] tag auto creation snapshot idle timeout (#2646)
43ddab54c is described below
commit 43ddab54c6087ac404ba73c6d33f031b11d95dc3
Author: monster <[email protected]>
AuthorDate: Thu Jan 25 16:36:39 2024 +0800
[core] tag auto creation snapshot idle timeout (#2646)
---
.../shortcodes/generated/core_configuration.html | 6 +++++
.../main/java/org/apache/paimon/CoreOptions.java | 11 ++++++++
.../org/apache/paimon/tag/TagAutoCreation.java | 30 ++++++++++++++++++----
.../org/apache/paimon/tag/TagAutoCreationTest.java | 20 +++++++++++++++
4 files changed, 62 insertions(+), 5 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/core_configuration.html
b/docs/layouts/shortcodes/generated/core_configuration.html
index c8fc47052..3abe40262 100644
--- a/docs/layouts/shortcodes/generated/core_configuration.html
+++ b/docs/layouts/shortcodes/generated/core_configuration.html
@@ -653,6 +653,12 @@ This config option does not affect the default filesystem
metastore.</td>
<td>Integer</td>
<td>The maximum number of tags to retain.</td>
</tr>
+ <tr>
+ <td><h5>snapshot.watermark-idle-timeout</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Duration</td>
+ <td>In watermarking, if a source remains idle beyond the specified
timeout duration, it triggers snapshot advancement and facilitates tag
creation.</td>
+ </tr>
<tr>
<td><h5>tag.period-formatter</h5></td>
<td style="word-wrap: break-word;">with_dashes</td>
diff --git a/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
b/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
index 7c8cfeaa4..49828ac11 100644
--- a/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
@@ -991,6 +991,13 @@ public class CoreOptions implements Serializable {
.noDefaultValue()
.withDescription("The maximum number of tags to retain.");
+ public static final ConfigOption<Duration> SNAPSHOT_WATERMARK_IDLE_TIMEOUT
=
+ key("snapshot.watermark-idle-timeout")
+ .durationType()
+ .noDefaultValue()
+ .withDescription(
+ "In watermarking, if a source remains idle beyond
the specified timeout duration, it triggers snapshot advancement and
facilitates tag creation.");
+
public static final ConfigOption<Integer> PARQUET_ENABLE_DICTIONARY =
key("parquet.enable.dictionary")
.intType()
@@ -1544,6 +1551,10 @@ public class CoreOptions implements Serializable {
return options.get(TAG_NUM_RETAINED_MAX);
}
+ public Duration snapshotWatermarkIdleTimeout() {
+ return options.get(SNAPSHOT_WATERMARK_IDLE_TIMEOUT);
+ }
+
public String sinkWatermarkTimeZone() {
return options.get(SINK_WATERMARK_TIME_ZONE);
}
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 27a8413bb..31a9b1997 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
@@ -23,13 +23,16 @@ import org.apache.paimon.Snapshot;
import org.apache.paimon.operation.TagDeletion;
import org.apache.paimon.table.sink.TagCallback;
import org.apache.paimon.tag.TagTimeExtractor.ProcessTimeExtractor;
+import org.apache.paimon.tag.TagTimeExtractor.WatermarkExtractor;
import org.apache.paimon.utils.SnapshotManager;
import org.apache.paimon.utils.TagManager;
import javax.annotation.Nullable;
import java.time.Duration;
+import java.time.Instant;
import java.time.LocalDateTime;
+import java.time.ZoneId;
import java.util.List;
import java.util.Optional;
import java.util.SortedMap;
@@ -49,6 +52,7 @@ public class TagAutoCreation {
private final Duration delay;
private final Integer numRetainedMax;
private final List<TagCallback> callbacks;
+ private final Duration idlenessTimeout;
private LocalDateTime nextTag;
private long nextSnapshot;
@@ -61,6 +65,7 @@ public class TagAutoCreation {
TagPeriodHandler periodHandler,
Duration delay,
Integer numRetainedMax,
+ Duration idlenessTimeout,
List<TagCallback> callbacks) {
this.snapshotManager = snapshotManager;
this.tagManager = tagManager;
@@ -70,6 +75,7 @@ public class TagAutoCreation {
this.delay = delay;
this.numRetainedMax = numRetainedMax;
this.callbacks = callbacks;
+ this.idlenessTimeout = idlenessTimeout;
this.periodHandler.validateDelay(delay);
@@ -89,11 +95,24 @@ public class TagAutoCreation {
}
public boolean forceCreatingSnapshot() {
- return timeExtractor instanceof ProcessTimeExtractor
- && (nextTag == null
- || isAfterOrEqual(
- LocalDateTime.now().minus(delay),
- periodHandler.nextTagTime(nextTag)));
+ if (timeExtractor instanceof WatermarkExtractor && idlenessTimeout !=
null) {
+ Snapshot latestSnapshot = snapshotManager.latestSnapshot();
+ if (latestSnapshot == null) {
+ return false;
+ }
+
+ LocalDateTime snapshotTime =
+ LocalDateTime.ofInstant(
+ Instant.ofEpochMilli(latestSnapshot.watermark()),
+ ZoneId.systemDefault());
+
+ return isAfterOrEqual(LocalDateTime.now().minus(idlenessTimeout),
snapshotTime);
+ } else if (timeExtractor instanceof ProcessTimeExtractor) {
+ return nextTag == null
+ || isAfterOrEqual(
+ LocalDateTime.now().minus(delay),
periodHandler.nextTagTime(nextTag));
+ }
+ return false;
}
public void run() {
@@ -179,6 +198,7 @@ public class TagAutoCreation {
TagPeriodHandler.create(options),
options.tagCreationDelay(),
options.tagNumRetainedMax(),
+ options.snapshotWatermarkIdleTimeout(),
callbacks);
}
}
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 4d5c3ebce..62f53aadb 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
@@ -37,6 +37,7 @@ 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.SNAPSHOT_WATERMARK_IDLE_TIMEOUT;
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;
@@ -301,6 +302,25 @@ public class TagAutoCreationTest extends
PrimaryKeyTableTestBase {
assertThat(tagManager.allTagNames()).contains("2023-07-18 12",
"many-tags-test");
}
+ @Test
+ public void testWatermarkIdleTimeoutForceCreatingSnapshot() throws
Exception {
+ Options options = new Options();
+ options.set(TAG_AUTOMATIC_CREATION, TagCreationMode.WATERMARK);
+ options.set(TAG_CREATION_PERIOD, TagCreationPeriod.HOURLY);
+ options.set(SINK_WATERMARK_TIME_ZONE,
ZoneId.systemDefault().toString());
+ options.set(SNAPSHOT_WATERMARK_IDLE_TIMEOUT.key(), "10 s");
+ FileStoreTable table = this.table.copy(options.toMap());
+ TableCommitImpl commit =
table.newCommit(commitUser).ignoreEmptyCommit(false);
+
+ commit.commit(new ManifestCommittable(0,
localZoneMills("2023-07-18T12:00:00")));
+ commit.commit(new ManifestCommittable(0,
localZoneMills("2023-07-18T12:00:10")));
+
+ Long watermark1 = table.snapshotManager().snapshot(1).watermark();
+ Long watermark2 = table.snapshotManager().snapshot(2).watermark();
+ assertThat(watermark2 - watermark1).isEqualTo(10000);
+ commit.close();
+ }
+
private long localZoneMills(String timestamp) {
return LocalDateTime.parse(timestamp)
.atZone(ZoneId.systemDefault())
