feat: impl show collation and show charset statements (#3753)

* feat: impl show collation and show charset statements

* docs: add api docs

src/frontend/src/instance.rs

@@ -491,11 +491,20 @@ pub fn check_permission(
         // database ops won't be checked
         Statement::CreateDatabase(_) | Statement::ShowDatabases(_) | Statement::DropDatabase(_) => {
         }
-        // show create table and alter are not supported yet
-        Statement::ShowCreateTable(_) | Statement::CreateExternalTable(_) | Statement::Alter(_) => {
+
+        Statement::ShowCreateTable(stmt) => {
+            validate_param(&stmt.table_name, query_ctx)?;
+        }
+        Statement::CreateExternalTable(stmt) => {
+            validate_param(&stmt.name, query_ctx)?;
+        }
+        Statement::Alter(stmt) => {
+            validate_param(stmt.table_name(), query_ctx)?;
         }
         // set/show variable now only alter/show variable in session
         Statement::SetVariables(_) | Statement::ShowVariables(_) => {}
+        // show charset and show collation won't be checked
+        Statement::ShowCharset(_) | Statement::ShowCollation(_) => {}
 
         Statement::Insert(insert) => {
             validate_param(insert.table_name(), query_ctx)?;

src/operator/src/statement.rs

@@ -118,6 +118,10 @@ impl StatementExecutor {
 
             Statement::ShowTables(stmt) => self.show_tables(stmt, query_ctx).await,
 
+            Statement::ShowCollation(kind) => self.show_collation(kind, query_ctx).await,
+
+            Statement::ShowCharset(kind) => self.show_charset(kind, query_ctx).await,
+
             Statement::Copy(sql::statements::copy::Copy::CopyTable(stmt)) => {
                 let req = to_copy_table_request(stmt, query_ctx.clone())?;
                 match req.direction {

src/operator/src/statement/show.rs

@@ -21,7 +21,9 @@ use session::context::QueryContextRef;
 use snafu::ResultExt;
 use sql::ast::Ident;
 use sql::statements::create::Partitions;
-use sql::statements::show::{ShowColumns, ShowDatabases, ShowIndex, ShowTables, ShowVariables};
+use sql::statements::show::{
+    ShowColumns, ShowDatabases, ShowIndex, ShowKind, ShowTables, ShowVariables,
+};
 use table::TableRef;
 
 use crate::error::{self, ExecuteStatementSnafu, Result};
@@ -97,6 +99,24 @@ impl StatementExecutor {
     pub fn show_variable(&self, stmt: ShowVariables, query_ctx: QueryContextRef) -> Result<Output> {
         query::sql::show_variable(stmt, query_ctx).context(error::ExecuteStatementSnafu)
     }
+
+    #[tracing::instrument(skip_all)]
+    pub async fn show_collation(
+        &self,
+        kind: ShowKind,
+        query_ctx: QueryContextRef,
+    ) -> Result<Output> {
+        query::sql::show_collations(kind, &self.query_engine, &self.catalog_manager, query_ctx)
+            .await
+            .context(error::ExecuteStatementSnafu)
+    }
+
+    #[tracing::instrument(skip_all)]
+    pub async fn show_charset(&self, kind: ShowKind, query_ctx: QueryContextRef) -> Result<Output> {
+        query::sql::show_charsets(kind, &self.query_engine, &self.catalog_manager, query_ctx)
+            .await
+            .context(error::ExecuteStatementSnafu)
+    }
 }
 
 pub(crate) fn create_partitions_stmt(partitions: Vec<PartitionInfo>) -> Result<Option<Partitions>> {

src/query/src/sql.rs

@@ -18,7 +18,8 @@ use std::collections::HashMap;
 use std::sync::Arc;
 
 use catalog::information_schema::{
-    columns, key_column_usage, schemata, tables, COLUMNS, KEY_COLUMN_USAGE, SCHEMATA, TABLES,
+    columns, key_column_usage, schemata, tables, CHARACTER_SETS, COLLATIONS, COLUMNS,
+    KEY_COLUMN_USAGE, SCHEMATA, TABLES,
 };
 use catalog::CatalogManagerRef;
 use common_catalog::consts::{
@@ -469,6 +470,76 @@ pub async fn show_tables(
     .await
 }
 
+/// Execute `SHOW COLLATION` statement and returns the `Output` if success.
+pub async fn show_collations(
+    kind: ShowKind,
+    query_engine: &QueryEngineRef,
+    catalog_manager: &CatalogManagerRef,
+    query_ctx: QueryContextRef,
+) -> Result<Output> {
+    // Refer to https://dev.mysql.com/doc/refman/8.0/en/show-collation.html
+    let projects = vec![
+        ("collation_name", "Collation"),
+        ("character_set_name", "Charset"),
+        ("id", "Id"),
+        ("is_default", "Default"),
+        ("is_compiled", "Compiled"),
+        ("sortlen", "Sortlen"),
+    ];
+
+    let filters = vec![];
+    let like_field = Some("collation_name");
+    let sort = vec![];
+
+    query_from_information_schema_table(
+        query_engine,
+        catalog_manager,
+        query_ctx,
+        COLLATIONS,
+        vec![],
+        projects,
+        filters,
+        like_field,
+        sort,
+        kind,
+    )
+    .await
+}
+
+/// Execute `SHOW CHARSET` statement and returns the `Output` if success.
+pub async fn show_charsets(
+    kind: ShowKind,
+    query_engine: &QueryEngineRef,
+    catalog_manager: &CatalogManagerRef,
+    query_ctx: QueryContextRef,
+) -> Result<Output> {
+    // Refer to https://dev.mysql.com/doc/refman/8.0/en/show-character-set.html
+    let projects = vec![
+        ("character_set_name", "Charset"),
+        ("description", "Description"),
+        ("default_collate_name", "Default collation"),
+        ("maxlen", "Maxlen"),
+    ];
+
+    let filters = vec![];
+    let like_field = Some("character_set_name");
+    let sort = vec![];
+
+    query_from_information_schema_table(
+        query_engine,
+        catalog_manager,
+        query_ctx,
+        CHARACTER_SETS,
+        vec![],
+        projects,
+        filters,
+        like_field,
+        sort,
+        kind,
+    )
+    .await
+}
+
 pub fn show_variable(stmt: ShowVariables, query_ctx: QueryContextRef) -> Result<Output> {
     let variable = stmt.variable.to_string().to_uppercase();
     let value = match variable.as_str() {

src/servers/src/mysql/federated.rs

@@ -35,8 +35,6 @@ static MYSQL_CONN_JAVA_PATTERN: Lazy<Regex> =
     Lazy::new(|| Regex::new("(?i)^(/\\* mysql-connector-j(.*))").unwrap());
 static SHOW_LOWER_CASE_PATTERN: Lazy<Regex> =
     Lazy::new(|| Regex::new("(?i)^(SHOW VARIABLES LIKE 'lower_case_table_names'(.*))").unwrap());
-static SHOW_COLLATION_PATTERN: Lazy<Regex> =
-    Lazy::new(|| Regex::new("(?i)^(show collation where(.*))").unwrap());
 static SHOW_VARIABLES_LIKE_PATTERN: Lazy<Regex> =
     Lazy::new(|| Regex::new("(?i)^(SHOW VARIABLES( LIKE (.*))?)").unwrap());
 
@@ -70,9 +68,6 @@ static OTHER_NOT_SUPPORTED_STMT: Lazy<RegexSet> = Lazy::new(|| {
         "(?i)^(SET @@(.*))",
         "(?i)^(SET PROFILING(.*))",
 
-        "(?i)^(SHOW COLLATION)",
-        "(?i)^(SHOW CHARSET)",
-
         // mysqlclient.
         "(?i)^(SELECT \\$\\$)",
 
@@ -96,8 +91,6 @@ static OTHER_NOT_SUPPORTED_STMT: Lazy<RegexSet> = Lazy::new(|| {
         "(?i)^(SHOW WARNINGS)",
         "(?i)^(/\\* ApplicationName=(.*)SHOW WARNINGS)",
         "(?i)^(/\\* ApplicationName=(.*)SHOW PLUGINS)",
-        "(?i)^(/\\* ApplicationName=(.*)SHOW COLLATION)",
-        "(?i)^(/\\* ApplicationName=(.*)SHOW CHARSET)",
         "(?i)^(/\\* ApplicationName=(.*)SHOW ENGINES)",
         "(?i)^(/\\* ApplicationName=(.*)SELECT @@(.*))",
         "(?i)^(/\\* ApplicationName=(.*)SHOW @@(.*))",
@@ -248,8 +241,7 @@ fn check_show_variables(query: &str) -> Option<Output> {
         Some(show_variables("sql_mode", "ONLY_FULL_GROUP_BY STRICT_TRANS_TABLES NO_ZERO_IN_DATE NO_ZERO_DATE ERROR_FOR_DIVISION_BY_ZERO NO_ENGINE_SUBSTITUTION"))
     } else if SHOW_LOWER_CASE_PATTERN.is_match(query) {
         Some(show_variables("lower_case_table_names", "0"))
-    } else if SHOW_COLLATION_PATTERN.is_match(query) || SHOW_VARIABLES_LIKE_PATTERN.is_match(query)
-    {
+    } else if SHOW_VARIABLES_LIKE_PATTERN.is_match(query) {
         Some(show_variables("", ""))
     } else {
         None
@@ -379,12 +371,6 @@ mod test {
 +------------------------+-------+";
         test(query, expected);
 
-        let query = "show collation";
-        let expected = "\
-++
-++"; // empty
-        test(query, expected);
-
         let query = "SELECT TIMEDIFF(NOW(), UTC_TIMESTAMP())";
         let expected = "\
 +----------------------------------+