From 6e4f5efa5f39738cd3024e5d7b68ae2cc36e5d8a Mon Sep 17 00:00:00 2001 From: sychen Date: Wed, 28 Oct 2026 20:18:49 +0800 Subject: [PATCH 1/2] [core] Expire TIMESTAMP partitions with the values-time strategy Partition expiration with the values-time strategy never expired a partition keyed by a TIMESTAMP or TIMESTAMP_LTZ column: the value was rendered with Timestamp#toString, whose ISO form (2023-01-01T00:00) the default partition time formatter cannot parse. Render such values as the SQL timestamp string for time extraction, and use the stored partition name when dropping them, as already done for DATE keys. --- .../operation/NormalPartitionExpire.java | 6 +- .../partition/PartitionExpireStrategy.java | 41 +++++++++++-- .../paimon/operation/PartitionExpireTest.java | 59 +++++++++++++++++++ 3 files changed, 99 insertions(+), 7 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/NormalPartitionExpire.java b/paimon-core/src/main/java/org/apache/paimon/operation/NormalPartitionExpire.java index aaceb6a17afa..4c6cec0ca57e 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/NormalPartitionExpire.java +++ b/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> doExpire( List> 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 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++; diff --git a/paimon-core/src/main/java/org/apache/paimon/partition/PartitionExpireStrategy.java b/paimon-core/src/main/java/org/apache/paimon/partition/PartitionExpireStrategy.java index fc00691c9a6a..bdbe76d416c6 100644 --- a/paimon-core/src/main/java/org/apache/paimon/partition/PartitionExpireStrategy.java +++ b/paimon-core/src/main/java/org/apache/paimon/partition/PartitionExpireStrategy.java @@ -22,10 +22,13 @@ 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.DataTypeRoot; +import org.apache.paimon.types.DataType; +import org.apache.paimon.types.DataTypeChecks; 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; @@ -78,13 +81,43 @@ public List 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: + // Timestamp#toString uses the ISO 'T' separator, which the default + // partition time formatter cannot parse. Render it as the SQL string. + values[index] = + DateTimeUtils.formatTimestamp( + (Timestamp) values[index], DataTypeChecks.getPrecision(type)); + break; + default: + break; } } return values; diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/PartitionExpireTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/PartitionExpireTest.java index e0a2372287f8..828397c299ab 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/PartitionExpireTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/operation/PartitionExpireTest.java @@ -321,6 +321,65 @@ 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 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(); + org.apache.paimon.data.Timestamp expiredTs = + org.apache.paimon.data.Timestamp.fromLocalDateTime(date(1)); + org.apache.paimon.data.Timestamp retainedTs = + org.apache.paimon.data.Timestamp.fromLocalDateTime(date(5)); + write(GenericRow.of(expiredTs, BinaryString.fromString("expired"))); + write(GenericRow.of(retainedTs, BinaryString.fromString("retained"))); + + Map 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> expectedExpired = new ArrayList<>(); + expectedExpired.add(Collections.singletonMap("f0", expiredName)); + LocalDateTime checkTime = date(6); + List 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 remaining = new ArrayList<>(); + table.newRead() + .createReader(table.newScan().plan().splits()) + .forEachRemaining(row -> remaining.add(row.getTimestamp(0, 0))); + assertThat(remaining).containsExactlyElementsOf(expectedRemaining); + } + @ParameterizedTest @ValueSource(strings = {"values-time", "update-time"}) public void testExpireDatePartitionWithoutMetastore(String expirationStrategy) From f6a0c2424c753c23f324bcc78870dba6fd905c5b Mon Sep 17 00:00:00 2001 From: sychen Date: Mon, 2 Nov 2026 21:08:50 +0800 Subject: [PATCH 2/2] [core] Respect LTZ time zone and custom formatters when expiring TIMESTAMP partitions --- .../partition/PartitionExpireStrategy.java | 33 ++++++- .../PartitionValuesTimeExpireStrategy.java | 2 +- .../paimon/operation/PartitionExpireTest.java | 97 +++++++++++++++++-- 3 files changed, 121 insertions(+), 11 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/partition/PartitionExpireStrategy.java b/paimon-core/src/main/java/org/apache/paimon/partition/PartitionExpireStrategy.java index bdbe76d416c6..f09466e9f17d 100644 --- a/paimon-core/src/main/java/org/apache/paimon/partition/PartitionExpireStrategy.java +++ b/paimon-core/src/main/java/org/apache/paimon/partition/PartitionExpireStrategy.java @@ -27,6 +27,7 @@ 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; @@ -45,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 { @@ -54,11 +56,20 @@ public abstract class PartitionExpireStrategy implements Serializable { protected final List 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 toPartitionString(Object[] array) { @@ -110,11 +121,8 @@ public Object[] convertPartition(BinaryRow partition) { break; case TIMESTAMP_WITHOUT_TIME_ZONE: case TIMESTAMP_WITH_LOCAL_TIME_ZONE: - // Timestamp#toString uses the ISO 'T' separator, which the default - // partition time formatter cannot parse. Render it as the SQL string. values[index] = - DateTimeUtils.formatTimestamp( - (Timestamp) values[index], DataTypeChecks.getPrecision(type)); + formatTimestampForTimeExtraction((Timestamp) values[index], type); break; default: break; @@ -123,6 +131,23 @@ public Object[] convertPartition(BinaryRow partition) { 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 selectExpiredPartitions( FileStoreScan scan, LocalDateTime expirationTime); diff --git a/paimon-core/src/main/java/org/apache/paimon/partition/PartitionValuesTimeExpireStrategy.java b/paimon-core/src/main/java/org/apache/paimon/partition/PartitionValuesTimeExpireStrategy.java index 0291bcc6c916..35a2b3a42f9e 100644 --- a/paimon-core/src/main/java/org/apache/paimon/partition/PartitionValuesTimeExpireStrategy.java +++ b/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); diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/PartitionExpireTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/PartitionExpireTest.java index 828397c299ab..e72a89425ea8 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/PartitionExpireTest.java +++ b/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; @@ -344,10 +348,8 @@ public void testExpireTimestampPartitionWithoutMetastore( tableOptions, "")); newTable(); - org.apache.paimon.data.Timestamp expiredTs = - org.apache.paimon.data.Timestamp.fromLocalDateTime(date(1)); - org.apache.paimon.data.Timestamp retainedTs = - org.apache.paimon.data.Timestamp.fromLocalDateTime(date(5)); + 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"))); @@ -365,7 +367,7 @@ public void testExpireTimestampPartitionWithoutMetastore( List> expectedExpired = new ArrayList<>(); expectedExpired.add(Collections.singletonMap("f0", expiredName)); LocalDateTime checkTime = date(6); - List expectedRemaining = singletonList(retainedTs); + List expectedRemaining = singletonList(retainedTs); if ("update-time".equals(expirationStrategy)) { checkTime = LocalDateTime.now().plusDays(3); expectedExpired.add(Collections.singletonMap("f0", retainedName)); @@ -373,13 +375,96 @@ public void testExpireTimestampPartitionWithoutMetastore( } assertThat(expire.expire(checkTime, Long.MAX_VALUE)) .containsExactlyElementsOf(expectedExpired); - List remaining = new ArrayList<>(); + List 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 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 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 readTimestamps() throws Exception { + List 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)