apache · PDGGK · Aug 17, 2026
diff --git a/...aimon-flink-1.18/src/main/java/org/apache/paimon/flink/procedure/ExpireTagsProcedure.java b/...aimon-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<String> expired = tagTimeExpire.expire();

diff --git a/...mon-flink-common/src/main/java/org/apache/paimon/flink/procedure/ExpireTagsProcedure.java b/...mon-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<String> expired = tagTimeExpire.expire();

diff --git a/...aimon-flink-common/src/test/java/org/apache/paimon/flink/action/ExpireTagsActionTest.java b/...aimon-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/...ink-common/src/test/java/org/apache/paimon/flink/procedure/ExpireTagsProcedureITCase.java b/...ink-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/...mon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ExpireTagsProcedure.java b/...mon-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<String> expired = tagTimeExpire.expire();