feat: support Create Table ... Like (#3372)

* feat: support `Create Table ... Like`

* fix: `check_permission` for `Create Table ... Like`

* style: renaming `name` -> `table_name` & `target` -> `source_name` and make `Create Table ... Like` testcase more complicated

* rebase

* avoid _ fn

Signed-off-by: tison <wander4096@gmail.com>

---------

Signed-off-by: tison <wander4096@gmail.com>
Co-authored-by: tison <wander4096@gmail.com>

src/operator/src/expr_factory.rs

@@ -181,7 +181,7 @@ pub(crate) async fn create_external_expr(
     Ok(expr)
 }
 
-/// Convert `CreateTable` statement to `CreateExpr` gRPC request.
+/// Convert `CreateTable` statement to [`CreateTableExpr`] gRPC request.
 pub fn create_to_expr(create: &CreateTable, query_ctx: QueryContextRef) -> Result<CreateTableExpr> {
     let (catalog_name, schema_name, table_name) =
         table_idents_to_full_name(&create.name, &query_ctx)

src/operator/src/statement.rs

@@ -153,6 +153,10 @@ impl StatementExecutor {
                 let _ = self.create_table(stmt, query_ctx).await?;
                 Ok(Output::AffectedRows(0))
             }
+            Statement::CreateTableLike(stmt) => {
+                let _ = self.create_table_like(stmt, query_ctx).await?;
+                Ok(Output::AffectedRows(0))
+            }
             Statement::CreateExternalTable(stmt) => {
                 let _ = self.create_external_table(stmt, query_ctx).await?;
                 Ok(Output::AffectedRows(0))
@@ -174,7 +178,6 @@ impl StatementExecutor {
                 let table_name = TableName::new(catalog, schema, table);
                 self.truncate_table(table_name).await
             }
-
             Statement::CreateDatabase(stmt) => {
                 self.create_database(
                     query_ctx.current_catalog(),
@@ -183,7 +186,6 @@ impl StatementExecutor {
                 )
                 .await
             }
-
             Statement::ShowCreateTable(show) => {
                 let (catalog, schema, table) =
                     table_idents_to_full_name(&show.table_name, &query_ctx)

src/operator/src/statement/ddl.rs

@@ -21,6 +21,7 @@ use catalog::CatalogManagerRef;
 use chrono::Utc;
 use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
 use common_catalog::format_full_table_name;
+use common_error::ext::BoxedError;
 use common_meta::cache_invalidator::Context;
 use common_meta::ddl::ExecutorContext;
 use common_meta::key::schema_name::{SchemaNameKey, SchemaNameValue};
@@ -34,11 +35,13 @@ use datatypes::prelude::ConcreteDataType;
 use datatypes::schema::RawSchema;
 use lazy_static::lazy_static;
 use partition::partition::{PartitionBound, PartitionDef};
+use query::sql::create_table_stmt;
 use regex::Regex;
 use session::context::QueryContextRef;
+use session::table_name::table_idents_to_full_name;
 use snafu::{ensure, IntoError, OptionExt, ResultExt};
 use sql::statements::alter::AlterTable;
-use sql::statements::create::{CreateExternalTable, CreateTable, Partitions};
+use sql::statements::create::{CreateExternalTable, CreateTable, CreateTableLike, Partitions};
 use table::dist_table::DistTable;
 use table::metadata::{self, RawTableInfo, RawTableMeta, TableId, TableInfo, TableType};
 use table::requests::{AlterKind, AlterTableRequest, TableOptions};
@@ -54,6 +57,7 @@ use crate::error::{
     UnrecognizedTableOptionSnafu,
 };
 use crate::expr_factory;
+use crate::statement::show::create_partitions_stmt;
 
 lazy_static! {
     static ref NAME_PATTERN_REG: Regex = Regex::new(&format!("^{NAME_PATTERN}$")).unwrap();
@@ -71,6 +75,46 @@ impl StatementExecutor {
             .await
     }
 
+    #[tracing::instrument(skip_all)]
+    pub async fn create_table_like(
+        &self,
+        stmt: CreateTableLike,
+        ctx: QueryContextRef,
+    ) -> Result<TableRef> {
+        let (catalog, schema, table) = table_idents_to_full_name(&stmt.source_name, &ctx)
+            .map_err(BoxedError::new)
+            .context(error::ExternalSnafu)?;
+        let table_ref = self
+            .catalog_manager
+            .table(&catalog, &schema, &table)
+            .await
+            .context(CatalogSnafu)?
+            .context(TableNotFoundSnafu { table_name: &table })?;
+        let partitions = self
+            .partition_manager
+            .find_table_partitions(table_ref.table_info().table_id())
+            .await
+            .context(error::FindTablePartitionRuleSnafu { table_name: table })?;
+
+        let quote_style = ctx.quote_style();
+        let mut create_stmt = create_table_stmt(&table_ref.table_info(), quote_style)
+            .context(error::ParseQuerySnafu)?;
+        create_stmt.name = stmt.table_name;
+        create_stmt.if_not_exists = false;
+
+        let partitions = create_partitions_stmt(partitions)?.and_then(|mut partitions| {
+            if !partitions.column_list.is_empty() {
+                partitions.set_quote(quote_style);
+                Some(partitions)
+            } else {
+                None
+            }
+        });
+
+        let create_expr = &mut expr_factory::create_to_expr(&create_stmt, ctx.clone())?;
+        self.create_table_inner(create_expr, partitions, &ctx).await
+    }
+
     #[tracing::instrument(skip_all)]
     pub async fn create_external_table(
         &self,
@@ -160,7 +204,7 @@ impl StatementExecutor {
 
         table_info.ident.table_id = table_id;
 
-        let table_info = Arc::new(table_info.try_into().context(error::CreateTableInfoSnafu)?);
+        let table_info = Arc::new(table_info.try_into().context(CreateTableInfoSnafu)?);
         create_table.table_id = Some(api::v1::TableId { id: table_id });
 
         let table = DistTable::table(table_info);

src/operator/src/statement/show.rs

@@ -76,7 +76,7 @@ impl StatementExecutor {
     }
 }
 
-fn create_partitions_stmt(partitions: Vec<PartitionInfo>) -> Result<Option<Partitions>> {
+pub(crate) fn create_partitions_stmt(partitions: Vec<PartitionInfo>) -> Result<Option<Partitions>> {
     if partitions.is_empty() {
         return Ok(None);
     }