feat: show create database (#4642)

* feat: show create database

* feat: add sqlness test

* chore: reorder mod and use

* feat: show create schema

* Update src/frontend/src/instance.rs

src/operator/src/error.rs

@@ -784,6 +784,12 @@ pub enum Error {
         #[snafu(implicit)]
         location: Location,
     },
+
+    #[snafu(display("Failed to upgrade catalog manager reference"))]
+    UpgradeCatalogManagerRef {
+        #[snafu(implicit)]
+        location: Location,
+    },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -931,6 +937,8 @@ impl ErrorExt for Error {
 
             Error::ExecuteAdminFunction { source, .. } => source.status_code(),
             Error::BuildRecordBatch { source, .. } => source.status_code(),
+
+            Error::UpgradeCatalogManagerRef { .. } => StatusCode::Internal,
         }
     }
 

src/operator/src/statement.rs

@@ -23,8 +23,10 @@ mod set;
 mod show;
 mod tql;
 
+use std::collections::HashMap;
 use std::sync::Arc;
 
+use catalog::kvbackend::KvBackendCatalogManager;
 use catalog::CatalogManagerRef;
 use client::RecordBatches;
 use common_error::ext::BoxedError;
@@ -32,6 +34,7 @@ use common_meta::cache::TableRouteCacheRef;
 use common_meta::cache_invalidator::CacheInvalidatorRef;
 use common_meta::ddl::ProcedureExecutorRef;
 use common_meta::key::flow::{FlowMetadataManager, FlowMetadataManagerRef};
+use common_meta::key::schema_name::SchemaNameKey;
 use common_meta::key::view_info::{ViewInfoManager, ViewInfoManagerRef};
 use common_meta::key::{TableMetadataManager, TableMetadataManagerRef};
 use common_meta::kv_backend::KvBackendRef;
@@ -60,7 +63,8 @@ use table::TableRef;
 use self::set::{set_bytea_output, set_datestyle, set_timezone, validate_client_encoding};
 use crate::error::{
     self, CatalogSnafu, ExecLogicalPlanSnafu, ExternalSnafu, InvalidSqlSnafu, NotSupportedSnafu,
-    PlanStatementSnafu, Result, SchemaNotFoundSnafu, TableNotFoundSnafu,
+    PlanStatementSnafu, Result, SchemaNotFoundSnafu, TableMetadataManagerSnafu, TableNotFoundSnafu,
+    UpgradeCatalogManagerRefSnafu,
 };
 use crate::insert::InserterRef;
 use crate::statement::copy_database::{COPY_DATABASE_TIME_END_KEY, COPY_DATABASE_TIME_START_KEY};
@@ -251,6 +255,29 @@ impl StatementExecutor {
                 )
                 .await
             }
+            Statement::ShowCreateDatabase(show) => {
+                let (catalog, database) =
+                    idents_to_full_database_name(&show.database_name, &query_ctx)
+                        .map_err(BoxedError::new)
+                        .context(ExternalSnafu)?;
+                let table_metadata_manager = self
+                    .catalog_manager
+                    .as_any()
+                    .downcast_ref::<KvBackendCatalogManager>()
+                    .map(|manager| manager.table_metadata_manager_ref().clone())
+                    .context(UpgradeCatalogManagerRefSnafu)?;
+                let opts: HashMap<String, String> = table_metadata_manager
+                    .schema_manager()
+                    .get(SchemaNameKey::new(&catalog, &database))
+                    .await
+                    .context(TableMetadataManagerSnafu)?
+                    .context(SchemaNotFoundSnafu {
+                        schema_info: &database,
+                    })?
+                    .into();
+
+                self.show_create_database(&database, opts.into()).await
+            }
             Statement::ShowCreateTable(show) => {
                 let (catalog, schema, table) =
                     table_idents_to_full_name(&show.table_name, &query_ctx)

src/operator/src/statement/show.rs

@@ -26,6 +26,7 @@ use sql::statements::show::{
     ShowColumns, ShowCreateFlow, ShowCreateView, ShowDatabases, ShowFlows, ShowIndex, ShowKind,
     ShowTableStatus, ShowTables, ShowVariables, ShowViews,
 };
+use sql::statements::OptionMap;
 use table::metadata::TableType;
 use table::table_name::TableName;
 use table::TableRef;
@@ -92,6 +93,15 @@ impl StatementExecutor {
             .context(ExecuteStatementSnafu)
     }
 
+    #[tracing::instrument(skip_all)]
+    pub async fn show_create_database(
+        &self,
+        database_name: &str,
+        opts: OptionMap,
+    ) -> Result<Output> {
+        query::sql::show_create_database(database_name, opts).context(ExecuteStatementSnafu)
+    }
+
     #[tracing::instrument(skip_all)]
     pub async fn show_create_table(
         &self,
@@ -118,8 +128,7 @@ impl StatementExecutor {
 
         let partitions = create_partitions_stmt(partitions)?;
 
-        query::sql::show_create_table(table, partitions, query_ctx)
-            .context(error::ExecuteStatementSnafu)
+        query::sql::show_create_table(table, partitions, query_ctx).context(ExecuteStatementSnafu)
     }
 
     #[tracing::instrument(skip_all)]