diff --git a/docs/en/administrator-guide/dynamic-partition.md b/docs/en/administrator-guide/dynamic-partition.md
index 4cdffb2b185f83..26dd3a85b2aa67 100644
--- a/docs/en/administrator-guide/dynamic-partition.md
+++ b/docs/en/administrator-guide/dynamic-partition.md
@@ -28,9 +28,7 @@ under the License.
 
 Dynamic partition is a new feature introduced in Doris verion 0.12. It's designed to manage partition's Time-to-Life (TTL), reducing the burden on users.
 
-The original design, implementation and effect can be referred to [ISSUE 2262](https://github.com/apache/incubator-doris/issues/2262)。
-
-Currently, the function of adding partitions dynamically is implemented, and the next version will support removing partitions dynamically.
+At present, the functions of dynamically adding partitions and dynamically deleting partitions are realized.
 
 ## Noun Interpretation
 
@@ -39,105 +37,245 @@ Currently, the function of adding partitions dynamically is implemented, and the
 
 ## Principle
 
-In some scenarios, the user will create partitions for the table according to the day and perform routine tasks regularly every day. In this case, the user needs to manually manage the partition, otherwise the data import may fail because the partition is forgot to create, which brings additional maintenance costs to the user.
-
-The design of implementation is that FE will starts a background thread that determines whether or not to start the thread and the scheduling frequency of the thread based on the parameters `dynamic_partition_enable` and `dynamic_partition_check_interval_seconds` in `fe.conf`.
+In some usage scenarios, the user will partition the table according to the day and perform routine tasks regularly every day. At this time, the user needs to manually manage the partition. Otherwise, the data load may fail because the user does not create a partition. This brings additional maintenance costs to the user.
 
-When create a olap table, the `dynamic_partition` properties will be assigned. FE will parse `dynamic_partition` properties and check the legitimacy of the input parameters firstly, and then persist the properties to FE metadata, register the table to the list of dynamic partition at the same time. Daemon thread will scan the dynamic partition list periodically according to the configuration parameters,
-read dynamic partition properties of the table, and doing the task of adding partitions. The scheduling information of each time will be kept in the memory of FE. You can check whether the scheduling task is successful through `SHOW DYNAMIC PARTITION TABLES`.
+Through the dynamic partitioning feature, users can set the rules of dynamic partitioning when building tables. FE will start a background thread to create or delete partitions according to the rules specified by the user. Users can also change existing rules at runtime.
 
 ## Usage
 
 ### Establishment of tables
 
-When creating a table, you can specify the attribute `dynamic_partition` in `PROPERTIES`, which means that the table is a dynamic partition table.
-    
-Examples:
+The rules for dynamic partitioning can be specified when the table is created or modified at runtime. Currently,dynamic partition rules can only be set for partition tables with single partition columns.    
 
-```
-CREATE TABLE example_db.dynamic_partition
-(
-k1 DATE,
-k2 INT,
-k3 SMALLINT,
-v1 VARCHAR(2048),
-v2 DATETIME DEFAULT "2014-02-04 15:36:00"
-)
-ENGINE=olap
-DUPLICATE KEY(k1, k2, k3)
-PARTITION BY RANGE (k1)
-(
-PARTITION p1 VALUES LESS THAN ("2014-01-01"),
-PARTITION p2 VALUES LESS THAN ("2014-06-01"),
-PARTITION p3 VALUES LESS THAN ("2014-12-01")
-)
-DISTRIBUTED BY HASH(k2) BUCKETS 32
-PROPERTIES(
-"storage_medium" = "SSD",
-"dynamic_partition.enable" = "true"
-"dynamic_partition.time_unit" = "DAY",
-"dynamic_partition.end" = "3",
-"dynamic_partition.prefix" = "p",
-"dynamic_partition.buckets" = "32"
- );
-```
-Create a dynamic partition table, specify enable dynamic partition features, take today is 2020-01-08 for example, at every time of scheduling, will create today and after 3 days in advance of four partitions 
-(if the partition is existed, the task will be ignored), partition name respectively according to the specified prefix `p20200108` `p20200109` `p20200110` `p20200111`, each partition to 32 the number of points barrels, each partition scope is as follows:
-```
-[types: [DATE]; keys: [2020-01-08]; ‥types: [DATE]; keys: [2020-01-09]; )
-[types: [DATE]; keys: [2020-01-09]; ‥types: [DATE]; keys: [2020-01-10]; )
-[types: [DATE]; keys: [2020-01-10]; ‥types: [DATE]; keys: [2020-01-11]; )
-[types: [DATE]; keys: [2020-01-11]; ‥types: [DATE]; keys: [2020-01-12]; )
-```
+* Specified when creating table
+
+    ```
+    CREATE TABLE tbl1
+    (...)
+    PROPERTIES
+    (
+        "dynamic_partition.prop1" = "value1",
+        "dynamic_partition.prop2" = "value2",
+        ...
+    )
+    ```
     
-### Enable Dynamic Partition Feature
+* Modify at runtime
+
+    ```
+    ALTER TABLE tbl1 SET
+    (
+        "dynamic_partition.prop1" = "value1",
+        "dynamic_partition.prop2" = "value2",
+        ...
+    )
+    ```
+    
+### Dynamic partition rule parameters
 
-1. First of all, `dynamic_partition_enable=true` needs to be set in fe.conf, which can be specified by modifying the configuration file when the cluster starts up, or dynamically modified by HTTP interface at run time
+The rules of dynamic partition are prefixed with `dynamic_partition.`:
 
-2. If you need to add dynamic partitioning properties to a table prior to version 0.12, you need to modify the properties of the table with the following command
+* `dynamic_partition.enable`
 
-```
-ALTER TABLE dynamic_partition set ("dynamic_partition.enable" = "true", "dynamic_partition.time_unit" = "DAY", "dynamic_partition.end" = "3", "dynamic_partition.prefix" = "p", "dynamic_partition.buckets" = "32");
-```
+    Whether to enable the dynamic partition feature. Can be specified as `TRUE` or` FALSE`. If not filled, the default is `TRUE`. If it is `FALSE`, Doris will ignore the dynamic partitioning rules of the table.
 
-### Disable Dynamic Partition Feature
+* `dynamic_partition.time_unit`
 
-If you need to stop dynamic partitioning for all dynamic partitioning tables in the cluster, you need to set 'dynamic_partition_enable=true' in fe.conf
+    The unit for dynamic partition scheduling. Can be specified as `DAY`,` WEEK`, and `MONTH`, means to create or delete partitions by day, week, and month, respectively.
 
-If you need to stop dynamic partitioning for a specified table, you can modify the properties of the table with the following command
+    When specified as `DAY`, the suffix format of the dynamically created partition name is `yyyyMMdd`, for example, `20200325`.
 
-```
-ALTER TABLE dynamic_partition set ("dynamic_partition.enable" = "false")
-```
+    When specified as `WEEK`, the suffix format of the dynamically created partition name is `yyyy_ww`. That is, the week of the year of current date. For example, the suffix of the partition created for `2020-03-25` is `2020_13`, indicating that it is currently the 13th week of 2020.
+
+    When specified as `MONTH`, the suffix format of the dynamically created partition name is `yyyyMM`, for example, `202003`.
+
+* `dynamic_partition.start`
+
+    The starting offset of the dynamic partition, usually a negative number. Depending on the `time_unit` attribute, based on the current day (week / month), the partitions with a partition range before this offset will be deleted. If not filled, the default is `-2147483648`, that is, the history partition will not be  deleted.
+    
+* `dynamic_partition.end`
+
+    The end offset of the dynamic partition, usually a positive number. According to the difference of the `time_unit` attribute, the partition of the corresponding range is created in advance based on the current day (week / month).
+
+* `dynamic_partition.prefix`
+
+    The dynamically created partition name prefix.
+
+* `dynamic_partition.buckets`
+
+    The number of buckets corresponding to the dynamically created partitions.
+    
+* `dynamic_partition.start_day_of_week`
+
+    When `time_unit` is` WEEK`, this parameter is used to specify the starting point of the week. The value ranges from 1 to 7. Where 1 is Monday and 7 is Sunday. The default is 1, which means that every week starts on Monday.
+    
+* `dynamic_partition.start_day_of_month`
+
+    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.
+
+### Example
+
+1. Table `tbl1` partition column k1, type is DATE, create a dynamic partition rule. By day partition, only the partitions of the last 7 days are kept, and the partitions of the next 3 days are created in advance.
+
+    ```
+    CREATA TABLE tbl1
+    (
+        k1 DATE,
+        ...
+    )
+    PARTITION BY RANGE(K1) ()
+    PROPERTIES
+    (
+        "dynamic_partition.enable" = "true",
+        "dynamic_partition.time_unit" = "DAY",
+        "dynamic_partition.start" = "-7",
+        "dynamic_partition.end" = "3",
+        "dynamic_partition.prefix" = "p",
+        "dynamic_partition.buckets" = "32"
+    );
+    ```
+
+    Suppose the current date is 2020-05-29. According to the above rules, tbl1 will produce the following partitions:
+    
+    ```
+    p20200529: ["2020-05-29", "2020-05-30")
+    p20200530: ["2020-05-30", "2020-05-31")
+    p20200531: ["2020-05-31", "2020-06-01")
+    p20200601: ["2020-06-01", "2020-06-02")
+    ```
+
+    On the next day, 2020-05-30, a new partition will be created `p20200602: [" 2020-06-02 "," 2020-06-03 ")`
+    
+    On 2020-06-06, because `dynamic_partition.start` is set to 7, the partition 7 days ago will be deleted, that is, the partition `p20200529` will be deleted.
+    
+2. Table tbl1 partition column k1, type is DATETIME, create a dynamic partition rule. Partition by week, only keep the partition of the last 2 weeks, and create the partition of the next 2 weeks in advance.
+
+    ```
+    CREATA TABLE tbl1
+    (
+        k1 DATETIME,
+        ...
+    )
+    PARTITION BY RANGE(K1) ()
+    PROPERTIES
+    (
+        "dynamic_partition.enable" = "true",
+        "dynamic_partition.time_unit" = "WEEK",
+        "dynamic_partition.start" = "-2",
+        "dynamic_partition.end" = "2",
+        "dynamic_partition.prefix" = "p",
+        "dynamic_partition.buckets" = "8"
+    );
+    ```
+
+    Suppose the current date is 2020-05-29, which is the 22nd week of 2020. The default week starts on Monday. Based on the above rules, tbl1 will produce the following partitions:
+    
+    ```
+    p2020_22: ["2020-05-25 00:00:00", "2020-06-01 00:00:00")
+    p2020_23: ["2020-06-01 00:00:00", "2020-06-08 00:00:00")
+    p2020_24: ["2020-06-08 00:00:00", "2020-06-15 00:00:00")
+    ```
+    
+    The start date of each partition is Monday of the week. At the same time, because the type of the partition column k1 is DATETIME, the partition value will fill the hour, minute and second fields, and all are 0.
+
+    On 2020-06-15, the 25th week, the partition 2 weeks ago will be deleted, ie `p2020_22` will be deleted.
+
+    In the above example, suppose the user specified the start day of the week as `"dynamic_partition.start_day_of_week" = "3"`, that is, set Wednesday as the start of week. The partition is as follows:
+    
+    ```
+    p2020_22: ["2020-05-27 00:00:00", "2020-06-03 00:00:00")
+    p2020_23: ["2020-06-03 00:00:00", "2020-06-10 00:00:00")
+    p2020_24: ["2020-06-10 00:00:00", "2020-06-17 00:00:00")
+    ```
+    
+    That is, the partition ranges from Wednesday of the current week to Tuesday of the next week.
+    
+    * Note: 2019-12-31 and 2020-01-01 are in same week, if the starting date of the partition is 2019-12-31, the partition name is `p2019_53`, if the starting date of the partition is 2020-01 -01, the partition name is `p2020_01`.
+
+3. Table tbl1 partition column k1, type is DATE, create a dynamic partition rule. Partition by month without deleting historical partitions, and create partitions for the next 2 months in advance. At the same time, set the starting date on the 3rd of each month.
+
+    ```
+    CREATA TABLE tbl1
+    (
+        k1 DATE,
+        ...
+    )
+    PARTITION BY RANGE(K1) ()
+    PROPERTIES
+    (
+        "dynamic_partition.enable" = "true",
+        "dynamic_partition.time_unit" = "MONTH",
+        "dynamic_partition.end" = "2",
+        "dynamic_partition.prefix" = "p",
+        "dynamic_partition.buckets" = "8",
+        "dynamic_partition.start_day_of_month" = "3"
+    );
+    ```
+    
+    Suppose the current date is 2020-05-29. Based on the above rules, tbl1 will produce the following partitions:
+    
+    ```
+    p202005: ["2020-05-03", "2020-06-03")
+    p202006: ["2020-06-03", "2020-07-03")
+    p202007: ["2020-07-03", "2020-08-03")
+    ```
+    
+    Because `dynamic_partition.start` is not set, the historical partition will not be deleted.
+
+    Assuming that today is 2020-05-20, and set 28th as the start of each month, the partition range is:
+
+    ```
+    p202004: ["2020-04-28", "2020-05-28")
+    p202005: ["2020-05-28", "2020-06-28")
+    p202006: ["2020-06-28", "2020-07-28")
+    ```
 
 ### Modify Dynamic Partition Properties
 
 You can modify the properties of the dynamic partition with the following command
 
 ```
-ALTER TABLE dynamic_partition set("key" = "value")
+ALTER TABLE tbl1 SET
+(
+    "dynamic_partition.prop1" = "value1",
+    ...
+);
 ```
 
+The modification of certain attributes may cause conflicts. Assume that the partition granularity was DAY and the following partitions have been created:
+
+```
+p20200519: ["2020-05-19", "2020-05-20")
+p20200520: ["2020-05-20", "2020-05-21")
+p20200521: ["2020-05-21", "2020-05-22")
+```
+
+If the partition granularity is changed to MONTH at this time, the system will try to create a partition with the range `["2020-05-01", "2020-06-01")`, and this range conflicts with the existing partition. So it cannot be created. And the partition with the range `["2020-06-01", "2020-07-01")` can be created normally. Therefore, the partition between 2020-05-22 and 2020-05-30 needs to be filled manually.
+
 ### Check Dynamic Partition Table Scheduling Status
 
 You can further view the scheduling of dynamic partitioned tables by using the following command:
 
 ```    
-SHOW DYNAMIC PARTITION TABLES;
-
-+-------------------+--------+----------+------+--------+---------+---------------------+---------------------+--------+------+
-| TableName         | Enable | TimeUnit | End  | Prefix | Buckets | LastUpdateTime      | LastSchedulerTime   | State  | Msg  |
-+-------------------+--------+----------+------+--------+---------+---------------------+---------------------+--------+------+
-| dynamic_partition | true   | DAY      | 3    | p      | 32      | 2020-01-08 20:19:09 | 2020-01-08 20:19:34 | NORMAL | N/A  |
-+-------------------+--------+----------+------+--------+---------+---------------------+---------------------+--------+------+
-1 row in set (0.00 sec)
-
+mysql> SHOW DYNAMIC PARTITION TABLES;
++-----------+--------+----------+-------------+------+--------+---------+-----------+----------------+---------------------+--------+------------------------+----------------------+
+| TableName | Enable | TimeUnit | Start       | End  | Prefix | Buckets | StartOf   | LastUpdateTime | LastSchedulerTime   | State  | LastCreatePartitionMsg | LastDropPartitionMsg |
++-----------+--------+----------+-------------+------+--------+---------+-----------+----------------+---------------------+--------+------------------------+----------------------+
+| d3        | true   | WEEK     | -3          | 3    | p      | 1       | MONDAY    | N/A            | 2020-05-25 14:29:24 | NORMAL | N/A                    | N/A                  |
+| d5        | true   | DAY      | -7          | 3    | p      | 32      | N/A       | N/A            | 2020-05-25 14:29:24 | NORMAL | N/A                    | N/A                  |
+| d4        | true   | WEEK     | -3          | 3    | p      | 1       | WEDNESDAY | N/A            | 2020-05-25 14:29:24 | NORMAL | N/A                    | N/A                  |
+| d6        | true   | MONTH    | -2147483648 | 2    | p      | 8       | 3rd       | N/A            | 2020-05-25 14:29:24 | NORMAL | N/A                    | N/A                  |
+| d2        | true   | DAY      | -3          | 3    | p      | 32      | N/A       | N/A            | 2020-05-25 14:29:24 | NORMAL | N/A                    | N/A                  |
+| d7        | true   | MONTH    | -2147483648 | 5    | p      | 8       | 24th      | N/A            | 2020-05-25 14:29:24 | NORMAL | N/A                    | N/A                  |
++-----------+--------+----------+-------------+------+--------+---------+-----------+----------------+---------------------+--------+------------------------+----------------------+
+7 rows in set (0.02 sec)
 ```
     
 * LastUpdateTime: The last time of modifying dynamic partition properties 
 * LastSchedulerTime:   The last time of performing dynamic partition scheduling
 * State:    The state of the last execution of dynamic partition scheduling
-* Msg:  Error message for the last time dynamic partition scheduling was performed 
+* LastCreatePartitionMsg: Error message of the last time to dynamically add partition scheduling
+* LastDropPartitionMsg: Error message of the last execution of dynamic deletion partition scheduling
 
 ## Advanced Operation
 
@@ -145,48 +283,27 @@ SHOW DYNAMIC PARTITION TABLES;
 
 * dynamic\_partition\_enable
 
-    Whether to enable Doris's dynamic partition feature. The default value is false, which is off. This parameter only affects the partitioning operation of dynamic partition tables, not normal tables.
+    Whether to enable Doris's dynamic partition feature. The default value is false, which is off. This parameter only affects the partitioning operation of dynamic partition tables, not normal tables. You can modify the parameters in `fe.conf` and restart FE to take effect. You can also execute the following commands at runtime to take effect:
     
-* dynamic\_partition\_check\_interval\_seconds
-
-    The execution frequency of dynamically partitioned threads, by default 3600(1 hour), which means scheduled every 1 hour.
+    MySQL protocal：
     
-### HTTP Restful API
-
-Doris provides an HTTP Restful API for modifying dynamic partition configuration parameters at run time.
-
-The API is implemented in FE, user can access it by `fe_host:fe_http_port`.The operation needs admin privilege.
-
-1. Set dynamic_partition_enable to true or false
+    `ADMIN SET FRONTEND CONFIG ("dynamic_partition_enable" = "true")`
     
-    * Set to true
+    HTTP protocal：
     
-        ```
-        GET /api/_set_config?dynamic_partition_enable=true
-        
-        For example: curl --location-trusted -u username:password -XGET http://fe_host:fe_http_port/api/_set_config?dynamic_partition_enable=true
-        
-        Return Code：200
-        ```
-        
-    * Set to false
+    `curl --location-trusted -u username:password -XGET http://fe_host:fe_http_port/api/_set_config?dynamic_partition_enable=true`
     
-        ```
-        GET /api/_set_config?dynamic_partition_enable=false
-        
-        For example: curl --location-trusted -u username:password -XGET http://fe_host:fe_http_port/api/_set_config?dynamic_partition_enable=false
-        
-        Return Code：200
-        ```
+    To turn off dynamic partitioning globally, set this parameter to false.
+    
+* dynamic\_partition\_check\_interval\_seconds
+
+    The execution frequency of dynamic partition threads defaults to 3600 (1 hour), that is, scheduling is performed every 1 hour. You can modify the parameters in `fe.conf` and restart FE to take effect. You can also modify the following commands at runtime:
+    
+    MySQL protocal：
+
+    `ADMIN SET FRONTEND CONFIG ("dynamic_partition_check_interval_seconds" = "7200")`
     
-2.  Set the scheduling frequency for dynamic partition 
+    HTTP protocal：
     
-    * Set schedule frequency to 12 hours.
-        
-        ```
-        GET /api/_set_config?dynamic_partition_check_interval_seconds=432000
-        
-        For example: curl --location-trusted -u username:password -XGET http://fe_host:fe_http_port/api/_set_config?dynamic_partition_check_interval_seconds=432000
-        
-        Return Code：200
-        ```
+    `curl --location-trusted -u username:password -XGET http://fe_host:fe_http_port/api/_set_config?dynamic_partition_check_interval_seconds=432000`
+    
\ No newline at end of file
diff --git a/docs/zh-CN/administrator-guide/dynamic-partition.md b/docs/zh-CN/administrator-guide/dynamic-partition.md
index 4705fbc109fdbd..3897742998f848 100644
--- a/docs/zh-CN/administrator-guide/dynamic-partition.md
+++ b/docs/zh-CN/administrator-guide/dynamic-partition.md
@@ -28,8 +28,6 @@ under the License.
 
 动态分区是在 Doris 0.12 版本中引入的新功能。旨在对表级别的分区实现生命周期管理(TTL)，减少用户的使用负担。
 
-最初的设计、实现和效果可以参阅 [ISSUE 2262](https://github.com/apache/incubator-doris/issues/2262)。
-
 目前实现了动态添加分区及动态删除分区的功能。
 
 ## 名词解释
@@ -41,117 +39,239 @@ under the License.
 
 在某些使用场景下，用户会将表按照天进行分区划分，每天定时执行例行任务，这时需要使用方手动管理分区，否则可能由于使用方没有创建分区导致数据导入失败，这给使用方带来了额外的维护成本。
 
-在实现方式上, FE会启动一个后台线程，根据fe.conf中`dynamic_partition_enable` 及 `dynamic_partition_check_interval_seconds`参数决定该线程是否启动以及该线程的调度频率。每次调度时，会在注册表中读取动态分区表的属性，并根据动态分区属性动态添加及删除分区。
-
-建表时，在properties中指定dynamic_partition属性，FE首先对动态分区属性进行解析，校验输入参数的合法性，然后将对应的属性持久化到FE的元数据中，并将该表注册到动态分区列表中，后台线程会根据配置参数定期对动态分区列表进行扫描，读取表的动态分区属性，执行添加分区及删除分区的任务，每次的调度信息会保留在FE的内存中（重启后则丢失），可以通过`SHOW DYNAMIC PARTITION TABLES`查看调度任务是否成功，如果存在分区创建或删除失败，会将失败信息输出。
+通过动态分区功能，用户可以在建表时设定动态分区的规则。FE 会启动一个后台线程，根据用户指定的规则创建或删除分区。用户也可以在运行时对现有规则进行变更。
 
 ## 使用方式
 
-### 动态分区属性参数说明:
+动态分区的规则可以在建表时指定，或者在运行时进行修改。当前仅支持对单分区列的分区表设定动态分区规则。
+
+* 建表时指定：
+
+    ```
+    CREATE TABLE tbl1
+    (...)
+    PROPERTIES
+    (
+        "dynamic_partition.prop1" = "value1",
+        "dynamic_partition.prop2" = "value2",
+        ...
+    )
+    ```
 
-`dynamic_partition.enable`: 是否开启动态分区特性，可指定为 `TRUE` 或 `FALSE`。如果不填写，默认为 `TRUE`。
+* 运行时修改
 
+    ```
+    ALTER TABLE tbl1 SET
+    (
+        "dynamic_partition.prop1" = "value1",
+        "dynamic_partition.prop2" = "value2",
+        ...
+    )
+    ```
 
-`dynamic_partition.time_unit`: 动态分区调度的单位，可指定为 `DAY` `WEEK` `MONTH`，当指定为 `DAY` 时，动态创建的分区名后缀格式为`yyyyMMdd`，例如`20200325`。当指定为 `WEEK` 时，动态创建的分区名后缀格式为`yyyy_ww`即当前日期属于这一年的第几周，例如 `2020-03-25` 创建的分区名后缀为 `2020_13`, 表明目前为2020年第13周。当指定为 `MONTH` 时，动态创建的分区名后缀格式为 `yyyyMM`，例如 `202003`。
+### 动态分区规则参数
 
-`dynamic_partition.start`: 动态分区的开始时间, 以当天为基准，超过该时间范围的分区将会被删除。如果不填写，则默认为`Integer.MIN_VALUE` 即 `-2147483648`。
+动态分区的规则参数都以 `dynamic_partition.` 为前缀：
 
+* `dynamic_partition.enable`
 
-`dynamic_partition.end`: 动态分区的结束时间, 以当天为基准，会提前创建N个单位的分区范围。
+    是否开启动态分区特性。可指定为 `TRUE` 或 `FALSE`。如果不填写，默认为 `TRUE`。如果为 `FALSE`，则 Doris 会忽略该表的动态分区规则。
 
-`dynamic_partition.prefix`: 动态创建的分区名前缀。
+* `dynamic_partition.time_unit`
 
-`dynamic_partition.buckets`: 动态创建的分区所对应的分桶数量。
+    动态分区调度的单位。可指定为 `DAY`、`WEEK`、`MONTH`。分别表示按天、按星期、按月进行分区创建或删除。
     
-### 建表
+    当指定为 `DAY` 时，动态创建的分区名后缀格式为 `yyyyMMdd`，例如`20200325`。
+    
+    当指定为 `WEEK` 时，动态创建的分区名后缀格式为`yyyy_ww`。即当前日期属于这一年的第几周，例如 `2020-03-25` 创建的分区名后缀为 `2020_13`, 表明目前为2020年第13周。
+    
+    当指定为 `MONTH` 时，动态创建的分区名后缀格式为 `yyyyMM`，例如 `202003`。
 
-建表时，可以在 `PROPERTIES` 中指定以下`dynamic_partition`属性，表示这个表是一个动态分区表。
+* `dynamic_partition.start`
 
-示例：
+    动态分区的起始偏移，为负数。根据 `time_unit` 属性的不同，以当天（星期/月）为基准，分区范围在此偏移之前的分区将会被删除。如果不填写，则默认为 `-2147483648`，即不删除历史分区。
+    
+* `dynamic_partition.end`
 
-```
-CREATE TABLE example_db.dynamic_partition
-(
-k1 DATE,
-k2 INT,
-k3 SMALLINT,
-v1 VARCHAR(2048),
-v2 DATETIME DEFAULT "2014-02-04 15:36:00"
-)
-ENGINE=olap
-DUPLICATE KEY(k1, k2, k3)
-PARTITION BY RANGE (k1)
-(
-PARTITION p20200321 VALUES LESS THAN ("2020-03-22"),
-PARTITION p20200322 VALUES LESS THAN ("2020-03-23"),
-PARTITION p20200323 VALUES LESS THAN ("2020-03-24"),
-PARTITION p20200324 VALUES LESS THAN ("2020-03-25")
-)
-DISTRIBUTED BY HASH(k2) BUCKETS 32
-PROPERTIES(
-"storage_medium" = "SSD",
-"dynamic_partition.enable" = "true",
-"dynamic_partition.time_unit" = "DAY",
-"dynamic_partition.start" = "-3",
-"dynamic_partition.end" = "3",
-"dynamic_partition.prefix" = "p",
-"dynamic_partition.buckets" = "32"
- );
-```
-创建一张动态分区表，指定开启动态分区特性，以当天为2020-03-25为例，在每次调度时，会删除分区上界小于 `2020-03-22` 的分区，为了避免删除非动态创建的分区，动态删除分区只会删除分区名符合动态创建分区规则的分区，例如分区名为a1, 则即使分区范围在待删除的分区范围内，也不会被删除。同时在调度时会提前创建今天以及以后3天（总共4天）的分区(若分区已存在则会忽略)，分区名根据指定前缀分别为`p20200325` `p20200326` `p20200327` `p20200328`,每个分区的分桶数量为32。同时会删除 `p20200321` 的分区，最终的分区范围如下:
-```
-[types: [DATE]; keys: [2020-03-22]; ‥types: [DATE]; keys: [2020-03-23]; )
-[types: [DATE]; keys: [2020-03-23]; ‥types: [DATE]; keys: [2020-03-24]; )
-[types: [DATE]; keys: [2020-03-24]; ‥types: [DATE]; keys: [2020-03-25]; )
-[types: [DATE]; keys: [2020-03-25]; ‥types: [DATE]; keys: [2020-03-26]; )
-[types: [DATE]; keys: [2020-03-26]; ‥types: [DATE]; keys: [2020-03-27]; )
-[types: [DATE]; keys: [2020-03-27]; ‥types: [DATE]; keys: [2020-03-28]; )
-[types: [DATE]; keys: [2020-03-28]; ‥types: [DATE]; keys: [2020-03-29]; )
-```
+    动态分区的结束偏移，为正数。根据 `time_unit` 属性的不同，以当天（星期/月）为基准，提前创建对应范围的分区。
+
+* `dynamic_partition.prefix`
+
+    动态创建的分区名前缀。
+
+* `dynamic_partition.buckets`
+
+    动态创建的分区所对应的分桶数量。
     
-### 开启动态分区功能
-1. 首先需要在fe.conf中设置`dynamic_partition_enable=true`，可以在集群启动时通过修改配置文件指定，或者通过MySQL连接后使用命令行 `ADMIN SET FRONTEND CONFIG ("dynamic_partition_enable" = "true")`修改，也可以在运行时通过http接口动态修改,修改方法查看高级操作部分
+* `dynamic_partition.start_day_of_week`
 
-2. 如果需要对0.12版本之前的表添加动态分区属性，则需要通过以下命令修改表的属性
-```
-ALTER TABLE dynamic_partition set ("dynamic_partition.enable" = "true", "dynamic_partition.time_unit" = "DAY", "dynamic_partition.end" = "3", "dynamic_partition.prefix" = "p", "dynamic_partition.buckets" = "32");
-```
+    当 `time_unit` 为 `WEEK` 时，该参数用于指定每周的起始点。取值为 1 到 7。其中 1 表示周一，7 表示周日。默认为 1，即表示每周以周一为起始点。
+    
+* `dynamic_partition.start_day_of_month`
+
+    当 `time_unit` 为 `MONTH` 时，该参数用于指定每月的起始日期。取值为 1 到 28。其中 1 表示每月1号，28 表示每月28号。默认为 1，即表示每月以1号位起始点。暂不支持以29、30、31号为起始日，以避免因闰年或闰月带来的歧义。
+  
+## 示例
+
+1. 表 tbl1 分区列 k1 类型为 DATE，创建一个动态分区规则。按天分区，只保留最近7天的分区，并且预先创建未来3天的分区。
+
+    ```
+    CREATA TABLE tbl1
+    (
+        k1 DATE,
+        ...
+    )
+    PARTITION BY RANGE(K1) ()
+    PROPERTIES
+    (
+        "dynamic_partition.enable" = "true",
+        "dynamic_partition.time_unit" = "DAY",
+        "dynamic_partition.start" = "-7",
+        "dynamic_partition.end" = "3",
+        "dynamic_partition.prefix" = "p",
+        "dynamic_partition.buckets" = "32"
+    );
+    ```
+    
+    假设当前日期为 2020-05-29。则根据以上规则，tbl1 会产生以下分区：
+    
+    ```
+    p20200529: ["2020-05-29", "2020-05-30")
+    p20200530: ["2020-05-30", "2020-05-31")
+    p20200531: ["2020-05-31", "2020-06-01")
+    p20200601: ["2020-06-01", "2020-06-02")
+    ```
+    
+    在第二天，即 2020-05-30，会创建新的分区 `p20200602: ["2020-06-02", "2020-06-03")`
+    
+    在 2020-06-06 时，因为 `dynamic_partition.start` 设置为 7，则将删除7天前的分区，即删除分区 `p20200529`。
+    
+2. 表 tbl1 分区列 k1 类型为 DATETIME，创建一个动态分区规则。按星期分区，只保留最近2个星期的分区，并且预先创建未来2个星期的分区。
+
+    ```
+    CREATA TABLE tbl1
+    (
+        k1 DATETIME,
+        ...
+    )
+    PARTITION BY RANGE(K1) ()
+    PROPERTIES
+    (
+        "dynamic_partition.enable" = "true",
+        "dynamic_partition.time_unit" = "WEEK",
+        "dynamic_partition.start" = "-2",
+        "dynamic_partition.end" = "2",
+        "dynamic_partition.prefix" = "p",
+        "dynamic_partition.buckets" = "8"
+    );
+    ```
+
+    假设当前日期为 2020-05-29，是 2020 年的第 22 周。默认每周起始为星期一。则根于以上规则，tbl1 会产生以下分区：
+    
+    ```
+    p2020_22: ["2020-05-25 00:00:00", "2020-06-01 00:00:00")
+    p2020_23: ["2020-06-01 00:00:00", "2020-06-08 00:00:00")
+    p2020_24: ["2020-06-08 00:00:00", "2020-06-15 00:00:00")
+    ```
+    
+    其中每个分区的起始日期为当周的周一。同时，因为分区列 k1 的类型为 DATETIME，则分区值会补全时分秒部分，且皆为 0。
+    
+    在 2020-06-15，即第25周时，会删除2周前的分区，即删除 `p2020_22`。
+    
+    在上面的例子中，假设用户指定了周起始日为 `"dynamic_partition.start_day_of_week" = "3"`，即以每周三为起始日。则分区如下：
+    
+    ```
+    p2020_22: ["2020-05-27 00:00:00", "2020-06-03 00:00:00")
+    p2020_23: ["2020-06-03 00:00:00", "2020-06-10 00:00:00")
+    p2020_24: ["2020-06-10 00:00:00", "2020-06-17 00:00:00")
+    ```
+    
+    即分区范围为当周的周三到下周的周二。
+    
+    * 注：2019-12-31 和 2020-01-01 在同一周内，如果分区的起始日期为 2019-12-31，则分区名为 `p2019_53`，如果分区的起始日期为 2020-01-01，则分区名为 `p2020_01`。
+    
+3. 表 tbl1 分区列 k1 类型为 DATE，创建一个动态分区规则。按月分区，不删除历史分区，并且预先创建未来2个月的分区。同时设定以每月3号为起始日。
+
+    ```
+    CREATA TABLE tbl1
+    (
+        k1 DATE,
+        ...
+    )
+    PARTITION BY RANGE(K1) ()
+    PROPERTIES
+    (
+        "dynamic_partition.enable" = "true",
+        "dynamic_partition.time_unit" = "MONTH",
+        "dynamic_partition.end" = "2",
+        "dynamic_partition.prefix" = "p",
+        "dynamic_partition.buckets" = "8",
+        "dynamic_partition.start_day_of_month" = "3"
+    );
+    ```
+    
+    假设当前日期为 2020-05-29。则根于以上规则，tbl1 会产生以下分区：
+    
+    ```
+    p202005: ["2020-05-03", "2020-06-03")
+    p202006: ["2020-06-03", "2020-07-03")
+    p202007: ["2020-07-03", "2020-08-03")
+    ```
 
-### 停止动态分区功能
+    因为没有设置 `dynamic_partition.start`，则不会删除历史分区。
+    
+    假设今天为 2020-05-20，并设置以每月28号为起始日，则分区范围为：
+    
+    ```
+    p202004: ["2020-04-28", "2020-05-28")
+    p202005: ["2020-05-28", "2020-06-28")
+    p202006: ["2020-06-28", "2020-07-28")
+    ```
+    
+## 修改动态分区属性
 
-如果需要对集群中所有动态分区表停止动态分区功能，则需要在fe.conf中设置`dynamic_partition_enable=false`
+通过如下命令可以修改动态分区的属性：
 
-如果需要对指定表停止动态分区功能，则可以通过以下命令修改表的属性
 ```
-ALTER TABLE dynamic_partition SET ("dynamic_partition.enable" = "false")
+ALTER TABLE tbl1 SET
+(
+    "dynamic_partition.prop1" = "value1",
+    ...
+);
 ```
 
-### 修改动态分区属性
+某些属性的修改可能会可能会产生冲突。假设之前分区粒度为 DAY，并且已经创建了如下分区：
 
-通过如下命令可以修改动态分区的属性
 ```
-ALTER TABLE dynamic_partition SET ("key" = "value")
+p20200519: ["2020-05-19", "2020-05-20")
+p20200520: ["2020-05-20", "2020-05-21")
+p20200521: ["2020-05-21", "2020-05-22")
 ```
 
+如果此时将分区粒度改为 MONTH，则系统会尝试创建范围为 `["2020-05-01", "2020-06-01")` 的分区，而该分区的分区范围和已有分区冲突，所以无法创建。而范围为 `["2020-06-01", "2020-07-01")` 的分区可以正常创建。因此，2020-05-22 到 2020-05-30 时间段的分区，需要自行填补。
+
 ### 查看动态分区表调度情况
 
-通过以下命令可以进一步查看动态分区表的调度情况：
+通过以下命令可以进一步查看当前数据库下，所有动态分区表的调度情况：
 
 ```    
-SHOW DYNAMIC PARTITION TABLES;
-
-+-------------------+--------+----------+-------+------+--------+---------+---------------------+---------------------+--------+------------------------+----------------------+
-| TableName         | Enable | TimeUnit | Start | End  | Prefix | Buckets | LastUpdateTime      | LastSchedulerTime   | State  | LastCreatePartitionMsg | LastDropPartitionMsg |
-+-------------------+--------+----------+-------+------+--------+---------+---------------------+---------------------+--------+------------------------+----------------------+
-| dynamic_partition | true   | DAY      | -3    | 3    | p      | 32      | 2020-03-12 17:25:47 | 2020-03-12 17:25:52 | NORMAL | N/A                    | N/A                  |
-+-------------------+--------+----------+-------+------+--------+---------+---------------------+---------------------+--------+------------------------+----------------------+
-1 row in set (0.00 sec)
-
+mysql> SHOW DYNAMIC PARTITION TABLES;
++-----------+--------+----------+-------------+------+--------+---------+-----------+----------------+---------------------+--------+------------------------+----------------------+
+| TableName | Enable | TimeUnit | Start       | End  | Prefix | Buckets | StartOf   | LastUpdateTime | LastSchedulerTime   | State  | LastCreatePartitionMsg | LastDropPartitionMsg |
++-----------+--------+----------+-------------+------+--------+---------+-----------+----------------+---------------------+--------+------------------------+----------------------+
+| d3        | true   | WEEK     | -3          | 3    | p      | 1       | MONDAY    | N/A            | 2020-05-25 14:29:24 | NORMAL | N/A                    | N/A                  |
+| d5        | true   | DAY      | -7          | 3    | p      | 32      | N/A       | N/A            | 2020-05-25 14:29:24 | NORMAL | N/A                    | N/A                  |
+| d4        | true   | WEEK     | -3          | 3    | p      | 1       | WEDNESDAY | N/A            | 2020-05-25 14:29:24 | NORMAL | N/A                    | N/A                  |
+| d6        | true   | MONTH    | -2147483648 | 2    | p      | 8       | 3rd       | N/A            | 2020-05-25 14:29:24 | NORMAL | N/A                    | N/A                  |
+| d2        | true   | DAY      | -3          | 3    | p      | 32      | N/A       | N/A            | 2020-05-25 14:29:24 | NORMAL | N/A                    | N/A                  |
+| d7        | true   | MONTH    | -2147483648 | 5    | p      | 8       | 24th      | N/A            | 2020-05-25 14:29:24 | NORMAL | N/A                    | N/A                  |
++-----------+--------+----------+-------------+------+--------+---------+-----------+----------------+---------------------+--------+------------------------+----------------------+
+7 rows in set (0.02 sec)
 ```
     
 * LastUpdateTime: 最后一次修改动态分区属性的时间 
 * LastSchedulerTime:   最后一次执行动态分区调度的时间
-* State:    最后一次执行动态分区调度的状态
+* State: 最后一次执行动态分区调度的状态
 * LastCreatePartitionMsg:  最后一次执行动态添加分区调度的错误信息 
 * LastDropPartitionMsg:  最后一次执行动态删除分区调度的错误信息 
 
@@ -161,48 +281,26 @@ SHOW DYNAMIC PARTITION TABLES;
 
 * dynamic\_partition\_enable
 
-    是否开启 Doris 的动态分区功能。默认为 false，即关闭。该参数只影响动态分区表的分区操作，不影响普通表。
+    是否开启 Doris 的动态分区功能。默认为 false，即关闭。该参数只影响动态分区表的分区操作，不影响普通表。可以通过修改 fe.conf 中的参数并重启 FE 生效。也可以在运行时执行以下命令生效：
     
-* dynamic\_partition\_check\_interval\_seconds
-
-    动态分区线程的执行频率，默认为3600(1个小时)，即每1个小时进行一次调度
+    MySQL 协议：
     
-### HTTP Restful API
-
-Doris 提供了修改动态分区配置参数的 HTTP Restful API，用于运行时修改动态分区配置参数。
-
-该 API 实现在 FE 端，使用 `fe_host:fe_http_port` 进行访问。需要 ADMIN 权限。
-
-1. 将 dynamic_partition_enable 设置为 true 或 false
+    `ADMIN SET FRONTEND CONFIG ("dynamic_partition_enable" = "true")`
+    
+    HTTP 协议：
+    
+    `curl --location-trusted -u username:password -XGET http://fe_host:fe_http_port/api/_set_config?dynamic_partition_enable=true`
     
-    * 标记为 true
+    若要全局关闭动态分区，则设置此参数为 false 即可。
     
-        ```
-        GET /api/_set_config?dynamic_partition_enable=true
-        
-        例如: curl --location-trusted -u username:password -XGET http://fe_host:fe_http_port/api/_set_config?dynamic_partition_enable=true
-        
-        返回：200
-        ```
-        
-    * 标记为 false
+* dynamic\_partition\_check\_interval\_seconds
+
+    动态分区线程的执行频率，默认为3600(1个小时)，即每1个小时进行一次调度。可以通过修改 fe.conf 中的参数并重启 FE 生效。也可以在运行时执行以下命令修改：
     
-        ```
-        GET /api/_set_config?dynamic_partition_enable=false
-        
-        例如: curl --location-trusted -u username:password -XGET http://fe_host:fe_http_port/api/_set_config?dynamic_partition_enable=false
-        
-        返回：200
-        ```
+    MySQL 协议：
+
+    `ADMIN SET FRONTEND CONFIG ("dynamic_partition_check_interval_seconds" = "7200")`
     
-2. 设置 dynamic partition 的调度频率
+    HTTP 协议：
     
-    * 设置调度时间为12小时调度一次
-        
-        ```
-        GET /api/_set_config?dynamic_partition_check_interval_seconds=432000
-        
-        例如: curl --location-trusted -u username:password -XGET http://fe_host:fe_http_port/api/_set_config?dynamic_partition_check_interval_seconds=432000
-        
-        返回：200
-        ```
+    `curl --location-trusted -u username:password -XGET http://fe_host:fe_http_port/api/_set_config?dynamic_partition_check_interval_seconds=432000`
\ No newline at end of file
diff --git a/fe/src/main/java/org/apache/doris/alter/MaterializedViewHandler.java b/fe/src/main/java/org/apache/doris/alter/MaterializedViewHandler.java
index e4827f1f653b31..921350e6e21a0e 100644
--- a/fe/src/main/java/org/apache/doris/alter/MaterializedViewHandler.java
+++ b/fe/src/main/java/org/apache/doris/alter/MaterializedViewHandler.java
@@ -117,7 +117,8 @@ protected void batchAddAlterJobV2(List<AlterJobV2> alterJobV2List) {
     // return true iff job is actually added this time
     private boolean addAlterJobV2ToTableNotFinalStateJobMap(AlterJobV2 alterJobV2) {
         if (alterJobV2.isDone()) {
-            LOG.warn("try to add a final job({}) to a unfinal set", alterJobV2.getJobId());
+            LOG.warn("try to add a final job({}) to a unfinal set. db: {}, tbl: {}",
+                    alterJobV2.getJobId(), alterJobV2.getDbId(), alterJobV2.getTableId());
             return false;
         }
 
diff --git a/fe/src/main/java/org/apache/doris/analysis/ShowDynamicPartitionStmt.java b/fe/src/main/java/org/apache/doris/analysis/ShowDynamicPartitionStmt.java
index d98b60bd733c38..4e98befc3ad093 100644
--- a/fe/src/main/java/org/apache/doris/analysis/ShowDynamicPartitionStmt.java
+++ b/fe/src/main/java/org/apache/doris/analysis/ShowDynamicPartitionStmt.java
@@ -17,7 +17,6 @@
 
 package org.apache.doris.analysis;
 
-import com.google.common.base.Strings;
 import org.apache.doris.catalog.Column;
 import org.apache.doris.catalog.ScalarType;
 import org.apache.doris.cluster.ClusterNamespace;
@@ -26,6 +25,8 @@
 import org.apache.doris.common.ErrorReport;
 import org.apache.doris.qe.ShowResultSetMetaData;
 
+import com.google.common.base.Strings;
+
 public class ShowDynamicPartitionStmt extends ShowStmt {
     private String db;
     private static final ShowResultSetMetaData SHOW_DYNAMIC_PARTITION_META_DATA =
@@ -37,6 +38,7 @@ public class ShowDynamicPartitionStmt extends ShowStmt {
                     .addColumn(new Column("End", ScalarType.createVarchar(20)))
                     .addColumn(new Column("Prefix", ScalarType.createVarchar(20)))
                     .addColumn(new Column("Buckets", ScalarType.createVarchar(20)))
+                    .addColumn(new Column("StartOf", ScalarType.createVarchar(20)))
                     .addColumn(new Column("LastUpdateTime", ScalarType.createVarchar(20)))
                     .addColumn(new Column("LastSchedulerTime", ScalarType.createVarchar(20)))
                     .addColumn(new Column("State", ScalarType.createVarchar(20)))
diff --git a/fe/src/main/java/org/apache/doris/catalog/DynamicPartitionProperty.java b/fe/src/main/java/org/apache/doris/catalog/DynamicPartitionProperty.java
index 567ca7d806e410..4bdd2fa5702dc2 100644
--- a/fe/src/main/java/org/apache/doris/catalog/DynamicPartitionProperty.java
+++ b/fe/src/main/java/org/apache/doris/catalog/DynamicPartitionProperty.java
@@ -17,7 +17,12 @@
 
 package org.apache.doris.catalog;
 
+import org.apache.doris.analysis.TimestampArithmeticExpr.TimeUnit;
+import org.apache.doris.common.util.DynamicPartitionUtil.StartOfDate;
+import org.apache.doris.common.util.TimeUtils;
+
 import java.util.Map;
+import java.util.TimeZone;
 
 public class DynamicPartitionProperty{
     public static final String TIME_UNIT = "dynamic_partition.time_unit";
@@ -26,6 +31,10 @@ public class DynamicPartitionProperty{
     public static final String PREFIX = "dynamic_partition.prefix";
     public static final String BUCKETS = "dynamic_partition.buckets";
     public static final String ENABLE = "dynamic_partition.enable";
+    public static final String START_DAY_OF_WEEK = "dynamic_partition.start_day_of_week";
+    public static final String START_DAY_OF_MONTH = "dynamic_partition.start_day_of_month";
+
+    public static final int MIN_START_OFFSET = Integer.MIN_VALUE;
 
     private boolean exist;
 
@@ -35,22 +44,44 @@ public class DynamicPartitionProperty{
     private int end;
     private String prefix;
     private int buckets;
+    private StartOfDate startOfWeek;
+    private StartOfDate startOfMonth;
+    // TODO: support setting timezone.
+    private TimeZone tz = TimeUtils.getDefaultTimeZone();
+
 
-    DynamicPartitionProperty(Map<String ,String> properties) {
+    public DynamicPartitionProperty(Map<String, String> properties) {
         if (properties != null && !properties.isEmpty()) {
             this.exist = true;
             this.enable = Boolean.parseBoolean(properties.get(ENABLE));
             this.timeUnit = properties.get(TIME_UNIT);
             // In order to compatible dynamic add partition version
-            this.start = Integer.parseInt(properties.getOrDefault(START, String.valueOf(Integer.MIN_VALUE)));
+            this.start = Integer.parseInt(properties.getOrDefault(START, String.valueOf(MIN_START_OFFSET)));
             this.end = Integer.parseInt(properties.get(END));
             this.prefix = properties.get(PREFIX);
             this.buckets = Integer.parseInt(properties.get(BUCKETS));
+            createStartOfs(properties);
         } else {
             this.exist = false;
         }
     }
 
+    private void createStartOfs(Map<String, String> properties) {
+        if (properties.containsKey(START_DAY_OF_WEEK)) {
+            startOfWeek = new StartOfDate(-1, -1, Integer.valueOf(properties.get(START_DAY_OF_WEEK)));
+        } else {
+            // default:
+            startOfWeek = new StartOfDate(-1, -1, 1 /* start from MONDAY */);
+        }
+
+        if (properties.containsKey(START_DAY_OF_MONTH)) {
+            startOfMonth = new StartOfDate(-1, Integer.valueOf(properties.get(START_DAY_OF_MONTH)), -1);
+        } else {
+            // default:
+            startOfMonth = new StartOfDate(-1, 1 /* 1st of month */, -1);
+        }
+    }
+
     public boolean isExist() {
         return exist;
     }
@@ -79,13 +110,42 @@ public boolean getEnable() {
         return enable;
     }
 
+    public StartOfDate getStartOfWeek() {
+        return startOfWeek;
+    }
+
+    public StartOfDate getStartOfMonth() {
+        return startOfMonth;
+    }
+
+    public String getStartOfInfo() {
+        if (getTimeUnit().equalsIgnoreCase(TimeUnit.WEEK.toString())) {
+            return startOfWeek.toDisplayInfo();
+        } else if (getTimeUnit().equalsIgnoreCase(TimeUnit.MONTH.toString())) {
+            return startOfMonth.toDisplayInfo();
+        } else {
+            return "N/A";
+        }
+    }
+
+    public TimeZone getTimeZone() {
+        return tz;
+    }
+
+
     @Override
     public String toString() {
-        return ",\n\"" + ENABLE + "\" = \"" + enable + "\"" +
+        String res = ",\n\"" + ENABLE + "\" = \"" + enable + "\"" +
                 ",\n\"" + TIME_UNIT + "\" = \"" + timeUnit + "\"" +
                 ",\n\"" + START + "\" = \"" + start + "\"" +
                 ",\n\"" + END + "\" = \"" + end + "\"" +
                 ",\n\"" + PREFIX + "\" = \"" + prefix + "\"" +
                 ",\n\"" + BUCKETS + "\" = \"" + buckets + "\"";
+        if (getTimeUnit().equalsIgnoreCase(TimeUnit.WEEK.toString())) {
+            res += ",\n\"" + START_DAY_OF_WEEK + "\" = \"" + startOfWeek.dayOfWeek + "\"";
+        } else if (getTimeUnit().equalsIgnoreCase(TimeUnit.MONTH.toString())) {
+            res += ",\n\"" + START_DAY_OF_MONTH + "\" = \"" + startOfMonth.day + "\"";
+        }
+        return res;
     }
 }
diff --git a/fe/src/main/java/org/apache/doris/clone/DynamicPartitionScheduler.java b/fe/src/main/java/org/apache/doris/clone/DynamicPartitionScheduler.java
index 5290ff90d81063..a311a68231e394 100644
--- a/fe/src/main/java/org/apache/doris/clone/DynamicPartitionScheduler.java
+++ b/fe/src/main/java/org/apache/doris/clone/DynamicPartitionScheduler.java
@@ -30,7 +30,6 @@
 import org.apache.doris.catalog.DynamicPartitionProperty;
 import org.apache.doris.catalog.HashDistributionInfo;
 import org.apache.doris.catalog.OlapTable;
-import org.apache.doris.catalog.PartitionInfo;
 import org.apache.doris.catalog.PartitionKey;
 import org.apache.doris.catalog.RangePartitionInfo;
 import org.apache.doris.catalog.Table;
@@ -127,20 +126,20 @@ private Map<String, String> createDefaultRuntimeInfo() {
         return defaultRuntimeInfo;
     }
 
-    private ArrayList<AddPartitionClause> getAddPartitionClause(OlapTable olapTable, Column partitionColumn, String partitionFormat) {
+    private ArrayList<AddPartitionClause> getAddPartitionClause(Database db, OlapTable olapTable,
+            Column partitionColumn, String partitionFormat) {
         ArrayList<AddPartitionClause> addPartitionClauses = new ArrayList<>();
-        Calendar calendar = Calendar.getInstance();
         DynamicPartitionProperty dynamicPartitionProperty = olapTable.getTableProperty().getDynamicPartitionProperty();
+        RangePartitionInfo rangePartitionInfo = (RangePartitionInfo) olapTable.getPartitionInfo();
+        Calendar currentDate = Calendar.getInstance(dynamicPartitionProperty.getTimeZone());
         for (int i = 0; i <= dynamicPartitionProperty.getEnd(); i++) {
-            String prevBorder = DynamicPartitionUtil.getPartitionRange(dynamicPartitionProperty.getTimeUnit(),
-                    i, (Calendar) calendar.clone(), partitionFormat);
-            // continue if partition already exists
-            String nextBorder = DynamicPartitionUtil.getPartitionRange(dynamicPartitionProperty.getTimeUnit(),
-                    i + 1, (Calendar) calendar.clone(), partitionFormat);
+            String prevBorder = DynamicPartitionUtil.getPartitionRangeString(dynamicPartitionProperty,
+                    (Calendar) currentDate.clone(), i, partitionFormat);
+            String nextBorder = DynamicPartitionUtil.getPartitionRangeString(dynamicPartitionProperty,
+                    (Calendar) currentDate.clone(), i + 1, partitionFormat);
             PartitionValue lowerValue = new PartitionValue(prevBorder);
             PartitionValue upperValue = new PartitionValue(nextBorder);
-            PartitionInfo partitionInfo = olapTable.getPartitionInfo();
-            RangePartitionInfo info = (RangePartitionInfo) (partitionInfo);
+
             boolean isPartitionExists = false;
             Range<PartitionKey> addPartitionKeyRange;
             try {
@@ -149,10 +148,11 @@ private ArrayList<AddPartitionClause> getAddPartitionClause(OlapTable olapTable,
                 addPartitionKeyRange = Range.closedOpen(lowerBound, upperBound);
             } catch (AnalysisException e) {
                 // keys.size is always equal to column.size, cannot reach this exception
-                LOG.warn("Keys size is not equal to column size. Error={}", e.getMessage());
+                LOG.warn("Keys size is not equal to column size. Error={}, db: {}, table: {}", e.getMessage(),
+                        db.getFullName(), olapTable.getName());
                 continue;
             }
-            for (Range<PartitionKey> partitionKeyRange : info.getIdToRange(false).values()) {
+            for (Range<PartitionKey> partitionKeyRange : rangePartitionInfo.getIdToRange(false).values()) {
                 // only support single column partition now
                 try {
                     RangeUtils.checkRangeIntersect(partitionKeyRange, addPartitionKeyRange);
@@ -161,7 +161,7 @@ private ArrayList<AddPartitionClause> getAddPartitionClause(OlapTable olapTable,
                     if (addPartitionKeyRange.equals(partitionKeyRange)) {
                         clearCreatePartitionFailedMsg(olapTable.getName());
                     } else {
-                        recordCreatePartitionFailedMsg(olapTable.getName(), e.getMessage());
+                        recordCreatePartitionFailedMsg(db.getFullName(), olapTable.getName(), e.getMessage());
                     }
                     break;
                 }
@@ -174,7 +174,8 @@ private ArrayList<AddPartitionClause> getAddPartitionClause(OlapTable olapTable,
             PartitionKeyDesc partitionKeyDesc = new PartitionKeyDesc(Collections.singletonList(lowerValue), Collections.singletonList(upperValue));
             HashMap<String, String> partitionProperties = new HashMap<>(1);
             partitionProperties.put("replication_num", String.valueOf(DynamicPartitionUtil.estimateReplicateNum(olapTable)));
-            String partitionName = dynamicPartitionProperty.getPrefix() + DynamicPartitionUtil.getFormattedPartitionName(prevBorder, dynamicPartitionProperty.getTimeUnit());
+            String partitionName = dynamicPartitionProperty.getPrefix() + DynamicPartitionUtil.getFormattedPartitionName(
+                    dynamicPartitionProperty.getTimeZone(), prevBorder, dynamicPartitionProperty.getTimeUnit());
             SingleRangePartitionDesc rangePartitionDesc = new SingleRangePartitionDesc(true, partitionName,
                     partitionKeyDesc, partitionProperties);
 
@@ -192,15 +193,23 @@ private ArrayList<AddPartitionClause> getAddPartitionClause(OlapTable olapTable,
         return addPartitionClauses;
     }
 
-    private ArrayList<DropPartitionClause> getDropPartitionClause(OlapTable olapTable, Column partitionColumn, String partitionFormat) {
+    /*
+     * 1. get the range of [start, 0) as a reserved range.
+     * 2. get DropPartitionClause of partitions which range are before this reserved range.
+     */
+    private ArrayList<DropPartitionClause> getDropPartitionClause(Database db, OlapTable olapTable, Column partitionColumn, String partitionFormat) {
         ArrayList<DropPartitionClause> dropPartitionClauses = new ArrayList<>();
-        Calendar calendar = Calendar.getInstance();
         DynamicPartitionProperty dynamicPartitionProperty = olapTable.getTableProperty().getDynamicPartitionProperty();
+        if (dynamicPartitionProperty.getStart() == DynamicPartitionProperty.MIN_START_OFFSET) {
+            // not set start offset, so not drop any partition
+            return dropPartitionClauses;
+        }
 
-        String lowerBorder = DynamicPartitionUtil.getPartitionRange(dynamicPartitionProperty.getTimeUnit(),
-                dynamicPartitionProperty.getStart(), (Calendar) calendar.clone(), partitionFormat);
-        String upperBorder = DynamicPartitionUtil.getPartitionRange(dynamicPartitionProperty.getTimeUnit(),
-                0, (Calendar) calendar.clone(), partitionFormat);
+        Calendar currentDate = Calendar.getInstance(dynamicPartitionProperty.getTimeZone());
+        String lowerBorder = DynamicPartitionUtil.getPartitionRangeString(dynamicPartitionProperty,
+                (Calendar) currentDate.clone(), dynamicPartitionProperty.getStart(), partitionFormat);
+        String upperBorder = DynamicPartitionUtil.getPartitionRangeString(dynamicPartitionProperty,
+                (Calendar) currentDate.clone(), 0, partitionFormat);
         PartitionValue lowerPartitionValue = new PartitionValue(lowerBorder);
         PartitionValue upperPartitionValue = new PartitionValue(upperBorder);
         Range<PartitionKey> reservePartitionKeyRange;
@@ -210,7 +219,8 @@ private ArrayList<DropPartitionClause> getDropPartitionClause(OlapTable olapTabl
             reservePartitionKeyRange = Range.closedOpen(lowerBound, upperBound);
         } catch (AnalysisException e) {
             // keys.size is always equal to column.size, cannot reach this exception
-            LOG.warn("Keys size is not equal to column size. Error={}", e.getMessage());
+            LOG.warn("Keys size is not equal to column size. Error={}, db: {}, table: {}", e.getMessage(),
+                    db.getFullName(), olapTable.getName());
             return dropPartitionClauses;
         }
         RangePartitionInfo info = (RangePartitionInfo) (olapTable.getPartitionInfo());
@@ -249,8 +259,8 @@ private void executeDynamicPartition() {
             ArrayList<DropPartitionClause> dropPartitionClauses;
             String tableName;
             boolean skipAddPartition = false;
-            db.readLock();
             OlapTable olapTable;
+            db.readLock();
             try {
                 olapTable = (OlapTable) db.getTable(tableId);
                 // Only OlapTable has DynamicPartitionProperty
@@ -264,8 +274,7 @@ private void executeDynamicPartition() {
                 if (olapTable.getState() != OlapTable.OlapTableState.NORMAL) {
                     String errorMsg = "Table[" + olapTable.getName() + "]'s state is not NORMAL."
                             + "Do not allow doing dynamic add partition. table state=" + olapTable.getState();
-                    recordCreatePartitionFailedMsg(olapTable.getName(), errorMsg);
-                    LOG.info(errorMsg);
+                    recordCreatePartitionFailedMsg(db.getFullName(), olapTable.getName(), errorMsg);
                     skipAddPartition = true;
                 }
 
@@ -275,19 +284,25 @@ private void executeDynamicPartition() {
                 // scheduler time should be record even no partition added
                 createOrUpdateRuntimeInfo(olapTable.getName(), LAST_SCHEDULER_TIME, TimeUtils.getCurrentFormatTime());
                 RangePartitionInfo rangePartitionInfo = (RangePartitionInfo) olapTable.getPartitionInfo();
+                if (rangePartitionInfo.getPartitionColumns().size() != 1) {
+                    // currently only support partition with single column.
+                    iterator.remove();
+                    continue;
+                }
+
                 Column partitionColumn = rangePartitionInfo.getPartitionColumns().get(0);
                 String partitionFormat;
                 try {
                     partitionFormat = DynamicPartitionUtil.getPartitionFormat(partitionColumn);
                 } catch (DdlException e) {
-                    recordCreatePartitionFailedMsg(olapTable.getName(), e.getMessage());
+                    recordCreatePartitionFailedMsg(db.getFullName(), olapTable.getName(), e.getMessage());
                     continue;
                 }
 
                 if (!skipAddPartition) {
-                    addPartitionClauses = getAddPartitionClause(olapTable, partitionColumn, partitionFormat);
+                    addPartitionClauses = getAddPartitionClause(db, olapTable, partitionColumn, partitionFormat);
                 }
-                dropPartitionClauses = getDropPartitionClause(olapTable, partitionColumn, partitionFormat);
+                dropPartitionClauses = getDropPartitionClause(db, olapTable, partitionColumn, partitionFormat);
                 tableName = olapTable.getName();
             } finally {
                 db.readUnlock();
@@ -299,7 +314,7 @@ private void executeDynamicPartition() {
                     Catalog.getCurrentCatalog().dropPartition(db, olapTable, dropPartitionClause);
                     clearDropPartitionFailedMsg(tableName);
                 } catch (DdlException e) {
-                    recordDropPartitionFailedMsg(tableName, e.getMessage());
+                    recordDropPartitionFailedMsg(db.getFullName(), tableName, e.getMessage());
                 } finally {
                     db.writeUnlock();
                 }
@@ -311,15 +326,15 @@ private void executeDynamicPartition() {
                         Catalog.getCurrentCatalog().addPartition(db, tableName, addPartitionClause);
                         clearCreatePartitionFailedMsg(tableName);
                     } catch (DdlException e) {
-                        recordCreatePartitionFailedMsg(tableName, e.getMessage());
+                        recordCreatePartitionFailedMsg(db.getFullName(), tableName, e.getMessage());
                     }
                 }
             }
         }
     }
 
-    private void recordCreatePartitionFailedMsg(String tableName, String msg) {
-        LOG.warn("dynamic add partition failed: " + msg);
+    private void recordCreatePartitionFailedMsg(String dbName, String tableName, String msg) {
+        LOG.warn("dynamic add partition failed: {}, db: {}, table: {}", msg, dbName, tableName);
         createOrUpdateRuntimeInfo(tableName, DYNAMIC_PARTITION_STATE, State.ERROR.toString());
         createOrUpdateRuntimeInfo(tableName, CREATE_PARTITION_MSG, msg);
     }
@@ -329,8 +344,8 @@ private void clearCreatePartitionFailedMsg(String tableName) {
         createOrUpdateRuntimeInfo(tableName, CREATE_PARTITION_MSG, DEFAULT_RUNTIME_VALUE);
     }
 
-    private void recordDropPartitionFailedMsg(String tableName, String msg) {
-        LOG.warn("dynamic drop partition failed: " + msg);
+    private void recordDropPartitionFailedMsg(String dbName, String tableName, String msg) {
+        LOG.warn("dynamic drop partition failed: {}, db: {}, table: {}", msg, dbName, tableName);
         createOrUpdateRuntimeInfo(tableName, DYNAMIC_PARTITION_STATE, State.ERROR.toString());
         createOrUpdateRuntimeInfo(tableName, DROP_PARTITION_MSG, msg);
     }
diff --git a/fe/src/main/java/org/apache/doris/common/util/DynamicPartitionUtil.java b/fe/src/main/java/org/apache/doris/common/util/DynamicPartitionUtil.java
index 806d5ebe105d00..c39a33caeb8984 100644
--- a/fe/src/main/java/org/apache/doris/common/util/DynamicPartitionUtil.java
+++ b/fe/src/main/java/org/apache/doris/common/util/DynamicPartitionUtil.java
@@ -36,6 +36,7 @@
 import org.apache.doris.common.ErrorReport;
 import org.apache.doris.common.FeNameFormat;
 
+import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
 
 import org.apache.logging.log4j.LogManager;
@@ -43,9 +44,12 @@
 
 import java.text.ParseException;
 import java.text.SimpleDateFormat;
+import java.time.DayOfWeek;
+import java.time.Month;
 import java.util.Calendar;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.TimeZone;
 
 public class DynamicPartitionUtil {
     private static final Logger LOG = LogManager.getLogger(DynamicPartitionUtil.class);
@@ -114,6 +118,36 @@ private static void checkEnable(String enable) throws DdlException {
         }
     }
 
+    private static void checkStartDayOfMonth(String val) throws DdlException {
+        if (Strings.isNullOrEmpty(val)) {
+            throw new DdlException("Invalid properties: " + DynamicPartitionProperty.START_DAY_OF_MONTH);
+        }
+        try {
+            int dayOfMonth = Integer.parseInt(val);
+            // only support from 1st to 28th, not allow 29th, 30th and 31th to avoid problems
+            // caused by lunar year and lunar month
+            if (dayOfMonth < 1 || dayOfMonth > 28) {
+                throw new DdlException(DynamicPartitionProperty.START_DAY_OF_MONTH + " should between 1 and 28");
+            }
+        } catch (NumberFormatException e) {
+            throw new DdlException("Invalid properties: " + DynamicPartitionProperty.START_DAY_OF_MONTH);
+        }
+    }
+
+    private static void checkStartDayOfWeek(String val) throws DdlException {
+        if (Strings.isNullOrEmpty(val)) {
+            throw new DdlException("Invalid properties: " + DynamicPartitionProperty.START_DAY_OF_WEEK);
+        }
+        try {
+            int dayOfWeek= Integer.parseInt(val);
+            if (dayOfWeek< 1 || dayOfWeek > 7) {
+                throw new DdlException(DynamicPartitionProperty.START_DAY_OF_WEEK + " should between 1 and 7");
+            }
+        } catch (NumberFormatException e) {
+            throw new DdlException("Invalid properties: " + DynamicPartitionProperty.START_DAY_OF_WEEK);
+        }
+    }
+
     public static boolean checkDynamicPartitionPropertiesExist(Map<String, String> properties) {
         if (properties == null) {
             return false;
@@ -123,10 +157,12 @@ public static boolean checkDynamicPartitionPropertiesExist(Map<String, String> p
                 properties.containsKey(DynamicPartitionProperty.END) ||
                 properties.containsKey(DynamicPartitionProperty.PREFIX) ||
                 properties.containsKey(DynamicPartitionProperty.BUCKETS) ||
-                properties.containsKey(DynamicPartitionProperty.ENABLE);
+                properties.containsKey(DynamicPartitionProperty.ENABLE) ||
+                properties.containsKey(DynamicPartitionProperty.START_DAY_OF_WEEK) ||
+                properties.containsKey(DynamicPartitionProperty.START_DAY_OF_MONTH);
     }
 
-    public static boolean checkInputDynamicPartitionProperties(Map<String, String> properties, PartitionInfo partitionInfo) throws DdlException{
+    private static boolean checkInputDynamicPartitionProperties(Map<String, String> properties, PartitionInfo partitionInfo) throws DdlException{
         if (properties == null || properties.isEmpty()) {
             return false;
         }
@@ -214,10 +250,24 @@ public static Map<String, String> analyzeDynamicPartition(Map<String, String> pr
         // If dynamic property is not specified.Use Integer.MIN_VALUE as default
         if (properties.containsKey(DynamicPartitionProperty.START)) {
             String startValue = properties.get(DynamicPartitionProperty.START);
-            checkStart(properties.get(DynamicPartitionProperty.START));
+            checkStart(startValue);
             properties.remove(DynamicPartitionProperty.START);
             analyzedProperties.put(DynamicPartitionProperty.START, startValue);
         }
+        
+        if (properties.containsKey(DynamicPartitionProperty.START_DAY_OF_MONTH)) {
+            String val = properties.get(DynamicPartitionProperty.START_DAY_OF_MONTH);
+            checkStartDayOfMonth(val);
+            properties.remove(DynamicPartitionProperty.START_DAY_OF_MONTH);
+            analyzedProperties.put(DynamicPartitionProperty.START_DAY_OF_MONTH, val);
+        }
+        
+        if (properties.containsKey(DynamicPartitionProperty.START_DAY_OF_WEEK)) {
+            String val = properties.get(DynamicPartitionProperty.START_DAY_OF_WEEK);
+            checkStartDayOfWeek(val);
+            properties.remove(DynamicPartitionProperty.START_DAY_OF_WEEK);
+            analyzedProperties.put(DynamicPartitionProperty.START_DAY_OF_WEEK, val);
+        }
         return analyzedProperties;
     }
 
@@ -273,34 +323,110 @@ public static String getPartitionFormat(Column column) throws DdlException {
         }
     }
 
-    public static String getFormattedPartitionName(String name, String timeUnit) {
-        name = name.replace("-", "").replace(":", "").replace(" ", "");
+    public static String getFormattedPartitionName(TimeZone tz, String formattedDateStr, String timeUnit) {
+        formattedDateStr = formattedDateStr.replace("-", "").replace(":", "").replace(" ", "");
         if (timeUnit.equalsIgnoreCase(TimeUnit.DAY.toString())) {
-            return name.substring(0, 8);
+            return formattedDateStr.substring(0, 8);
         } else if (timeUnit.equalsIgnoreCase(TimeUnit.MONTH.toString())) {
-            return name.substring(0, 6);
+            return formattedDateStr.substring(0, 6);
         } else {
-            name = name.substring(0, 8);
-            Calendar calendar = Calendar.getInstance();
+            formattedDateStr = formattedDateStr.substring(0, 8);
+            Calendar calendar = Calendar.getInstance(tz);
             try {
-                calendar.setTime(new SimpleDateFormat("yyyyMMdd").parse(name));
+                calendar.setTime(new SimpleDateFormat("yyyyMMdd").parse(formattedDateStr));
             } catch (ParseException e) {
                 LOG.warn("Format dynamic partition name error. Error={}", e.getMessage());
-                return name;
+                return formattedDateStr;
+            }
+            int weekOfYear = calendar.get(Calendar.WEEK_OF_YEAR);
+            if (weekOfYear <= 1 && calendar.get(Calendar.MONTH) >= 11) {
+                // eg: JDK think 2019-12-30 as the first week of year 2020, we need to handle this.
+                // to make it as the 53rd week of year 2019.
+                weekOfYear += 52;
             }
-            return String.format("%s_%02d", calendar.get(Calendar.YEAR), calendar.get(Calendar.WEEK_OF_YEAR));
+            return String.format("%s_%02d", calendar.get(Calendar.YEAR), weekOfYear);
         }
     }
 
-    public static String getPartitionRange(String timeUnit, int offset, Calendar calendar, String format) {
+    // return the partition range date string formatted as yyyy-MM-dd[ HH:mm::ss]
+    // TODO: support HOUR and YEAR
+    public static String getPartitionRangeString(DynamicPartitionProperty property, Calendar current,
+            int offset, String format) {
+        String timeUnit = property.getTimeUnit();
+        TimeZone tz = property.getTimeZone();
         if (timeUnit.equalsIgnoreCase(TimeUnit.DAY.toString())) {
-            calendar.add(Calendar.DAY_OF_MONTH, offset);
+            return getPartitionRangeOfDay(current, offset, tz, format);
         } else if (timeUnit.equalsIgnoreCase(TimeUnit.WEEK.toString())) {
-            calendar.add(Calendar.WEEK_OF_MONTH, offset);
-        } else {
-            calendar.add(Calendar.MONTH, offset);
+            return getPartitionRangeOfWeek(current, offset, property.getStartOfWeek(), tz, format);
+        } else { // MONTH
+            return getPartitionRangeOfMonth(current, offset, property.getStartOfMonth(), tz, format);
+        }
+    }
+    
+    /*
+     * return formatted string of partition range in DAY granularity.
+     * offset: The offset from the current day. 0 means current day, 1 means tomorrow, -1 means yesterday.
+     * format: the format of the return date string.
+     * 
+     * Eg:
+     *  Today is 2020-05-24, offset = -1
+     *  It will return 2020-05-23
+     */
+    private static String getPartitionRangeOfDay(Calendar current, int offset, TimeZone tz, String format) {
+        current.add(Calendar.DATE, offset);
+        return getFormattedTimeWithoutHourMinuteSecond(current, format);
+    }
+
+    /*
+     * return formatted string of partition range in WEEK granularity.
+     * offset: The offset from the current week. 0 means current week, 1 means next week, -1 means last week.
+     * startOf: Define the start day of each week. 1 means MONDAY, 7 means SUNDAY.
+     * format: the format of the return date string.
+     * 
+     * Eg:
+     *  Today is 2020-05-24, offset = -1, startOf.dayOfWeek = 3
+     *  It will return 2020-05-20  (Wednesday of last week)
+     */
+    private static String getPartitionRangeOfWeek(Calendar current, int offset, StartOfDate startOf, TimeZone tz,
+            String format) {
+        Preconditions.checkArgument(startOf.isStartOfWeek());
+        // 1. get the offset week
+        current.add(Calendar.WEEK_OF_YEAR, offset);
+        // 2. get the date of `startOf` week
+        int day = current.get(Calendar.DAY_OF_WEEK);
+        // SUNDAY will return 1, we will set it to 7, and make MONDAY to 1, and so on
+        day = (day == 1 ? 7 : day - 1);
+        current.add(Calendar.DATE, (startOf.dayOfWeek - day));
+        return getFormattedTimeWithoutHourMinuteSecond(current, format);
+    }
+
+    /*
+     * return formatted string of partition range in MONTH granularity.
+     * offset: The offset from the current month. 0 means current month, 1 means next month, -1 means last month.
+     * startOf: Define the start date of each month. 1 means start on the 1st of every month.
+     * format: the format of the return date string.
+     * 
+     * Eg:
+     *  Today is 2020-05-24, offset = 1, startOf.month = 3
+     *  It will return 2020-06-03 
+     */
+    private static String getPartitionRangeOfMonth(Calendar current, int offset, StartOfDate startOf, TimeZone tz,
+            String format) {
+        Preconditions.checkArgument(startOf.isStartOfMonth());
+        // 1. Get the offset date.
+        int realOffset = offset;
+        int currentDay = current.get(Calendar.DATE);
+        if (currentDay < startOf.day) {
+            // eg: today is 2020-05-20, `startOf.day` is 25, and offset is 0.
+            // we should return 2020-04-25, which is the last month.
+            realOffset -= 1;
         }
-        // dynamic partition's time accuracy is DAY
+        current.add(Calendar.MONTH, realOffset);
+        current.set(Calendar.DATE, startOf.day);
+        return getFormattedTimeWithoutHourMinuteSecond(current, format);
+    }
+
+    private static String getFormattedTimeWithoutHourMinuteSecond(Calendar calendar, String format) {
         calendar.set(Calendar.HOUR_OF_DAY, 0);
         calendar.set(Calendar.MINUTE, 0);
         calendar.set(Calendar.SECOND, 0);
@@ -319,4 +445,52 @@ public static int estimateReplicateNum(OlapTable table) {
         }
         return replicateNum;
     }
-}
\ No newline at end of file
+
+    /*
+     * Used to indicate the start date.
+     * Taking the year as the granularity, it can indicate the month and day as the start date.
+     * Taking the month as the granularity, it can indicate the date of as the start date.
+     * Taking the week as the granularity, it can indicate the day of the week as the starting date.
+     */
+    public static class StartOfDate {
+        public int month;
+        public int day;
+        public int dayOfWeek;
+
+        public StartOfDate(int month, int day, int dayOfWeek) {
+            this.month = month;
+            this.day = day;
+            this.dayOfWeek = dayOfWeek;
+        }
+
+        public boolean isStartOfYear() {
+            return this.month != -1 && this.day != -1 && this.dayOfWeek == -1;
+        }
+
+        public boolean isStartOfMonth() {
+            return this.month == -1 && this.day != -1 && this.dayOfWeek == -1;
+        }
+
+        public boolean isStartOfWeek() {
+            return this.month == -1 && this.day == -1 && this.dayOfWeek != -1;
+        }
+
+        public String toDisplayInfo() {
+            if (isStartOfWeek()) {
+                return DayOfWeek.of(dayOfWeek).name();
+            } else if (isStartOfMonth()) {
+                return Util.ordinal(day);
+            } else if (isStartOfYear()) {
+                return Month.of(month) + " " + Util.ordinal(day);
+            } else {
+                return "N/A";
+            }
+        }
+
+        @Override
+        public String toString() {
+            // TODO Auto-generated method stub
+            return super.toString();
+        }
+    }
+}
diff --git a/fe/src/main/java/org/apache/doris/common/util/TimeUtils.java b/fe/src/main/java/org/apache/doris/common/util/TimeUtils.java
index 789895779eeb5a..f8c0c1db3afa24 100644
--- a/fe/src/main/java/org/apache/doris/common/util/TimeUtils.java
+++ b/fe/src/main/java/org/apache/doris/common/util/TimeUtils.java
@@ -78,9 +78,6 @@ public class TimeUtils {
     public static Date MIN_DATETIME = null;
     public static Date MAX_DATETIME = null;
 
-    public static int MIN_TIME;
-    public static int MAX_TIME;
-
     static {
         TIME_ZONE = new SimpleTimeZone(8 * 3600 * 1000, "");
         
@@ -128,6 +125,10 @@ public static TimeZone getTimeZone() {
         return TimeZone.getTimeZone(ZoneId.of(timezone, timeZoneAliasMap));
     }
 
+    public static TimeZone getDefaultTimeZone() {
+        return TimeZone.getTimeZone(ZoneId.of(DEFAULT_TIME_ZONE, timeZoneAliasMap));
+    }
+
     public static String longToTimeString(long timeStamp, SimpleDateFormat dateFormat) {
         if (timeStamp <= 0L) {
             return "N/A";
diff --git a/fe/src/main/java/org/apache/doris/common/util/Util.java b/fe/src/main/java/org/apache/doris/common/util/Util.java
index d92c3ab0abbb5e..faeb29af37cde7 100644
--- a/fe/src/main/java/org/apache/doris/common/util/Util.java
+++ b/fe/src/main/java/org/apache/doris/common/util/Util.java
@@ -53,6 +53,8 @@ public class Util {
 
     private static final long DEFAULT_EXEC_CMD_TIMEOUT_MS = 600000L;
 
+    private static final String[] ORDINAL_SUFFIX = new String[] { "th", "st", "nd", "rd", "th", "th", "th", "th", "th", "th" };
+
     static {
         TYPE_STRING_MAP.put(PrimitiveType.TINYINT, "tinyint(4)");
         TYPE_STRING_MAP.put(PrimitiveType.SMALLINT, "smallint(6)");
@@ -443,6 +445,17 @@ public static long decodeVarint64(DataInput in) throws IOException {
 
         return result;
     }
-
+    
+    // return the ordinal string of an Integer
+    public static String ordinal(int i) {
+        switch (i % 100) {
+            case 11:
+            case 12:
+            case 13:
+                return i + "th";
+            default:
+                return i + ORDINAL_SUFFIX[i % 10];
+        }
+    }
 }
 
diff --git a/fe/src/main/java/org/apache/doris/load/Load.java b/fe/src/main/java/org/apache/doris/load/Load.java
index 341d3d37cac950..1f36885c42cc11 100644
--- a/fe/src/main/java/org/apache/doris/load/Load.java
+++ b/fe/src/main/java/org/apache/doris/load/Load.java
@@ -98,11 +98,12 @@
 import org.apache.doris.thrift.TPriority;
 import org.apache.doris.transaction.PartitionCommitInfo;
 import org.apache.doris.transaction.TableCommitInfo;
+import org.apache.doris.transaction.TransactionNotFoundException;
 import org.apache.doris.transaction.TransactionState;
-import org.apache.doris.transaction.TransactionStatus;
 import org.apache.doris.transaction.TransactionState.LoadJobSourceType;
-import org.apache.doris.transaction.TransactionState.TxnSourceType;
 import org.apache.doris.transaction.TransactionState.TxnCoordinator;
+import org.apache.doris.transaction.TransactionState.TxnSourceType;
+import org.apache.doris.transaction.TransactionStatus;
 
 import com.google.common.base.Joiner;
 import com.google.common.base.Preconditions;
@@ -2831,6 +2832,10 @@ private boolean processCancelled(LoadJob job, CancelType cancelType, String msg,
                     job.getDbId(),
                     job.getTransactionId(),
                     job.getFailMsg().toString());
+        } catch (TransactionNotFoundException e) {
+            // the transaction may already be aborted due to timeout by transaction manager.
+            // just print a log and continue to cancel the job.
+            LOG.info("transaction not found when try to abort it: {}", e.getTransactionId());
         } catch (Exception e) {
             LOG.info("errors while abort transaction", e);
             if (failedMsg != null) {
diff --git a/fe/src/main/java/org/apache/doris/qe/ShowExecutor.java b/fe/src/main/java/org/apache/doris/qe/ShowExecutor.java
index 3104cacf524a12..576738859cbc03 100644
--- a/fe/src/main/java/org/apache/doris/qe/ShowExecutor.java
+++ b/fe/src/main/java/org/apache/doris/qe/ShowExecutor.java
@@ -1496,6 +1496,7 @@ private void handleShowDynamicPartition() {
                             String.valueOf(dynamicPartitionProperty.getEnd()),
                             dynamicPartitionProperty.getPrefix(),
                             String.valueOf(dynamicPartitionProperty.getBuckets()),
+                            dynamicPartitionProperty.getStartOfInfo(),
                             dynamicPartitionScheduler.getRuntimeInfo(tableName, DynamicPartitionScheduler.LAST_UPDATE_TIME),
                             dynamicPartitionScheduler.getRuntimeInfo(tableName, DynamicPartitionScheduler.LAST_SCHEDULER_TIME),
                             dynamicPartitionScheduler.getRuntimeInfo(tableName, DynamicPartitionScheduler.DYNAMIC_PARTITION_STATE),
diff --git a/fe/src/main/java/org/apache/doris/transaction/DatabaseTransactionMgr.java b/fe/src/main/java/org/apache/doris/transaction/DatabaseTransactionMgr.java
index 8f10fa0348d1c3..b0c51f4bcbea02 100644
--- a/fe/src/main/java/org/apache/doris/transaction/DatabaseTransactionMgr.java
+++ b/fe/src/main/java/org/apache/doris/transaction/DatabaseTransactionMgr.java
@@ -852,7 +852,7 @@ public void abortTransaction(String label, String reason) throws UserException {
         try {
             Set<Long> existingTxns = unprotectedGetTxnIdsByLabel(label);
             if (existingTxns == null || existingTxns.isEmpty()) {
-                throw new UserException("transaction not found, label=" + label);
+                throw new TransactionNotFoundException("transaction not found, label=" + label);
             }
             // find PREPARE txn. For one load label, there should be only one PREPARE txn.
             TransactionState prepareTxn = null;
@@ -865,7 +865,7 @@ public void abortTransaction(String label, String reason) throws UserException {
             }
 
             if (prepareTxn == null) {
-                throw new UserException("running transaction not found, label=" + label);
+                throw new TransactionNotFoundException("running transaction not found, label=" + label);
             }
 
             transactionId = prepareTxn.getTransactionId();
@@ -888,7 +888,7 @@ public void abortTransaction(long transactionId, String reason, TxnCommitAttachm
             readUnlock();
         }
         if (transactionState == null) {
-            throw new UserException("transaction not found");
+            throw new TransactionNotFoundException("transaction not found", transactionId);
         }
 
         // update transaction state extra if exists
@@ -921,7 +921,7 @@ private boolean unprotectAbortTransaction(long transactionId, String reason)
             throws UserException {
         TransactionState transactionState = unprotectedGetTransactionState(transactionId);
         if (transactionState == null) {
-            throw new UserException("transaction not found");
+            throw new TransactionNotFoundException("transaction not found", transactionId);
         }
         if (transactionState.getTransactionStatus() == TransactionStatus.ABORTED) {
             return false;
diff --git a/fe/src/main/java/org/apache/doris/transaction/TransactionException.java b/fe/src/main/java/org/apache/doris/transaction/TransactionException.java
index 2fdc5066d43a32..fe450c30397149 100644
--- a/fe/src/main/java/org/apache/doris/transaction/TransactionException.java
+++ b/fe/src/main/java/org/apache/doris/transaction/TransactionException.java
@@ -21,8 +21,8 @@
 
 public class TransactionException extends UserException {
     
-    private long transactionId;
-    
+    private long transactionId = -1;
+
     public TransactionException(String msg) {
         super(msg);
     }
@@ -35,7 +35,7 @@ public TransactionException(String msg, long transactionId) {
         super(msg);
         this.transactionId = transactionId;
     }
-    
+
     public long getTransactionId() {
         return transactionId;
     }
diff --git a/fe/src/main/java/org/apache/doris/transaction/TransactionNotFoundException.java b/fe/src/main/java/org/apache/doris/transaction/TransactionNotFoundException.java
new file mode 100644
index 00000000000000..bc3351ee6ac7d8
--- /dev/null
+++ b/fe/src/main/java/org/apache/doris/transaction/TransactionNotFoundException.java
@@ -0,0 +1,33 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.transaction;
+
+public class TransactionNotFoundException extends TransactionException {
+
+    public TransactionNotFoundException(String msg) {
+        super(msg);
+    }
+
+    public TransactionNotFoundException(String msg, Throwable e) {
+        super(msg, e);
+    }
+
+    public TransactionNotFoundException(String msg, long transactionId) {
+        super(msg, transactionId);
+    }
+}
diff --git a/fe/src/test/java/org/apache/doris/common/util/DynamicPartitionUtilTest.java b/fe/src/test/java/org/apache/doris/common/util/DynamicPartitionUtilTest.java
new file mode 100644
index 00000000000000..3339c159e356c5
--- /dev/null
+++ b/fe/src/test/java/org/apache/doris/common/util/DynamicPartitionUtilTest.java
@@ -0,0 +1,209 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.common.util;
+
+import org.apache.doris.catalog.DynamicPartitionProperty;
+
+import com.google.common.collect.Maps;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.util.Calendar;
+import java.util.Date;
+import java.util.Map;
+
+public class DynamicPartitionUtilTest {
+
+    private static final String FORMAT = "yyyy-MM-dd";
+
+    private static Map<String, String> getDynamProp(String timeUnit, int start, int end, int startOfWeek,
+            int startOfMonth) {
+        Map<String, String> prop = Maps.newHashMap();
+        prop.put(DynamicPartitionProperty.ENABLE, "true");
+        prop.put(DynamicPartitionProperty.TIME_UNIT, timeUnit);
+        prop.put(DynamicPartitionProperty.START, String.valueOf(start));
+        prop.put(DynamicPartitionProperty.END, String.valueOf(end));
+        prop.put(DynamicPartitionProperty.PREFIX, "p");
+        prop.put(DynamicPartitionProperty.BUCKETS, "1");
+        if (startOfWeek > 0) {
+            prop.put(DynamicPartitionProperty.START_DAY_OF_WEEK, String.valueOf(startOfWeek));
+        }
+
+        if (startOfMonth > 0) {
+            prop.put(DynamicPartitionProperty.START_DAY_OF_MONTH, String.valueOf(startOfMonth));
+        }
+        return prop;
+    }
+
+    private static Calendar getCalendarWithDate(String dateStr) throws ParseException {
+        SimpleDateFormat sdf = new SimpleDateFormat(FORMAT);
+        Calendar calendar = Calendar.getInstance(TimeUtils.getDefaultTimeZone());
+        Date date = sdf.parse(dateStr);
+        calendar.setTime(date);
+        return calendar;
+    }
+
+    @Test
+    public void testGetPartitionRangeString() throws ParseException {
+        // TimeUnit: DAY
+        
+        // 1. 2020-05-25, offset -7
+        DynamicPartitionProperty property = new DynamicPartitionProperty(getDynamProp("DAY", -3, 3, -1, -1));
+        String res = DynamicPartitionUtil.getPartitionRangeString(property, getCalendarWithDate("2020-05-25"), -7,
+                FORMAT);
+        Assert.assertEquals("2020-05-18", res);
+        String partName = DynamicPartitionUtil.getFormattedPartitionName(TimeUtils.getDefaultTimeZone(), res, "DAY");
+        Assert.assertEquals("20200518", partName);
+        // 2. 2020-05-25, offset 0
+        res = DynamicPartitionUtil.getPartitionRangeString(property, getCalendarWithDate("2020-05-25"), 0,
+                FORMAT);
+        Assert.assertEquals("2020-05-25", res);
+        partName = DynamicPartitionUtil.getFormattedPartitionName(TimeUtils.getDefaultTimeZone(), res, "DAY");
+        Assert.assertEquals("20200525", partName);
+        // 3. 2020-05-25, offset 7
+        res = DynamicPartitionUtil.getPartitionRangeString(property, getCalendarWithDate("2020-05-25"), 7,
+                FORMAT);
+        Assert.assertEquals("2020-06-01", res);
+        partName = DynamicPartitionUtil.getFormattedPartitionName(TimeUtils.getDefaultTimeZone(), res, "DAY");
+        Assert.assertEquals("20200601", partName);
+        // 4. 2020-02-28, offset 3
+        res = DynamicPartitionUtil.getPartitionRangeString(property, getCalendarWithDate("2020-02-28"), 3,
+                FORMAT);
+        Assert.assertEquals("2020-03-02", res);
+        partName = DynamicPartitionUtil.getFormattedPartitionName(TimeUtils.getDefaultTimeZone(), res, "DAY");
+        Assert.assertEquals("20200302", partName);
+
+        // TimeUnit: WEEK
+        // 1. 2020-05-25, start day: MONDAY, offset 0
+        property = new DynamicPartitionProperty(getDynamProp("WEEK", -3, 3, 1, -1));
+        res = DynamicPartitionUtil.getPartitionRangeString(property, getCalendarWithDate("2020-05-25"), 0,
+                FORMAT);
+        Assert.assertEquals("2020-05-25", res);
+        partName = DynamicPartitionUtil.getFormattedPartitionName(TimeUtils.getDefaultTimeZone(), res, "WEEK");
+        Assert.assertEquals("2020_22", partName);
+
+        // 2. 2020-05-28, start day: MONDAY, offset 0
+        property = new DynamicPartitionProperty(getDynamProp("WEEK", -3, 3, 1, -1));
+        res = DynamicPartitionUtil.getPartitionRangeString(property, getCalendarWithDate("2020-05-28"), 0,
+                FORMAT);
+        Assert.assertEquals("2020-05-25", res);
+        partName = DynamicPartitionUtil.getFormattedPartitionName(TimeUtils.getDefaultTimeZone(), res, "WEEK");
+        Assert.assertEquals("2020_22", partName);
+
+        // 3. 2020-05-25, start day: SUNDAY, offset 0
+        property = new DynamicPartitionProperty(getDynamProp("WEEK", -3, 3, 7, -1));
+        res = DynamicPartitionUtil.getPartitionRangeString(property, getCalendarWithDate("2020-05-25"), 0,
+                FORMAT);
+        Assert.assertEquals("2020-05-31", res);
+        partName = DynamicPartitionUtil.getFormattedPartitionName(TimeUtils.getDefaultTimeZone(), res, "WEEK");
+        Assert.assertEquals("2020_23", partName);
+
+        // 4. 2020-05-25, start day: MONDAY, offset -2
+        property = new DynamicPartitionProperty(getDynamProp("WEEK", -3, 3, 1, -1));
+        res = DynamicPartitionUtil.getPartitionRangeString(property, getCalendarWithDate("2020-05-25"), -2,
+                FORMAT);
+        Assert.assertEquals("2020-05-11", res);
+        partName = DynamicPartitionUtil.getFormattedPartitionName(TimeUtils.getDefaultTimeZone(), res, "WEEK");
+        Assert.assertEquals("2020_20", partName);
+
+        // 5. 2020-02-29, start day: WED, offset 0
+        property = new DynamicPartitionProperty(getDynamProp("WEEK", -3, 3, 3, -1));
+        res = DynamicPartitionUtil.getPartitionRangeString(property, getCalendarWithDate("2020-02-29"), 0,
+                FORMAT);
+        Assert.assertEquals("2020-02-26", res);
+        partName = DynamicPartitionUtil.getFormattedPartitionName(TimeUtils.getDefaultTimeZone(), res, "WEEK");
+        Assert.assertEquals("2020_09", partName);
+
+        // 6. 2020-02-29, start day: TUS, offset 1
+        property = new DynamicPartitionProperty(getDynamProp("WEEK", -3, 3, 2, -1));
+        res = DynamicPartitionUtil.getPartitionRangeString(property, getCalendarWithDate("2020-02-29"), 1,
+                FORMAT);
+        Assert.assertEquals("2020-03-03", res);
+        partName = DynamicPartitionUtil.getFormattedPartitionName(TimeUtils.getDefaultTimeZone(), res, "WEEK");
+        Assert.assertEquals("2020_10", partName);
+
+        // 6. 2020-01-01, start day: MONDAY, offset -1
+        property = new DynamicPartitionProperty(getDynamProp("WEEK", -3, 3, 1, -1));
+        res = DynamicPartitionUtil.getPartitionRangeString(property, getCalendarWithDate("2020-01-01"), -1,
+                FORMAT);
+        Assert.assertEquals("2019-12-23", res);
+        partName = DynamicPartitionUtil.getFormattedPartitionName(TimeUtils.getDefaultTimeZone(), res, "WEEK");
+        Assert.assertEquals("2019_52", partName);
+
+        // 6. 2020-01-01, start day: MONDAY, offset 0
+        property = new DynamicPartitionProperty(getDynamProp("WEEK", -3, 3, 1, -1));
+        res = DynamicPartitionUtil.getPartitionRangeString(property, getCalendarWithDate("2020-01-01"), 0,
+                FORMAT);
+        Assert.assertEquals("2019-12-30", res);
+        partName = DynamicPartitionUtil.getFormattedPartitionName(TimeUtils.getDefaultTimeZone(), res, "WEEK");
+        Assert.assertEquals("2019_53", partName);
+
+        // TimeUnit: MONTH
+        // 1. 2020-05-25, start day: 1, offset 0
+        property = new DynamicPartitionProperty(getDynamProp("MONTH", -3, 3, -1, 1));
+        res = DynamicPartitionUtil.getPartitionRangeString(property, getCalendarWithDate("2020-05-25"), 0,
+                FORMAT);
+        Assert.assertEquals("2020-05-01", res);
+        partName = DynamicPartitionUtil.getFormattedPartitionName(TimeUtils.getDefaultTimeZone(), res, "MONTH");
+        Assert.assertEquals("202005", partName);
+
+        // 2. 2020-05-25, start day: 26, offset 0
+        property = new DynamicPartitionProperty(getDynamProp("MONTH", -3, 3, -1, 26));
+        res = DynamicPartitionUtil.getPartitionRangeString(property, getCalendarWithDate("2020-05-25"), 0,
+                FORMAT);
+        Assert.assertEquals("2020-04-26", res);
+        partName = DynamicPartitionUtil.getFormattedPartitionName(TimeUtils.getDefaultTimeZone(), res, "MONTH");
+        Assert.assertEquals("202004", partName);
+
+        // 3. 2020-05-25, start day: 26, offset -1
+        property = new DynamicPartitionProperty(getDynamProp("MONTH", -3, 3, -1, 26));
+        res = DynamicPartitionUtil.getPartitionRangeString(property, getCalendarWithDate("2020-05-25"), -1,
+                FORMAT);
+        Assert.assertEquals("2020-03-26", res);
+        partName = DynamicPartitionUtil.getFormattedPartitionName(TimeUtils.getDefaultTimeZone(), res, "MONTH");
+        Assert.assertEquals("202003", partName);
+
+        // 4. 2020-02-29, start day: 26, offset 3
+        property = new DynamicPartitionProperty(getDynamProp("MONTH", -3, 3, -1, 26));
+        res = DynamicPartitionUtil.getPartitionRangeString(property, getCalendarWithDate("2020-02-29"), 3,
+                FORMAT);
+        Assert.assertEquals("2020-05-26", res);
+        partName = DynamicPartitionUtil.getFormattedPartitionName(TimeUtils.getDefaultTimeZone(), res, "MONTH");
+        Assert.assertEquals("202005", partName);
+
+        // 5. 2020-02-29, start day: 27, offset 0
+        property = new DynamicPartitionProperty(getDynamProp("MONTH", -3, 3, -1, 27));
+        res = DynamicPartitionUtil.getPartitionRangeString(property, getCalendarWithDate("2020-02-29"), 0,
+                FORMAT);
+        Assert.assertEquals("2020-02-27", res);
+        partName = DynamicPartitionUtil.getFormattedPartitionName(TimeUtils.getDefaultTimeZone(), res, "MONTH");
+        Assert.assertEquals("202002", partName);
+
+        // 6. 2020-02-29, start day: 27, offset -3
+        property = new DynamicPartitionProperty(getDynamProp("MONTH", -3, 3, -1, 27));
+        res = DynamicPartitionUtil.getPartitionRangeString(property, getCalendarWithDate("2020-02-29"), -3,
+                FORMAT);
+        Assert.assertEquals("2019-11-27", res);
+        partName = DynamicPartitionUtil.getFormattedPartitionName(TimeUtils.getDefaultTimeZone(), res, "MONTH");
+        Assert.assertEquals("201911", partName);
+    }
+
+}