[core] Validate DATE partition timestamps as dates

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) {

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());