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..bfc6e32f09c9 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; @@ -51,11 +54,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) { @@ -78,18 +90,53 @@ 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: + 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: + values[index] = + formatTimestampForTimeExtraction((Timestamp) values[index], type); + break; + default: + break; } } return values; } + private String formatTimestampForTimeExtraction(Timestamp timestamp, DataType type) { + 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 e0a2372287f8..5f8166076ce1 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; @@ -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 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 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(booleans = {false, true}) + public void testLocalZonedTimestampPartitionIsNotExpiredByDefault(boolean legacyPartitionName) + throws Exception { + TimeZone defaultZone = TimeZone.getDefault(); + TimeZone.setDefault(TimeZone.getTimeZone("America/New_York")); + try { + // TIMESTAMP_LTZ values are not normalized for time extraction: a legacy name is a UTC + // spelling that a partition spec would parse back in the default time zone + createTimestampPartitionedTable( + DataTypes.TIMESTAMP_WITH_LOCAL_TIME_ZONE(0), legacyPartitionName); + Timestamp oldTs = Timestamp.fromLocalDateTime(date(1)); + Timestamp freshTs = Timestamp.fromLocalDateTime(date(1).plusHours(5)); + write(GenericRow.of(oldTs, BinaryString.fromString("old"))); + write(GenericRow.of(freshTs, BinaryString.fromString("fresh"))); + + NormalPartitionExpire expire = newValuesTimeExpire("1 d", null); + assertThat(expire.expire(date(1).plusHours(22), Long.MAX_VALUE)).isEmpty(); + assertThat(readTimestamps()).containsExactlyInAnyOrder(oldTs, freshTs); + } 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)