[feature](mtmv)(3)Implementing multi table materialized views

be/src/vec/exec/scan/vmeta_scanner.cpp

@@ -231,6 +231,9 @@ Status VMetaScanner::_fetch_metadata(const TMetaScanRange& meta_scan_range) {
     case TMetadataType::CATALOGS:
         RETURN_IF_ERROR(_build_catalogs_metadata_request(meta_scan_range, &request));
         break;
+    case TMetadataType::MATERIALIZED_VIEWS:
+        RETURN_IF_ERROR(_build_materialized_views_metadata_request(meta_scan_range, &request));
+        break;
     case TMetadataType::QUERIES:
         RETURN_IF_ERROR(_build_queries_metadata_request(meta_scan_range, &request));
         break;
@@ -375,6 +378,27 @@ Status VMetaScanner::_build_catalogs_metadata_request(const TMetaScanRange& meta
     return Status::OK();
 }
 
+Status VMetaScanner::_build_materialized_views_metadata_request(
+        const TMetaScanRange& meta_scan_range, TFetchSchemaTableDataRequest* request) {
+    VLOG_CRITICAL << "VMetaScanner::_build_materialized_views_metadata_request";
+    if (!meta_scan_range.__isset.materialized_views_params) {
+        return Status::InternalError(
+                "Can not find TMaterializedViewsMetadataParams from meta_scan_range.");
+    }
+
+    // create request
+    request->__set_schema_table_name(TSchemaTableName::METADATA_TABLE);
+
+    // create TMetadataTableRequestParams
+    TMetadataTableRequestParams metadata_table_params;
+    metadata_table_params.__set_metadata_type(TMetadataType::MATERIALIZED_VIEWS);
+    metadata_table_params.__set_materialized_views_metadata_params(
+            meta_scan_range.materialized_views_params);
+
+    request->__set_metada_table_params(metadata_table_params);
+    return Status::OK();
+}
+
 Status VMetaScanner::_build_queries_metadata_request(const TMetaScanRange& meta_scan_range,
                                                      TFetchSchemaTableDataRequest* request) {
     VLOG_CRITICAL << "VMetaScanner::_build_queries_metadata_request";

be/src/vec/exec/scan/vmeta_scanner.h

@@ -81,6 +81,8 @@ class VMetaScanner : public VScanner {
                                                    TFetchSchemaTableDataRequest* request);
     Status _build_catalogs_metadata_request(const TMetaScanRange& meta_scan_range,
                                             TFetchSchemaTableDataRequest* request);
+    Status _build_materialized_views_metadata_request(const TMetaScanRange& meta_scan_range,
+                                                      TFetchSchemaTableDataRequest* request);
     Status _build_queries_metadata_request(const TMetaScanRange& meta_scan_range,
                                            TFetchSchemaTableDataRequest* request);
     bool _meta_eos;

fe/fe-common/src/main/java/org/apache/doris/common/Config.java

@@ -1581,15 +1581,16 @@ public class Config extends ConfigBase {
     @ConfField(description = {"任务堆积时用于存放定时任务的队列大小", "The number of timer jobs that can be queued."})
     public static int job_dispatch_timer_job_queue_size = 1024;
 
-    /**
-     * The number of threads used to consume insert tasks.
-     * if you have a lot of insert jobs,and the average execution frequency is relatively high you need to increase
-     * this value or increase the number of {@code @job_insert_task_queue_size}
-     * The value should be greater than 0, if it is 0 or <=0, set it to 5
-     */
-    @ConfField(description = {"用于执行 Insert 任务的线程数", "The number of threads used to consume insert tasks."})
+    @ConfField(description = {"用于执行 Insert 任务的线程数,值应该大于0，否则默认为5",
+            "The number of threads used to consume Insert tasks, "
+                    + "the value should be greater than 0, if it is <=0, default is 5."})
     public static int job_insert_task_consumer_thread_num = 10;
 
+    @ConfField(description = {"用于执行 MTMV 任务的线程数,值应该大于0，否则默认为5",
+            "The number of threads used to consume mtmv tasks, "
+                    + "the value should be greater than 0, if it is <=0, default is 5."})
+    public static int job_mtmv_task_consumer_thread_num = 10;
+
     /*---------------------- JOB CONFIG END------------------------*/
     /**
      * The number of async tasks that can be queued. @See TaskDisruptor

fe/fe-core/src/main/cup/sql_parser.cup

@@ -2595,16 +2595,28 @@ resume_job_stmt ::=
 show_job_stmt ::=
     KW_SHOW KW_JOBS
     {:
-        RESULT = new ShowJobStmt(null,null,null);
+        RESULT = new ShowJobStmt(null,null);
     :}
-    | KW_SHOW  KW_JOB KW_FOR job_label:jobLabel
+    | KW_SHOW KW_MTMV KW_JOBS
+    {:
+        RESULT = new ShowJobStmt(null,org.apache.doris.job.common.JobType.MTMV);
+    :}
+    | KW_SHOW KW_MTMV KW_JOB KW_FOR job_label:jobLabel
     {:
-        RESULT = new ShowJobStmt(null,jobLabel,null);
+        RESULT = new ShowJobStmt(jobLabel,org.apache.doris.job.common.JobType.MTMV);
     :}
+    | KW_SHOW  KW_JOB KW_FOR job_label:jobLabel
+    {:
+        RESULT = new ShowJobStmt(jobLabel,org.apache.doris.job.common.JobType.INSERT);
+    :}    
     | KW_SHOW KW_JOB KW_TASKS KW_FOR job_label:jobLabel
     {:
-        RESULT = new ShowJobTaskStmt(null,jobLabel);
+        RESULT = new ShowJobTaskStmt(jobLabel,org.apache.doris.job.common.JobType.INSERT);
     :}    
+    | KW_SHOW KW_MTMV KW_JOB KW_TASKS KW_FOR job_label:jobLabel
+    {:
+        RESULT = new ShowJobTaskStmt(jobLabel,org.apache.doris.job.common.JobType.MTMV);
+    :}       
     ;
 pause_job_stmt ::=
     KW_PAUSE KW_JOB KW_FOR job_label:jobLabel
@@ -2978,10 +2990,6 @@ drop_stmt ::=
     {:
         RESULT = new AlterTableStmt(tableName, Lists.newArrayList(new DropIndexClause(indexName, ifExists, tableName, false)));
     :}
-    | KW_DROP KW_MATERIALIZED KW_VIEW opt_if_exists:ifExists table_name:mvName
-    {:
-        RESULT = new DropMaterializedViewStmt(ifExists, mvName);
-    :}
     | KW_DROP KW_MATERIALIZED KW_VIEW opt_if_exists:ifExists ident:mvName KW_ON table_name:tableName
     {:
         RESULT = new DropMaterializedViewStmt(ifExists, mvName, tableName);

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

@@ -28,7 +28,6 @@
 import org.apache.doris.analysis.DropMaterializedViewStmt;
 import org.apache.doris.analysis.DropPartitionClause;
 import org.apache.doris.analysis.DropPartitionFromIndexClause;
-import org.apache.doris.analysis.DropTableStmt;
 import org.apache.doris.analysis.ModifyColumnCommentClause;
 import org.apache.doris.analysis.ModifyDistributionClause;
 import org.apache.doris.analysis.ModifyEngineClause;
@@ -45,6 +44,7 @@
 import org.apache.doris.catalog.DataProperty;
 import org.apache.doris.catalog.Database;
 import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.MTMV;
 import org.apache.doris.catalog.MysqlTable;
 import org.apache.doris.catalog.OdbcTable;
 import org.apache.doris.catalog.OlapTable;
@@ -62,6 +62,8 @@
 import org.apache.doris.common.util.DynamicPartitionUtil;
 import org.apache.doris.common.util.MetaLockUtils;
 import org.apache.doris.common.util.PropertyAnalyzer;
+import org.apache.doris.nereids.trees.plans.commands.info.TableNameInfo;
+import org.apache.doris.persist.AlterMTMV;
 import org.apache.doris.persist.AlterViewInfo;
 import org.apache.doris.persist.BatchModifyPartitionsInfo;
 import org.apache.doris.persist.ModifyCommentOperationLog;
@@ -120,26 +122,13 @@ public void processCreateMaterializedView(CreateMaterializedViewStmt stmt)
     }
 
     public void processDropMaterializedView(DropMaterializedViewStmt stmt) throws DdlException, MetaNotFoundException {
-        if (!stmt.isForMTMV() && stmt.getTableName() == null) {
-            throw new DdlException("Drop materialized view without table name is unsupported : " + stmt.toSql());
-        }
-
-        // drop materialized view
-        if (!stmt.isForMTMV()) {
-            TableName tableName = stmt.getTableName();
-
-            // check db
-            String dbName = tableName.getDb();
-            Database db = Env.getCurrentInternalCatalog().getDbOrDdlException(dbName);
+        TableName tableName = stmt.getTableName();
+        String dbName = tableName.getDb();
+        Database db = Env.getCurrentInternalCatalog().getDbOrDdlException(dbName);
 
-            String name = tableName.getTbl();
-            OlapTable olapTable = (OlapTable) db.getTableOrMetaException(name, TableType.OLAP);
-            ((MaterializedViewHandler) materializedViewHandler).processDropMaterializedView(stmt, db, olapTable);
-        } else {
-            DropTableStmt dropTableStmt = new DropTableStmt(stmt.isIfExists(), stmt.getMTMVName(), false);
-            dropTableStmt.setMaterializedView(true);
-            Env.getCurrentInternalCatalog().dropTable(dropTableStmt);
-        }
+        String name = tableName.getTbl();
+        OlapTable olapTable = (OlapTable) db.getTableOrMetaException(name, TableType.OLAP);
+        ((MaterializedViewHandler) materializedViewHandler).processDropMaterializedView(stmt, db, olapTable);
     }
 
     private boolean processAlterOlapTable(AlterTableStmt stmt, OlapTable olapTable, List<AlterClause> alterClauses,
@@ -155,6 +144,10 @@ private boolean processAlterOlapTable(AlterTableStmt stmt, OlapTable olapTable,
         AlterOperations currentAlterOps = new AlterOperations();
         currentAlterOps.checkConflict(alterClauses);
 
+        if (olapTable instanceof MTMV) {
+            currentAlterOps.checkMTMVAllow(alterClauses);
+        }
+
         // check cluster capacity and db quota, only need to check once.
         if (currentAlterOps.needCheckCapacity()) {
             Env.getCurrentSystemInfo().checkAvailableCapacity();
@@ -201,6 +194,8 @@ private boolean processAlterOlapTable(AlterTableStmt stmt, OlapTable olapTable,
         } else if (currentAlterOps.hasSchemaChangeOp()) {
             // if modify storage type to v2, do schema change to convert all related tablets to segment v2 format
             schemaChangeHandler.process(stmt.toSql(), alterClauses, clusterName, db, olapTable);
+            // if base table schemaChanged, need change mtmv status
+            Env.getCurrentEnv().getMtmvService().alterTable(olapTable);
         } else if (currentAlterOps.hasRollupOp()) {
             materializedViewHandler.process(alterClauses, clusterName, db, olapTable);
         } else if (currentAlterOps.hasPartitionOp()) {
@@ -256,6 +251,7 @@ private boolean processAlterOlapTable(AlterTableStmt stmt, OlapTable olapTable,
             }
         } else if (currentAlterOps.hasRenameOp()) {
             processRename(db, olapTable, alterClauses);
+            Env.getCurrentEnv().getMtmvService().alterTable(olapTable);
         } else if (currentAlterOps.hasReplaceTableOp()) {
             processReplaceTable(db, olapTable, alterClauses);
         } else if (currentAlterOps.contains(AlterOpType.MODIFY_TABLE_PROPERTY_SYNC)) {
@@ -877,4 +873,45 @@ public AlterHandler getMaterializedViewHandler() {
     public AlterHandler getClusterHandler() {
         return clusterHandler;
     }
+
+    public void processAlterMTMV(AlterMTMV alterMTMV, boolean isReplay) {
+        TableNameInfo tbl = alterMTMV.getMvName();
+        MTMV mtmv = null;
+        try {
+            Database db = Env.getCurrentInternalCatalog().getDbOrDdlException(tbl.getDb());
+            mtmv = (MTMV) db.getTableOrMetaException(tbl.getTbl(), TableType.MATERIALIZED_VIEW);
+
+            mtmv.writeLock();
+            switch (alterMTMV.getOpType()) {
+                case ALTER_REFRESH_INFO:
+                    mtmv.alterRefreshInfo(alterMTMV.getRefreshInfo());
+                    break;
+                case ALTER_STATUS:
+                    mtmv.alterStatus(alterMTMV.getStatus());
+                    break;
+                case ALTER_PROPERTY:
+                    mtmv.alterMvProperties(alterMTMV.getMvProperties());
+                    break;
+                case ADD_TASK:
+                    mtmv.addTaskResult(alterMTMV.getTask(), alterMTMV.getCache());
+                    Env.getCurrentEnv().getMtmvService()
+                            .refreshComplete(mtmv, alterMTMV.getCache(), alterMTMV.getTask());
+                    break;
+                default:
+                    throw new RuntimeException("Unknown type value: " + alterMTMV.getOpType());
+            }
+            // 4. log it and replay it in the follower
+            if (!isReplay) {
+                Env.getCurrentEnv().getMtmvService().alterMTMV(mtmv, alterMTMV);
+                Env.getCurrentEnv().getEditLog().logAlterMTMV(alterMTMV);
+            }
+        } catch (UserException e) {
+            // if MTMV has been dropped, ignore this exception
+            LOG.warn(e);
+        } finally {
+            if (mtmv != null) {
+                mtmv.writeUnlock();
+            }
+        }
+    }
 }

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

@@ -71,4 +71,9 @@ public boolean needCheckCapacity() {
         return this == ADD_ROLLUP || this == SCHEMA_CHANGE || this == ADD_PARTITION || this == ENABLE_FEATURE;
     }
 
+    public boolean mtmvAllowOp() {
+        return this == MODIFY_TABLE_PROPERTY || this == MODIFY_DISTRIBUTION || this == MODIFY_TABLE_COMMENT
+                || this == ADD_PARTITION || this == DROP_PARTITION || this == REPLACE_PARTITION
+                || this == MODIFY_PARTITION;
+    }
 }

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

@@ -48,6 +48,14 @@ public void checkConflict(List<AlterClause> alterClauses) throws DdlException {
         }
     }
 
+    public void checkMTMVAllow(List<AlterClause> alterClauses) throws DdlException {
+        for (AlterClause alterClause : alterClauses) {
+            if (!alterClause.getOpType().mtmvAllowOp()) {
+                throw new DdlException("Alter operation " + alterClause.getOpType() + " Not allowed to MTMV");
+            }
+        }
+    }
+
     // some operations take up disk space. so we need to check the disk capacity before processing.
     // return true if we see these kind of operations.
     public boolean needCheckCapacity() {

fe/fe-core/src/main/java/org/apache/doris/analysis/CreateJobStmt.java

@@ -83,8 +83,6 @@ public class CreateJobStmt extends DdlStmt {
     private final String comment;
     private JobExecuteType executeType;
 
-    private String timezone = TimeUtils.DEFAULT_TIME_ZONE;
-
     private static final ImmutableSet<Class<? extends DdlStmt>> supportStmtSuperClass
             = new ImmutableSet.Builder<Class<? extends DdlStmt>>().add(InsertStmt.class)
             .add(UpdateStmt.class).build();
@@ -114,7 +112,7 @@ public void analyze(Analyzer analyzer) throws UserException {
         Env.getCurrentInternalCatalog().getDbOrAnalysisException(dbName);
         analyzerSqlStmt();
         // check its insert stmt,currently only support insert stmt
-        //todo used InsertIntoCommand if job is InsertJob
+        //todo when support other stmt,need to check stmt type and generate jobInstance
         InsertJob job = new InsertJob();
         JobExecutionConfiguration jobExecutionConfiguration = new JobExecutionConfiguration();
         jobExecutionConfiguration.setExecuteType(executeType);
@@ -144,10 +142,12 @@ public void analyze(Analyzer analyzer) throws UserException {
         job.setJobName(labelName.getLabelName());
         job.setCreateUser(ConnectContext.get().getCurrentUserIdentity());
         job.setJobStatus(JobStatus.RUNNING);
-        job.checkJobParams();
+        job.setJobId(Env.getCurrentEnv().getNextId());
         String originStmt = getOrigStmt().originStmt;
         String executeSql = parseExecuteSql(originStmt);
         job.setExecuteSql(executeSql);
+
+        //job.checkJobParams();
         jobInstance = job;
     }
 

fe/fe-core/src/main/java/org/apache/doris/analysis/CreateMTMVStmt.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.analysis;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.Index;
+import org.apache.doris.mtmv.EnvInfo;
+import org.apache.doris.mtmv.MTMVRefreshInfo;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+public class CreateMTMVStmt extends CreateTableStmt {
+    private final MTMVRefreshInfo refreshInfo;
+    private final String querySql;
+    private final EnvInfo envInfo;
+    private Map<String, String> mvProperties;
+
+    public CreateMTMVStmt(boolean ifNotExists, TableName mvName, List<Column> columns,
+            MTMVRefreshInfo refreshInfo, KeysDesc keyDesc, DistributionDesc distributionDesc,
+            Map<String, String> properties, Map<String, String> mvProperties, String querySql, String comment,
+            EnvInfo envInfo) {
+        super(ifNotExists, false, mvName, columns, new ArrayList<Index>(), DEFAULT_ENGINE_NAME, keyDesc, null,
+                distributionDesc, properties, null, comment, null, null);
+        this.refreshInfo = refreshInfo;
+        this.querySql = querySql;
+        this.envInfo = envInfo;
+        this.mvProperties = mvProperties;
+    }
+
+    public MTMVRefreshInfo getRefreshInfo() {
+        return refreshInfo;
+    }
+
+    public String getQuerySql() {
+        return querySql;
+    }
+
+    public EnvInfo getEnvInfo() {
+        return envInfo;
+    }
+
+    public Map<String, String> getMvProperties() {
+        return mvProperties;
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/analysis/CreateTableStmt.java

@@ -67,7 +67,7 @@ public class CreateTableStmt extends DdlStmt {
 
     protected static final String DEFAULT_ENGINE_NAME = "olap";
 
-    private boolean ifNotExists;
+    protected boolean ifNotExists;
     private boolean isExternal;
     protected TableName tableName;
     protected List<ColumnDef> columnDefs;
@@ -244,7 +244,7 @@ public CreateTableStmt(boolean ifNotExists,
         this.extProperties = extProperties;
         this.columnDefs = Lists.newArrayList();
         this.comment = Strings.nullToEmpty(comment);
-        this.rollupAlterClauseList = rollupAlterClauseList;
+        this.rollupAlterClauseList = (rollupAlterClauseList == null) ? Lists.newArrayList() : rollupAlterClauseList;
     }
 
     public void addColumnDef(ColumnDef columnDef) {