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)