[core] Expire TIMESTAMP partitions with the values-time strategy

paimon-core/src/main/java/org/apache/paimon/operation/NormalPartitionExpire.java

@@ -34,7 +34,6 @@
 import javax.annotation.Nullable;
 
 import java.time.Duration;
-import java.time.LocalDate;
 import java.time.LocalDateTime;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -161,14 +160,15 @@ private List<Map<String, String>> doExpire(
         List<List<String>> expiredPartValues = new ArrayList<>(partitionEntries.size());
         for (PartitionEntry partition : partitionEntries) {
             Object[] array = strategy.convertPartition(partition.partition());
-            // DATE values are normalized for time extraction; use the stored names for dropping.
+            // DATE and TIMESTAMP values are normalized for time extraction; use the stored
+            // names for dropping.
             Map<String, String> partitionSpec =
                     commit.pathFactory()
                             .partitionComputer()
                             .generatePartValues(partition.partition());
             int index = 0;
             for (String value : partitionSpec.values()) {
-                if (array[index] instanceof LocalDate) {
+                if (array[index] != null && strategy.isNormalizedForTimeExtraction(index)) {
                     array[index] = value;
                 }
                 index++;

paimon-core/src/main/java/org/apache/paimon/partition/PartitionExpireStrategy.java

@@ -22,10 +22,14 @@
 import org.apache.paimon.catalog.CatalogLoader;
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.Timestamp;
 import org.apache.paimon.manifest.PartitionEntry;
 import org.apache.paimon.operation.FileStoreScan;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypeChecks;
 import org.apache.paimon.types.DataTypeRoot;
 import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.DateTimeUtils;
 import org.apache.paimon.utils.RowDataToObjectArrayConverter;
 
 import org.apache.paimon.shade.guava30.com.google.common.primitives.UnsignedBytes;
@@ -42,6 +46,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.TimeZone;
 
 /** Strategy for partition expiration. */
 public abstract class PartitionExpireStrategy implements Serializable {
@@ -51,11 +56,20 @@ public abstract class PartitionExpireStrategy implements Serializable {
     protected final List<String> partitionKeys;
     protected final String partitionDefaultName;
     private final RowDataToObjectArrayConverter toObjectArrayConverter;
+    @Nullable private final String timestampFormatter;
 
     public PartitionExpireStrategy(RowType partitionType, String partitionDefaultName) {
+        this(partitionType, partitionDefaultName, null);
+    }
+
+    public PartitionExpireStrategy(
+            RowType partitionType,
+            String partitionDefaultName,
+            @Nullable String timestampFormatter) {
         this.toObjectArrayConverter = new RowDataToObjectArrayConverter(partitionType);
         this.partitionKeys = partitionType.getFieldNames();
         this.partitionDefaultName = partitionDefaultName;
+        this.timestampFormatter = timestampFormatter;
     }
 
     public Map<String, String> toPartitionString(Object[] array) {
@@ -78,18 +92,62 @@ public List<String> toPartitionValue(Object[] array) {
         return list;
     }
 
+    /**
+     * Whether {@link #convertPartition} renders the value at this index in a form meant for time
+     * extraction that can differ from the stored partition name.
+     */
+    public boolean isNormalizedForTimeExtraction(int index) {
+        switch (toObjectArrayConverter.rowType().getTypeAt(index).getTypeRoot()) {
+            case DATE:
+            case TIMESTAMP_WITHOUT_TIME_ZONE:
+            case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+                return true;
+            default:
+                return false;
+        }
+    }
+
     public Object[] convertPartition(BinaryRow partition) {
         Object[] values = toObjectArrayConverter.convert(partition);
         RowType partitionType = toObjectArrayConverter.rowType();
         for (int index = 0; index < values.length; index++) {
-            if (values[index] != null
-                    && partitionType.getTypeAt(index).getTypeRoot() == DataTypeRoot.DATE) {
-                values[index] = LocalDate.ofEpochDay((Integer) values[index]);
+            if (values[index] == null) {
+                continue;
+            }
+            DataType type = partitionType.getTypeAt(index);
+            switch (type.getTypeRoot()) {
+                case DATE:
+                    values[index] = LocalDate.ofEpochDay((Integer) values[index]);
+                    break;
+                case TIMESTAMP_WITHOUT_TIME_ZONE:
+                case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+                    values[index] =
+                            formatTimestampForTimeExtraction((Timestamp) values[index], type);
+                    break;
+                default:
+                    break;
             }
         }
         return values;
     }
 
+    private String formatTimestampForTimeExtraction(Timestamp timestamp, DataType type) {
+        if (type.is(DataTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE)) {
+            // Use the local wall-clock time, consistent with the current partition names and
+            // with the local current time used for expiration.
+            timestamp =
+                    DateTimeUtils.timestampWithLocalZoneToTimestamp(
+                            timestamp, TimeZone.getDefault());
+        }
+        if (timestampFormatter != null) {
+            // Keep the Timestamp#toString input an explicitly configured formatter expects.
+            return timestamp.toString();
+        }
+        // Timestamp#toString uses the ISO 'T' separator, which the default partition time
+        // formatter cannot parse. Render it as the SQL timestamp string.
+        return DateTimeUtils.formatTimestamp(timestamp, DataTypeChecks.getPrecision(type));
+    }
+
     public abstract List<PartitionEntry> selectExpiredPartitions(
             FileStoreScan scan, LocalDateTime expirationTime);
 

paimon-core/src/main/java/org/apache/paimon/partition/PartitionValuesTimeExpireStrategy.java

@@ -55,7 +55,7 @@ public class PartitionValuesTimeExpireStrategy extends PartitionExpireStrategy {
     private final boolean retainLatest;
 
     public PartitionValuesTimeExpireStrategy(CoreOptions options, RowType partitionType) {
-        super(partitionType, options.partitionDefaultName());
+        super(partitionType, options.partitionDefaultName(), options.partitionTimestampFormatter());
         String timePattern = options.partitionTimestampPattern();
         String timeFormatter = options.partitionTimestampFormatter();
         this.timeExtractor = new PartitionTimeExtractor(timePattern, timeFormatter);

paimon-core/src/test/java/org/apache/paimon/operation/PartitionExpireTest.java

@@ -24,6 +24,7 @@
 import org.apache.paimon.data.BinaryRow;
 import org.apache.paimon.data.BinaryString;
 import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.Timestamp;
 import org.apache.paimon.data.serializer.InternalRowSerializer;
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.fs.local.LocalFileIO;
@@ -67,6 +68,8 @@
 import org.junit.jupiter.params.provider.CsvSource;
 import org.junit.jupiter.params.provider.ValueSource;
 
+import javax.annotation.Nullable;
+
 import java.io.IOException;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
@@ -83,6 +86,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.TimeZone;
 import java.util.UUID;
 import java.util.concurrent.ThreadLocalRandom;
 
@@ -321,6 +325,146 @@ public void testDateAndHourPartitionValues() {
         assertThat(strategy.isExpired(partitionDate.atTime(13, 0), partition)).isTrue();
     }
 
+    @ParameterizedTest
+    @CsvSource({
+        "values-time, false",
+        "values-time, true",
+        "update-time, false",
+        "update-time, true"
+    })
+    public void testExpireTimestampPartitionWithoutMetastore(
+            String expirationStrategy, boolean legacyPartitionName) throws Exception {
+        SchemaManager schemaManager = new FileSystemSchemaManager(LocalFileIO.create(), path);
+        Map<String, String> tableOptions = new HashMap<>();
+        tableOptions.put(METASTORE_PARTITIONED_TABLE.key(), "false");
+        tableOptions.put(
+                CoreOptions.PARTITION_GENERATE_LEGACY_NAME.key(),
+                String.valueOf(legacyPartitionName));
+        schemaManager.createTable(
+                new Schema(
+                        RowType.of(DataTypes.TIMESTAMP(0), VarCharType.STRING_TYPE).getFields(),
+                        singletonList("f0"),
+                        emptyList(),
+                        tableOptions,
+                        ""));
+        newTable();
+        Timestamp expiredTs = Timestamp.fromLocalDateTime(date(1));
+        Timestamp retainedTs = Timestamp.fromLocalDateTime(date(5));
+        write(GenericRow.of(expiredTs, BinaryString.fromString("expired")));
+        write(GenericRow.of(retainedTs, BinaryString.fromString("retained")));
+
+        Map<String, String> options = new HashMap<>();
+        options.put(PARTITION_EXPIRATION_TIME.key(), "2 d");
+        options.put(CoreOptions.PARTITION_EXPIRATION_STRATEGY.key(), expirationStrategy);
+        options.put(CoreOptions.PARTITION_TIMESTAMP_PATTERN.key(), "$f0");
+        table = table.copy(options);
+        NormalPartitionExpire expire =
+                (NormalPartitionExpire) table.store().newPartitionExpire("", table);
+        expire.setLastCheck(date(1));
+        // the legacy name is Timestamp#toString, the new name is the SQL timestamp string
+        String expiredName = legacyPartitionName ? "2023-01-01T00:00" : "2023-01-01 00:00:00";
+        String retainedName = legacyPartitionName ? "2023-01-05T00:00" : "2023-01-05 00:00:00";
+        List<Map<String, String>> expectedExpired = new ArrayList<>();
+        expectedExpired.add(Collections.singletonMap("f0", expiredName));
+        LocalDateTime checkTime = date(6);
+        List<Timestamp> expectedRemaining = singletonList(retainedTs);
+        if ("update-time".equals(expirationStrategy)) {
+            checkTime = LocalDateTime.now().plusDays(3);
+            expectedExpired.add(Collections.singletonMap("f0", retainedName));
+            expectedRemaining = emptyList();
+        }
+        assertThat(expire.expire(checkTime, Long.MAX_VALUE))
+                .containsExactlyElementsOf(expectedExpired);
+        List<Timestamp> remaining = new ArrayList<>();
+        table.newRead()
+                .createReader(table.newScan().plan().splits())
+                .forEachRemaining(row -> remaining.add(row.getTimestamp(0, 0)));
+        assertThat(remaining).containsExactlyElementsOf(expectedRemaining);
+    }
+
+    @Test
+    public void testExpireLocalZonedTimestampPartitionInNonUtcZone() throws Exception {
+        TimeZone defaultZone = TimeZone.getDefault();
+        TimeZone.setDefault(TimeZone.getTimeZone("Asia/Shanghai"));
+        try {
+            // current partition names render LTZ values in the default time zone
+            createTimestampPartitionedTable(DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(0), false);
+            // instant 2023-01-01T00:00Z is local 2023-01-01 08:00:00
+            Timestamp expiredTs = Timestamp.fromLocalDateTime(date(1));
+            // instant 2023-01-01T20:00Z is local 2023-01-02 04:00:00
+            Timestamp retainedTs = Timestamp.fromLocalDateTime(date(1).plusHours(20));
+            write(GenericRow.of(expiredTs, BinaryString.fromString("expired")));
+            write(GenericRow.of(retainedTs, BinaryString.fromString("retained")));
+
+            NormalPartitionExpire expire = newValuesTimeExpire("1 d", null);
+            // local 2023-01-03 02:00:00: the retained partition is only 22 hours old
+            assertThat(expire.expire(date(3).plusHours(2), Long.MAX_VALUE))
+                    .containsExactly(Collections.singletonMap("f0", "2023-01-01 08:00:00"));
+            assertThat(readTimestamps()).containsExactly(retainedTs);
+        } finally {
+            TimeZone.setDefault(defaultZone);
+        }
+    }
+
+    @ParameterizedTest
+    @ValueSource(booleans = {false, true})
+    public void testExpireTimestampPartitionWithCustomFormatter(boolean legacyPartitionName)
+            throws Exception {
+        createTimestampPartitionedTable(DataTypes.TIMESTAMP(0), legacyPartitionName);
+        Timestamp expiredTs = Timestamp.fromLocalDateTime(date(1));
+        Timestamp retainedTs = Timestamp.fromLocalDateTime(date(5));
+        write(GenericRow.of(expiredTs, BinaryString.fromString("expired")));
+        write(GenericRow.of(retainedTs, BinaryString.fromString("retained")));
+
+        // the formatter matches the Timestamp#toString input it has always received
+        NormalPartitionExpire expire = newValuesTimeExpire("2 d", "yyyy-MM-dd'T'HH:mm");
+        String expiredName = legacyPartitionName ? "2023-01-01T00:00" : "2023-01-01 00:00:00";
+        assertThat(expire.expire(date(6), Long.MAX_VALUE))
+                .containsExactly(Collections.singletonMap("f0", expiredName));
+        assertThat(readTimestamps()).containsExactly(retainedTs);
+    }
+
+    private void createTimestampPartitionedTable(DataType partitionType, boolean legacyName)
+            throws Exception {
+        SchemaManager schemaManager = new FileSystemSchemaManager(LocalFileIO.create(), path);
+        Map<String, String> tableOptions = new HashMap<>();
+        tableOptions.put(METASTORE_PARTITIONED_TABLE.key(), "false");
+        tableOptions.put(
+                CoreOptions.PARTITION_GENERATE_LEGACY_NAME.key(), String.valueOf(legacyName));
+        schemaManager.createTable(
+                new Schema(
+                        RowType.of(partitionType, VarCharType.STRING_TYPE).getFields(),
+                        singletonList("f0"),
+                        emptyList(),
+                        tableOptions,
+                        ""));
+        newTable();
+    }
+
+    private NormalPartitionExpire newValuesTimeExpire(
+            String expirationTime, @Nullable String formatter) {
+        Map<String, String> options = new HashMap<>();
+        options.put(PARTITION_EXPIRATION_TIME.key(), expirationTime);
+        options.put(CoreOptions.PARTITION_EXPIRATION_STRATEGY.key(), "values-time");
+        options.put(CoreOptions.PARTITION_TIMESTAMP_PATTERN.key(), "$f0");
+        if (formatter != null) {
+            options.put(CoreOptions.PARTITION_TIMESTAMP_FORMATTER.key(), formatter);
+        }
+        table = table.copy(options);
+        NormalPartitionExpire expire =
+                (NormalPartitionExpire) table.store().newPartitionExpire("", table);
+        expire.setLastCheck(date(1));
+        return expire;
+    }
+
+    private List<Timestamp> readTimestamps() throws Exception {
+        List<Timestamp> timestamps = new ArrayList<>();
+        table.newRead()
+                .createReader(table.newScan().plan().splits())
+                .forEachRemaining(row -> timestamps.add(row.getTimestamp(0, 0)));
+        return timestamps;
+    }
+
     @ParameterizedTest
     @ValueSource(strings = {"values-time", "update-time"})
     public void testExpireDatePartitionWithoutMetastore(String expirationStrategy)