Support create table as select

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

@@ -1236,6 +1236,21 @@ create_stmt ::=
         RESULT = new CreateTableStmt(ifNotExists, isExternal, name, columns, indexes, engineName, keys, partition,
         distribution, tblProperties, extProperties, tableComment, index);
     :}
+    | KW_CREATE opt_external:isExternal KW_TABLE opt_if_not_exists:ifNotExists table_name:name
+            opt_col_list:columns
+            opt_engine:engineName
+            opt_keys:keys
+            opt_comment:tableComment
+            opt_partition:partition
+            opt_distribution:distribution
+            opt_rollup:index
+            opt_properties:tblProperties
+            opt_ext_properties:extProperties
+            KW_AS query_stmt:query_def
+    {:
+        RESULT = new CreateTableAsSelectStmt(new CreateTableStmt(ifNotExists, isExternal, name, null, engineName, keys, partition,
+        distribution, tblProperties, extProperties, tableComment, index), columns, query_def);
+    :}
     /* User */
     | KW_CREATE KW_USER opt_if_not_exists:ifNotExists grant_user:user opt_user_role:userRole
     {:

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

@@ -17,12 +17,11 @@
 
 package org.apache.doris.analysis;
 
-import org.apache.doris.catalog.Table;
-import org.apache.doris.common.AnalysisException;
 import org.apache.doris.common.ErrorCode;
 import org.apache.doris.common.ErrorReport;
 import org.apache.doris.common.UserException;
 
+import java.util.ArrayList;
 import java.util.List;
 
 /**
@@ -31,78 +30,49 @@
  *      CREATE TABLE table_name [( column_name_list )]
  *          opt_engine opt_partition opt_properties KW_AS query_stmt
  */
-public class CreateTableAsSelectStmt extends StatementBase {
+public class CreateTableAsSelectStmt extends DdlStmt {
     private final CreateTableStmt createTableStmt;
     private final List<String> columnNames;
-    private final QueryStmt queryStmt;
-    private final InsertStmt insertStmt;
-
+    private QueryStmt queryStmt;
+
     public CreateTableAsSelectStmt(CreateTableStmt createTableStmt,
-                                   List<String> columnNames,
-                                   QueryStmt queryStmt) {
+                                   List<String> columnNames, QueryStmt queryStmt) {
         this.createTableStmt = createTableStmt;
         this.columnNames = columnNames;
         this.queryStmt = queryStmt;
-        this.insertStmt = new InsertStmt(createTableStmt.getDbTbl(), queryStmt);
+        // Insert is not currently supported
     }
-
+    
     @Override
-    public void analyze(Analyzer analyzer) throws UserException, AnalysisException {
+    public void analyze(Analyzer analyzer) throws UserException {
         // first: we analyze queryStmt before create table.
         // To avoid duplicate registrations of table/colRefs,
         // create a new root analyzer and clone the query statement for this initial pass.
         Analyzer dummyRootAnalyzer = new Analyzer(analyzer.getCatalog(), analyzer.getContext());
         QueryStmt tmpStmt = queryStmt.clone();
         tmpStmt.analyze(dummyRootAnalyzer);
-
-        // TODO(zc): support char, varchar and decimal
-        for (Expr expr : tmpStmt.getResultExprs()) {
-            if (expr.getType().isDecimalV2() || expr.getType().isStringType()) {
+        this.queryStmt = tmpStmt;
+        ArrayList<Expr> resultExprs = getQueryStmt().getResultExprs();
+        // TODO: support decimal
+        for (Expr expr : resultExprs) {
+            if (expr.getType().isDecimalV2()) {
                 ErrorReport.reportAnalysisException(ErrorCode.ERR_UNSUPPORTED_TYPE_IN_CTAS, expr.getType());
             }
         }
-
-        // Check columnNames
-        if (columnNames != null) {
-            if (columnNames.size() != tmpStmt.getColLabels().size()) {
-                ErrorReport.report(ErrorCode.ERR_COL_NUMBER_NOT_MATCH);
-            }
-            for (int i = 0; i < columnNames.size(); ++i) {
-                createTableStmt.addColumnDef(new ColumnDef(
-                        columnNames.get(i), new TypeDef(tmpStmt.getResultExprs().get(i).getType())));
-            }
-        } else {
-            for (int i = 0; i < tmpStmt.getColLabels().size(); ++i) {
-                createTableStmt.addColumnDef(new ColumnDef(
-                        tmpStmt.getColLabels().get(i), new TypeDef(tmpStmt.getResultExprs().get(i).getType())));
-            }
+        if (columnNames != null && columnNames.size() != resultExprs.size()) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_COL_NUMBER_NOT_MATCH);
         }
-
-        // Analyze create table statement
-        createTableStmt.analyze(analyzer);
-
-        // Analyze insert
-        Table newTable = null;
-        insertStmt.setTargetTable(newTable);
-        insertStmt.analyze(analyzer);
     }
-
-    public void createTable(Analyzer analyzer) throws AnalysisException {
-        // TODO(zc): Support create table later.
-        // Create table
-        try {
-            analyzer.getCatalog().createTable(createTableStmt);
-        } catch (UserException e) {
-            throw new AnalysisException(e.getMessage());
-        }
+
+    public CreateTableStmt getCreateTableStmt() {
+        return createTableStmt;
     }
-
-    public InsertStmt getInsertStmt() {
-        return insertStmt;
+    
+    public List<String> getColumnNames() {
+        return columnNames;
     }
-
-    @Override
-    public RedirectStatus getRedirectStatus() {
-        return RedirectStatus.FORWARD_WITH_SYNC;
+
+    public QueryStmt getQueryStmt() {
+        return queryStmt;
     }
 }

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

@@ -203,6 +203,10 @@ public DistributionDesc getDistributionDesc() {
         return this.distributionDesc;
     }
 
+    public void setDistributionDesc(DistributionDesc desc) {
+        this.distributionDesc = desc;
+    }
+
     public Map<String, String> getProperties() {
         return this.properties;
     }

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

@@ -40,15 +40,19 @@
 import org.apache.doris.analysis.AlterSystemStmt;
 import org.apache.doris.analysis.AlterTableStmt;
 import org.apache.doris.analysis.AlterViewStmt;
+import org.apache.doris.analysis.Analyzer;
 import org.apache.doris.analysis.BackupStmt;
 import org.apache.doris.analysis.CancelAlterSystemStmt;
 import org.apache.doris.analysis.CancelAlterTableStmt;
 import org.apache.doris.analysis.CancelBackupStmt;
+import org.apache.doris.analysis.ColumnDef;
+import org.apache.doris.analysis.ColumnDef.DefaultValue;
 import org.apache.doris.analysis.ColumnRenameClause;
 import org.apache.doris.analysis.CreateClusterStmt;
 import org.apache.doris.analysis.CreateDbStmt;
 import org.apache.doris.analysis.CreateFunctionStmt;
 import org.apache.doris.analysis.CreateMaterializedViewStmt;
+import org.apache.doris.analysis.CreateTableAsSelectStmt;
 import org.apache.doris.analysis.CreateTableLikeStmt;
 import org.apache.doris.analysis.CreateTableStmt;
 import org.apache.doris.analysis.CreateUserStmt;
@@ -62,14 +66,17 @@
 import org.apache.doris.analysis.DropMaterializedViewStmt;
 import org.apache.doris.analysis.DropPartitionClause;
 import org.apache.doris.analysis.DropTableStmt;
+import org.apache.doris.analysis.Expr;
 import org.apache.doris.analysis.FunctionName;
+import org.apache.doris.analysis.HashDistributionDesc;
 import org.apache.doris.analysis.InstallPluginStmt;
 import org.apache.doris.analysis.KeysDesc;
 import org.apache.doris.analysis.LinkDbStmt;
 import org.apache.doris.analysis.MigrateDbStmt;
 import org.apache.doris.analysis.ModifyDistributionClause;
 import org.apache.doris.analysis.PartitionDesc;
 import org.apache.doris.analysis.PartitionRenameClause;
+import org.apache.doris.analysis.QueryStmt;
 import org.apache.doris.analysis.RecoverDbStmt;
 import org.apache.doris.analysis.RecoverPartitionStmt;
 import org.apache.doris.analysis.RecoverTableStmt;
@@ -82,6 +89,7 @@
 import org.apache.doris.analysis.TableRef;
 import org.apache.doris.analysis.TableRenameClause;
 import org.apache.doris.analysis.TruncateTableStmt;
+import org.apache.doris.analysis.TypeDef;
 import org.apache.doris.analysis.UninstallPluginStmt;
 import org.apache.doris.analysis.UserDesc;
 import org.apache.doris.analysis.UserIdentity;
@@ -113,6 +121,7 @@
 import org.apache.doris.common.ErrorReport;
 import org.apache.doris.common.FeConstants;
 import org.apache.doris.common.FeMetaVersion;
+import org.apache.doris.common.FeNameFormat;
 import org.apache.doris.common.MarkedCountDownLatch;
 import org.apache.doris.common.MetaHeader;
 import org.apache.doris.common.MetaNotFoundException;
@@ -3086,6 +3095,54 @@ public void createTableLike(CreateTableLikeStmt stmt) throws DdlException {
             throw new DdlException("Failed to execute CREATE TABLE LIKE " + stmt.getExistedTableName() + ". Reason: " + e.getMessage());
         }
     }
+
+    public void createTableAsSelect(CreateTableAsSelectStmt stmt) throws DdlException {
+        try {
+            List<String> columnNames = stmt.getColumnNames();
+            CreateTableStmt createTableStmt = stmt.getCreateTableStmt();
+            QueryStmt queryStmt = stmt.getQueryStmt();
+            ArrayList<Expr> resultExprs = queryStmt.getResultExprs();
+            ArrayList<String> colLabels = queryStmt.getColLabels();
+            int size = resultExprs.size();
+            // Check columnNames
+            int colNameIndex = 0;
+            for (int i = 0; i < size; ++i) {
+                String name;
+                if (columnNames != null) {
+                    // use custom column names
+                    name = columnNames.get(i);
+                } else {
+                    name = colLabels.get(i);
+                }
+                try {
+                    FeNameFormat.checkColumnName(name);
+                } catch (AnalysisException exception) {
+                    name = "_col" + (colNameIndex++);
+                }
+                TypeDef typeDef;
+                Expr resultExpr = resultExprs.get(i);
+                // varchar/char transfer to string
+                if (resultExpr.getType().isStringType()) {
+                    typeDef = new TypeDef(Type.STRING);
+                } else {
+                    typeDef = new TypeDef(resultExpr.getType());
+                }
+                createTableStmt.addColumnDef(new ColumnDef(name, typeDef, false,
+                        null, true,
+                        new DefaultValue(false, null),
+                        ""));
+                // set first column as default distribution
+                if (createTableStmt.getDistributionDesc() == null && i == 0) {
+                    createTableStmt.setDistributionDesc(new HashDistributionDesc(10, Lists.newArrayList(name)));
+                }
+            }
+            Analyzer dummyRootAnalyzer = new Analyzer(this, ConnectContext.get());
+            createTableStmt.analyze(dummyRootAnalyzer);
+            createTable(createTableStmt);
+        } catch (UserException e) {
+            throw new DdlException("Failed to execute CREATE TABLE AS SELECT Reason: " + e.getMessage());
+        }
+    }
 
     public void addPartition(Database db, String tableName, AddPartitionClause addPartitionClause) throws DdlException {
         SinglePartitionDesc singlePartitionDesc = addPartitionClause.getSingeRangePartitionDesc();

fe/fe-core/src/main/java/org/apache/doris/qe/DdlExecutor.java

@@ -51,6 +51,7 @@
 import org.apache.doris.analysis.CreateRoleStmt;
 import org.apache.doris.analysis.CreateRoutineLoadStmt;
 import org.apache.doris.analysis.CreateSqlBlockRuleStmt;
+import org.apache.doris.analysis.CreateTableAsSelectStmt;
 import org.apache.doris.analysis.CreateTableLikeStmt;
 import org.apache.doris.analysis.CreateTableStmt;
 import org.apache.doris.analysis.CreateUserStmt;
@@ -126,6 +127,8 @@ public static void execute(Catalog catalog, DdlStmt ddlStmt) throws Exception {
             catalog.createTable((CreateTableStmt) ddlStmt);
         } else if (ddlStmt instanceof CreateTableLikeStmt) {
             catalog.createTableLike((CreateTableLikeStmt) ddlStmt);
+        } else if (ddlStmt instanceof CreateTableAsSelectStmt) {
+            catalog.createTableAsSelect((CreateTableAsSelectStmt) ddlStmt);
         } else if (ddlStmt instanceof DropTableStmt) {
             catalog.dropTable((DropTableStmt) ddlStmt);
         } else if (ddlStmt instanceof CreateMaterializedViewStmt) {

fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java

@@ -367,8 +367,6 @@ public void execute(TUniqueId queryId) throws Exception {
                 handleUseStmt();
             } else if (parsedStmt instanceof TransactionStmt) {
                 handleTransactionStmt();
-            } else if (parsedStmt instanceof CreateTableAsSelectStmt) {
-                handleInsertStmt();
             } else if (parsedStmt instanceof InsertStmt) { // Must ahead of DdlStmt because InserStmt is its subclass
                 try {
                     handleInsertStmt();
@@ -511,13 +509,12 @@ public void analyze(TQueryOptions tQueryOptions) throws UserException {
             if (parsedStmt instanceof QueryStmt) {
                 queryStmt = (QueryStmt) parsedStmt;
                 queryStmt.getTables(analyzer, tableMap, parentViewNameSet);
+            } else if (parsedStmt instanceof CreateTableAsSelectStmt) {
+                CreateTableAsSelectStmt parsedStmt = (CreateTableAsSelectStmt) this.parsedStmt;
+                queryStmt = parsedStmt.getQueryStmt();
+                queryStmt.getTables(analyzer, tableMap, parentViewNameSet);
             } else {
-                InsertStmt insertStmt;
-                if (parsedStmt instanceof InsertStmt) {
-                    insertStmt = (InsertStmt) parsedStmt;
-                } else {
-                    insertStmt = ((CreateTableAsSelectStmt) parsedStmt).getInsertStmt();
-                }
+                InsertStmt insertStmt = (InsertStmt) parsedStmt;
                 insertStmt.getTables(analyzer, tableMap, parentViewNameSet);
             }
             // table id in tableList is in ascending order because that table map is a sorted map
@@ -641,9 +638,6 @@ private void analyzeAndGenerateQueryPlan(TQueryOptions tQueryOptions) throws Use
         planner = new Planner();
         if (parsedStmt instanceof QueryStmt || parsedStmt instanceof InsertStmt) {
             planner.plan(parsedStmt, analyzer, tQueryOptions);
-        } else {
-            planner.plan(((CreateTableAsSelectStmt) parsedStmt).getInsertStmt(),
-                    analyzer, new TQueryOptions());
         }
         // TODO(zc):
         // Preconditions.checkState(!analyzer.hasUnassignedConjuncts());
@@ -1155,14 +1149,7 @@ private void handleInsertStmt() throws Exception {
         // Every time set no send flag and clean all data in buffer
         context.getMysqlChannel().reset();
         // create plan
-        InsertStmt insertStmt = null;
-        if (parsedStmt instanceof CreateTableAsSelectStmt) {
-            // Create table here
-            ((CreateTableAsSelectStmt) parsedStmt).createTable(analyzer);
-            insertStmt = ((CreateTableAsSelectStmt) parsedStmt).getInsertStmt();
-        } else {
-            insertStmt = (InsertStmt) parsedStmt;
-        }
+        InsertStmt insertStmt = (InsertStmt) parsedStmt;
 
         if (insertStmt.getQueryStmt().hasOutFileClause()) {
             throw new DdlException("Not support OUTFILE clause in INSERT statement");