From 47635b166117c277b0620eb4a75234339d961885 Mon Sep 17 00:00:00 2001 From: sychen Date: Wed, 28 Oct 2026 20:18:49 +0800 Subject: [PATCH] [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 | 60 +++++++++++++++++++ 3 files changed, 100 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 26ca00f2f7f5..be44f049bf30 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 javax.annotation.Nullable; @@ -74,13 +77,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 57dabaea4c6a..3bee6ecd3bef 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 @@ -61,6 +61,7 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; import org.junit.jupiter.params.provider.ValueSource; import java.io.IOException; @@ -313,6 +314,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)