[Bug][Refactor] Fix the conflict of temp partition and dynamic partition operations (#3201)

The bug is described in issue: #3200.

This CL solve the problem by:
1. Refactor the alter operation conflict checking logic by introducing new classes `AlterOperations` and `AlterOpType`.
2. Allow add/drop temporary partition when dynamic partition feature is enabled.
3. Allow modifying table's property when there is temporary partition in table.
4. Make the properties `dynamic_partition.enable` optional, and default is true.

docs/documentation/cn/administrator-guide/dynamic-partition.md

@@ -42,7 +42,7 @@ under the License.
 
 ### 动态分区属性参数说明:
 
-`dynamic_partition.enable`: 是否开启动态分区特性，可指定为 `TRUE` 或 `FALSE`。
+`dynamic_partition.enable`: 是否开启动态分区特性，可指定为 `TRUE` 或 `FALSE`。如果不填写，默认为 `TRUE`。
 
 
 `dynamic_partition.time_unit`: 动态分区调度的单位，可指定为 `DAY` `WEEK` `MONTH`，当指定为 `DAY` 时，动态创建的分区名后缀格式为`yyyyMMdd`，例如`20200325`。当指定为 `WEEK` 时，动态创建的分区名后缀格式为`yyyy_ww`即当前日期属于这一年的第几周，例如 `2020-03-25` 创建的分区名后缀为 `2020_13`, 表明目前为2020年第13周。当指定为 `MONTH` 时，动态创建的分区名后缀格式为 `yyyyMM`，例如 `202003`。

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

@@ -257,13 +257,15 @@ under the License.
       PROPERTIES (
           "dynamic_partition.enable" = "true|false",
           "dynamic_partition.time_unit" = "DAY|WEEK|MONTH",
+          "dynamic_partition.start" = "${integer_value}",
           "dynamic_partitoin.end" = "${integer_value}",
           "dynamic_partition.prefix" = "${string_value}",
           "dynamic_partition.buckets" = "${integer_value}
 ```
-    dynamic_partition.enable: 用于指定表级别的动态分区功能是否开启
+    dynamic_partition.enable: 用于指定表级别的动态分区功能是否开启。默认为 true。
     dynamic_partition.time_unit: 用于指定动态添加分区的时间单位，可选择为DAY（天），WEEK(周)，MONTH（月）
-    dynamic_partition.end: 用于指定提前创建的分区数量
+    dynamic_partition.start: 用于指定向前删除多少个分区。值必须小于0。默认为 Integer.MIN_VALUE。
+    dynamic_partition.end: 用于指定提前创建的分区数量。值必须大于0。
     dynamic_partition.prefix: 用于指定创建的分区名前缀，例如分区名前缀为p，则自动创建分区名为p20200108
     dynamic_partition.buckets: 用于指定自动创建的分区分桶数量
 
@@ -524,7 +526,7 @@ under the License.
     PROPERTIES ("storage_type"="column");
 ```
 
-11. 创建一个动态分区表(需要在FE配置中开启动态分区功能)，该表每天提前创建3天的分区，例如今天为`2020-01-08`，则会创建分区名为`p20200108`, `p20200109`, `p20200110`, `p20200111`的分区. 分区范围分别为: 
+11. 创建一个动态分区表(需要在FE配置中开启动态分区功能)，该表每天提前创建3天的分区，并删除3天前的分区。例如今天为`2020-01-08`，则会创建分区名为`p20200108`, `p20200109`, `p20200110`, `p20200111`的分区. 分区范围分别为: 
 
 ```
 [types: [DATE]; keys: [2020-01-08]; ‥types: [DATE]; keys: [2020-01-09]; )
@@ -554,6 +556,7 @@ under the License.
     PROPERTIES(
     "storage_medium" = "SSD",
     "dynamic_partition.time_unit" = "DAY",
+    "dynamic_partition.start" = "-3",
     "dynamic_partition.end" = "3",
     "dynamic_partition.prefix" = "p",
     "dynamic_partition.buckets" = "32"

fe/src/main/java/org/apache/doris/alter/Alter.java

@@ -17,29 +17,18 @@
 
 package org.apache.doris.alter;
 
-import org.apache.doris.analysis.AddColumnClause;
-import org.apache.doris.analysis.AddColumnsClause;
 import org.apache.doris.analysis.AddPartitionClause;
-import org.apache.doris.analysis.AddRollupClause;
 import org.apache.doris.analysis.AlterClause;
 import org.apache.doris.analysis.AlterSystemStmt;
-import org.apache.doris.analysis.AlterTableClause;
 import org.apache.doris.analysis.AlterTableStmt;
 import org.apache.doris.analysis.AlterViewStmt;
 import org.apache.doris.analysis.ColumnRenameClause;
-import org.apache.doris.analysis.CreateIndexClause;
 import org.apache.doris.analysis.CreateMaterializedViewStmt;
-import org.apache.doris.analysis.DropColumnClause;
-import org.apache.doris.analysis.DropIndexClause;
 import org.apache.doris.analysis.DropMaterializedViewStmt;
 import org.apache.doris.analysis.DropPartitionClause;
-import org.apache.doris.analysis.DropRollupClause;
-import org.apache.doris.analysis.IndexDef;
-import org.apache.doris.analysis.ModifyColumnClause;
 import org.apache.doris.analysis.ModifyPartitionClause;
 import org.apache.doris.analysis.ModifyTablePropertiesClause;
 import org.apache.doris.analysis.PartitionRenameClause;
-import org.apache.doris.analysis.ReorderColumnsClause;
 import org.apache.doris.analysis.ReplacePartitionClause;
 import org.apache.doris.analysis.RollupRenameClause;
 import org.apache.doris.analysis.TableName;
@@ -47,7 +36,6 @@ import org.apache.doris.analysis.TableRenameClause;
 import org.apache.doris.catalog.Catalog;
 import org.apache.doris.catalog.Column;
 import org.apache.doris.catalog.Database;
-import org.apache.doris.catalog.Index;
 import org.apache.doris.catalog.OlapTable;
 import org.apache.doris.catalog.OlapTable.OlapTableState;
 import org.apache.doris.catalog.Table;
@@ -72,8 +60,6 @@ import org.apache.logging.log4j.Logger;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
-import java.util.TreeSet;
 
 public class Alter {
     private static final Logger LOG = LogManager.getLogger(Alter.class);
@@ -167,141 +153,19 @@ public class Alter {
             ErrorReport.reportDdlException(ErrorCode.ERR_BAD_DB_ERROR, dbName);
         }
 
-        // check cluster capacity
-        Catalog.getCurrentSystemInfo().checkClusterCapacity(clusterName);
-
-        // schema change ops can appear several in one alter stmt without other alter ops entry
-        boolean hasSchemaChange = false;
-        // materialized view ops (include rollup), if has, should appear one and only one add or drop mv entry
-        boolean hasAddMaterializedView = false;
-        boolean hasDropRollup = false;
-        // partition ops, if has, should appear one and only one entry
-        boolean hasPartition = false;
-        // rename ops, if has, should appear one and only one entry
-        boolean hasRename = false;
-        // modify properties ops, if has, should appear one and only one entry
-        boolean hasModifyProp = false;
-
         // check conflict alter ops first
         List<AlterClause> alterClauses = stmt.getOps();
-        // check conflict alter ops first
+        AlterOperations currentAlterOps = new AlterOperations();
+        currentAlterOps.checkConflict(alterClauses);
 
-        // if all alter clauses are DropPartitionClause or DropRollupClause, no need to check quota.
-        boolean allIsDropOps = true;
-        for (AlterClause alterClause : alterClauses) {
-            if (!(alterClause instanceof DropPartitionClause)
-                    && !(alterClause instanceof DropRollupClause)) {
-                allIsDropOps = false;
-                break;
-            }
-        }
-
-        if (!allIsDropOps) {
-            // check db quota
+        // check cluster capacity and db quota, only need to check once.
+        if (currentAlterOps.needCheckCapacity()) {
+            Catalog.getCurrentSystemInfo().checkClusterCapacity(clusterName);
             db.checkQuota();
         }
 
-        // synchronized operation must handle outside db write lock
-        boolean needSynchronized = false;
-        boolean needTableStable = false;
-        for (AlterClause alterClause : alterClauses) {
-            if (!needTableStable) {
-                needTableStable = ((AlterTableClause) alterClause).isNeedTableStable();
-            }
-            if ((alterClause instanceof AddColumnClause
-                    || alterClause instanceof AddColumnsClause
-                    || alterClause instanceof DropColumnClause
-                    || alterClause instanceof ModifyColumnClause
-                    || alterClause instanceof ReorderColumnsClause
-                    || alterClause instanceof CreateIndexClause
-                    || alterClause instanceof DropIndexClause)
-                    && !hasAddMaterializedView && !hasDropRollup && !hasPartition && !hasRename) {
-                hasSchemaChange = true;
-                if (alterClause instanceof CreateIndexClause) {
-                    Table table = db.getTable(dbTableName.getTbl());
-                    if (!(table instanceof OlapTable)) {
-                        throw new AnalysisException("create index only support in olap table at current version.");
-                    }
-                    List<Index> indexes = ((OlapTable) table).getIndexes();
-                    IndexDef indexDef = ((CreateIndexClause) alterClause).getIndexDef();
-                    Set<String> newColset = new TreeSet<>(String.CASE_INSENSITIVE_ORDER);
-                    newColset.addAll(indexDef.getColumns());
-                    for (Index idx : indexes) {
-                        if (idx.getIndexName().equalsIgnoreCase(indexDef.getIndexName())) {
-                            throw new AnalysisException("index `" + indexDef.getIndexName() + "` already exist.");
-                        }
-                        Set<String> idxSet = new TreeSet<>(String.CASE_INSENSITIVE_ORDER);
-                        idxSet.addAll(idx.getColumns());
-                        if (newColset.equals(idxSet)) {
-                            throw new AnalysisException("index for columns (" + String
-                                    .join(",", indexDef.getColumns()) + " ) already exist.");
-                        }
-                    }
-                    OlapTable olapTable = (OlapTable) table;
-                    for (String col : indexDef.getColumns()) {
-                        Column column = olapTable.getColumn(col);
-                        if (column != null) {
-                            indexDef.checkColumn(column, olapTable.getKeysType());
-                        } else {
-                            throw new AnalysisException("BITMAP column does not exist in table. invalid column: "
-                                    + col);
-                        }
-                    }
-                } else if (alterClause instanceof DropIndexClause) {
-                    Table table = db.getTable(dbTableName.getTbl());
-                    if (!(table instanceof OlapTable)) {
-                        throw new AnalysisException("drop index only support in olap table at current version.");
-                    }
-                    String indexName = ((DropIndexClause) alterClause).getIndexName();
-                    List<Index> indexes = ((OlapTable) table).getIndexes();
-                    Index found = null;
-                    for (Index idx : indexes) {
-                        if (idx.getIndexName().equalsIgnoreCase(indexName)) {
-                            found = idx;
-                            break;
-                        }
-                    }
-                    if (found == null) {
-                            throw new AnalysisException("index " + indexName + " does not exist");
-                        }
-                }
-            } else if ((alterClause instanceof AddRollupClause)
-                    && !hasSchemaChange && !hasDropRollup
-                    && !hasPartition && !hasRename && !hasModifyProp) {
-                hasAddMaterializedView = true;
-            } else if (alterClause instanceof DropRollupClause && !hasSchemaChange && !hasAddMaterializedView
-                    && !hasPartition && !hasRename && !hasModifyProp) {
-                hasDropRollup = true;
-            } else if (alterClause instanceof AddPartitionClause && !hasSchemaChange && !hasAddMaterializedView
-                    && !hasDropRollup && !hasPartition && !hasRename && !hasModifyProp) {
-                hasPartition = true;
-            } else if (alterClause instanceof DropPartitionClause && !hasSchemaChange && !hasAddMaterializedView && !hasDropRollup
-                    && !hasPartition && !hasRename && !hasModifyProp) {
-                hasPartition = true;
-            } else if (alterClause instanceof ModifyPartitionClause && !hasSchemaChange && !hasAddMaterializedView
-                    && !hasDropRollup && !hasPartition && !hasRename && !hasModifyProp) {
-                hasPartition = true;
-            } else if ((alterClause instanceof TableRenameClause || alterClause instanceof RollupRenameClause
-                    || alterClause instanceof PartitionRenameClause || alterClause instanceof ColumnRenameClause)
-                    && !hasSchemaChange && !hasAddMaterializedView && !hasDropRollup && !hasPartition && !hasRename
-                    && !hasModifyProp) {
-                hasRename = true;
-            } else if (alterClause instanceof ReplacePartitionClause && !hasSchemaChange && !hasAddMaterializedView
-                    && !hasDropRollup && !hasPartition && !hasRename && !hasModifyProp) {
-                hasPartition = true;
-            } else if (alterClause instanceof ModifyTablePropertiesClause && !hasSchemaChange && !hasAddMaterializedView
-                    && !hasDropRollup && !hasPartition && !hasRename && !hasModifyProp) {
-                Map<String, String> properties = alterClause.getProperties();
-                if (properties.containsKey(PropertyAnalyzer.PROPERTIES_INMEMORY)) {
-                    needSynchronized = true;
-                } else  {
-                    hasModifyProp = true;
-                }
-            } else {
-                throw new DdlException("Conflicting alter clauses. see help for more information");
-            }
-        } // end for alter clauses
-
+        // some operations will take long time to process, need to be done outside the databse lock
+        boolean needProcessOutsideDatabaseLock = false;
         String tableName = dbTableName.getTbl();
         db.writeLock();
         try {
@@ -313,44 +177,29 @@ public class Alter {
             if (table.getType() != TableType.OLAP) {
                 throw new DdlException("Do not support alter non-OLAP table[" + tableName + "]");
             }
-
             OlapTable olapTable = (OlapTable) table;
 
-            if (olapTable.getPartitions().size() == 0 && !hasPartition) {
-                throw new DdlException("table with empty parition cannot do schema change. [" + tableName + "]");
+            if (olapTable.getPartitions().size() == 0 && !currentAlterOps.hasPartitionOp()) {
+                throw new DdlException("Table with empty parition cannot do schema change. [" + tableName + "]");
             }
 
             if (olapTable.getState() != OlapTableState.NORMAL) {
-                throw new DdlException("Table[" + table.getName() + "]'s state is not NORMAL. Do not allow doing ALTER ops");
+                throw new DdlException(
+                        "Table[" + table.getName() + "]'s state is not NORMAL. Do not allow doing ALTER ops");
             }
 
-            // schema change job will wait until table become stable
-            if (needTableStable && !hasSchemaChange && !hasAddMaterializedView) {
-                // check if all tablets are healthy, and no tablet is in tablet scheduler
-                boolean isStable = olapTable.isStable(Catalog.getCurrentSystemInfo(),
-                        Catalog.getCurrentCatalog().getTabletScheduler(),
-                        db.getClusterName());
-                if (!isStable) {
-                    throw new DdlException("table [" + olapTable.getName() + "] is not stable."
-                            + " Some tablets of this table may not be healthy or are being scheduled."
-                            + " You need to repair the table first"
-                            + " or stop cluster balance. See 'help admin;'.");
-                }
-            }
-
-            if (hasSchemaChange || hasModifyProp) {
+            if (currentAlterOps.hasSchemaChangeOp()) {
                 // if modify storage type to v2, do schema change to convert all related tablets to segment v2 format
                 schemaChangeHandler.process(alterClauses, clusterName, db, olapTable);
-            } else if (hasAddMaterializedView || hasDropRollup) {
+            } else if (currentAlterOps.hasRollupOp()) {
                 materializedViewHandler.process(alterClauses, clusterName, db, olapTable);
-            } else if (hasPartition) {
+            } else if (currentAlterOps.hasPartitionOp()) {
                 Preconditions.checkState(alterClauses.size() == 1);
-                // when this is a dynamic partition table, do not allow doing partition operation.
-                // TODO(cmy): although some of operation can be done with dynamic partition,
-                // but currently we check it strictly to avoid some unexpected exception.
-                DynamicPartitionUtil.checkAlterAllowed(olapTable);
                 AlterClause alterClause = alterClauses.get(0);
                 if (alterClause instanceof DropPartitionClause) {
+                    if (!((DropPartitionClause) alterClause).isTempPartition()) {
+                        DynamicPartitionUtil.checkAlterAllowed((OlapTable) db.getTable(tableName));
+                    }
                     Catalog.getInstance().dropPartition(db, olapTable, ((DropPartitionClause) alterClause));
                 } else if (alterClause instanceof ReplacePartitionClause) {
                     Catalog.getCurrentCatalog().replaceTempPartition(db, tableName, (ReplacePartitionClause) alterClause);
@@ -358,45 +207,50 @@ public class Alter {
                     ModifyPartitionClause clause = ((ModifyPartitionClause) alterClause);
                     Map<String, String> properties = clause.getProperties();
                     if (properties.containsKey(PropertyAnalyzer.PROPERTIES_INMEMORY)) {
-                        needSynchronized = true;
+                        needProcessOutsideDatabaseLock = true;
                     } else {
                         String partitionName = clause.getPartitionName();
-                        Catalog.getInstance().modifyPartition(db, olapTable, partitionName, properties);
+                        Catalog.getInstance().modifyPartitionProperty(db, olapTable, partitionName, properties);
                     }
+                } else if (alterClause instanceof AddPartitionClause) {
+                    needProcessOutsideDatabaseLock = true;
                 } else {
-                    // add (temp) partition
-                    needSynchronized = true;
+                    throw new DdlException("Invalid alter opertion: " + alterClause.getOpType());
                 }
-            } else if (hasRename) {
+            } else if (currentAlterOps.hasRenameOp()) {
                 processRename(db, olapTable, alterClauses);
+            } else if (currentAlterOps.contains(AlterOpType.MODIFY_TABLE_PROPERTY_SYNC)) {
+                needProcessOutsideDatabaseLock = true;
+            } else {
+                throw new DdlException("Invalid alter operations: " + currentAlterOps);
             }
         } finally {
             db.writeUnlock();
         }
 
         // the following ops should done outside db lock. because it contain synchronized create operation
-        if (needSynchronized) {
+        if (needProcessOutsideDatabaseLock) {
             Preconditions.checkState(alterClauses.size() == 1);
             AlterClause alterClause = alterClauses.get(0);
             if (alterClause instanceof AddPartitionClause) {
-                DynamicPartitionUtil.checkAlterAllowed((OlapTable) db.getTable(tableName));
-                Catalog.getInstance().addPartition(db, tableName, (AddPartitionClause) alterClause);
+                if (!((AddPartitionClause) alterClause).isTempPartition()) {
+                    DynamicPartitionUtil.checkAlterAllowed((OlapTable) db.getTable(tableName));
+                }
+                Catalog.getCurrentCatalog().addPartition(db, tableName, (AddPartitionClause) alterClause);
             } else if (alterClause instanceof ModifyPartitionClause) {
                 ModifyPartitionClause clause = ((ModifyPartitionClause) alterClause);
                 Map<String, String> properties = clause.getProperties();
                 String partitionName = clause.getPartitionName();
                 // currently, only in memory property could reach here
                 Preconditions.checkState(properties.containsKey(PropertyAnalyzer.PROPERTIES_INMEMORY));
-                if (properties.containsKey(PropertyAnalyzer.PROPERTIES_INMEMORY)) {
-                    boolean isInMemory = Boolean.parseBoolean(properties.get(PropertyAnalyzer.PROPERTIES_INMEMORY));
-                    ((SchemaChangeHandler)schemaChangeHandler).updatePartitionInMemoryMeta(
-                            db, tableName, partitionName, isInMemory);
-                }
+                boolean isInMemory = Boolean.parseBoolean(properties.get(PropertyAnalyzer.PROPERTIES_INMEMORY));
+                ((SchemaChangeHandler) schemaChangeHandler).updatePartitionInMemoryMeta(
+                        db, tableName, partitionName, isInMemory);
 
                 db.writeLock();
                 try {
-                    OlapTable olapTable = (OlapTable)db.getTable(tableName);
-                    Catalog.getInstance().modifyPartition(db, olapTable, partitionName, properties);
+                    OlapTable olapTable = (OlapTable) db.getTable(tableName);
+                    Catalog.getCurrentCatalog().modifyPartitionProperty(db, olapTable, partitionName, properties);
                 } finally {
                     db.writeUnlock();
                 }
@@ -404,9 +258,9 @@ public class Alter {
                 Map<String, String> properties = alterClause.getProperties();
                 // currently, only in memory property could reach here
                 Preconditions.checkState(properties.containsKey(PropertyAnalyzer.PROPERTIES_INMEMORY));
-                ((SchemaChangeHandler)schemaChangeHandler).updateTableInMemoryMeta(db, tableName, properties);
+                ((SchemaChangeHandler) schemaChangeHandler).updateTableInMemoryMeta(db, tableName, properties);
             } else {
-                Preconditions.checkState(false);
+                throw new DdlException("Invalid alter opertion: " + alterClause.getOpType());
             }
         }
     }

fe/src/main/java/org/apache/doris/alter/AlterOpType.java

@@ -0,0 +1,62 @@
+// 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.alter;
+
+public enum AlterOpType {
+    // rollup
+    ADD_ROLLUP,
+    DROP_ROLLUP,
+    // schema change
+    SCHEMA_CHANGE,
+    // partition
+    ADD_PARTITION,
+    DROP_PARTITION,
+    REPLACE_PARTITION,
+    MODIFY_PARTITION,
+    // rename
+    RENAME,
+    // table property
+    MODIFY_TABLE_PROPERTY,
+    MODIFY_TABLE_PROPERTY_SYNC, // Some operations are performed synchronously, so we distinguish them by suffix _SYNC
+    // others operation, such as add/drop backend. currently we do not care about them
+    ALTER_OTHER,
+
+    INVALID_OP;
+
+    // true means 2 operations have no conflict.
+    public static Boolean[][] COMPATIBITLITY_MATRIX;
+    static {
+        COMPATIBITLITY_MATRIX = new Boolean[INVALID_OP.ordinal() + 1][INVALID_OP.ordinal() + 1];
+        for (int i = 0; i < INVALID_OP.ordinal(); i++) {
+            for (int j = 0; j < INVALID_OP.ordinal(); j++) {
+                COMPATIBITLITY_MATRIX[i][j] = false;
+            }
+        }
+
+        // rollup can be added or dropped in batch
+        COMPATIBITLITY_MATRIX[ADD_ROLLUP.ordinal()][ADD_ROLLUP.ordinal()] = true;
+        COMPATIBITLITY_MATRIX[DROP_ROLLUP.ordinal()][DROP_ROLLUP.ordinal()] = true;
+        // schema change, such as add/modify/drop columns can be processed in batch
+        COMPATIBITLITY_MATRIX[SCHEMA_CHANGE.ordinal()][SCHEMA_CHANGE.ordinal()] = true;
+    }
+
+    public boolean needCheckCapacity() {
+        return this == ADD_ROLLUP || this == SCHEMA_CHANGE || this == ADD_PARTITION;
+    }
+
+}

fe/src/main/java/org/apache/doris/alter/AlterOperations.java

@@ -0,0 +1,103 @@
+// 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.alter;
+
+import org.apache.doris.analysis.AlterClause;
+import org.apache.doris.common.DdlException;
+
+import com.google.common.base.Joiner;
+import com.google.common.collect.Sets;
+
+import java.util.List;
+import java.util.Set;
+
+/*
+ * AlterOperations contains a set alter operations generated from a AlterStmt's alter clause.
+ * This class is mainly used to integrate these operation types and check whether they have conflicts.
+ */
+public class AlterOperations {
+    private Set<AlterOpType> currentOps = Sets.newHashSet();
+    
+    public AlterOperations() {
+    }
+
+    public Set<AlterOpType> getCurrentOps() {
+        return currentOps;
+    }
+
+    // check the conflicts of the given list of alter clauses
+    public void checkConflict(List<AlterClause> alterClauses) throws DdlException {
+        for (AlterClause alterClause : alterClauses) {
+            checkOp(alterClause.getOpType());
+        }
+    }
+
+    // some operations take up disk space. so we need to check the disk capacity before processing.
+    // return true if we see these kind of opertions.
+    public boolean needCheckCapacity() {
+        for (AlterOpType currentOp : currentOps) {
+            if (currentOp.needCheckCapacity()) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    public boolean hasPartitionOp() {
+        return currentOps.contains(AlterOpType.ADD_PARTITION) || currentOps.contains(AlterOpType.DROP_PARTITION)
+                || currentOps.contains(AlterOpType.REPLACE_PARTITION) || currentOps.contains(AlterOpType.MODIFY_PARTITION);
+    }
+
+    // MODIFY_TABLE_PROPERTY is also processed by SchemaChangeHandler
+    public boolean hasSchemaChangeOp() {
+        return currentOps.contains(AlterOpType.SCHEMA_CHANGE) || currentOps.contains(AlterOpType.MODIFY_TABLE_PROPERTY);
+    }
+
+    public boolean hasRollupOp() {
+        return currentOps.contains(AlterOpType.ADD_ROLLUP) || currentOps.contains(AlterOpType.DROP_ROLLUP);
+    }
+
+    public boolean hasRenameOp() {
+        return currentOps.contains(AlterOpType.RENAME);
+    }
+
+    public boolean contains(AlterOpType op) {
+        return currentOps.contains(op);
+    }
+
+    // throw exception if the given operation has conflict with current operations.,
+    private void checkOp(AlterOpType opType) throws DdlException {
+        if (currentOps.isEmpty()) {
+            currentOps.add(opType);
+            return;
+        }
+
+        for (AlterOpType currentOp : currentOps) {
+            if (!AlterOpType.COMPATIBITLITY_MATRIX[currentOp.ordinal()][opType.ordinal()]) {
+                throw new DdlException("Alter operation " + opType + " conflicts with operation " + currentOp);
+            }
+        }
+
+        currentOps.add(opType);
+    }
+
+    @Override
+    public String toString() {
+        return Joiner.on(", ").join(currentOps);
+    }
+}

fe/src/main/java/org/apache/doris/alter/SchemaChangeHandler.java

@@ -27,6 +27,7 @@ import org.apache.doris.analysis.ColumnPosition;
 import org.apache.doris.analysis.CreateIndexClause;
 import org.apache.doris.analysis.DropColumnClause;
 import org.apache.doris.analysis.DropIndexClause;
+import org.apache.doris.analysis.IndexDef;
 import org.apache.doris.analysis.ModifyColumnClause;
 import org.apache.doris.analysis.ModifyTablePropertiesClause;
 import org.apache.doris.analysis.ReorderColumnsClause;
@@ -1324,10 +1325,6 @@ public class SchemaChangeHandler extends AlterHandler {
     public void process(List<AlterClause> alterClauses, String clusterName, Database db, OlapTable olapTable)
             throws UserException {
 
-        if (olapTable.existTempPartitions()) {
-            throw new DdlException("Can not alter table when there are temp partitions in table");
-        }
-
         // index id -> index schema
         Map<Long, LinkedList<Column>> indexSchemaMap = new HashMap<>();
         for (Map.Entry<Long, List<Column>> entry : olapTable.getIndexIdToSchema().entrySet()) {
@@ -1369,6 +1366,11 @@ public class SchemaChangeHandler extends AlterHandler {
                 }
             }
 
+            // the following operations can not be done when there are temp partitions exist.
+            if (olapTable.existTempPartitions()) {
+                throw new DdlException("Can not alter table when there are temp partitions in table");
+            }
+
             if (alterClause instanceof AddColumnClause) {
                 // add column
                 processAddColumn((AddColumnClause) alterClause, olapTable, indexSchemaMap);
@@ -1388,9 +1390,9 @@ public class SchemaChangeHandler extends AlterHandler {
                 // modify table properties
                 // do nothing, properties are already in propertyMap
             } else if (alterClause instanceof CreateIndexClause) {
-                processAddIndex((CreateIndexClause) alterClause, newIndexes);
+                processAddIndex((CreateIndexClause) alterClause, olapTable, newIndexes);
             } else if (alterClause instanceof DropIndexClause) {
-                processDropIndex((DropIndexClause) alterClause, newIndexes);
+                processDropIndex((DropIndexClause) alterClause, olapTable, newIndexes);
             } else {
                 Preconditions.checkState(false);
             }
@@ -1590,13 +1592,54 @@ public class SchemaChangeHandler extends AlterHandler {
         }
     }
 
-    private void processAddIndex(CreateIndexClause alterClause, List<Index> indexes) {
-        if (alterClause.getIndex() != null) {
-            indexes.add(alterClause.getIndex());
+    private void processAddIndex(CreateIndexClause alterClause, OlapTable olapTable, List<Index> newIndexes)
+            throws UserException {
+        if (alterClause.getIndex() == null) {
+            return;
         }
+
+        List<Index> existedIndexes = olapTable.getIndexes();
+        IndexDef indexDef = alterClause.getIndexDef();
+        Set<String> newColset = Sets.newTreeSet(String.CASE_INSENSITIVE_ORDER);
+        newColset.addAll(indexDef.getColumns());
+        for (Index existedIdx : existedIndexes) {
+            if (existedIdx.getIndexName().equalsIgnoreCase(indexDef.getIndexName())) {
+                throw new DdlException("index `" + indexDef.getIndexName() + "` already exist.");
+            }
+            Set<String> existedIdxColSet = Sets.newTreeSet(String.CASE_INSENSITIVE_ORDER);
+            existedIdxColSet.addAll(existedIdx.getColumns());
+            if (newColset.equals(existedIdxColSet)) {
+                throw new DdlException(
+                        "index for columns (" + String.join(",", indexDef.getColumns()) + " ) already exist.");
+            }
+        }
+
+        for (String col : indexDef.getColumns()) {
+            Column column = olapTable.getColumn(col);
+            if (column != null) {
+                indexDef.checkColumn(column, olapTable.getKeysType());
+            } else {
+                throw new DdlException("BITMAP column does not exist in table. invalid column: " + col);
+            }
+        }
+
+        newIndexes.add(alterClause.getIndex());
     }
 
-    private void processDropIndex(DropIndexClause alterClause, List<Index> indexes) {
+    private void processDropIndex(DropIndexClause alterClause, OlapTable olapTable, List<Index> indexes) throws DdlException {
+        String indexName = alterClause.getIndexName();
+        List<Index> existedIndexes = olapTable.getIndexes();
+        Index found = null;
+        for (Index existedIdx : existedIndexes) {
+            if (existedIdx.getIndexName().equalsIgnoreCase(indexName)) {
+                found = existedIdx;
+                break;
+            }
+        }
+        if (found == null) {
+            throw new DdlException("index " + indexName + " does not exist");
+        }
+
         Iterator<Index> itr = indexes.iterator();
         while (itr.hasNext()) {
             Index idx  = itr.next();

fe/src/main/java/org/apache/doris/analysis/AddColumnClause.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris.analysis;
 
+import org.apache.doris.alter.AlterOpType;
 import org.apache.doris.catalog.Column;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.ErrorCode;
@@ -54,6 +55,7 @@ public class AddColumnClause extends AlterTableClause {
 
     public AddColumnClause(ColumnDef columnDef, ColumnPosition colPos, String rollupName,
                            Map<String, String> properties) {
+        super(AlterOpType.SCHEMA_CHANGE);
         this.columnDef = columnDef;
         this.colPos = colPos;
         this.rollupName = rollupName;

fe/src/main/java/org/apache/doris/analysis/AddColumnsClause.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris.analysis;
 
+import org.apache.doris.alter.AlterOpType;
 import org.apache.doris.catalog.Column;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.ErrorCode;
@@ -46,6 +47,7 @@ public class AddColumnsClause extends AlterTableClause {
     }
 
     public AddColumnsClause(List<ColumnDef> columnDefs, String rollupName, Map<String, String> properties) {
+        super(AlterOpType.SCHEMA_CHANGE);
         this.columnDefs = columnDefs;
         this.rollupName = rollupName;
         this.properties = properties;

fe/src/main/java/org/apache/doris/analysis/AddPartitionClause.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris.analysis;
 
+import org.apache.doris.alter.AlterOpType;
 import org.apache.doris.common.AnalysisException;
 
 import java.util.Map;
@@ -46,11 +47,12 @@ public class AddPartitionClause extends AlterTableClause {
                               DistributionDesc distributionDesc,
                               Map<String, String> properties,
                               boolean isTempPartition) {
+        super(AlterOpType.ADD_PARTITION);
         this.partitionDesc = partitionDesc;
         this.distributionDesc = distributionDesc;
         this.properties = properties;
         this.isTempPartition = isTempPartition;
-        
+
         this.needTableStable = false;
     }
 

fe/src/main/java/org/apache/doris/analysis/AddRollupClause.java

@@ -17,14 +17,13 @@
 
 package org.apache.doris.analysis;
 
+import org.apache.doris.alter.AlterOpType;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.ErrorCode;
 import org.apache.doris.common.ErrorReport;
 import org.apache.doris.common.FeNameFormat;
 
 import com.google.common.base.Strings;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
 
 import java.util.List;
@@ -43,11 +42,6 @@ public class AddRollupClause extends AlterTableClause {
 
     private Map<String, String> properties;
 
-    public AddRollupClause() {
-        columnNames = Lists.newArrayList();
-        properties = Maps.newHashMap();
-    }
-
     public String getRollupName() {
         return rollupName;
     }
@@ -67,6 +61,7 @@ public class AddRollupClause extends AlterTableClause {
     public AddRollupClause(String rollupName, List<String> columnNames,
                            List<String> dupKeys, String baseRollupName,
                            Map<String, String> properties) {
+        super(AlterOpType.ADD_ROLLUP);
         this.rollupName = rollupName;
         this.columnNames = columnNames;
         this.dupKeys = dupKeys;

fe/src/main/java/org/apache/doris/analysis/AlterClause.java

@@ -17,6 +17,8 @@
 
 package org.apache.doris.analysis;
 
+import org.apache.doris.alter.AlterOpType;
+
 import org.apache.commons.lang.NotImplementedException;
 
 import java.util.Map;
@@ -24,7 +26,17 @@ import java.util.Map;
 // Alter clause.
 public abstract class AlterClause implements ParseNode {
 
+    protected AlterOpType opType;
+
+    public AlterClause(AlterOpType opType) {
+        this.opType = opType;
+    }
+
     public Map<String, String> getProperties() {
         throw new NotImplementedException();
     }
+
+    public AlterOpType getOpType() {
+        return opType;
+    }
 }

fe/src/main/java/org/apache/doris/analysis/AlterClusterClause.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris.analysis;
 
+import org.apache.doris.alter.AlterOpType;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.Config;
 import org.apache.doris.common.ErrorCode;
@@ -34,6 +35,7 @@ public class AlterClusterClause extends AlterClause {
     private String password;
 
     public AlterClusterClause(AlterClusterType type, Map<String, String> properties) {
+        super(AlterOpType.ALTER_OTHER);
         this.type = type;
         this.properties = properties;
         instanceNum = 0;

fe/src/main/java/org/apache/doris/analysis/AlterLoadErrorUrlClause.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris.analysis;
 
+import org.apache.doris.alter.AlterOpType;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.util.PrintableMap;
 import org.apache.doris.load.LoadErrorHub;
@@ -38,6 +39,7 @@ public class AlterLoadErrorUrlClause extends AlterClause {
     private LoadErrorHub.Param param;
 
     public AlterLoadErrorUrlClause(Map<String, String> properties) {
+        super(AlterOpType.ALTER_OTHER);
         this.properties = properties;
     }
 

fe/src/main/java/org/apache/doris/analysis/AlterTableClause.java

@@ -17,8 +17,15 @@
 
 package org.apache.doris.analysis;
 
+import org.apache.doris.alter.AlterOpType;
+
 // alter table clause
 public abstract class AlterTableClause extends AlterClause {
+
+    public AlterTableClause(AlterOpType opType) {
+        super(opType);
+    }
+
     // if set to true, the corresponding table should be stable before processing this operation on it.
     protected boolean needTableStable = true;
 

fe/src/main/java/org/apache/doris/analysis/AlterUserClause.java

@@ -17,11 +17,9 @@
 
 package org.apache.doris.analysis;
 
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
+import org.apache.doris.alter.AlterOpType;
 import org.apache.doris.common.AnalysisException;
+
 import com.google.common.base.Strings;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
@@ -30,6 +28,10 @@ import org.apache.commons.lang.NotImplementedException;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
 public class AlterUserClause extends AlterClause {
     private static final Logger LOG = LogManager.getLogger(AlterUserClause.class);
     private List<String> hostOrIps;
@@ -40,6 +42,7 @@ public class AlterUserClause extends AlterClause {
     private AlterUserType type;
     
     public AlterUserClause(AlterUserType type, List<String> hostOrIps) {
+        super(AlterOpType.ALTER_OTHER);
         this.type = type;
         this.hostOrIps = hostOrIps;
         this.ips = Lists.newArrayList();

fe/src/main/java/org/apache/doris/analysis/BackendClause.java

@@ -17,9 +17,11 @@
 
 package org.apache.doris.analysis;
 
+import org.apache.doris.alter.AlterOpType;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.Pair;
 import org.apache.doris.system.SystemInfoService;
+
 import com.google.common.base.Preconditions;
 
 import org.apache.commons.lang.NotImplementedException;
@@ -34,6 +36,7 @@ public class BackendClause extends AlterClause {
     protected List<Pair<String, Integer>> hostPortPairs;
 
     protected BackendClause(List<String> hostPorts) {
+        super(AlterOpType.ALTER_OTHER);
         this.hostPorts = hostPorts;
         this.hostPortPairs = new LinkedList<Pair<String, Integer>>();
     }

fe/src/main/java/org/apache/doris/analysis/ColumnRenameClause.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris.analysis;
 
+import org.apache.doris.alter.AlterOpType;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.FeNameFormat;
 
@@ -30,6 +31,7 @@ public class ColumnRenameClause extends AlterTableClause {
     private String newColName;
 
     public ColumnRenameClause(String colName, String newColName) {
+        super(AlterOpType.SCHEMA_CHANGE);
         this.colName = colName;
         this.newColName = newColName;
         this.needTableStable = false;

fe/src/main/java/org/apache/doris/analysis/CreateIndexClause.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris.analysis;
 
+import org.apache.doris.alter.AlterOpType;
 import org.apache.doris.catalog.Index;
 import org.apache.doris.common.AnalysisException;
 
@@ -36,10 +37,10 @@ public class CreateIndexClause extends AlterTableClause {
     private Index index;
 
     public CreateIndexClause(TableName tableName, IndexDef indexDef, boolean alter) {
+        super(AlterOpType.SCHEMA_CHANGE);
         this.tableName = tableName;
         this.indexDef = indexDef;
         this.alter = alter;
-        this.needTableStable = true;
     }
 
     @Override

fe/src/main/java/org/apache/doris/analysis/DropColumnClause.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris.analysis;
 
+import org.apache.doris.alter.AlterOpType;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.ErrorCode;
 import org.apache.doris.common.ErrorReport;
@@ -41,6 +42,7 @@ public class DropColumnClause extends AlterTableClause {
     }
 
     public DropColumnClause(String colName, String rollupName, Map<String, String> properties) {
+        super(AlterOpType.SCHEMA_CHANGE);
         this.colName = colName;
         this.rollupName = rollupName;
         this.properties = properties;

fe/src/main/java/org/apache/doris/analysis/DropIndexClause.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris.analysis;
 
+import org.apache.doris.alter.AlterOpType;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.UserException;
 
@@ -30,10 +31,10 @@ public class DropIndexClause extends AlterTableClause {
     private boolean alter;
 
     public DropIndexClause(String indexName, TableName tableName, boolean alter) {
+        super(AlterOpType.SCHEMA_CHANGE);
         this.indexName = indexName;
         this.tableName = tableName;
         this.alter = alter;
-        this.needTableStable = true;
     }
 
     public String getIndexName() {

fe/src/main/java/org/apache/doris/analysis/DropPartitionClause.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris.analysis;
 
+import org.apache.doris.alter.AlterOpType;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.ErrorCode;
 import org.apache.doris.common.ErrorReport;
@@ -33,6 +34,7 @@ public class DropPartitionClause extends AlterTableClause {
     private boolean isTempPartition;
 
     public DropPartitionClause(boolean ifExists, String partitionName, boolean isTempPartition) {
+        super(AlterOpType.DROP_PARTITION);
         this.ifExists = ifExists;
         this.partitionName = partitionName;
         this.isTempPartition = isTempPartition;

fe/src/main/java/org/apache/doris/analysis/DropRollupClause.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris.analysis;
 
+import org.apache.doris.alter.AlterOpType;
 import org.apache.doris.common.AnalysisException;
 
 import com.google.common.base.Strings;
@@ -29,6 +30,7 @@ public class DropRollupClause extends AlterTableClause {
     private Map<String, String> properties;
 
     public DropRollupClause(String rollupName, Map<String, String> properties) {
+        super(AlterOpType.DROP_ROLLUP);
         this.rollupName = rollupName;
         this.properties = properties;
         this.needTableStable = false;

fe/src/main/java/org/apache/doris/analysis/FrontendClause.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris.analysis;
 
+import org.apache.doris.alter.AlterOpType;
 import org.apache.doris.catalog.Catalog;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.ErrorCode;
@@ -41,6 +42,7 @@ public class FrontendClause extends AlterClause {
     protected FrontendNodeType role;
 
     protected FrontendClause(String hostPort, FrontendNodeType role) {
+        super(AlterOpType.ALTER_OTHER);
         this.hostPort = hostPort;
         this.role = role;
     }

fe/src/main/java/org/apache/doris/analysis/IndexDef.java

@@ -17,9 +17,6 @@
 
 package org.apache.doris.analysis;
 
-import java.util.List;
-import java.util.TreeSet;
-
 import org.apache.doris.catalog.Column;
 import org.apache.doris.catalog.KeysType;
 import org.apache.doris.catalog.PrimitiveType;
@@ -27,6 +24,9 @@ import org.apache.doris.common.AnalysisException;
 
 import com.google.common.base.Strings;
 
+import java.util.List;
+import java.util.TreeSet;
+
 public class IndexDef {
     private String indexName;
     private List<String> columns;
@@ -136,6 +136,8 @@ public class IndexDef {
                         "BITMAP index only used in columns of DUP_KEYS table or key columns of"
                                 + " UNIQUE_KEYS/AGG_KEYS table. invalid column: " + indexColName);
             }
+        } else {
+            throw new AnalysisException("Unsupported index type: " + indexType);
         }
     }
 

fe/src/main/java/org/apache/doris/analysis/ModifyBrokerClause.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris.analysis;
 
+import org.apache.doris.alter.AlterOpType;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.Pair;
 import org.apache.doris.system.SystemInfoService;
@@ -24,6 +25,7 @@ import org.apache.doris.system.SystemInfoService;
 import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
 import com.google.common.collect.Sets;
+
 import org.apache.commons.lang.NotImplementedException;
 
 import java.util.List;
@@ -44,6 +46,7 @@ public class ModifyBrokerClause extends AlterClause {
     protected Set<Pair<String, Integer>> hostPortPairs;
 
     public ModifyBrokerClause(ModifyOp op, String brokerName, List<String> hostPorts) {
+        super(AlterOpType.ALTER_OTHER);
         this.op = op;
         this.brokerName = brokerName;
         this.hostPorts = hostPorts;

fe/src/main/java/org/apache/doris/analysis/ModifyColumnClause.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris.analysis;
 
+import org.apache.doris.alter.AlterOpType;
 import org.apache.doris.catalog.Column;
 import org.apache.doris.common.AnalysisException;
 
@@ -48,6 +49,7 @@ public class ModifyColumnClause extends AlterTableClause {
 
     public ModifyColumnClause(ColumnDef columnDef, ColumnPosition colPos, String rollup,
                               Map<String, String> properties) {
+        super(AlterOpType.SCHEMA_CHANGE);
         this.columnDef = columnDef;
         this.colPos = colPos;
         this.rollupName = rollup;

fe/src/main/java/org/apache/doris/analysis/ModifyPartitionClause.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris.analysis;
 
+import org.apache.doris.alter.AlterOpType;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.util.PrintableMap;
 
@@ -35,13 +36,14 @@ public class ModifyPartitionClause extends AlterTableClause {
     }
 
     public ModifyPartitionClause(String partitionName, Map<String, String> properties) {
+        super(AlterOpType.MODIFY_PARTITION);
         this.partitionName = partitionName;
         this.properties = properties;
         // ATTN: currently, modify partition only allow 3 kinds of operations:
         // 1. modify replication num
         // 2. modify data property
         // 3. modify in memory
-        // And these 2 operations does not require table to be stable.
+        // And these 3 operations does not require table to be stable.
         // If other kinds of operations be added later, "needTableStable" may be changed.
         this.needTableStable = false;
     }

fe/src/main/java/org/apache/doris/analysis/ModifyTablePropertiesClause.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris.analysis;
 
+import org.apache.doris.alter.AlterOpType;
 import org.apache.doris.catalog.TableProperty;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.Config;
@@ -32,6 +33,7 @@ public class ModifyTablePropertiesClause extends AlterTableClause {
     private Map<String, String> properties;
 
     public ModifyTablePropertiesClause(Map<String, String> properties) {
+        super(AlterOpType.MODIFY_TABLE_PROPERTY);
         this.properties = properties;
     }
 
@@ -87,6 +89,7 @@ public class ModifyTablePropertiesClause extends AlterTableClause {
             properties.remove(defaultReplicationNumName);
         } else if (properties.containsKey(PropertyAnalyzer.PROPERTIES_INMEMORY)) {
             this.needTableStable = false;
+            this.opType = AlterOpType.MODIFY_TABLE_PROPERTY_SYNC;
         } else {
             throw new AnalysisException("Unknown table property: " + properties.keySet());
         }

fe/src/main/java/org/apache/doris/analysis/PartitionRenameClause.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris.analysis;
 
+import org.apache.doris.alter.AlterOpType;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.FeNameFormat;
 
@@ -30,6 +31,7 @@ public class PartitionRenameClause extends AlterTableClause {
     private String newPartitionName;
 
     public PartitionRenameClause(String partitionName, String newPartitionName) {
+        super(AlterOpType.RENAME);
         this.partitionName = partitionName;
         this.newPartitionName = newPartitionName;
         this.needTableStable = false;

fe/src/main/java/org/apache/doris/analysis/ReorderColumnsClause.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris.analysis;
 
+import org.apache.doris.alter.AlterOpType;
 import org.apache.doris.common.AnalysisException;
 
 import com.google.common.base.Strings;
@@ -40,6 +41,7 @@ public class ReorderColumnsClause extends AlterTableClause {
     }
 
     public ReorderColumnsClause(List<String> cols, String rollup, Map<String, String> properties) {
+        super(AlterOpType.SCHEMA_CHANGE);
         this.columnsByPos = cols;
         this.rollupName = rollup;
         this.properties = properties;

fe/src/main/java/org/apache/doris/analysis/ReplacePartitionClause.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris.analysis;
 
+import org.apache.doris.alter.AlterOpType;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.util.PropertyAnalyzer;
 
@@ -53,6 +54,7 @@ public class ReplacePartitionClause extends AlterTableClause {
 
     public ReplacePartitionClause(PartitionNames partitionNames, PartitionNames tempPartitionNames,
             Map<String, String> properties) {
+        super(AlterOpType.REPLACE_PARTITION);
         this.partitionNames = partitionNames;
         this.tempPartitionNames = tempPartitionNames;
         this.needTableStable = false;

fe/src/main/java/org/apache/doris/analysis/RollupRenameClause.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris.analysis;
 
+import org.apache.doris.alter.AlterOpType;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.FeNameFormat;
 
@@ -30,6 +31,7 @@ public class RollupRenameClause extends AlterTableClause {
     private String newRollupName;
 
     public RollupRenameClause(String rollupName, String newRollupName) {
+        super(AlterOpType.RENAME);
         this.rollupName = rollupName;
         this.newRollupName = newRollupName;
         this.needTableStable = false;

fe/src/main/java/org/apache/doris/analysis/TableRenameClause.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris.analysis;
 
+import org.apache.doris.alter.AlterOpType;
 import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.FeNameFormat;
 
@@ -29,6 +30,7 @@ public class TableRenameClause extends AlterTableClause {
     private String newTableName;
 
     public TableRenameClause(String newTableName) {
+        super(AlterOpType.RENAME);
         this.newTableName = newTableName;
         this.needTableStable = false;
     }

fe/src/main/java/org/apache/doris/catalog/Catalog.java

@@ -3229,7 +3229,7 @@ public class Catalog {
         }
     }
 
-    public void modifyPartition(Database db, OlapTable olapTable, String partitionName, Map<String, String> properties)
+    public void modifyPartitionProperty(Database db, OlapTable olapTable, String partitionName, Map<String, String> properties)
             throws DdlException {
         Preconditions.checkArgument(db.isWriteLockHeldByCurrentThread());
         if (olapTable.getState() != OlapTableState.NORMAL) {

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

@@ -84,7 +84,6 @@ public class DynamicPartitionScheduler extends MasterDaemon {
         ERROR
     }
 
-
     public DynamicPartitionScheduler(String name, long intervalMs) {
         super(name, intervalMs);
         this.initialize = false;

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

@@ -210,7 +210,7 @@ public enum ErrorCode {
             "Table %s is not a colocated table"),
     ERR_INVALID_OPERATION(5065, new byte[] { '4', '2', '0', '0', '0' }, "Operation %s is invalid"),
     ERROR_DYNAMIC_PARTITION_TIME_UNIT(5065, new byte[] {'4', '2', '0', '0', '0'},
-            "Unsupported time unit %s. Expect DAY WEEK MONTH."),
+            "Unsupported time unit %s. Expect DAY/WEEK/MONTH."),
     ERROR_DYNAMIC_PARTITION_START_ZERO(5066, new byte[] {'4', '2', '0', '0', '0'},
             "Dynamic partition start must less than 0"),
     ERROR_DYNAMIC_PARTITION_START_FORMAT(5066, new byte[] {'4', '2', '0', '0', '0'},

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

@@ -36,9 +36,10 @@ import org.apache.doris.common.ErrorCode;
 import org.apache.doris.common.ErrorReport;
 import org.apache.doris.common.FeNameFormat;
 
+import com.google.common.base.Strings;
+
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-import com.google.common.base.Strings;
 
 import java.text.ParseException;
 import java.text.SimpleDateFormat;
@@ -145,7 +146,7 @@ public class DynamicPartitionUtil {
                 Strings.isNullOrEmpty(end) &&
                 Strings.isNullOrEmpty(buckets)))) {
             if (Strings.isNullOrEmpty(enable)) {
-                throw new DdlException("Must assign dynamic_partition.enable properties");
+                properties.put(DynamicPartitionProperty.ENABLE, "true");
             }
             if (Strings.isNullOrEmpty(timeUnit)) {
                 throw new DdlException("Must assign dynamic_partition.time_unit properties");
@@ -225,7 +226,7 @@ public class DynamicPartitionUtil {
         if (tableProperty != null && tableProperty.getDynamicPartitionProperty() != null &&
                 tableProperty.getDynamicPartitionProperty().isExist() &&
                 tableProperty.getDynamicPartitionProperty().getEnable()) {
-            throw new DdlException("Cannot modify partition on a Dynamic Partition Table, set `dynamic_partition.enable` to false firstly.");
+            throw new DdlException("Cannot add/drop partition on a Dynamic Partition Table, set `dynamic_partition.enable` to false firstly.");
         }
     }
 

fe/src/test/java/org/apache/doris/alter/AlterTest.java

@@ -0,0 +1,190 @@
+// 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.alter;
+
+import org.apache.doris.analysis.AlterTableStmt;
+import org.apache.doris.analysis.CreateDbStmt;
+import org.apache.doris.analysis.CreateTableStmt;
+import org.apache.doris.catalog.Catalog;
+import org.apache.doris.catalog.Database;
+import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.FeConstants;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.utframe.UtFrameUtils;
+
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.io.File;
+import java.util.Map;
+import java.util.UUID;
+
+public class AlterTest {
+
+    private static String runningDir = "fe/mocked/AlterTest/" + UUID.randomUUID().toString() + "/";
+
+    private static ConnectContext connectContext;
+
+    @BeforeClass
+    public static void beforeClass() throws Exception {
+        FeConstants.runningUnitTest = true;
+        FeConstants.default_scheduler_interval_millisecond = 100;
+        Config.dynamic_partition_enable = true;
+        UtFrameUtils.createMinDorisCluster(runningDir);
+
+        // create connect context
+        connectContext = UtFrameUtils.createDefaultCtx();
+        // create database
+        String createDbStmtStr = "create database test;";
+        CreateDbStmt createDbStmt = (CreateDbStmt) UtFrameUtils.parseAndAnalyzeStmt(createDbStmtStr, connectContext);
+        Catalog.getCurrentCatalog().createDb(createDbStmt);
+
+        createTable("CREATE TABLE test.tbl1\n" + 
+                "(\n" + 
+                "    k1 date,\n" + 
+                "    k2 int,\n" + 
+                "    v1 int sum\n" + 
+                ")\n" + 
+                "PARTITION BY RANGE(k1)\n" + 
+                "(\n" + 
+                "    PARTITION p1 values less than('2020-02-01'),\n" + 
+                "    PARTITION p2 values less than('2020-03-01')\n" + 
+                ")\n" + 
+                "DISTRIBUTED BY HASH(k2) BUCKETS 3\n" + 
+                "PROPERTIES('replication_num' = '1');");
+    }
+
+    @AfterClass
+    public static void tearDown() {
+        File file = new File(runningDir);
+        file.delete();
+    }
+
+    private static void createTable(String sql) throws Exception {
+        CreateTableStmt createTableStmt = (CreateTableStmt) UtFrameUtils.parseAndAnalyzeStmt(sql, connectContext);
+        Catalog.getCurrentCatalog().createTable(createTableStmt);
+    }
+
+    private static void alterTable(String sql, boolean expectedException) throws Exception {
+        AlterTableStmt alterTableStmt = (AlterTableStmt) UtFrameUtils.parseAndAnalyzeStmt(sql, connectContext);
+        try {
+            Catalog.getCurrentCatalog().alterTable(alterTableStmt);
+            if (expectedException) {
+                Assert.fail();
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+            if (!expectedException) {
+                Assert.fail();
+            }
+        }
+    }
+
+    @Test
+    public void testConflictAlterOperations() throws Exception {
+        String stmt = "alter table test.tbl1 add partition p3 values less than('2020-04-01'), add partition p4 values less than('2020-05-01')";
+        alterTable(stmt, true);
+
+        stmt = "alter table test.tbl1 add partition p3 values less than('2020-04-01'), drop partition p4";
+        alterTable(stmt, true);
+
+        stmt = "alter table test.tbl1 drop partition p3, drop partition p4";
+        alterTable(stmt, true);
+
+        stmt = "alter table test.tbl1 drop partition p3, add column k3 int";
+        alterTable(stmt, true);
+
+        // no conflict
+        stmt = "alter table test.tbl1 add column k3 int, add column k4 int";
+        alterTable(stmt, false);
+        waitSchemaChangeJobDone(false);
+        
+        stmt = "alter table test.tbl1 add rollup r1 (k1)";
+        alterTable(stmt, false);
+        waitSchemaChangeJobDone(true);
+
+        stmt = "alter table test.tbl1 add rollup r2 (k1), r3 (k1)";
+        alterTable(stmt, false);
+        waitSchemaChangeJobDone(true);
+        
+        // enable dynamic partition
+        stmt = "alter table test.tbl1 set (\n" + 
+                "'dynamic_partition.enable' = 'true',\n" + 
+                "'dynamic_partition.time_unit' = 'DAY',\n" + 
+                "'dynamic_partition.start' = '-3',\n" + 
+                "'dynamic_partition.end' = '3',\n" + 
+                "'dynamic_partition.prefix' = 'p',\n" + 
+                "'dynamic_partition.buckets' = '3'\n" + 
+                " );";
+        alterTable(stmt, false);
+        Database db = Catalog.getCurrentCatalog().getDb("default_cluster:test");
+        OlapTable tbl = (OlapTable)db.getTable("tbl1");
+        Assert.assertTrue(tbl.getTableProperty().getDynamicPartitionProperty().getEnable());
+        Assert.assertEquals(4, tbl.getIndexIdToSchema().size());
+        
+        // add partition when dynamic partition is enable
+        stmt = "alter table test.tbl1 add partition p3 values less than('2020-04-01') distributed by hash(k2) buckets 4 PROPERTIES ('replication_num' = '1')";
+        alterTable(stmt, true);
+
+        // add temp partition when dynamic partition is enable
+        stmt = "alter table test.tbl1 add temporary partition tp3 values less than('2020-04-01') distributed by hash(k2) buckets 4 PROPERTIES ('replication_num' = '1')";
+        alterTable(stmt, false);
+        Assert.assertEquals(1, tbl.getTempPartitions().size());
+
+        // disable the dynamic partition
+        stmt = "alter table test.tbl1 set ('dynamic_partition.enable' = 'false')";
+        alterTable(stmt, false);
+        Assert.assertFalse(tbl.getTableProperty().getDynamicPartitionProperty().getEnable());
+        
+        // add partition when dynamic partition is disable
+        stmt = "alter table test.tbl1 add partition p3 values less than('2020-04-01') distributed by hash(k2) buckets 4";
+        alterTable(stmt, false);
+
+        // set table's default replication num
+        Assert.assertEquals(Short.valueOf("1"), tbl.getDefaultReplicationNum());
+        stmt = "alter table test.tbl1 set ('default.replication_num' = '3');";
+        alterTable(stmt, false);
+        Assert.assertEquals(Short.valueOf("3"), tbl.getDefaultReplicationNum());
+
+        // add partition without set replication num
+        stmt = "alter table test.tbl1 add partition p4 values less than('2020-05-01')";
+        alterTable(stmt, true);
+
+        // add partition when dynamic partition is disable
+        stmt = "alter table test.tbl1 add partition p4 values less than('2020-05-01') ('replication_num' = '1')";
+        alterTable(stmt, false);
+    }
+
+    private void waitSchemaChangeJobDone(boolean rollupJob) throws InterruptedException {
+        Map<Long, AlterJobV2> alterJobs = Catalog.getCurrentCatalog().getSchemaChangeHandler().getAlterJobsV2();
+        if (rollupJob) {
+            alterJobs = Catalog.getCurrentCatalog().getRollupHandler().getAlterJobsV2();
+        }
+        for (AlterJobV2 alterJobV2 : alterJobs.values()) {
+            while (!alterJobV2.getJobState().isFinalState()) {
+                System.out.println("alter job " + alterJobV2.getJobId() + " is running. state: " + alterJobV2.getJobState());
+                Thread.sleep(1000);
+            }
+            System.out.println(alterJobV2.getType() + " alter job " + alterJobV2.getJobId() + " is done. state: " + alterJobV2.getJobState());
+            Assert.assertEquals(AlterJobV2.JobState.FINISHED, alterJobV2.getJobState());
+        }
+    }
+}

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

@@ -1,10 +1,5 @@
 package org.apache.doris.catalog;
 
-import com.google.common.collect.Lists;
-import mockit.Expectations;
-import mockit.Injectable;
-import mockit.Mock;
-import mockit.MockUp;
 import org.apache.doris.analysis.Analyzer;
 import org.apache.doris.analysis.ColumnDef;
 import org.apache.doris.analysis.CreateTableStmt;
@@ -24,6 +19,9 @@ import org.apache.doris.persist.EditLog;
 import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.system.SystemInfoService;
 import org.apache.doris.task.AgentBatchTask;
+
+import com.google.common.collect.Lists;
+
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
@@ -36,6 +34,11 @@ import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
+import mockit.Expectations;
+import mockit.Injectable;
+import mockit.Mock;
+import mockit.MockUp;
+
 public class DynamicPartitionTableTest {
     private TableName dbTableName;
     private String dbName = "testDb";
@@ -153,53 +156,6 @@ public class DynamicPartitionTableTest {
         catalog.createTable(stmt);
     }
 
-    @Test
-    public void testMissEnable(@Injectable SystemInfoService systemInfoService,
-                               @Injectable PaloAuth paloAuth,
-                               @Injectable EditLog editLog) throws UserException {
-        new Expectations(catalog) {
-            {
-                catalog.getDb(dbTableName.getDb());
-                minTimes = 0;
-                result = db;
-
-                Catalog.getCurrentSystemInfo();
-                minTimes = 0;
-                result = systemInfoService;
-
-                systemInfoService.checkClusterCapacity(anyString);
-                minTimes = 0;
-                systemInfoService.seqChooseBackendIds(anyInt, true, true, anyString);
-                minTimes = 0;
-                result = beIds;
-
-                catalog.getAuth();
-                minTimes = 0;
-                result = paloAuth;
-                paloAuth.checkTblPriv((ConnectContext) any, anyString, anyString, PrivPredicate.CREATE);
-                minTimes = 0;
-                result = true;
-
-                catalog.getEditLog();
-                minTimes = 0;
-                result = editLog;
-            }
-        };
-
-        properties.remove(DynamicPartitionProperty.ENABLE);
-
-        CreateTableStmt stmt = new CreateTableStmt(false, false, dbTableName, columnDefs, "olap",
-                new KeysDesc(KeysType.AGG_KEYS, columnNames),
-                new RangePartitionDesc(Lists.newArrayList("key1"), singleRangePartitionDescs),
-                new HashDistributionDesc(1, Lists.newArrayList("key1")), properties, null, "");
-        stmt.analyze(analyzer);
-
-        expectedEx.expect(DdlException.class);
-        expectedEx.expectMessage("Must assign dynamic_partition.enable properties");
-
-        catalog.createTable(stmt);
-    }
-
     @Test
     public void testMissPrefix(@Injectable SystemInfoService systemInfoService,
                                @Injectable PaloAuth paloAuth,

fe/src/test/java/org/apache/doris/utframe/DemoTest.java

@@ -109,10 +109,10 @@ public class DemoTest {
         Assert.assertEquals(1, alterJobs.size());
         for (AlterJobV2 alterJobV2 : alterJobs.values()) {
             while (!alterJobV2.getJobState().isFinalState()) {
-                System.out.println("alter job " + alterJobV2.getDbId() + " is running. state: " + alterJobV2.getJobState());
+                System.out.println("alter job " + alterJobV2.getJobId() + " is running. state: " + alterJobV2.getJobState());
                 Thread.sleep(1000);
             }
-            System.out.println("alter job " + alterJobV2.getDbId() + " is done. state: " + alterJobV2.getJobState());
+            System.out.println("alter job " + alterJobV2.getJobId() + " is done. state: " + alterJobV2.getJobState());
             Assert.assertEquals(AlterJobV2.JobState.FINISHED, alterJobV2.getJobState());
         }
         db.readLock();