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

docs/en/administrator-guide/dynamic-partition.md

@@ -131,6 +131,10 @@ The rules of dynamic partition are prefixed with `dynamic_partition.`:
 
     When the `start` attribute is not specified, this parameter has no effect.
 
+* `dynamic_partition.history_partition_num`
+
+   When `create_history_partition` is `true`, this parameter is used to specify the number of history partitions.
+
 * `dynamic_partition.hot_partition_num`
 
     Specify how many of the latest partitions are hot partitions. For hot partition, the system will automatically set its `storage_medium` parameter to SSD, and set `storage_cooldown_time`.

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

@@ -303,16 +303,14 @@ Syntax:
         )    
        ```
 
-       Dynamic_partition. Enable: specifies whether dynamic partitioning at the table level is enabled
-
-       Dynamic_partition. Time_unit: used to specify the time unit for dynamically adding partitions, which can be selected as HOUR, DAY, WEEK, and MONTH.
+       dynamic_partition.enable: specifies whether dynamic partitioning at the table level is enabled
+       dynamic_partition.time_unit: used to specify the time unit for dynamically adding partitions, which can be selected as HOUR, DAY, WEEK, and MONTH.
                                      Attention: When the time unit is HOUR, the data type of partition column cannot be DATE.
-
-       Dynamic_partition. End: used to specify the number of partitions created in advance
-
-       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.end: used to specify the number of partitions created in advance
+       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
        ```
     5)  You can create multiple Rollups in bulk when building a table
     grammar:

docs/zh-CN/administrator-guide/dynamic-partition.md

@@ -129,6 +129,10 @@ under the License.
 
     当不指定 `start` 属性时，该参数不生效。
 
+* `dynamic_partition.history_partition_num`
+
+   当 `create_history_partition` 为 `true` 时，该参数用于指定创建历史分区数量。
+
 * `dynamic_partition.hot_partition_num`
 
     指定最新的多少个分区为热分区。对于热分区，系统会自动设置其 `storage_medium` 参数为SSD，并且设置 `storage_cooldown_time`。

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

@@ -332,6 +332,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: 当开启创建历史分区功能时，用于指定创建历史分区数量。
 
     5) 建表时可以批量创建多个 Rollup
     语法：

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

@@ -37,11 +37,13 @@ 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 NOT_SET_REPLICATION_NUM = -1;
+    public static final int NOT_SET_HISTORY_PARTITION_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(NOT_SET_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

@@ -143,9 +143,20 @@ 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;
+        int start = dynamicPartitionProperty.getStart();
+        int historyPartitionNum = dynamicPartitionProperty.getHistoryPartitionNum();
+        // When enable create_history_partition, will check the valid value from start and history_partition_num.
+        if (createHistoryPartition) {
+            if (historyPartitionNum == DynamicPartitionProperty.NOT_SET_HISTORY_PARTITION_NUM) {
+                idx = start;
+            } else {
+                idx = Math.max(start, -historyPartitionNum);
+            }
+        } else {
+            idx = 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,9 @@ 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");
 
     ErrorCode(int code, byte[] sqlState, String errorMsg) {
         this.code = code;

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

@@ -79,7 +79,7 @@ public static void checkTimeUnit(String timeUnit, PartitionInfo partitionInfo) t
                     + TimeUnit.HOUR.toString() + " when type of partition column "
                     + partitionColumn.getDisplayName() + " is " + PrimitiveType.DATE.toString());
         } else if (PrimitiveType.getIntegerTypes().contains(partitionColumn.getDataType())
-           && timeUnit.equalsIgnoreCase(TimeUnit.HOUR.toString())) {
+                && timeUnit.equalsIgnoreCase(TimeUnit.HOUR.toString())) {
             // The partition column's type is INT, not support HOUR
             ErrorReport.reportDdlException(DynamicPartitionProperty.TIME_UNIT + " could not be "
                     + TimeUnit.HOUR.toString() + " when type of partition column "
@@ -152,6 +152,20 @@ private static boolean checkCreateHistoryPartition(String create) throws DdlExce
         return Boolean.valueOf(create);
     }
 
+    private static void 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);
+            }
+        } catch (NumberFormatException e) {
+            throw new DdlException("Invalid properties: " + DynamicPartitionProperty.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,13 +362,35 @@ public static Map<String, String> analyzeDynamicPartition(Map<String, String> pr
             analyzedProperties.put(DynamicPartitionProperty.CREATE_HISTORY_PARTITION, val);
         }
 
+        if (properties.containsKey(DynamicPartitionProperty.HISTORY_PARTITION_NUM)) {
+            String val = properties.get(DynamicPartitionProperty.HISTORY_PARTITION_NUM);
+            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 create_history_partition is true, will pre-create history partition according the valid value from
+        // start and history_partition_num.
+        int expectCreatePartitionNum;
         if (!createHistoryPartition) {
             start = 0;
+            expectCreatePartitionNum = end - start;
+        } else {
+            int historyPartitionNum = Integer.valueOf(analyzedProperties.getOrDefault(DynamicPartitionProperty.HISTORY_PARTITION_NUM,
+                    String.valueOf(DynamicPartitionProperty.NOT_SET_HISTORY_PARTITION_NUM)));
+            if (historyPartitionNum != DynamicPartitionProperty.NOT_SET_HISTORY_PARTITION_NUM) {
+                expectCreatePartitionNum = end - Math.max(start, -historyPartitionNum);
+            } else {
+                if (start == Integer.MIN_VALUE) {
+                    throw new DdlException("Provide start or create_history_partition property when creating history partition");
+                }
+                expectCreatePartitionNum = end - start;
+            }
         }
-        if (hasEnd && (end - start > Config.max_dynamic_partition_num)) {
-            throw new DdlException("Too many dynamic partitions: " + (end - start) + ". Limit: " + Config.max_dynamic_partition_num);
+        if (hasEnd && (expectCreatePartitionNum > Config.max_dynamic_partition_num)) {
+            throw new DdlException("Too many dynamic partitions: " + expectCreatePartitionNum + ". Limit: " + Config.max_dynamic_partition_num);
         }
 
         if (properties.containsKey(DynamicPartitionProperty.START_DAY_OF_MONTH)) {

fe/fe-core/src/test/java/org/apache/doris/catalog/DynamicPartitionTableTest.java

@@ -586,6 +586,33 @@ public void testFillHistoryDynamicPartition2() throws Exception {
         createTable(createOlapTblStmt);
     }
 
+    @Test
+    public void testFillHistoryDynamicPartitionWithHistoryPartitionNum() throws Exception {
+        String createOlapTblStmt = "CREATE TABLE test.`history_dynamic_partition_day` (\n" +
+                "  `k1` datetime NULL COMMENT \"\",\n" +
+                "  `k2` int NULL COMMENT \"\"\n" +
+                ") ENGINE=OLAP\n" +
+                "PARTITION BY RANGE(`k1`)\n" +
+                "()\n" +
+                "DISTRIBUTED BY HASH(`k2`) BUCKETS 3\n" +
+                "PROPERTIES (\n" +
+                "\"replication_num\" = \"1\",\n" +
+                "\"dynamic_partition.enable\" = \"true\",\n" +
+                "\"dynamic_partition.end\" = \"3\",\n" +
+                "\"dynamic_partition.create_history_partition\" = \"true\",\n" +
+                "\"dynamic_partition.history_partition_num\" = \"10\",\n" +
+                "\"dynamic_partition.time_unit\" = \"day\",\n" +
+                "\"dynamic_partition.prefix\" = \"p\",\n" +
+                "\"dynamic_partition.buckets\" = \"1\"\n" +
+                ");";
+        createTable(createOlapTblStmt);
+        OlapTable emptyDynamicTable = (OlapTable) Catalog.getCurrentCatalog().getDb("default_cluster:test").getTable("history_dynamic_partition_day");
+        Map<String, String> tableProperties = emptyDynamicTable.getTableProperty().getProperties();
+        Assert.assertEquals(14, emptyDynamicTable.getAllPartitions().size());
+        // never delete the old partitions
+        Assert.assertEquals(Integer.parseInt(tableProperties.get("dynamic_partition.start")), Integer.MIN_VALUE);
+    }
+
     @Test
     public void testAllTypeDynamicPartition() throws Exception {
         String createOlapTblStmt = "CREATE TABLE test.`hour_dynamic_partition` (\n" +