[Bug] Create dynamic partition table failed with enable create_history_partition and not specify the start value

docs/en/sql-reference/sql-statements/Data Definition/CREATE TABLE.md

@@ -312,6 +312,10 @@ Syntax:
        Dynamic_partition. Prefix: used to specify the partition name prefix to be created, such as the partition name prefix p, automatically creates the partition name p20200108
 
        Dynamic_partition. Buckets: specifies the number of partition buckets that are automatically created
+
+       Dynamic_partition. Create_history_partition: specifies whether create history partitions, default value is false
+
+       Dynamic_partition. History_partition_num: used to specify the number of history partitions when enable create_history_partition, default value is 10
        ```
     5)  You can create multiple Rollups in bulk when building a table
     grammar:

docs/zh-CN/sql-reference/sql-statements/Data Definition/CREATE TABLE.md

@@ -331,6 +331,8 @@ under the License.
     dynamic_partition.end: 用于指定提前创建的分区数量。值必须大于0。
     dynamic_partition.prefix: 用于指定创建的分区名前缀，例如分区名前缀为p，则自动创建分区名为p20200108
     dynamic_partition.buckets: 用于指定自动创建的分区分桶数量
+    dynamic_partition.create_history_partition: 用于创建历史分区功能是否开启。默认为 false。
+    dynamic_partition.history_partition_num: 当开启创建历史分区功能时，用于指定创建历史分区数量。默认为10。
 
     5) 建表时可以批量创建多个 Rollup
     语法：

fe/fe-core/src/main/java/org/apache/doris/catalog/DynamicPartitionProperty.java

@@ -37,10 +37,12 @@ public class DynamicPartitionProperty {
     public static final String TIME_ZONE = "dynamic_partition.time_zone";
     public static final String REPLICATION_NUM = "dynamic_partition.replication_num";
     public static final String CREATE_HISTORY_PARTITION = "dynamic_partition.create_history_partition";
+    public static final String HISTORY_PARTITION_NUM = "dynamic_partition.history_partition_num";
     public static final String HOT_PARTITION_NUM = "dynamic_partition.hot_partition_num";
 
     public static final int MIN_START_OFFSET = Integer.MIN_VALUE;
     public static final int MAX_END_OFFSET = Integer.MAX_VALUE;
+    public static final int DEFAULT_HISTORY_PARTITION_NUM = 10;
     public static final int NOT_SET_REPLICATION_NUM = -1;
 
     private boolean exist;
@@ -56,6 +58,7 @@ public class DynamicPartitionProperty {
     private TimeZone tz = TimeUtils.getSystemTimeZone();
     private int replicationNum;
     private boolean createHistoryPartition = false;
+    private int historyPartitionNum;
     // This property are used to describe the number of partitions that need to be reserved on the high-speed storage.
     // If not set, default is 0
     private int hotPartitionNum;
@@ -73,6 +76,7 @@ public DynamicPartitionProperty(Map<String, String> properties) {
             this.buckets = Integer.parseInt(properties.get(BUCKETS));
             this.replicationNum = Integer.parseInt(properties.getOrDefault(REPLICATION_NUM, String.valueOf(NOT_SET_REPLICATION_NUM)));
             this.createHistoryPartition = Boolean.parseBoolean(properties.get(CREATE_HISTORY_PARTITION));
+            this.historyPartitionNum = Integer.parseInt(properties.getOrDefault(HISTORY_PARTITION_NUM, String.valueOf(DEFAULT_HISTORY_PARTITION_NUM)));
             this.hotPartitionNum = Integer.parseInt(properties.getOrDefault(HOT_PARTITION_NUM, "0"));
             createStartOfs(properties);
         } else {
@@ -136,6 +140,10 @@ public boolean isCreateHistoryPartition() {
         return createHistoryPartition;
     }
 
+    public int getHistoryPartitionNum() {
+        return historyPartitionNum;
+    }
+
     public int getHotPartitionNum() {
         return hotPartitionNum;
     }
@@ -175,6 +183,7 @@ public String getProperties(int tableReplicationNum) {
                 ",\n\"" + REPLICATION_NUM + "\" = \"" + useReplicationNum + "\"" +
                 ",\n\"" + BUCKETS + "\" = \"" + buckets + "\"" +
                 ",\n\"" + CREATE_HISTORY_PARTITION + "\" = \"" + createHistoryPartition + "\"" +
+                ",\n\"" + HISTORY_PARTITION_NUM + "\" = \"" + historyPartitionNum + "\"" +
                 ",\n\"" + HOT_PARTITION_NUM + "\" = \"" + hotPartitionNum + "\"";
         if (getTimeUnit().equalsIgnoreCase(TimeUnit.WEEK.toString())) {
             res += ",\n\"" + START_DAY_OF_WEEK + "\" = \"" + startOfWeek.dayOfWeek + "\"";

fe/fe-core/src/main/java/org/apache/doris/clone/DynamicPartitionScheduler.java

@@ -146,9 +146,8 @@ private ArrayList<AddPartitionClause> getAddPartitionClause(Database db, OlapTab
         ZonedDateTime now = ZonedDateTime.now(dynamicPartitionProperty.getTimeZone().toZoneId());
 
         boolean createHistoryPartition = dynamicPartitionProperty.isCreateHistoryPartition();
-        int idx = createHistoryPartition ? dynamicPartitionProperty.getStart() : 0;
+        int idx = createHistoryPartition ? -dynamicPartitionProperty.getHistoryPartitionNum() : 0;
         int hotPartitionNum = dynamicPartitionProperty.getHotPartitionNum();
-        String timeUnit = dynamicPartitionProperty.getTimeUnit();
 
         for (; idx <= dynamicPartitionProperty.getEnd(); idx++) {
             String prevBorder = DynamicPartitionUtil.getPartitionRangeString(dynamicPartitionProperty, now, idx, partitionFormat);

fe/fe-core/src/main/java/org/apache/doris/common/ErrorCode.java

@@ -241,7 +241,11 @@ public enum ErrorCode {
     ERROR_CREATE_TABLE_LIKE_EMPTY(5073, new byte[] {'4', '2', '0', '0', '0'},
             "Origin create table stmt is empty"),
     ERROR_DYNAMIC_PARTITION_CREATE_HISTORY_PARTITION(5074, new byte[]{'4', '2', '0', '0', '0'},
-            "Invalid dynamic partition create_history_partition: %s. Expected true or false");
+            "Invalid dynamic partition create_history_partition: %s. Expected true or false"),
+    ERROR_DYNAMIC_PARTITION_HISTORY_PARTITION_NUM_ZERO(5075, new byte[] {'4', '2', '0', '0', '0'},
+            "Dynamic history partition num must greater than 0"),
+    ERROR_DYNAMIC_PARTITION_HISTORY_PARTITION_NUM_FORMAT(5076, new byte[] {'4', '2', '0', '0', '0'},
+            "Invalid dynamic history partition num: %s.");
 
     ErrorCode(int code, byte[] sqlState, String errorMsg) {
         this.code = code;

fe/fe-core/src/main/java/org/apache/doris/common/util/DynamicPartitionUtil.java

@@ -152,6 +152,22 @@ private static boolean checkCreateHistoryPartition(String create) throws DdlExce
         return Boolean.valueOf(create);
     }
 
+    private static int checkHistoryPartitionNum(String val) throws DdlException {
+        if (Strings.isNullOrEmpty(val)) {
+            throw new DdlException("Invalid properties: " + DynamicPartitionProperty.HISTORY_PARTITION_NUM);
+        }
+        try {
+            int historyPartitionNum = Integer.parseInt(val);
+            if (historyPartitionNum <= 0) {
+                ErrorReport.reportDdlException(ErrorCode.ERROR_DYNAMIC_PARTITION_HISTORY_PARTITION_NUM_ZERO);
+            }
+            return historyPartitionNum;
+        } catch (NumberFormatException e) {
+            ErrorReport.reportDdlException(ErrorCode.ERROR_DYNAMIC_PARTITION_HISTORY_PARTITION_NUM_FORMAT, val);
+        }
+        return DynamicPartitionProperty.DEFAULT_HISTORY_PARTITION_NUM;
+    }
+
     private static void checkStartDayOfMonth(String val) throws DdlException {
         if (Strings.isNullOrEmpty(val)) {
             throw new DdlException("Invalid properties: " + DynamicPartitionProperty.START_DAY_OF_MONTH);
@@ -348,10 +364,20 @@ public static Map<String, String> analyzeDynamicPartition(Map<String, String> pr
             analyzedProperties.put(DynamicPartitionProperty.CREATE_HISTORY_PARTITION, val);
         }
 
+        int historyPartitionNum = 0;
+        if (properties.containsKey(DynamicPartitionProperty.HISTORY_PARTITION_NUM)) {
+            String val = properties.get(DynamicPartitionProperty.HISTORY_PARTITION_NUM);
+            historyPartitionNum = checkHistoryPartitionNum(val);
+            properties.remove(DynamicPartitionProperty.HISTORY_PARTITION_NUM);
+            analyzedProperties.put(DynamicPartitionProperty.HISTORY_PARTITION_NUM, val);
+        }
+
         // Check the number of dynamic partitions that need to be created to avoid creating too many partitions at once.
         // If create_history_partition is false, history partition is not considered.
         if (!createHistoryPartition) {
             start = 0;
+        } else {
+            start = -historyPartitionNum;
         }
         if (hasEnd && (end - start > Config.max_dynamic_partition_num)) {
             throw new DdlException("Too many dynamic partitions: " + (end - start) + ". Limit: " + Config.max_dynamic_partition_num);