[bugfix](schema change) fix multi alter clauses for light schema change. (#2)

SWJTU-ZhangLei · Lchangliang · commit fc455df21d61 · 2022-06-21T19:57:38.000+08:00
diff --git a/fe/fe-core/src/main/java/org/apache/doris/alter/RollupJobV2.java b/fe/fe-core/src/main/java/org/apache/doris/alter/RollupJobV2.java
@@ -381,7 +381,7 @@ protected void runWaitingTxnJob() throws AlterCancelException {
                         AlterReplicaTask rollupTask = new AlterReplicaTask(rollupReplica.getBackendId(), dbId, tableId,
                                 partitionId, rollupIndexId, baseIndexId, rollupTabletId, baseTabletId,
                                 rollupReplica.getId(), rollupSchemaHash, baseSchemaHash, visibleVersion, jobId,
-                                JobType.ROLLUP, defineExprs, descTable);
+                                JobType.ROLLUP, defineExprs, descTable, null);
                         rollupBatchTask.addTask(rollupTask);
                     }
                 }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeHandler.java b/fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeHandler.java
@@ -1547,6 +1547,8 @@ public void process(List<AlterClause> alterClauses, String clusterName, Database
             throws UserException {
         olapTable.writeLockOrDdlException();
         try {
+            //alterClauses can or cannot light schema change
+            boolean ligthSchemaChange = true;
             // index id -> index schema
             Map<Long, List<Column>> indexSchemaMap = new HashMap<>();
             for (Map.Entry<Long, List<Column>> entry : olapTable.getIndexIdToSchema(true).entrySet()) {
@@ -1610,55 +1612,59 @@ public void process(List<AlterClause> alterClauses, String clusterName, Database
 
                 if (alterClause instanceof AddColumnClause) {
                     // add column
-                    boolean ligthSchemaChange = processAddColumn((AddColumnClause) alterClause, olapTable, indexSchemaMap);
-                    LOG.debug("processAddColumn, table: {}({}), getMaxColUniqueId(): {}, ligthSchemaChange: {}", olapTable.getName(), olapTable.getId(), 
-                        olapTable.getMaxColUniqueId(), ligthSchemaChange);
-                    if (ligthSchemaChange) {
-                        //for schema change add column optimize, direct modify table meta.
-                        Catalog.getCurrentCatalog().modifyTableAddOrDropColumns(db, olapTable, indexSchemaMap, newIndexes, false);
-                        return;
+                    boolean clauseCanLigthSchemaChange = processAddColumn((AddColumnClause) alterClause, olapTable, indexSchemaMap);
+                    if (clauseCanLigthSchemaChange == false) {
+                        ligthSchemaChange = false;
                     }
                 } else if (alterClause instanceof AddColumnsClause) {
                     // add columns
-                    boolean ligthSchemaChange = processAddColumns((AddColumnsClause) alterClause, olapTable, indexSchemaMap, false);
-                    LOG.debug("processAddColumns, table: {}({}), getMaxColUniqueId(): {}, ligthSchemaChange: {}", olapTable.getName(), olapTable.getId(),
-                        olapTable.getMaxColUniqueId(), ligthSchemaChange);
-                    if (ligthSchemaChange) {
-                        //for schema change add column optimize, direct modify table meta.
-                        Catalog.getCurrentCatalog().modifyTableAddOrDropColumns(db, olapTable, indexSchemaMap, newIndexes, false);
-                        return;
+                    boolean clauseCanLigthSchemaChange = processAddColumns((AddColumnsClause) alterClause, olapTable, indexSchemaMap, false);
+                    if (clauseCanLigthSchemaChange == false) {
+                        ligthSchemaChange = false;
                     }
                 } else if (alterClause instanceof DropColumnClause) {
                     // drop column and drop indexes on this column
-                    boolean ligthSchemaChange = processDropColumn((DropColumnClause) alterClause, olapTable, indexSchemaMap, newIndexes);
-                    LOG.debug("processDropColumn, table: {}({}), getMaxColUniqueId(): {}", olapTable.getName(), olapTable.getId(), olapTable.getMaxColUniqueId());
-                    if (ligthSchemaChange) {
-                        //for schema change add column optimize, direct modify table meta.
-                        Catalog.getCurrentCatalog().modifyTableAddOrDropColumns(db, olapTable, indexSchemaMap, newIndexes, false);
-                        return;
+                    boolean clauseCanLigthSchemaChange = processDropColumn((DropColumnClause) alterClause, olapTable, indexSchemaMap, newIndexes);
+                    if (clauseCanLigthSchemaChange == false) {
+                        ligthSchemaChange = false;
                     }
                 } else if (alterClause instanceof ModifyColumnClause) {
                     // modify column
                     processModifyColumn((ModifyColumnClause) alterClause, olapTable, indexSchemaMap);
+                    ligthSchemaChange = false;
                 } else if (alterClause instanceof ReorderColumnsClause) {
                     // reorder column
                     processReorderColumn((ReorderColumnsClause) alterClause, olapTable, indexSchemaMap);
+                    ligthSchemaChange = false;
                 } else if (alterClause instanceof ModifyTablePropertiesClause) {
                     // modify table properties
                     // do nothing, properties are already in propertyMap
+                    ligthSchemaChange = false;
                 } else if (alterClause instanceof CreateIndexClause) {
                     if (processAddIndex((CreateIndexClause) alterClause, olapTable, newIndexes)) {
                         return;
                     }
+                    ligthSchemaChange = false;
                 } else if (alterClause instanceof DropIndexClause) {
                     if (processDropIndex((DropIndexClause) alterClause, olapTable, newIndexes)) {
                         return;
                     }
+                    ligthSchemaChange = false;
                 } else {
                     Preconditions.checkState(false);
                 }
-            } // end for alter clausesnnnnnn
-            createJob(db.getId(), olapTable, indexSchemaMap, propertyMap, newIndexes);
+            } // end for alter clauses
+
+            LOG.debug("processAddColumns, table: {}({}), getMaxColUniqueId(): {}, ligthSchemaChange: {}", olapTable.getName(),
+                        olapTable.getId(), olapTable.getMaxColUniqueId(), ligthSchemaChange);
+
+            if (ligthSchemaChange) {
+                //for schema change add/drop value column optimize, direct modify table meta.
+                Catalog.getCurrentCatalog().modifyTableAddOrDropColumns(db, olapTable, indexSchemaMap, newIndexes, false);
+                return;
+            } else {
+                createJob(db.getId(), olapTable, indexSchemaMap, propertyMap, newIndexes);
+            }
         } finally {
             olapTable.writeUnlock();
         }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeJobV2.java b/fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeJobV2.java
@@ -431,7 +431,7 @@ protected void runWaitingTxnJob() throws AlterCancelException {
                     long originIdxId = indexIdMap.get(shadowIdxId);
                     int shadowSchemaHash = indexSchemaVersionAndHashMap.get(shadowIdxId).schemaHash;
                     int originSchemaHash = tbl.getSchemaHashByIndexId(indexIdMap.get(shadowIdxId));
-
+                    List<Column> originSchemaColumns = tbl.getSchemaByIndexId(originIdxId);
                     for (Tablet shadowTablet : shadowIdx.getTablets()) {
                         long shadowTabletId = shadowTablet.getId();
                         long originTabletId = partitionIndexTabletMap.get(partitionId, shadowIdxId).get(shadowTabletId);
@@ -440,7 +440,7 @@ protected void runWaitingTxnJob() throws AlterCancelException {
                             AlterReplicaTask rollupTask = new AlterReplicaTask(shadowReplica.getBackendId(), dbId,
                                     tableId, partitionId, shadowIdxId, originIdxId, shadowTabletId, originTabletId,
                                     shadowReplica.getId(), shadowSchemaHash, originSchemaHash, visibleVersion, jobId,
-                                    JobType.SCHEMA_CHANGE, defineExprs, descTable);
+                                    JobType.SCHEMA_CHANGE, defineExprs, descTable, originSchemaColumns);
                             schemaChangeBatchTask.addTask(rollupTask);
                         }
                     }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/SlotRef.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/SlotRef.java
@@ -71,7 +71,7 @@ public SlotRef(TableName tblName, String col) {
     public SlotRef(SlotDescriptor desc) {
         super();
         this.tblName = null;
-        this.col = null;
+        this.col = desc.getColumn().getName();
         this.desc = desc;
         this.type = desc.getType();
         // TODO(zc): label is meaningful
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapTableSink.java b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapTableSink.java
@@ -189,7 +189,11 @@ private TOlapTableSchemaParam createSchema(long dbId, OlapTable table) {
             List<String> columns = Lists.newArrayList();
             List<TColumn> columns_desc = Lists.newArrayList();
             columns.addAll(indexMeta.getSchema().stream().map(Column::getName).collect(Collectors.toList()));
-            columns_desc.addAll(indexMeta.getSchema().stream().map(Column::toThrift).collect(Collectors.toList()));
+            for (Column column : indexMeta.getSchema()) {
+                TColumn tColumn = column.toThrift();
+                column.setIndexFlag(tColumn, table.getIndexes());
+                columns_desc.add(tColumn);
+            }
             TOlapTableIndexSchema indexSchema = new TOlapTableIndexSchema(pair.getKey(), columns,
                     indexMeta.getSchemaHash(), columns_desc);
             schemaParam.addToIndexes(indexSchema);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/task/AlterReplicaTask.java b/fe/fe-core/src/main/java/org/apache/doris/task/AlterReplicaTask.java
@@ -21,12 +21,15 @@
 import org.apache.doris.analysis.DescriptorTable;
 import org.apache.doris.analysis.Expr;
 import org.apache.doris.analysis.SlotRef;
+import org.apache.doris.catalog.Column;
 import org.apache.doris.thrift.TAlterMaterializedViewParam;
 import org.apache.doris.thrift.TAlterTabletReqV2;
+import org.apache.doris.thrift.TColumn;
 import org.apache.doris.thrift.TTaskType;
 
 import com.google.common.collect.Lists;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 
@@ -48,6 +51,7 @@ public class AlterReplicaTask extends AgentTask {
 
     private Map<String, Expr> defineExprs;
     private DescriptorTable descTable;
+    private List<Column> baseSchemaColumns;
 
     /**
      * AlterReplicaTask constructor.
@@ -56,7 +60,7 @@ public class AlterReplicaTask extends AgentTask {
     public AlterReplicaTask(long backendId, long dbId, long tableId, long partitionId, long rollupIndexId,
             long baseIndexId, long rollupTabletId, long baseTabletId, long newReplicaId, int newSchemaHash,
             int baseSchemaHash, long version, long jobId, AlterJobV2.JobType jobType, Map<String, Expr> defineExprs,
-            DescriptorTable descTable) {
+            DescriptorTable descTable, List<Column> baseSchemaColumns) {
         super(null, backendId, TTaskType.ALTER, dbId, tableId, partitionId, rollupIndexId, rollupTabletId);
 
         this.baseTabletId = baseTabletId;
@@ -71,6 +75,7 @@ public AlterReplicaTask(long backendId, long dbId, long tableId, long partitionI
         this.jobType = jobType;
         this.defineExprs = defineExprs;
         this.descTable = descTable;
+        this.baseSchemaColumns = baseSchemaColumns;
     }
 
     public long getBaseTabletId() {
@@ -115,6 +120,14 @@ public TAlterTabletReqV2 toThrift() {
             }
         }
         req.setDescTbl(descTable.toThrift());
+
+        if (baseSchemaColumns != null) {
+            List<TColumn> columns = new ArrayList<TColumn>();
+            for (Column column : baseSchemaColumns) {
+                columns.add(column.toThrift());
+            }
+            req.setColumns(columns);
+        }
         return req;
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -381,7 +381,7 @@ protected void runWaitingTxnJob() throws AlterCancelException {`
`381`	`381`	`AlterReplicaTask rollupTask = new AlterReplicaTask(rollupReplica.getBackendId(), dbId, tableId,`
`382`	`382`	`partitionId, rollupIndexId, baseIndexId, rollupTabletId, baseTabletId,`
`383`	`383`	`rollupReplica.getId(), rollupSchemaHash, baseSchemaHash, visibleVersion, jobId,`
`384`		`- JobType.ROLLUP, defineExprs, descTable);`
	`384`	`+ JobType.ROLLUP, defineExprs, descTable, null);`
`385`	`385`	`rollupBatchTask.addTask(rollupTask);`
`386`	`386`	`}`
`387`	`387`	`}`