[DynamicPartition] Support specifying hot data partition

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

@@ -131,6 +131,22 @@ The rules of dynamic partition are prefixed with `dynamic_partition.`:
 
     When the `start` attribute is not specified, this parameter has no effect.
 
+* `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`.
+
+    Let us give an example. Suppose today is 2021-05-20, partition by day, and the properties of dynamic partition are set to: hot_partition_num=2, end=3, start=-3. Then the system will automatically create the following partitions, and set the `storage_medium` and `storage_cooldown_time` properties:
+
+    ```
+    p20210517: ["2021-05-17", "2021-05-18") storage_medium=HDD storage_cooldown_time=9999-12-31 23:59:59
+    p20210518: ["2021-05-18", "2021-05-19") storage_medium=HDD storage_cooldown_time=9999-12-31 23:59:59
+    p20210519: ["2021-05-19", "2021-05-20") storage_medium=SSD storage_cooldown_time=2021-05-21 00:00:00
+    p20210520: ["2021-05-20", "2021-05-21") storage_medium=SSD storage_cooldown_time=2021-05-22 00:00:00
+    p20210521: ["2021-05-21", "2021-05-22") storage_medium=SSD storage_cooldown_time=2021-05-23 00:00:00
+    p20210522: ["2021-05-22", "2021-05-23") storage_medium=SSD storage_cooldown_time=2021-05-24 00:00:00
+    p20210523: ["2021-05-23", "2021-05-24") storage_medium=SSD storage_cooldown_time=2021-05-25 00:00:00
+    ```
+
 ### 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/dynamic-partition.md

@@ -128,6 +128,22 @@ under the License.
     默认为 false。当置为 true 时，Doris 会自动创建由 start 到 end 的所有分区。同时，FE 的参数 `max_dynamic_partition_num` 会限制总分区数量，以避免一次性创建过多分区。当 end - start 的值大于 `max_dynamic_partition_num` 值时，操作将被禁止。
 
     当不指定 `start` 属性时，该参数不生效。
+
+* `dynamic_partition.hot_partition_num`
+
+    指定最新的多少个分区为热分区。对于热分区，系统会自动设置其 `storage_medium` 参数为SSD，并且设置 `storage_cooldown_time`。
+
+    我们举例说明。假设今天是 2021-05-20，按天分区，动态分区的属性设置为：hot_partition_num=2, end=3, start=-3。则系统会自动创建以下分区，并且设置 `storage_medium` 和 `storage_cooldown_time` 参数：
+
+    ```
+    p20210517：["2021-05-17", "2021-05-18") storage_medium=HDD storage_cooldown_time=9999-12-31 23:59:59
+    p20210518：["2021-05-18", "2021-05-19") storage_medium=HDD storage_cooldown_time=9999-12-31 23:59:59
+    p20210519：["2021-05-19", "2021-05-20") storage_medium=SSD storage_cooldown_time=2021-05-21 00:00:00
+    p20210520：["2021-05-20", "2021-05-21") storage_medium=SSD storage_cooldown_time=2021-05-22 00:00:00
+    p20210521：["2021-05-21", "2021-05-22") storage_medium=SSD storage_cooldown_time=2021-05-23 00:00:00
+    p20210522：["2021-05-22", "2021-05-23") storage_medium=SSD storage_cooldown_time=2021-05-24 00:00:00
+    p20210523：["2021-05-23", "2021-05-24") storage_medium=SSD storage_cooldown_time=2021-05-25 00:00:00
+    ```
 
 ### 注意事项 
 

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

@@ -37,6 +37,7 @@ 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 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;
@@ -55,6 +56,9 @@ public class DynamicPartitionProperty {
     private TimeZone tz = TimeUtils.getSystemTimeZone();
     private int replicationNum;
     private boolean createHistoryPartition = false;
+    // 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;
 
     public DynamicPartitionProperty(Map<String, String> properties) {
         if (properties != null && !properties.isEmpty()) {
@@ -69,6 +73,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.hotPartitionNum = Integer.parseInt(properties.getOrDefault(HOT_PARTITION_NUM, "0"));
             createStartOfs(properties);
         } else {
             this.exist = false;
@@ -131,6 +136,10 @@ public boolean isCreateHistoryPartition() {
         return createHistoryPartition;
     }
 
+    public int getHotPartitionNum() {
+        return hotPartitionNum;
+    }
+
     public String getStartOfInfo() {
         if (getTimeUnit().equalsIgnoreCase(TimeUnit.WEEK.toString())) {
             return startOfWeek.toDisplayInfo();
@@ -165,7 +174,8 @@ public String getProperties(int tableReplicationNum) {
                 ",\n\"" + PREFIX + "\" = \"" + prefix + "\"" +
                 ",\n\"" + REPLICATION_NUM + "\" = \"" + useReplicationNum + "\"" +
                 ",\n\"" + BUCKETS + "\" = \"" + buckets + "\"" +
-                ",\n\"" + CREATE_HISTORY_PARTITION + "\" = \"" + createHistoryPartition + "\"";
+                ",\n\"" + CREATE_HISTORY_PARTITION + "\" = \"" + createHistoryPartition + "\"" +
+                ",\n\"" + HOT_PARTITION_NUM + "\" = \"" + hotPartitionNum + "\"";
         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/catalog/TableProperty.java

@@ -92,6 +92,7 @@ public TableProperty buildProperty(short opCode) {
         }
         return this;
     }
+
     public TableProperty buildDynamicProperty() throws DdlException {
         if (properties.containsKey(DynamicPartitionProperty.ENABLE)
                 && Boolean.valueOf(properties.get(DynamicPartitionProperty.ENABLE))

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

@@ -42,8 +42,10 @@
 import org.apache.doris.common.Pair;
 import org.apache.doris.common.util.DynamicPartitionUtil;
 import org.apache.doris.common.util.MasterDaemon;
+import org.apache.doris.common.util.PropertyAnalyzer;
 import org.apache.doris.common.util.RangeUtils;
 import org.apache.doris.common.util.TimeUtils;
+import org.apache.doris.thrift.TStorageMedium;
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -145,6 +147,8 @@ private ArrayList<AddPartitionClause> getAddPartitionClause(Database db, OlapTab
 
         boolean createHistoryPartition = dynamicPartitionProperty.isCreateHistoryPartition();
         int idx = createHistoryPartition ? dynamicPartitionProperty.getStart() : 0;
+        int hotPartitionNum = dynamicPartitionProperty.getHotPartitionNum();
+        String timeUnit = dynamicPartitionProperty.getTimeUnit();
 
         for (; idx <= dynamicPartitionProperty.getEnd(); idx++) {
             String prevBorder = DynamicPartitionUtil.getPartitionRangeString(dynamicPartitionProperty, now, idx, partitionFormat);
@@ -185,12 +189,19 @@ private ArrayList<AddPartitionClause> getAddPartitionClause(Database db, OlapTab
 
             // construct partition desc
             PartitionKeyDesc partitionKeyDesc = PartitionKeyDesc.createFixed(Collections.singletonList(lowerValue), Collections.singletonList(upperValue));
-            HashMap<String, String> partitionProperties = new HashMap<>(1);
+            HashMap<String, String> partitionProperties = Maps.newHashMap();
             if (dynamicPartitionProperty.getReplicationNum() == DynamicPartitionProperty.NOT_SET_REPLICATION_NUM) {
                 partitionProperties.put("replication_num", String.valueOf(olapTable.getDefaultReplicationNum()));
             } else {
                 partitionProperties.put("replication_num", String.valueOf(dynamicPartitionProperty.getReplicationNum()));
             }
+
+            if (hotPartitionNum > 0) {
+                // set storage_medium and storage_cooldown_time based on dynamic_partition.hot_partition_num
+                setStorageMediumProperty(partitionProperties, dynamicPartitionProperty, now, hotPartitionNum, idx);
+            }
+
+
             String partitionName = dynamicPartitionProperty.getPrefix() + DynamicPartitionUtil.getFormattedPartitionName(
                     dynamicPartitionProperty.getTimeZone(), prevBorder, dynamicPartitionProperty.getTimeUnit());
             SinglePartitionDesc rangePartitionDesc = new SinglePartitionDesc(true, partitionName,
@@ -210,6 +221,17 @@ private ArrayList<AddPartitionClause> getAddPartitionClause(Database db, OlapTab
         return addPartitionClauses;
     }
 
+    private void setStorageMediumProperty(HashMap<String, String> partitionProperties, DynamicPartitionProperty property,
+                                          ZonedDateTime now, int hotPartitionNum, int offset) {
+        if (offset + hotPartitionNum <= 0) {
+            return;
+        }
+        partitionProperties.put(PropertyAnalyzer.PROPERTIES_STORAGE_MEDIUM, TStorageMedium.SSD.name());
+        String cooldownTime = DynamicPartitionUtil.getPartitionRangeString(property, now, offset + hotPartitionNum,
+                DynamicPartitionUtil.DATETIME_FORMAT);
+        partitionProperties.put(PropertyAnalyzer.PROPERTIES_STORAGE_COLDOWN_TIME, cooldownTime);
+    }
+
     /**
      * 1. get the range of [start, 0) as a reserved range.
      * 2. get DropPartitionClause of partitions which range are before this reserved range.

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

@@ -57,9 +57,9 @@
 public class DynamicPartitionUtil {
     private static final Logger LOG = LogManager.getLogger(DynamicPartitionUtil.class);
 
-    private static final String TIMESTAMP_FORMAT = "yyyyMMdd";
-    private static final String DATE_FORMAT = "yyyy-MM-dd";
-    private static final String DATETIME_FORMAT = "yyyy-MM-dd HH:mm:ss";
+    public static final String TIMESTAMP_FORMAT = "yyyyMMdd";
+    public static final String DATE_FORMAT = "yyyy-MM-dd";
+    public static final String DATETIME_FORMAT = "yyyy-MM-dd HH:mm:ss";
 
     public static void checkTimeUnit(String timeUnit, PartitionInfo partitionInfo) throws DdlException {
         if (Strings.isNullOrEmpty(timeUnit)
@@ -195,6 +195,19 @@ private static void checkReplicationNum(String val) throws DdlException {
         }
     }
 
+    private static void checkHotPartitionNum(String val) throws DdlException {
+        if (Strings.isNullOrEmpty(val)) {
+            throw new DdlException("Invalid properties: " + DynamicPartitionProperty.HOT_PARTITION_NUM);
+        }
+        try {
+            if (Integer.parseInt(val) < 0) {
+                throw new DdlException(DynamicPartitionProperty.HOT_PARTITION_NUM + " must larger than 0.");
+            }
+        } catch (NumberFormatException e) {
+            throw new DdlException("Invalid " + DynamicPartitionProperty.HOT_PARTITION_NUM + " value");
+        }
+    }
+
     public static boolean checkDynamicPartitionPropertiesExist(Map<String, String> properties) {
         if (properties == null) {
             return false;
@@ -208,7 +221,8 @@ public static boolean checkDynamicPartitionPropertiesExist(Map<String, String> p
                 properties.containsKey(DynamicPartitionProperty.REPLICATION_NUM) ||
                 properties.containsKey(DynamicPartitionProperty.ENABLE) ||
                 properties.containsKey(DynamicPartitionProperty.START_DAY_OF_WEEK) ||
-                properties.containsKey(DynamicPartitionProperty.START_DAY_OF_MONTH);
+                properties.containsKey(DynamicPartitionProperty.START_DAY_OF_MONTH) ||
+                properties.containsKey(DynamicPartitionProperty.HOT_PARTITION_NUM);
     }
 
     public static boolean checkInputDynamicPartitionProperties(Map<String, String> properties, PartitionInfo partitionInfo) throws DdlException {
@@ -369,6 +383,14 @@ public static Map<String, String> analyzeDynamicPartition(Map<String, String> pr
             properties.remove(DynamicPartitionProperty.REPLICATION_NUM);
             analyzedProperties.put(DynamicPartitionProperty.REPLICATION_NUM, val);
         }
+
+        if (properties.containsKey(DynamicPartitionProperty.HOT_PARTITION_NUM)) {
+            String val = properties.get(DynamicPartitionProperty.HOT_PARTITION_NUM);
+            checkHotPartitionNum(val);
+            properties.remove(DynamicPartitionProperty.HOT_PARTITION_NUM);
+            analyzedProperties.put(DynamicPartitionProperty.HOT_PARTITION_NUM, val);
+        }
+
         return analyzedProperties;
     }
 
@@ -482,7 +504,7 @@ public static String getPartitionRangeString(DynamicPartitionProperty property,
      * Today is 2020-05-24 00, offset = 1
      * It will return 2020-05-24 01:00:00
      */
-    private static String getPartitionRangeOfHour(ZonedDateTime current, int offset, String format) {
+    public static String getPartitionRangeOfHour(ZonedDateTime current, int offset, String format) {
         return getFormattedTimeWithoutMinuteSecond(current.plusHours(offset), format);
     }