[Feature] Support create history dynamic partition

be/test/exec/tablet_sink_test.cpp

@@ -453,7 +453,7 @@ TEST_F(OlapTableSinkTest, normal) {
     ASSERT_TRUE(st.ok());
     // close
     st = sink.close(&state, Status::OK());
-    ASSERT_TRUE(st.ok() || st.to_string() == "Internal error: already stopped, skip waiting for close. cancelled/!eos: : 1/1") << st.to_string();
+    ASSERT_TRUE(st.ok() || st.to_string() == "Internal error: wait close failed. ") << st.to_string();
 
     // each node has a eof
     ASSERT_EQ(2, service->_eof_counters);
@@ -586,7 +586,7 @@ TEST_F(OlapTableSinkTest, convert) {
     ASSERT_TRUE(st.ok());
     // close
     st = sink.close(&state, Status::OK());
-    ASSERT_TRUE(st.ok() || st.to_string() == "Internal error: already stopped, skip waiting for close. cancelled/!eos: : 1/1") << st.to_string();
+    ASSERT_TRUE(st.ok() || st.to_string() == "Internal error: wait close failed. ") << st.to_string();
 
     // each node has a eof
     ASSERT_EQ(2, service->_eof_counters);
@@ -966,7 +966,7 @@ TEST_F(OlapTableSinkTest, decimal) {
     ASSERT_TRUE(st.ok());
     // close
     st = sink.close(&state, Status::OK());
-    ASSERT_TRUE(st.ok() || st.to_string() == "Internal error: already stopped, skip waiting for close. cancelled/!eos: : 1/1") << st.to_string();
+    ASSERT_TRUE(st.ok() || st.to_string() == "Internal error: wait close failed. ") << st.to_string();
 
     ASSERT_EQ(2, output_set.size());
     ASSERT_TRUE(output_set.count("[(12 12.3)]") > 0);

docs/en/administrator-guide/config/fe_config.md

@@ -789,3 +789,8 @@ And the new UI will also try to get this base path first to assemble the URL.
 Only valid when `enable_http_server_v2` is true.
 
 The default is empty, that is, not set.
+
+### `max_dynamic_partition_num`
+
+When creating a dynamic partition table, the maximum number of partitions allowed to be automatically created. To prevent creating too many partitions at once.
+The default is 500.

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

@@ -125,6 +125,12 @@ The rules of dynamic partition are prefixed with `dynamic_partition.`:
 
     When `time_unit` is` MONTH`, this parameter is used to specify the start date of each month. The value ranges from 1 to 28. 1 means the 1st of every month, and 28 means the 28th of every month. The default is 1, which means that every month starts at 1st. The 29, 30 and 31 are not supported at the moment to avoid ambiguity caused by lunar years or months.
 
+* `dynamic_partition.create_history_partition`
+
+    The default is false. When set to true, Doris will automatically create all partitions from start to end. At the same time, the parameter `max_dynamic_partition_num` of FE will limit the total number of partitions to avoid creating too many partitions at once. If (end - start) is larger than `max_dynamic_partition_num`, the operation will fail.
+
+    When the `start` attribute is not specified, this parameter has no effect.
+
 ### Notice
 
 If some partitions between `dynamic_partition.start` and `dynamic_partition.end` are lost due to some unexpected circumstances when using dynamic partition, the lost partitions between the current time and `dynamic_partition.end` will be recreated, but the lost partitions between `dynamic_partition.start` and the current time will not be recreated.

docs/zh-CN/administrator-guide/config/fe_config.md

@@ -794,3 +794,8 @@ thrift_client_timeout_ms 的值被设置为大于0来避免线程卡在java.net.
 仅在 `enable_http_server_v2` 为 true 的情况下才有效。
 
 默认为空，即不设置。
+
+### `max_dynamic_partition_num`
+
+在创建动态分区表时，允许自动创建的最大分区数量。以防止一次性创建过多的分区。
+默认为 500。

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

@@ -110,7 +110,7 @@ under the License.
 * `dynamic_partition.buckets`
 
     动态创建的分区所对应的分桶数量。
-  
+
 * `dynamic_partition.replication_num`
 
     动态创建的分区所对应的副本数量，如果不填写，则默认为该表创建时指定的副本数量。
@@ -122,6 +122,12 @@ under the License.
 * `dynamic_partition.start_day_of_month`
 
     当 `time_unit` 为 `MONTH` 时，该参数用于指定每月的起始日期。取值为 1 到 28。其中 1 表示每月1号，28 表示每月28号。默认为 1，即表示每月以1号位起始点。暂不支持以29、30、31号为起始日，以避免因闰年或闰月带来的歧义。
+
+* `dynamic_partition.create_history_partition`
+
+    默认为 false。当置为 true 时，Doris 会自动创建由 start 到 end 的所有分区。同时，FE 的参数 `max_dynamic_partition_num` 会限制总分区数量，以避免一次性创建过多分区。当 end - start 的值大于 `max_dynamic_partition_num` 值时，操作将被禁止。
+
+    当不指定 `start` 属性时，该参数不生效。
 
 ### 注意事项 
 
@@ -346,6 +352,6 @@ mysql> SHOW DYNAMIC PARTITION TABLES;
 
 1. 创建动态分区表后提示 ```Could not create table with dynamic partition when fe config dynamic_partition_enable is false```
 
-	由于动态分区的总开关，也就是 FE 的配置 ```dynamic_partition_enable``` 为 false，导致无法创建动态分区表。
+    由于动态分区的总开关，也就是 FE 的配置 ```dynamic_partition_enable``` 为 false，导致无法创建动态分区表。
 
-        这时候请修改 FE 的配置文件，增加一行 ```dynamic_partition_enable=true```，并重启 FE。或者执行命令 ADMIN SET FRONTEND CONFIG ("dynamic_partition_enable" = "true") 将动态分区开关打开即可。
+    这时候请修改 FE 的配置文件，增加一行 ```dynamic_partition_enable=true```，并重启 FE。或者执行命令 ADMIN SET FRONTEND CONFIG ("dynamic_partition_enable" = "true") 将动态分区开关打开即可。

fe/fe-core/pom.xml

@@ -737,6 +737,7 @@ under the License.
                             <sources>
                                 <!-- add arbitrary num of src dirs here -->
                                 <source>${basedir}/target/generated-sources/build/</source>
+                                <source>${basedir}/target/generated-sources/</source>
                             </sources>
                         </configuration>
                     </execution>

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

@@ -36,8 +36,10 @@ public class DynamicPartitionProperty {
     public static final String START_DAY_OF_MONTH = "dynamic_partition.start_day_of_month";
     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 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;
 
     private boolean exist;
@@ -52,6 +54,7 @@ public class DynamicPartitionProperty {
     private StartOfDate startOfMonth;
     private TimeZone tz = TimeUtils.getSystemTimeZone();
     private int replicationNum;
+    private boolean createHistoryPartition = false;
 
     public DynamicPartitionProperty(Map<String, String> properties) {
         if (properties != null && !properties.isEmpty()) {
@@ -65,6 +68,7 @@ public DynamicPartitionProperty(Map<String, String> properties) {
             this.prefix = properties.get(PREFIX);
             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));
             createStartOfs(properties);
         } else {
             this.exist = false;
@@ -123,6 +127,10 @@ public StartOfDate getStartOfMonth() {
         return startOfMonth;
     }
 
+    public boolean isCreateHistoryPartition() {
+        return createHistoryPartition;
+    }
+
     public String getStartOfInfo() {
         if (getTimeUnit().equalsIgnoreCase(TimeUnit.WEEK.toString())) {
             return startOfWeek.toDisplayInfo();
@@ -156,7 +164,8 @@ public String getProperties(int tableReplicationNum) {
                 ",\n\"" + END + "\" = \"" + end + "\"" +
                 ",\n\"" + PREFIX + "\" = \"" + prefix + "\"" +
                 ",\n\"" + REPLICATION_NUM + "\" = \"" + useReplicationNum + "\"" +
-                ",\n\"" + BUCKETS + "\" = \"" + buckets + "\"";
+                ",\n\"" + BUCKETS + "\" = \"" + buckets + "\"" +
+                ",\n\"" + CREATE_HISTORY_PARTITION + "\" = \"" + createHistoryPartition + "\"";
         if (getTimeUnit().equalsIgnoreCase(TimeUnit.WEEK.toString())) {
             res += ",\n\"" + START_DAY_OF_WEEK + "\" = \"" + startOfWeek.dayOfWeek + "\"";
         } else if (getTimeUnit().equalsIgnoreCase(TimeUnit.MONTH.toString())) {

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

@@ -45,14 +45,14 @@
 import org.apache.doris.common.util.RangeUtils;
 import org.apache.doris.common.util.TimeUtils;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Range;
 import com.google.common.collect.Sets;
 
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
-
 import java.time.ZonedDateTime;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -142,9 +142,13 @@ private ArrayList<AddPartitionClause> getAddPartitionClause(Database db, OlapTab
         DynamicPartitionProperty dynamicPartitionProperty = olapTable.getTableProperty().getDynamicPartitionProperty();
         RangePartitionInfo rangePartitionInfo = (RangePartitionInfo) olapTable.getPartitionInfo();
         ZonedDateTime now = ZonedDateTime.now(dynamicPartitionProperty.getTimeZone().toZoneId());
-        for (int i = 0; i <= dynamicPartitionProperty.getEnd(); i++) {
-            String prevBorder = DynamicPartitionUtil.getPartitionRangeString(dynamicPartitionProperty, now, i, partitionFormat);
-            String nextBorder = DynamicPartitionUtil.getPartitionRangeString(dynamicPartitionProperty, now, i + 1, partitionFormat);
+
+        boolean createHistoryPartition = dynamicPartitionProperty.isCreateHistoryPartition();
+        int idx = createHistoryPartition ? dynamicPartitionProperty.getStart() : 0;
+
+        for (; idx <= dynamicPartitionProperty.getEnd(); idx++) {
+            String prevBorder = DynamicPartitionUtil.getPartitionRangeString(dynamicPartitionProperty, now, idx, partitionFormat);
+            String nextBorder = DynamicPartitionUtil.getPartitionRangeString(dynamicPartitionProperty, now, idx + 1, partitionFormat);
             PartitionValue lowerValue = new PartitionValue(prevBorder);
             PartitionValue upperValue = new PartitionValue(nextBorder);
 

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

@@ -1379,4 +1379,12 @@ public class Config extends ConfigBase {
      */
     @ConfField
     public static int grpc_max_message_size_bytes = 1 * 1024 * 1024 * 1024; // 1GB
+
+    /**
+     * Used to limit the maximum number of partitions that can be created when creating a dynamic partition table,
+     * to avoid creating too many partitions at one time.
+     * The number is determined by "start" and "end" in the dynamic partition parameters.
+     */
+    @ConfField(mutable = true, masterOnly = true)
+    public static int max_dynamic_partition_num = 500;
 }

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

@@ -239,7 +239,9 @@ public enum ErrorCode {
     ERROR_DYNAMIC_PARTITION_REPLICATION_NUM_FORMAT(5072, new byte[] {'4', '2', '0', '0', '0'},
             "Invalid dynamic partition replication num: %s."),
     ERROR_CREATE_TABLE_LIKE_EMPTY(5073, new byte[] {'4', '2', '0', '0', '0'},
-            "Origin create table stmt is empty");
+            "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");
 
     ErrorCode(int code, byte[] sqlState, String errorMsg) {
         this.code = code;