This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new b9a4671484 [core] Read expire_tags older_than as a wall clock, not as
an instant (#9263)
b9a4671484 is described below
commit b9a467148459eb0e9c39b2dd0603a38cfd926ba0
Author: ZIHAN DAI <[email protected]>
AuthorDate: Thu Aug 20 12:22:09 2026 +1000
[core] Read expire_tags older_than as a wall clock, not as an instant
(#9263)
---
.../paimon/flink/procedure/ExpireTagsProcedure.java | 5 +----
.../paimon/flink/procedure/ExpireTagsProcedure.java | 5 +----
.../paimon/flink/action/ExpireTagsActionTest.java | 8 ++++----
.../flink/procedure/ExpireTagsProcedureITCase.java | 18 +++++++++---------
.../paimon/spark/procedure/ExpireTagsProcedure.java | 5 +----
.../spark/procedure/ExpireTagsProcedureTest.scala | 16 +++++++---------
6 files changed, 23 insertions(+), 34 deletions(-)
diff --git
a/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/ExpireTagsProcedure.java
b/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/ExpireTagsProcedure.java
index 037c4bb71d..6a3c2cc59c 100644
---
a/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/ExpireTagsProcedure.java
+++
b/paimon-flink/paimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/ExpireTagsProcedure.java
@@ -29,7 +29,6 @@ import org.apache.flink.table.procedure.ProcedureContext;
import java.time.LocalDateTime;
import java.util.Collections;
import java.util.List;
-import java.util.TimeZone;
/** A procedure to expire tags by time. */
public class ExpireTagsProcedure extends ProcedureBase {
@@ -51,9 +50,7 @@ public class ExpireTagsProcedure extends ProcedureBase {
TagTimeExpire tagTimeExpire =
fileStoreTable.store().newTagAutoManager(fileStoreTable).getTagTimeExpire();
if (olderThanStr != null) {
- LocalDateTime olderThanTime =
- DateTimeUtils.parseTimestampData(olderThanStr, 3,
TimeZone.getDefault())
- .toLocalDateTime();
+ LocalDateTime olderThanTime =
DateTimeUtils.toLocalDateTime(olderThanStr, 3);
tagTimeExpire.withOlderThanTime(olderThanTime);
}
List<String> expired = tagTimeExpire.expire();
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/ExpireTagsProcedure.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/ExpireTagsProcedure.java
index 2c03cf31d1..9b0ca3af87 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/ExpireTagsProcedure.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/ExpireTagsProcedure.java
@@ -35,7 +35,6 @@ import javax.annotation.Nullable;
import java.time.LocalDateTime;
import java.util.Collections;
import java.util.List;
-import java.util.TimeZone;
/** A procedure to expire tags by time. */
public class ExpireTagsProcedure extends ProcedureBase {
@@ -61,9 +60,7 @@ public class ExpireTagsProcedure extends ProcedureBase {
TagTimeExpire tagTimeExpire =
fileStoreTable.store().newTagAutoManager(fileStoreTable).getTagTimeExpire();
if (olderThanStr != null) {
- LocalDateTime olderThanTime =
- DateTimeUtils.parseTimestampData(olderThanStr, 3,
TimeZone.getDefault())
- .toLocalDateTime();
+ LocalDateTime olderThanTime =
DateTimeUtils.toLocalDateTime(olderThanStr, 3);
tagTimeExpire.withOlderThanTime(olderThanTime);
}
List<String> expired = tagTimeExpire.expire();
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/ExpireTagsActionTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/ExpireTagsActionTest.java
index b07cf50816..5d711e822f 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/ExpireTagsActionTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/ExpireTagsActionTest.java
@@ -18,7 +18,6 @@
package org.apache.paimon.flink.action;
-import org.apache.paimon.data.Timestamp;
import org.apache.paimon.table.FileStoreTable;
import org.junit.jupiter.api.BeforeEach;
@@ -29,6 +28,7 @@ import org.junit.jupiter.params.provider.ValueSource;
import java.nio.file.Path;
import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
import java.util.concurrent.ThreadLocalRandom;
import static org.apache.paimon.flink.util.ReadWriteTableTestUtil.bEnv;
@@ -128,8 +128,8 @@ public class ExpireTagsActionTest extends ActionITCaseBase {
// tag-3 as the base older_than time
LocalDateTime olderThanTime =
table.tagManager().getOrThrow("tag-3").getTagCreateTime();
- java.sql.Timestamp timestamp =
- new
java.sql.Timestamp(Timestamp.fromLocalDateTime(olderThanTime).getMillisecond());
+ String timestamp =
+ DateTimeFormatter.ofPattern("yyyy-MM-dd
HH:mm:ss.SSS").format(olderThanTime);
createAction(
ExpireTagsAction.class,
@@ -141,7 +141,7 @@ public class ExpireTagsActionTest extends ActionITCaseBase {
"--table",
"T",
"--older_than",
- timestamp.toString(),
+ timestamp,
"--force_start_flink_job",
Boolean.toString(forceStartFlinkJob))
.run();
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/ExpireTagsProcedureITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/ExpireTagsProcedureITCase.java
index e44769e648..1c59fbbb5c 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/ExpireTagsProcedureITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/ExpireTagsProcedureITCase.java
@@ -18,7 +18,6 @@
package org.apache.paimon.flink.procedure;
-import org.apache.paimon.data.Timestamp;
import org.apache.paimon.flink.CatalogITCaseBase;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.utils.SnapshotManager;
@@ -28,6 +27,7 @@ import org.junit.jupiter.api.Test;
import java.io.IOException;
import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;
@@ -106,13 +106,11 @@ public class ExpireTagsProcedureITCase extends
CatalogITCaseBase {
// tag-2 as the base older_than time.
// tag-1 expired by its file creation time.
LocalDateTime olderThanTime1 =
table.tagManager().getOrThrow("tag-2").getTagCreateTime();
- java.sql.Timestamp timestamp1 =
- new java.sql.Timestamp(
-
Timestamp.fromLocalDateTime(olderThanTime1).getMillisecond());
+ String timestamp1 = WALL_CLOCK.format(olderThanTime1);
assertThat(
sql(
"CALL sys.expire_tags(`table` => 'default.T',
older_than => '"
- + timestamp1.toString()
+ + timestamp1
+ "')"))
.containsExactlyInAnyOrder(Row.of("tag-1"));
@@ -123,19 +121,21 @@ public class ExpireTagsProcedureITCase extends
CatalogITCaseBase {
// tag-4 as the base older_than time.
// tag-2,tag-3,tag-5 expired, tag-5 reached its tagTimeRetained.
LocalDateTime olderThanTime2 =
table.tagManager().getOrThrow("tag-4").getTagCreateTime();
- java.sql.Timestamp timestamp2 =
- new java.sql.Timestamp(
-
Timestamp.fromLocalDateTime(olderThanTime2).getMillisecond());
+ String timestamp2 = WALL_CLOCK.format(olderThanTime2);
assertThat(
sql(
"CALL sys.expire_tags(`table` => 'default.T',
older_than => '"
- + timestamp2.toString()
+ + timestamp2
+ "')"))
.containsExactlyInAnyOrder(Row.of("tag-2"), Row.of("tag-3"),
Row.of("tag-5"));
assertThat(sql("select tag_name from
`T$tags`")).containsExactly(Row.of("tag-4"));
}
+ /** The plain wall clock a user types, as the documented example does. */
+ private static final DateTimeFormatter WALL_CLOCK =
+ DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS");
+
private void checkSnapshots(SnapshotManager sm, int earliest, int latest)
throws IOException {
assertThat(sm.snapshotCount()).isEqualTo(latest - earliest + 1);
assertThat(sm.earliestSnapshotId()).isEqualTo(earliest);
diff --git
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ExpireTagsProcedure.java
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ExpireTagsProcedure.java
index f8e685cf2e..3f36105f39 100644
---
a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ExpireTagsProcedure.java
+++
b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ExpireTagsProcedure.java
@@ -34,7 +34,6 @@ import org.apache.spark.unsafe.types.UTF8String;
import java.time.LocalDateTime;
import java.util.Collections;
import java.util.List;
-import java.util.TimeZone;
import static org.apache.spark.sql.types.DataTypes.StringType;
@@ -86,9 +85,7 @@ public class ExpireTagsProcedure extends BaseProcedure {
.getTagTimeExpire();
if (olderThanStr != null) {
LocalDateTime olderThanTime =
- DateTimeUtils.parseTimestampData(
- olderThanStr, 3,
TimeZone.getDefault())
- .toLocalDateTime();
+ DateTimeUtils.toLocalDateTime(olderThanStr, 3);
tagTimeExpire.withOlderThanTime(olderThanTime);
}
List<String> expired = tagTimeExpire.expire();
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/ExpireTagsProcedureTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/ExpireTagsProcedureTest.scala
index d1e1d3f52d..86019d856f 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/ExpireTagsProcedureTest.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/ExpireTagsProcedureTest.scala
@@ -18,7 +18,6 @@
package org.apache.paimon.spark.procedure;
-import org.apache.paimon.data.Timestamp
import org.apache.paimon.spark.PaimonSparkTestBase
import org.apache.paimon.utils.SnapshotManager
@@ -27,6 +26,9 @@ import org.assertj.core.api.Assertions.assertThat
class ExpireTagsProcedureTest extends PaimonSparkTestBase {
+ /** The plain wall clock a user types, as the documented example does. */
+ private val WALL_CLOCK =
java.time.format.DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS")
+
test("Paimon procedure: expire tags that reached its timeRetained") {
val tagTimeExpireEnabled = scala.util.Random.nextBoolean()
spark.sql(s"""
@@ -100,11 +102,9 @@ class ExpireTagsProcedureTest extends PaimonSparkTestBase {
// tag-2 as the base older_than time.
// tag-1 expired by its file creation time.
val olderThanTime1 =
table.tagManager().getOrThrow("tag-2").getTagCreateTime
- val timestamp1 =
- new
java.sql.Timestamp(Timestamp.fromLocalDateTime(olderThanTime1).getMillisecond)
+ val timestamp1 = WALL_CLOCK.format(olderThanTime1)
checkAnswer(
- spark.sql(
- s"CALL paimon.sys.expire_tags(table => 'test.T', older_than =>
'${timestamp1.toString}')"),
+ spark.sql(s"CALL paimon.sys.expire_tags(table => 'test.T', older_than =>
'$timestamp1')"),
Row("tag-1") :: Nil
)
@@ -115,11 +115,9 @@ class ExpireTagsProcedureTest extends PaimonSparkTestBase {
// tag-4 as the base older_than time.
// tag-2,tag-3,tag-5 expired, tag-5 reached its tagTimeRetained.
val olderThanTime2 =
table.tagManager().getOrThrow("tag-4").getTagCreateTime
- val timestamp2 =
- new
java.sql.Timestamp(Timestamp.fromLocalDateTime(olderThanTime2).getMillisecond)
+ val timestamp2 = WALL_CLOCK.format(olderThanTime2)
checkAnswer(
- spark.sql(
- s"CALL paimon.sys.expire_tags(table => 'test.T', older_than =>
'${timestamp2.toString}')"),
+ spark.sql(s"CALL paimon.sys.expire_tags(table => 'test.T', older_than =>
'$timestamp2')"),
Row("tag-2") :: Row("tag-3") :: Row("tag-5") :: Nil
)
