Fix errors when alter materialized view which based on dup table

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

@@ -129,7 +129,7 @@ private void processAddColumn(AddColumnClause alterClause, OlapTable olapTable,
 
         Set<String> newColNameSet = Sets.newHashSet(column.getName());
         addColumnInternal(olapTable, column, columnPos, targetIndexId, baseIndexId,
-                          baseIndexName, indexSchemaMap, newColNameSet);
+                          indexSchemaMap, newColNameSet);
     }
 
     private void processAddColumns(AddColumnsClause alterClause, OlapTable olapTable,
@@ -154,7 +154,7 @@ private void processAddColumns(AddColumnsClause alterClause, OlapTable olapTable
 
         for (Column column : columns) {
             addColumnInternal(olapTable, column, null, targetIndexId, baseIndexId,
-                    baseIndexName, indexSchemaMap, newColNameSet);
+                    indexSchemaMap, newColNameSet);
         }
     }
 
@@ -531,7 +531,7 @@ private void processReorderColumn(ReorderColumnsClause alterClause, OlapTable ol
      * Modified schema will be saved in 'indexSchemaMap'
      */
     private void addColumnInternal(OlapTable olapTable, Column newColumn, ColumnPosition columnPos,
-                                   long targetIndexId, long baseIndexId, String baseIndexName,
+                                   long targetIndexId, long baseIndexId,
                                    Map<Long, LinkedList<Column>> indexSchemaMap,
                                    Set<String> newColNameSet) throws DdlException {
 
@@ -556,6 +556,10 @@ private void addColumnInternal(OlapTable olapTable, Column newColumn, ColumnPosi
                 throw new DdlException("Can not assign aggregation method on column in Duplicate data model table: " + newColName);
             }
             if (!newColumn.isKey()) {
+                if (targetIndexId != -1L &&
+                        olapTable.getIndexMetaByIndexId(targetIndexId).getKeysType() == KeysType.AGG_KEYS) {
+                    throw new DdlException("Please add non-key column on base table directly");
+                }
                 newColumn.setAggregationType(AggregateType.NONE, true);
             }
         }

fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeJobV2.java

@@ -21,6 +21,7 @@
 import org.apache.doris.catalog.Column;
 import org.apache.doris.catalog.Database;
 import org.apache.doris.catalog.Index;
+import org.apache.doris.catalog.KeysType;
 import org.apache.doris.catalog.MaterializedIndex;
 import org.apache.doris.catalog.MaterializedIndex.IndexState;
 import org.apache.doris.catalog.OlapTable;
@@ -237,7 +238,9 @@ protected void runPendingJob() throws AlterCancelException {
                     short shadowShortKeyColumnCount = indexShortKeyMap.get(shadowIdxId);
                     List<Column> shadowSchema = indexSchemaMap.get(shadowIdxId);
                     int shadowSchemaHash = indexSchemaVersionAndHashMap.get(shadowIdxId).schemaHash;
-                    int originSchemaHash = tbl.getSchemaHashByIndexId(indexIdMap.get(shadowIdxId));
+                    long originIndexId = indexIdMap.get(shadowIdxId);
+                    int originSchemaHash = tbl.getSchemaHashByIndexId(originIndexId);
+                    KeysType originKeysType = tbl.getKeysTypeByIndexId(originIndexId);
 
                     for (Tablet shadowTablet : shadowIdx.getTablets()) {
                         long shadowTabletId = shadowTablet.getId();
@@ -249,7 +252,7 @@ protected void runPendingJob() throws AlterCancelException {
                                     backendId, dbId, tableId, partitionId, shadowIdxId, shadowTabletId,
                                     shadowShortKeyColumnCount, shadowSchemaHash,
                                     Partition.PARTITION_INIT_VERSION, Partition.PARTITION_INIT_VERSION_HASH,
-                                    tbl.getKeysType(), TStorageType.COLUMN, storageMedium,
+                                    originKeysType, TStorageType.COLUMN, storageMedium,
                                     shadowSchema, bfColumns, bfFpp, countDownLatch, indexes,
                                     tbl.isInMemory(),
                                     tbl.getPartitionInfo().getTabletType(partitionId));
@@ -338,7 +341,8 @@ private void addShadowIndexToCatalog(OlapTable tbl) {
             tbl.setIndexMeta(shadowIdxId, indexIdToName.get(shadowIdxId), indexSchemaMap.get(shadowIdxId),
                     indexSchemaVersionAndHashMap.get(shadowIdxId).schemaVersion,
                     indexSchemaVersionAndHashMap.get(shadowIdxId).schemaHash,
-                    indexShortKeyMap.get(shadowIdxId), TStorageType.COLUMN, null);
+                    indexShortKeyMap.get(shadowIdxId), TStorageType.COLUMN,
+                             tbl.getKeysTypeByIndexId(indexIdMap.get(shadowIdxId)));
         }
 
         tbl.rebuildFullSchema();

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

@@ -556,7 +556,9 @@ public KeysType getKeysType() {
     }
 
     public KeysType getKeysTypeByIndexId(long indexId) {
-        return indexIdToMeta.get(indexId).getKeysType();
+        MaterializedIndexMeta indexMeta = indexIdToMeta.get(indexId);
+        Preconditions.checkNotNull(indexMeta, "index id:" + indexId + " meta is null");
+         return indexMeta.getKeysType();
     }
 
     public PartitionInfo getPartitionInfo() {

fe/fe-core/src/test/java/org/apache/doris/alter/SchemaChangeHandlerTest.java

@@ -0,0 +1,64 @@
+// 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.ColumnPosition;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.KeysType;
+import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.common.jmockit.Deencapsulation;
+
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import mockit.Expectations;
+import mockit.Injectable;
+
+public class SchemaChangeHandlerTest {
+
+    @Test
+    public void testAddValueColumnOnAggMV(@Injectable OlapTable olapTable, @Injectable Column newColumn,
+                                          @Injectable ColumnPosition columnPosition) {
+        SchemaChangeHandler schemaChangeHandler = new SchemaChangeHandler();
+        new Expectations() {
+            {
+                olapTable.getKeysType();
+                result = KeysType.DUP_KEYS;
+                newColumn.getAggregationType();
+                result = null;
+                olapTable.getIndexMetaByIndexId(2).getKeysType();
+                result = KeysType.AGG_KEYS;
+                newColumn.isKey();
+                result = false;
+            }
+        };
+
+        try {
+            Deencapsulation.invoke(schemaChangeHandler, "addColumnInternal", olapTable, newColumn, columnPosition,
+                                   new Long(2), new Long(1),
+                                   Maps.newHashMap(), Sets.newHashSet());
+            Assert.fail();
+        } catch (Exception e) {
+            System.out.println(e.getMessage());
+        }
+
+    }
+}