[feature-wip](MTMV) Support table aliases when creating a materialized view with multiple tables (#15849)

## Use Case

mysql> CREATE TABLE t_user (
    ->   event_day DATE,
    ->   id bigint,
    ->   username varchar(20)
    -> )
    -> DISTRIBUTED BY HASH(id) BUCKETS 10
    -> PROPERTIES ('replication_num' = '1');
Query OK, 0 rows affected (0.07 sec)

mysql> CREATE TABLE t_user_pv(
    ->   event_day DATE,
    ->   id bigint,
    ->   pv bigint
    -> )
    -> DISTRIBUTED BY HASH(id) BUCKETS 10
    -> PROPERTIES ('replication_num' = '1');
Query OK, 0 rows affected (0.09 sec)

mysql> CREATE MATERIALIZED VIEW mv
    -> BUILD IMMEDIATE REFRESH COMPLETE
    -> START WITH "2022-10-27 19:35:00"
    -> NEXT 1 SECOND
    -> KEY (username)
    -> DISTRIBUTED BY HASH(username) BUCKETS 10
    -> PROPERTIES ('replication_num' = '1')
    -> AS SELECT t1.username ,t2.pv FROM t_user t1 LEFT JOIN t_user_pv t2 on t1.id = t2.id;
Query OK, 0 rows affected (0.10 sec)

mysql> DESC mv;
+----------+-------------+------+-------+---------+-------+
| Field    | Type        | Null | Key   | Default | Extra |
+----------+-------------+------+-------+---------+-------+
| username | VARCHAR(20) | Yes  | true  | NULL    |       |
| pv       | BIGINT      | Yes  | false | NULL    | NONE  |
+----------+-------------+------+-------+---------+-------+
2 rows in set (0.02 sec)

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

@@ -57,6 +57,7 @@ public class CreateMultiTableMaterializedViewStmt extends CreateTableStmt {
         this.partitionDesc = partitionDesc;
         this.distributionDesc = distributionDesc;
         this.properties = properties;
+        engineName = DEFAULT_ENGINE_NAME;
     }
 
     @Override
@@ -83,12 +84,13 @@ public class CreateMultiTableMaterializedViewStmt extends CreateTableStmt {
             }
             OlapTable table = (OlapTable) database.getTableOrAnalysisException(tableRef.getName().getTbl());
             olapTables.put(table.getName(), table);
+            olapTables.put(tableRef.getAlias(), table);
         }
         columnDefs = generateColumnDefinitions(selectStmt.getSelectList());
     }
 
     private List<ColumnDef> generateColumnDefinitions(SelectList selectList) throws AnalysisException, DdlException {
-        List<MVColumnItem> mvColumnItems = generateMVColumnItems(olapTables, selectList);
+        List<MVColumnItem> mvColumnItems = generateMVColumnItems(selectList);
         List<Column> schema = generateSchema(mvColumnItems);
         return schema.stream()
                 .map(column -> new ColumnDef(
@@ -111,7 +113,7 @@ public class CreateMultiTableMaterializedViewStmt extends CreateTableStmt {
         return columns;
     }
 
-    private List<MVColumnItem> generateMVColumnItems(Map<String, OlapTable> olapTables, SelectList selectList)
+    private List<MVColumnItem> generateMVColumnItems(SelectList selectList)
             throws AnalysisException {
         Map<String, MVColumnItem> uniqueMVColumnItems = Maps.newLinkedHashMap();
         for (SelectListItem item : selectList.getItems()) {
@@ -148,13 +150,13 @@ public class CreateMultiTableMaterializedViewStmt extends CreateTableStmt {
         StringBuilder sb = new StringBuilder();
         sb.append("CREATE MATERIALIZED VIEW ").append(mvName).append(" BUILD ").append(buildMode.toString());
         if (refreshInfo != null) {
-            sb.append(" ").append(refreshInfo.toString());
+            sb.append(" ").append(refreshInfo);
         }
         if (partitionDesc != null) {
-            sb.append(" ").append(partitionDesc.toString());
+            sb.append(" ").append(partitionDesc);
         }
         if (distributionDesc != null) {
-            sb.append(" ").append(distributionDesc.toString());
+            sb.append(" ").append(distributionDesc);
         }
         if (properties != null && !properties.isEmpty()) {
             sb.append("\nPROPERTIES (");

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

@@ -58,7 +58,7 @@ import java.util.stream.Collectors;
 public class CreateTableStmt extends DdlStmt {
     private static final Logger LOG = LogManager.getLogger(CreateTableStmt.class);
 
-    private static final String DEFAULT_ENGINE_NAME = "olap";
+    protected static final String DEFAULT_ENGINE_NAME = "olap";
 
     private boolean ifNotExists;
     private boolean isExternal;
@@ -70,7 +70,7 @@ public class CreateTableStmt extends DdlStmt {
     protected DistributionDesc distributionDesc;
     protected Map<String, String> properties;
     private Map<String, String> extProperties;
-    private String engineName;
+    protected String engineName;
     private String comment;
     private List<AlterClause> rollupAlterClauseList = Lists.newArrayList();
 

fe/fe-core/src/test/java/org/apache/doris/catalog/MultiTableMaterializedViewTest.java

@@ -304,4 +304,33 @@ public class MultiTableMaterializedViewTest extends TestWithFeService {
                 connectContext.getState().getErrorMessage()
                         .contains("The partition columns doesn't match the ones in base table"));
     }
+
+    @Test
+    public void testCreateWithTableAliases() throws Exception {
+        createTable("CREATE TABLE t_user ("
+                + "  event_day DATE,"
+                + "  id bigint,"
+                + "  username varchar(20)"
+                + ")"
+                + "DISTRIBUTED BY HASH(id) BUCKETS 10 "
+                + "PROPERTIES ('replication_num' = '1')"
+        );
+        createTable("CREATE TABLE t_user_pv("
+                + "  event_day DATE,"
+                + "  id bigint,"
+                + "  pv bigint"
+                + ")"
+                + "DISTRIBUTED BY HASH(id) BUCKETS 10 "
+                + "PROPERTIES ('replication_num' = '1')"
+        );
+        new StmtExecutor(connectContext, "CREATE MATERIALIZED VIEW mv "
+                + "BUILD IMMEDIATE REFRESH COMPLETE "
+                + "START WITH \"2022-10-27 19:35:00\" "
+                + "NEXT 1 SECOND "
+                + "KEY (username) "
+                + "DISTRIBUTED BY HASH(username) BUCKETS 10 "
+                + "PROPERTIES ('replication_num' = '1') "
+                + "AS SELECT t1.username ,t2.pv FROM t_user t1 LEFT JOIN t_user_pv t2 on t1.id = t2.id").execute();
+        Assertions.assertNull(connectContext.getState().getErrorCode(), connectContext.getState().getErrorMessage());
+    }
 }