From 78180d8eeedc283a20adbe5bd7e864c308b7e2f7 Mon Sep 17 00:00:00 2001 From: chenjian2664 Date: Fri, 30 Oct 2026 12:48:10 +0800 Subject: [PATCH] [core] Validate DATE partition timestamps as dates --- .../operation/AbstractFileStoreWrite.java | 9 +++ .../paimon/table/sink/TableWriteTest.java | 66 +++++++++++++++++++ 2 files changed, 75 insertions(+) diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java index 0c0738c6173e..343318077e02 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java @@ -40,6 +40,7 @@ import org.apache.paimon.partition.PartitionTimeExtractor; import org.apache.paimon.table.sink.CommitMessage; import org.apache.paimon.table.sink.CommitMessageImpl; +import org.apache.paimon.types.DataTypeRoot; import org.apache.paimon.types.RowType; import org.apache.paimon.utils.CommitIncrement; import org.apache.paimon.utils.ExecutorThreadFactory; @@ -53,6 +54,7 @@ import javax.annotation.Nullable; +import java.time.LocalDate; import java.time.format.DateTimeParseException; import java.util.ArrayList; import java.util.Arrays; @@ -820,6 +822,13 @@ private static PartitionTimestampValidator create( private void validate(BinaryRow partition) { Object[] array = partitionConverter.convert(partition); + RowType partitionType = partitionConverter.rowType(); + for (int i = 0; i < array.length; i++) { + if (array[i] != null + && partitionType.getTypeAt(i).getTypeRoot() == DataTypeRoot.DATE) { + array[i] = LocalDate.ofEpochDay((Integer) array[i]); + } + } try { timeExtractor.extract(partitionKeys, Arrays.asList(array)); } catch (DateTimeParseException e) { diff --git a/paimon-core/src/test/java/org/apache/paimon/table/sink/TableWriteTest.java b/paimon-core/src/test/java/org/apache/paimon/table/sink/TableWriteTest.java index 3441dadecf89..3d530832e21c 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/sink/TableWriteTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/sink/TableWriteTest.java @@ -51,7 +51,10 @@ import org.junit.jupiter.api.BeforeEach; 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 java.time.LocalDate; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -317,6 +320,69 @@ public void testPartitionTimestampValidation() throws Exception { write.close(); } + @ParameterizedTest + @CsvSource({ + "true, yyyy-MM-dd, $dt", + "true, yyyy-MM-dd,", + "true, , $dt", + "false, yyyy-MM-dd, $dt", + "false, yyyy-MM-dd,", + "false, , $dt" + }) + public void testDatePartitionTimestampValidation( + boolean hasPrimaryKey, String formatter, String pattern) throws Exception { + Options conf = new Options(); + conf.set(CoreOptions.BUCKET_KEY, "id"); + conf.set(CoreOptions.PARTITION_TIMESTAMP_FORMAT_STRICT, true); + if (formatter != null) { + conf.set(CoreOptions.PARTITION_TIMESTAMP_FORMATTER, formatter); + } + if (pattern != null) { + conf.set(CoreOptions.PARTITION_TIMESTAMP_PATTERN, pattern); + } + RowType rowType = + RowType.of( + new DataType[] {DataTypes.DATE(), DataTypes.INT()}, + new String[] {"dt", "id"}); + FileStoreTable table = + createFileStoreTable( + rowType, + Collections.singletonList("dt"), + hasPrimaryKey ? Arrays.asList("dt", "id") : Collections.emptyList(), + conf); + + try (TableWriteImpl write = table.newWrite(commitUser)) { + for (String date : + Arrays.asList("1969-12-31", "1970-01-01", "2024-02-29", "2026-10-29")) { + write.write(GenericRow.of((int) LocalDate.parse(date).toEpochDay(), 1)); + } + } + } + + @Test + public void testDateAndHourPartitionTimestampValidation() throws Exception { + Options conf = new Options(); + conf.set(CoreOptions.BUCKET_KEY, "id"); + conf.set(CoreOptions.PARTITION_TIMESTAMP_FORMAT_STRICT, true); + conf.set(CoreOptions.PARTITION_TIMESTAMP_FORMATTER, "yyyy-MM-dd HH:mm:ss"); + conf.set(CoreOptions.PARTITION_TIMESTAMP_PATTERN, "$dt $hour:00:00"); + RowType rowType = + RowType.of( + new DataType[] {DataTypes.INT(), DataTypes.DATE(), DataTypes.INT()}, + new String[] {"hour", "dt", "id"}); + FileStoreTable table = + createFileStoreTable( + rowType, Arrays.asList("hour", "dt"), Collections.emptyList(), conf); + int epochDay = (int) LocalDate.of(2024, 2, 29).toEpochDay(); + + try (TableWriteImpl write = table.newWrite(commitUser)) { + write.write(GenericRow.of(12, epochDay, 1)); + assertThatThrownBy(() -> write.write(GenericRow.of(25, epochDay, 1))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Partition hour=25, dt=2024-02-29"); + } + } + @Test public void testDeduplicateDeleteWithPrimaryKeyOnly() throws Exception { FileStoreTable table = createNotNullFileStoreTable(new Options());