diff --git a/docs/en/administrator-guide/dynamic-partition.md b/docs/en/administrator-guide/dynamic-partition.md
index a0afe4e68ed18e..11125cc746a8a0 100644
--- a/docs/en/administrator-guide/dynamic-partition.md
+++ b/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.
diff --git a/docs/zh-CN/administrator-guide/dynamic-partition.md b/docs/zh-CN/administrator-guide/dynamic-partition.md
index ddbd55f6964ec5..80d316364e5ac1 100644
--- a/docs/zh-CN/administrator-guide/dynamic-partition.md
+++ b/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
+    ```
   
 ### 注意事项 
  
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/DynamicPartitionProperty.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/DynamicPartitionProperty.java
index 4fa23197685c75..50892a0d040fd0 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/DynamicPartitionProperty.java
+++ b/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())) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/TableProperty.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/TableProperty.java
index bbc2aa8a5f4e77..bf94346dcb81d4 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/TableProperty.java
+++ b/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))
diff --git a/fe/fe-core/src/main/java/org/apache/doris/clone/DynamicPartitionScheduler.java b/fe/fe-core/src/main/java/org/apache/doris/clone/DynamicPartitionScheduler.java
index 78e02cf6208f09..b61168984287bd 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/clone/DynamicPartitionScheduler.java
+++ b/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.
diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/util/DynamicPartitionUtil.java b/fe/fe-core/src/main/java/org/apache/doris/common/util/DynamicPartitionUtil.java
index b824d8bb07452a..4454087514da23 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/common/util/DynamicPartitionUtil.java
+++ b/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);
     }
 
diff --git a/fe/fe-core/src/test/java/org/apache/doris/catalog/DynamicPartitionTableTest.java b/fe/fe-core/src/test/java/org/apache/doris/catalog/DynamicPartitionTableTest.java
index 28e890b30fc7b1..64fcb778d0f545 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/catalog/DynamicPartitionTableTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/catalog/DynamicPartitionTableTest.java
@@ -22,7 +22,9 @@
 import org.apache.doris.common.Config;
 import org.apache.doris.common.DdlException;
 import org.apache.doris.common.FeConstants;
+import org.apache.doris.common.util.DynamicPartitionUtil;
 import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.thrift.TStorageMedium;
 import org.apache.doris.utframe.UtFrameUtils;
 
 import com.clearspring.analytics.util.Lists;
@@ -41,6 +43,8 @@
 import java.util.GregorianCalendar;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
 import java.util.UUID;
 
 public class DynamicPartitionTableTest {
@@ -772,4 +776,245 @@ public void testHourDynamicPartitionWithIntType() throws Exception {
         createTable(createOlapTblStmt);
     }
 
+    @Test
+    public void testHotPartitionNum() throws Exception {
+        Database testDb = Catalog.getCurrentCatalog().getDb("default_cluster:test");
+        // 1. hour
+        String createOlapTblStmt = "CREATE TABLE test.`hot_partition_hour_tbl1` (\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.start\" = \"-3\",\n" +
+                "\"dynamic_partition.end\" = \"3\",\n" +
+                "\"dynamic_partition.create_history_partition\" = \"true\",\n" +
+                "\"dynamic_partition.time_unit\" = \"hour\",\n" +
+                "\"dynamic_partition.prefix\" = \"p\",\n" +
+                "\"dynamic_partition.buckets\" = \"1\",\n" +
+                "\"dynamic_partition.hot_partition_num\" = \"1\"\n" +
+                ");";
+        createTable(createOlapTblStmt);
+        OlapTable tbl = (OlapTable)testDb.getTable("hot_partition_hour_tbl1");
+        RangePartitionInfo partitionInfo = (RangePartitionInfo) tbl.getPartitionInfo();
+        Map<Long, DataProperty> idToDataProperty = new TreeMap<>(partitionInfo.idToDataProperty);
+        Assert.assertEquals(7, idToDataProperty.size());
+        int count = 0;
+        for (DataProperty dataProperty : idToDataProperty.values()) {
+            if (count < 3) {
+                Assert.assertEquals(TStorageMedium.HDD, dataProperty.getStorageMedium());
+            } else {
+                Assert.assertEquals(TStorageMedium.SSD, dataProperty.getStorageMedium());
+            }
+            ++count;
+        }
+
+        createOlapTblStmt = "CREATE TABLE test.`hot_partition_hour_tbl2` (\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.start\" = \"-3\",\n" +
+                "\"dynamic_partition.end\" = \"3\",\n" +
+                "\"dynamic_partition.create_history_partition\" = \"true\",\n" +
+                "\"dynamic_partition.time_unit\" = \"hour\",\n" +
+                "\"dynamic_partition.prefix\" = \"p\",\n" +
+                "\"dynamic_partition.buckets\" = \"1\",\n" +
+                "\"dynamic_partition.hot_partition_num\" = \"0\"\n" +
+                ");";
+        createTable(createOlapTblStmt);
+        tbl = (OlapTable)testDb.getTable("hot_partition_hour_tbl2");
+        partitionInfo = (RangePartitionInfo) tbl.getPartitionInfo();
+        idToDataProperty = new TreeMap<>(partitionInfo.idToDataProperty);
+        Assert.assertEquals(7, idToDataProperty.size());
+        for (DataProperty dataProperty : idToDataProperty.values()) {
+            Assert.assertEquals(TStorageMedium.HDD, dataProperty.getStorageMedium());
+        }
+
+        createOlapTblStmt = "CREATE TABLE test.`hot_partition_hour_tbl3` (\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.start\" = \"-3\",\n" +
+                "\"dynamic_partition.end\" = \"3\",\n" +
+                "\"dynamic_partition.create_history_partition\" = \"true\",\n" +
+                "\"dynamic_partition.time_unit\" = \"hour\",\n" +
+                "\"dynamic_partition.prefix\" = \"p\",\n" +
+                "\"dynamic_partition.buckets\" = \"1\",\n" +
+                "\"dynamic_partition.hot_partition_num\" = \"3\"\n" +
+                ");";
+        createTable(createOlapTblStmt);
+        tbl = (OlapTable)testDb.getTable("hot_partition_hour_tbl3");
+        partitionInfo = (RangePartitionInfo) tbl.getPartitionInfo();
+        idToDataProperty = new TreeMap<>(partitionInfo.idToDataProperty);
+        Assert.assertEquals(7, idToDataProperty.size());
+        count = 0;
+        for (DataProperty dataProperty : idToDataProperty.values()) {
+            if (count < 1) {
+                Assert.assertEquals(TStorageMedium.HDD, dataProperty.getStorageMedium());
+            } else {
+                Assert.assertEquals(TStorageMedium.SSD, dataProperty.getStorageMedium());
+            }
+            ++count;
+        }
+
+        // 2. day
+        createOlapTblStmt = "CREATE TABLE test.`hot_partition_day_tbl1` (\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.start\" = \"-3\",\n" +
+                "\"dynamic_partition.end\" = \"3\",\n" +
+                "\"dynamic_partition.time_unit\" = \"day\",\n" +
+                "\"dynamic_partition.prefix\" = \"p\",\n" +
+                "\"dynamic_partition.buckets\" = \"1\",\n" +
+                "\"dynamic_partition.hot_partition_num\" = \"2\"\n" +
+                ");";
+        createTable(createOlapTblStmt);
+        tbl = (OlapTable)testDb.getTable("hot_partition_day_tbl1");
+        partitionInfo = (RangePartitionInfo) tbl.getPartitionInfo();
+        idToDataProperty = new TreeMap<>(partitionInfo.idToDataProperty);
+        Assert.assertEquals(4, idToDataProperty.size());
+        for (DataProperty dataProperty : idToDataProperty.values()) {
+            Assert.assertEquals(TStorageMedium.SSD, dataProperty.getStorageMedium());
+        }
+
+        createOlapTblStmt = "CREATE TABLE test.`hot_partition_day_tbl2` (\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.start\" = \"-3\",\n" +
+                "\"dynamic_partition.end\" = \"4\",\n" +
+                "\"dynamic_partition.create_history_partition\" = \"true\",\n" +
+                "\"dynamic_partition.time_unit\" = \"day\",\n" +
+                "\"dynamic_partition.prefix\" = \"p\",\n" +
+                "\"dynamic_partition.buckets\" = \"1\",\n" +
+                "\"dynamic_partition.hot_partition_num\" = \"2\"\n" +
+                ");";
+        createTable(createOlapTblStmt);
+        tbl = (OlapTable)testDb.getTable("hot_partition_day_tbl2");
+        partitionInfo = (RangePartitionInfo) tbl.getPartitionInfo();
+        idToDataProperty = new TreeMap<>(partitionInfo.idToDataProperty);
+        Assert.assertEquals(8, idToDataProperty.size());
+        count = 0;
+        for (DataProperty dataProperty : idToDataProperty.values()) {
+            if (count < 2) {
+                Assert.assertEquals(TStorageMedium.HDD, dataProperty.getStorageMedium());
+            } else {
+                Assert.assertEquals(TStorageMedium.SSD, dataProperty.getStorageMedium());
+            }
+            ++count;
+        }
+        // 3. week
+        createOlapTblStmt = "CREATE TABLE test.`hot_partition_week_tbl1` (\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.start\" = \"-3\",\n" +
+                "\"dynamic_partition.end\" = \"4\",\n" +
+                "\"dynamic_partition.create_history_partition\" = \"true\",\n" +
+                "\"dynamic_partition.time_unit\" = \"week\",\n" +
+                "\"dynamic_partition.prefix\" = \"p\",\n" +
+                "\"dynamic_partition.buckets\" = \"1\",\n" +
+                "\"dynamic_partition.hot_partition_num\" = \"1\"\n" +
+                ");";
+        createTable(createOlapTblStmt);
+        tbl = (OlapTable)testDb.getTable("hot_partition_week_tbl1");
+        partitionInfo = (RangePartitionInfo) tbl.getPartitionInfo();
+        idToDataProperty = new TreeMap<>(partitionInfo.idToDataProperty);
+        Assert.assertEquals(8, idToDataProperty.size());
+        count = 0;
+        for (DataProperty dataProperty : idToDataProperty.values()) {
+            if (count < 3) {
+                Assert.assertEquals(TStorageMedium.HDD, dataProperty.getStorageMedium());
+            } else {
+                Assert.assertEquals(TStorageMedium.SSD, dataProperty.getStorageMedium());
+            }
+            ++count;
+        }
+        // 4. month
+        createOlapTblStmt = "CREATE TABLE test.`hot_partition_month_tbl1` (\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.start\" = \"-3\",\n" +
+                "\"dynamic_partition.end\" = \"4\",\n" +
+                "\"dynamic_partition.create_history_partition\" = \"true\",\n" +
+                "\"dynamic_partition.time_unit\" = \"month\",\n" +
+                "\"dynamic_partition.prefix\" = \"p\",\n" +
+                "\"dynamic_partition.buckets\" = \"1\",\n" +
+                "\"dynamic_partition.hot_partition_num\" = \"4\"\n" +
+                ");";
+        createTable(createOlapTblStmt);
+        tbl = (OlapTable)testDb.getTable("hot_partition_month_tbl1");
+        partitionInfo = (RangePartitionInfo) tbl.getPartitionInfo();
+        idToDataProperty = new TreeMap<>(partitionInfo.idToDataProperty);
+        Assert.assertEquals(8, idToDataProperty.size());
+        for (DataProperty dataProperty : idToDataProperty.values()) {
+            Assert.assertEquals(TStorageMedium.SSD, dataProperty.getStorageMedium());
+        }
+    }
+
+    @Test(expected = DdlException.class)
+    public void testHotPartitionNumAbnormal() throws Exception {
+        // dynamic_partition.hot_partition_num must larger than 0.
+        String createOlapTblStmt = "CREATE TABLE test.`hot_partition_hour_tbl1` (\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.start\" = \"-3\",\n" +
+                "\"dynamic_partition.end\" = \"3\",\n" +
+                "\"dynamic_partition.create_history_partition\" = \"true\",\n" +
+                "\"dynamic_partition.time_unit\" = \"hour\",\n" +
+                "\"dynamic_partition.prefix\" = \"p\",\n" +
+                "\"dynamic_partition.buckets\" = \"1\",\n" +
+                "\"dynamic_partition.hot_partition_num\" = \"-1\"\n" +
+                ");";
+        createTable(createOlapTblStmt);
+    }
 }