From 57da4488b4e0cc291c3c8c75b906d37fee6908c1 Mon Sep 17 00:00:00 2001 From: Zihan Dai Date: Mon, 17 Aug 2026 15:36:09 +1000 Subject: [PATCH] [core] Read expire_tags older_than as a wall clock, not as an instant parseTimestampData(s, p, tz) returns a Timestamp holding a true epoch milli, and Timestamp.toLocalDateTime() renders that milli as a UTC wall clock. Composing them shifts the cutoff by the JVM's UTC offset, while what it is compared against -- TagManager stores LocalDateTime.now() -- is a local wall clock. For the documented example the cutoff becomes 15:00 in America/New_York and 03:00 in Asia/Shanghai instead of the 11:00 the user typed, deleting tags that were meant to be kept or keeping ones meant to expire. The existing tests fed the procedure a string produced by fromLocalDateTime().getMillisecond() rendered through java.sql.Timestamp, which pre-shifts by exactly the offset the procedure re-applies, so they passed in every timezone. They now pass the plain wall clock a user types. --- .../flink/procedure/ExpireTagsProcedure.java | 5 +---- .../flink/procedure/ExpireTagsProcedure.java | 5 +---- .../flink/action/ExpireTagsActionTest.java | 8 ++++---- .../procedure/ExpireTagsProcedureITCase.java | 18 +++++++++--------- .../spark/procedure/ExpireTagsProcedure.java | 5 +---- .../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 037c4bb71d0f..6a3c2cc59c21 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 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 String[] call(ProcedureContext procedureContext, String tableId, String o 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 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 2c03cf31d15a..9b0ca3af8709 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 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 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 b07cf508163d..5d711e822fe3 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 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 void expireTags(boolean forceStartFlinkJob) throws Exception { // 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 void expireTags(boolean forceStartFlinkJob) throws Exception { "--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 e44769e6486a..1c59fbbb5c79 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 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 void testExpireTagsByOlderThanTime() throws Exception { // 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 void testExpireTagsByOlderThanTime() throws Exception { // 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 f8e685cf2e54..3f36105f39ae 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 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 InternalRow[] call(InternalRow args) { .getTagTimeExpire(); if (olderThanStr != null) { LocalDateTime olderThanTime = - DateTimeUtils.parseTimestampData( - olderThanStr, 3, TimeZone.getDefault()) - .toLocalDateTime(); + DateTimeUtils.toLocalDateTime(olderThanStr, 3); tagTimeExpire.withOlderThanTime(olderThanTime); } List 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 d1e1d3f52dba..86019d856f82 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 )