[ISSUE-2457] Support create table like statement

query/src/sql/parser/ast/ast_display_test.rs

@@ -56,6 +56,8 @@ mod test {
             }],
             engine: "".to_string(),
             options: vec![],
+            like_db: None,
+            like_table: None,
         };
         assert_eq!(
             format!("{}", stmt),

query/src/sql/parser/ast/statement.rs

@@ -54,6 +54,8 @@ pub enum Statement {
         // Thus we have to check validity of engine in parser.
         engine: String,
         options: Vec<SQLProperty>,
+        like_db: Option<Identifier>,
+        like_table: Option<Identifier>,
     },
     // Describe schema of a table
     // Like `SHOW CREATE TABLE`

query/src/sql/parser/transformer/transform_sqlparser.rs

@@ -98,8 +98,8 @@ impl TransformerSqlparser {
     fn transform_ddl(&self, orig_ast: &SqlparserStatement) -> Result<Statement> {
         match orig_ast {
             SqlparserStatement::Truncate { table_name, .. } => {
-                let (db, tpl) = self.get_database_and_table_from_idents(&table_name.0);
-                if let Some(table) = tpl {
+                let (db, tbl) = self.get_database_and_table_from_idents(&table_name.0);
+                if let Some(table) = tbl {
                     Ok(Statement::TruncateTable {
                         database: db,
                         table,
@@ -115,16 +115,26 @@ impl TransformerSqlparser {
                 if_not_exists,
                 name,
                 columns,
+                like,
                 ..
             } => {
-                let (db, tpl) = self.get_database_and_table_from_idents(&name.0);
-                let tpl = tpl.ok_or_else(|| {
+                let (db, tbl) = self.get_database_and_table_from_idents(&name.0);
+                let tbl = tbl.ok_or_else(|| {
                     ErrorCode::SyntaxException(format!(
                         "Unsupported SQL statement: {}",
                         self.orig_stmt
                     ))
                 })?;
-                if columns.is_empty() {
+
+                // statement of 'CREATE TABLE db1.table1 LIKE db2.table2'
+                let (like_db, like_table) = match like {
+                    Some(like_table_name) => {
+                        self.get_database_and_table_from_idents(&like_table_name.0)
+                    }
+                    None => (None, None),
+                };
+
+                if like_table.is_none() && columns.is_empty() {
                     return Err(ErrorCode::SyntaxException(format!(
                         "Unsupported SQL statement: {}",
                         self.orig_stmt
@@ -176,13 +186,16 @@ impl TransformerSqlparser {
                     };
                     cols.push(col);
                 }
+
                 Ok(Statement::CreateTable {
                     if_not_exists: *if_not_exists,
                     database: db,
-                    table: tpl,
+                    table: tbl,
                     columns: cols,
                     engine: "".to_string(),
                     options: vec![],
+                    like_db,
+                    like_table,
                 })
             }
             SqlparserStatement::AlterTable { .. } => {

query/src/sql/sql_parser.rs

@@ -841,7 +841,19 @@ impl<'a> DfParser<'a> {
             self.parser
                 .parse_keywords(&[Keyword::IF, Keyword::NOT, Keyword::EXISTS]);
         let table_name = self.parser.parse_object_name()?;
+
+        // Parse the table which we copy schema from. This is for create table like statement.
+        // https://dev.mysql.com/doc/refman/8.0/en/create-table-like.html
+        let mut table_like = None;
+        if self.parser.parse_keyword(Keyword::LIKE) {
+            table_like = Some(self.parser.parse_object_name()?);
+        }
+
         let (columns, _) = self.parse_columns()?;
+        if !columns.is_empty() && table_like.is_some() {
+            return parser_err!("mix create table like statement and column definition.");
+        }
+
         let engine = self.parse_table_engine()?;
 
         // parse table options: https://dev.mysql.com/doc/refman/8.0/en/create-table.html
@@ -853,6 +865,7 @@ impl<'a> DfParser<'a> {
             columns,
             engine,
             options,
+            like: table_like,
         };
 
         Ok(DfStatement::CreateTable(create))

query/src/sql/sql_parser_test.rs

@@ -143,6 +143,7 @@ fn create_table() -> Result<()> {
             name: Ident::new("location".to_string()),
             value: Value::SingleQuotedString("/data/33.csv".into()),
         }],
+        like: None,
     });
     expect_parse_ok(sql, expected)?;
 
@@ -167,6 +168,22 @@ fn create_table() -> Result<()> {
                 value: Value::SingleQuotedString("foo".into()),
             },
         ],
+        like: None,
+    });
+    expect_parse_ok(sql, expected)?;
+
+    // create table like statement
+    let sql = "CREATE TABLE db1.test1 LIKE db2.test2 ENGINE = Parquet location = 'batcave'";
+    let expected = DfStatement::CreateTable(DfCreateTable {
+        if_not_exists: false,
+        name: ObjectName(vec![Ident::new("db1"), Ident::new("test1")]),
+        columns: vec![],
+        engine: "Parquet".to_string(),
+        options: vec![SqlOption {
+            name: Ident::new("location".to_string()),
+            value: Value::SingleQuotedString("batcave".into()),
+        }],
+        like: Some(ObjectName(vec![Ident::new("db2"), Ident::new("test2")])),
     });
     expect_parse_ok(sql, expected)?;
 

query/src/sql/statements/statement_create_table.rs

@@ -43,6 +43,9 @@ pub struct DfCreateTable {
     pub columns: Vec<ColumnDef>,
     pub engine: String,
     pub options: Vec<SqlOption>,
+
+    // The table name after "create .. like" statement.
+    pub like: Option<ObjectName>,
 }
 
 #[async_trait::async_trait]
@@ -51,7 +54,7 @@ impl AnalyzableStatement for DfCreateTable {
     async fn analyze(&self, ctx: Arc<QueryContext>) -> Result<AnalyzedResult> {
         let table_meta = self.table_meta(ctx.clone()).await?;
         let if_not_exists = self.if_not_exists;
-        let (db, table) = self.resolve_table(ctx)?;
+        let (db, table) = Self::resolve_table(ctx, &self.name)?;
 
         Ok(AnalyzedResult::SimpleQuery(Box::new(
             PlanNode::CreateTable(CreateTablePlan {
@@ -65,11 +68,8 @@ impl AnalyzableStatement for DfCreateTable {
 }
 
 impl DfCreateTable {
-    fn resolve_table(&self, ctx: Arc<QueryContext>) -> Result<(String, String)> {
-        let DfCreateTable {
-            name: ObjectName(idents),
-            ..
-        } = self;
+    fn resolve_table(ctx: Arc<QueryContext>, table_name: &ObjectName) -> Result<(String, String)> {
+        let idents = &table_name.0;
         match idents.len() {
             0 => Err(ErrorCode::SyntaxException("Create table name is empty")),
             1 => Ok((ctx.get_current_database(), idents[0].value.clone())),
@@ -108,30 +108,45 @@ impl DfCreateTable {
     }
 
     async fn table_schema(&self, ctx: Arc<QueryContext>) -> Result<DataSchemaRef> {
-        let expr_analyzer = ExpressionAnalyzer::create(ctx);
-        let mut fields = Vec::with_capacity(self.columns.len());
+        match &self.like {
+            // For create table like statement, for example 'CREATE TABLE test2 LIKE db1.test1',
+            // we use the original table's schema.
+            Some(like_table_name) => {
+                // resolve database and table name from 'like statement'
+                let (origin_db_name, origin_table_name) =
+                    Self::resolve_table(ctx.clone(), like_table_name)?;
 
-        for column in &self.columns {
-            let mut nullable = true;
-            let mut default_expr = None;
-            for opt in &column.options {
-                match &opt.option {
-                    ColumnOption::NotNull => {
-                        nullable = false;
-                    }
-                    ColumnOption::Default(expr) => {
-                        let expr = expr_analyzer.analyze(expr).await?;
-                        default_expr = Some(serde_json::to_vec(&expr)?);
+                // use the origin table's schema for the table to create
+                let origin_table = ctx.get_table(&origin_db_name, &origin_table_name).await?;
+                Ok(origin_table.schema())
+            }
+            None => {
+                let expr_analyzer = ExpressionAnalyzer::create(ctx);
+                let mut fields = Vec::with_capacity(self.columns.len());
+
+                for column in &self.columns {
+                    let mut nullable = true;
+                    let mut default_expr = None;
+                    for opt in &column.options {
+                        match &opt.option {
+                            ColumnOption::NotNull => {
+                                nullable = false;
+                            }
+                            ColumnOption::Default(expr) => {
+                                let expr = expr_analyzer.analyze(expr).await?;
+                                default_expr = Some(serde_json::to_vec(&expr)?);
+                            }
+                            _ => {}
+                        }
                     }
-                    _ => {}
+                    let field = SQLCommon::make_data_type(&column.data_type).map(|data_type| {
+                        DataField::new(&column.name.value, data_type, nullable)
+                            .with_default_expr(default_expr)
+                    })?;
+                    fields.push(field);
                 }
+                Ok(DataSchemaRefExt::create(fields))
             }
-            let field = SQLCommon::make_data_type(&column.data_type).map(|data_type| {
-                DataField::new(&column.name.value, data_type, nullable)
-                    .with_default_expr(default_expr)
-            })?;
-            fields.push(field);
         }
-        Ok(DataSchemaRefExt::create(fields))
     }
 }

tests/suites/0_stateless/05_0000_ddl_create_tables.result

@@ -1,3 +1,4 @@
 1
 2
 4
+8

tests/suites/0_stateless/05_0000_ddl_create_tables.sql

@@ -15,3 +15,12 @@ create table t2(a int,b int) engine=NotExists; -- {ErrorCode 4003}
 
 DROP TABLE IF EXISTS t;
 DROP TABLE IF EXISTS t2;
+
+CREATE DATABASE db1;
+CREATE DATABASE db2;
+CREATE TABLE db1.test1(a INT, b INT);
+CREATE TABLE db2.test2 LIKE db1.test1;
+INSERT INTO db2.test2 VALUES (3, 5);
+SELECT a+b FROM db2.test2;
+DROP DATABASE db1;
+DROP DATABASE db2;

website/databend/docs/user/sqlstatement/data-definition-language-ddl/ddl-create-table.md

@@ -14,6 +14,10 @@ CREATE TABLE [IF NOT EXISTS] [db.]table_name
     ...
 ) ENGINE = engine
 ```
+```sql
+CREATE TABLE [IF NOT EXISTS] [db.]table_name
+LIKE [db.]origin_table_name ENGINE = engine
+```
 
 :::note
 Local engine is one of `Memory`, `Parquet`, `JSONEachRow`, `Null` or `CSV`, data will be stored in the DatabendQuery memory/disk locally.
@@ -37,4 +41,15 @@ mysql> SELECT * FROM test;
 +------+---------+
 |  888 |  stars  |
 +------+---------+
+
+mysql> CREATE TABLE test2 LIKE test Engine = Memory;
+
+mysql> INSERT INTO test2(a,b) values(0, 'sun');
+
+mysql> SELECT * FROM test2;
++------+------+
+| a    | b    |
++------+------+
+|    0 | sun  |
++------+------+
 ```