[core][bug] Fix the incorrect partition key format generated from the…

… DATE type.

paimon-core/src/main/java/org/apache/paimon/utils/RowDataPartitionComputer.java

@@ -19,23 +19,28 @@
 package org.apache.paimon.utils;
 
 import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypeRoot;
 import org.apache.paimon.types.RowType;
 
 import java.util.Arrays;
 import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Objects;
 
 /** PartitionComputer for {@link InternalRow}. */
 public class RowDataPartitionComputer {
 
     protected final String defaultPartValue;
     protected final String[] partitionColumns;
+    protected final List<DataType> fieldTypes;
     protected final InternalRow.FieldGetter[] partitionFieldGetters;
 
     public RowDataPartitionComputer(
             String defaultPartValue, RowType rowType, String[] partitionColumns) {
         this.defaultPartValue = defaultPartValue;
         this.partitionColumns = partitionColumns;
+        this.fieldTypes = rowType.getFieldTypes();
         List<String> columnList = rowType.getFieldNames();
         this.partitionFieldGetters =
                 Arrays.stream(partitionColumns)
@@ -52,7 +57,18 @@ public LinkedHashMap<String, String> generatePartValues(InternalRow in) {
 
         for (int i = 0; i < partitionFieldGetters.length; i++) {
             Object field = partitionFieldGetters[i].getFieldOrNull(in);
-            String partitionValue = field != null ? field.toString() : null;
+            String partitionValue;
+
+            if (Objects.nonNull(field)) {
+                if (fieldTypes.get(i).is(DataTypeRoot.DATE)) {
+                    partitionValue = DateTimeUtils.toLocalDate((Integer) field).toString();
+                } else {
+                    partitionValue = field.toString();
+                }
+            } else {
+                partitionValue = null;
+            }
+
             if (StringUtils.isNullOrWhitespaceOnly(partitionValue)) {
                 partitionValue = defaultPartValue;
             }

paimon-core/src/test/java/org/apache/paimon/utils/RowDataPartitionComputerTest.java

@@ -0,0 +1,52 @@
+package org.apache.paimon.utils;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryRowWriter;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DateType;
+import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.LocalDate;
+import java.time.LocalTime;
+import java.util.LinkedHashMap;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/** Tests for {@link RowDataPartitionComputer}. */
+public class RowDataPartitionComputerTest {
+
+    @Test
+    public void testGeneratePartitionValues() {
+        RowDataPartitionComputer partitionComputer =
+                FileStorePathFactory.getPartitionComputer(
+                        RowType.of(
+                                new DataType[] {new DateType(), new IntType()},
+                                new String[] {"dt", "hour"}),
+                        FileStorePathFactory.PARTITION_DEFAULT_NAME.defaultValue());
+        BinaryRow binaryRow = buildPartitionBinaryRow(LocalDate.now(), LocalTime.now().getHour());
+        LinkedHashMap<String, String> map = partitionComputer.generatePartValues(binaryRow);
+
+        assertEquals(map.get("dt"), LocalDate.now().toString());
+        assertEquals(map.get("hour"), String.valueOf(LocalTime.now().getHour()));
+    }
+
+    public BinaryRow buildPartitionBinaryRow(LocalDate dt, Integer hour) {
+        BinaryRow partition = new BinaryRow(2);
+        BinaryRowWriter writer = new BinaryRowWriter(partition);
+        if (dt != null) {
+            writer.writeInt(0, (int) dt.toEpochDay());
+        } else {
+            writer.setNullAt(0);
+        }
+        if (hour != null) {
+            writer.writeInt(1, hour);
+        } else {
+            writer.setNullAt(1);
+        }
+        writer.complete();
+        return partition;
+    }
+}