feat: impl show index and show columns

src/catalog/src/information_schema.rs

@@ -12,8 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-mod columns;
-mod key_column_usage;
+pub mod columns;
+pub mod key_column_usage;
 mod memory_table;
 mod partitions;
 mod predicate;

src/catalog/src/information_schema/columns.rs

@@ -48,16 +48,16 @@ pub(super) struct InformationSchemaColumns {
     catalog_manager: Weak<dyn CatalogManager>,
 }
 
-const TABLE_CATALOG: &str = "table_catalog";
-const TABLE_SCHEMA: &str = "table_schema";
-const TABLE_NAME: &str = "table_name";
-const COLUMN_NAME: &str = "column_name";
-const DATA_TYPE: &str = "data_type";
-const SEMANTIC_TYPE: &str = "semantic_type";
-const COLUMN_DEFAULT: &str = "column_default";
-const IS_NULLABLE: &str = "is_nullable";
+pub const TABLE_CATALOG: &str = "table_catalog";
+pub const TABLE_SCHEMA: &str = "table_schema";
+pub const TABLE_NAME: &str = "table_name";
+pub const COLUMN_NAME: &str = "column_name";
+pub const DATA_TYPE: &str = "data_type";
+pub const SEMANTIC_TYPE: &str = "semantic_type";
+pub const COLUMN_DEFAULT: &str = "column_default";
+pub const IS_NULLABLE: &str = "is_nullable";
 const COLUMN_TYPE: &str = "column_type";
-const COLUMN_COMMENT: &str = "column_comment";
+pub const COLUMN_COMMENT: &str = "column_comment";
 const INIT_CAPACITY: usize = 42;
 
 impl InformationSchemaColumns {

src/catalog/src/information_schema/key_column_usage.rs

@@ -37,13 +37,16 @@ use crate::error::{
 use crate::information_schema::{InformationTable, Predicates};
 use crate::CatalogManager;
 
-const CONSTRAINT_SCHEMA: &str = "constraint_schema";
-const CONSTRAINT_NAME: &str = "constraint_name";
-const TABLE_CATALOG: &str = "table_catalog";
-const TABLE_SCHEMA: &str = "table_schema";
-const TABLE_NAME: &str = "table_name";
-const COLUMN_NAME: &str = "column_name";
-const ORDINAL_POSITION: &str = "ordinal_position";
+pub const CONSTRAINT_SCHEMA: &str = "constraint_schema";
+pub const CONSTRAINT_NAME: &str = "constraint_name";
+// It's always `def` in MySQL
+pub const TABLE_CATALOG: &str = "table_catalog";
+// The real catalog name for this key column.
+pub const REAL_TABLE_CATALOG: &str = "real_table_catalog";
+pub const TABLE_SCHEMA: &str = "table_schema";
+pub const TABLE_NAME: &str = "table_name";
+pub const COLUMN_NAME: &str = "column_name";
+pub const ORDINAL_POSITION: &str = "ordinal_position";
 const INIT_CAPACITY: usize = 42;
 
 /// The virtual table implementation for `information_schema.KEY_COLUMN_USAGE`.
@@ -76,6 +79,11 @@ impl InformationSchemaKeyColumnUsage {
             ),
             ColumnSchema::new(CONSTRAINT_NAME, ConcreteDataType::string_datatype(), false),
             ColumnSchema::new(TABLE_CATALOG, ConcreteDataType::string_datatype(), false),
+            ColumnSchema::new(
+                REAL_TABLE_CATALOG,
+                ConcreteDataType::string_datatype(),
+                false,
+            ),
             ColumnSchema::new(TABLE_SCHEMA, ConcreteDataType::string_datatype(), false),
             ColumnSchema::new(TABLE_NAME, ConcreteDataType::string_datatype(), false),
             ColumnSchema::new(COLUMN_NAME, ConcreteDataType::string_datatype(), false),
@@ -158,6 +166,7 @@ struct InformationSchemaKeyColumnUsageBuilder {
     constraint_schema: StringVectorBuilder,
     constraint_name: StringVectorBuilder,
     table_catalog: StringVectorBuilder,
+    real_table_catalog: StringVectorBuilder,
     table_schema: StringVectorBuilder,
     table_name: StringVectorBuilder,
     column_name: StringVectorBuilder,
@@ -179,6 +188,7 @@ impl InformationSchemaKeyColumnUsageBuilder {
             constraint_schema: StringVectorBuilder::with_capacity(INIT_CAPACITY),
             constraint_name: StringVectorBuilder::with_capacity(INIT_CAPACITY),
             table_catalog: StringVectorBuilder::with_capacity(INIT_CAPACITY),
+            real_table_catalog: StringVectorBuilder::with_capacity(INIT_CAPACITY),
             table_schema: StringVectorBuilder::with_capacity(INIT_CAPACITY),
             table_name: StringVectorBuilder::with_capacity(INIT_CAPACITY),
             column_name: StringVectorBuilder::with_capacity(INIT_CAPACITY),
@@ -223,6 +233,7 @@ impl InformationSchemaKeyColumnUsageBuilder {
                                 &predicates,
                                 &schema_name,
                                 "TIME INDEX",
+                                &catalog_name,
                                 &schema_name,
                                 &table_name,
                                 &column.name,
@@ -231,6 +242,7 @@ impl InformationSchemaKeyColumnUsageBuilder {
                         }
                         if keys.contains(&idx) {
                             primary_constraints.push((
+                                catalog_name.clone(),
                                 schema_name.clone(),
                                 table_name.clone(),
                                 column.name.clone(),
@@ -244,13 +256,14 @@ impl InformationSchemaKeyColumnUsageBuilder {
             }
         }
 
-        for (i, (schema_name, table_name, column_name)) in
+        for (i, (catalog_name, schema_name, table_name, column_name)) in
             primary_constraints.into_iter().enumerate()
         {
             self.add_key_column_usage(
                 &predicates,
                 &schema_name,
                 "PRIMARY",
+                &catalog_name,
                 &schema_name,
                 &table_name,
                 &column_name,
@@ -269,6 +282,7 @@ impl InformationSchemaKeyColumnUsageBuilder {
         predicates: &Predicates,
         constraint_schema: &str,
         constraint_name: &str,
+        table_catalog: &str,
         table_schema: &str,
         table_name: &str,
         column_name: &str,
@@ -277,6 +291,7 @@ impl InformationSchemaKeyColumnUsageBuilder {
         let row = [
             (CONSTRAINT_SCHEMA, &Value::from(constraint_schema)),
             (CONSTRAINT_NAME, &Value::from(constraint_name)),
+            (REAL_TABLE_CATALOG, &Value::from(table_catalog)),
             (TABLE_SCHEMA, &Value::from(table_schema)),
             (TABLE_NAME, &Value::from(table_name)),
             (COLUMN_NAME, &Value::from(column_name)),
@@ -291,6 +306,7 @@ impl InformationSchemaKeyColumnUsageBuilder {
         self.constraint_schema.push(Some(constraint_schema));
         self.constraint_name.push(Some(constraint_name));
         self.table_catalog.push(Some("def"));
+        self.real_table_catalog.push(Some(table_catalog));
         self.table_schema.push(Some(table_schema));
         self.table_name.push(Some(table_name));
         self.column_name.push(Some(column_name));
@@ -310,6 +326,7 @@ impl InformationSchemaKeyColumnUsageBuilder {
             Arc::new(self.constraint_schema.finish()),
             Arc::new(self.constraint_name.finish()),
             Arc::new(self.table_catalog.finish()),
+            Arc::new(self.real_table_catalog.finish()),
             Arc::new(self.table_schema.finish()),
             Arc::new(self.table_name.finish()),
             Arc::new(self.column_name.finish()),

src/frontend/src/instance.rs

@@ -455,6 +455,17 @@ impl PrometheusHandler for Instance {
     }
 }
 
+/// Validate `stmt.database` permission if it's presented.
+macro_rules! validate_db_permission {
+    ($stmt: expr, $query_ctx: expr) => {
+        if let Some(database) = &$stmt.database {
+            validate_catalog_and_schema($query_ctx.current_catalog(), database, $query_ctx)
+                .map_err(BoxedError::new)
+                .context(SqlExecInterceptedSnafu)?;
+        }
+    };
+}
+
 pub fn check_permission(
     plugins: Plugins,
     stmt: &Statement,
@@ -495,11 +506,13 @@ pub fn check_permission(
             validate_param(drop_stmt.table_name(), query_ctx)?;
         }
         Statement::ShowTables(stmt) => {
-            if let Some(database) = &stmt.database {
-                validate_catalog_and_schema(query_ctx.current_catalog(), database, query_ctx)
-                    .map_err(BoxedError::new)
-                    .context(SqlExecInterceptedSnafu)?;
-            }
+            validate_db_permission!(stmt, query_ctx);
+        }
+        Statement::ShowColumns(stmt) => {
+            validate_db_permission!(stmt, query_ctx);
+        }
+        Statement::ShowIndex(stmt) => {
+            validate_db_permission!(stmt, query_ctx);
         }
         Statement::DescribeTable(stmt) => {
             validate_param(stmt.name(), query_ctx)?;

src/operator/src/statement.rs

@@ -239,6 +239,10 @@ impl StatementExecutor {
                 Ok(Output::new_with_affected_rows(0))
             }
             Statement::ShowVariables(show_variable) => self.show_variable(show_variable, query_ctx),
+            Statement::ShowColumns(show_columns) => {
+                self.show_columns(show_columns, query_ctx).await
+            }
+            Statement::ShowIndex(show_index) => self.show_index(show_index, query_ctx).await,
         }
     }
 

src/operator/src/statement/show.rs

@@ -21,7 +21,7 @@ use session::context::QueryContextRef;
 use snafu::ResultExt;
 use sql::ast::Ident;
 use sql::statements::create::Partitions;
-use sql::statements::show::{ShowDatabases, ShowTables, ShowVariables};
+use sql::statements::show::{ShowColumns, ShowDatabases, ShowIndex, ShowTables, ShowVariables};
 use table::TableRef;
 
 use crate::error::{self, ExecuteStatementSnafu, Result};
@@ -50,6 +50,28 @@ impl StatementExecutor {
             .context(ExecuteStatementSnafu)
     }
 
+    #[tracing::instrument(skip_all)]
+    pub(super) async fn show_columns(
+        &self,
+        stmt: ShowColumns,
+        query_ctx: QueryContextRef,
+    ) -> Result<Output> {
+        query::sql::show_columns(stmt, &self.query_engine, &self.catalog_manager, query_ctx)
+            .await
+            .context(ExecuteStatementSnafu)
+    }
+
+    #[tracing::instrument(skip_all)]
+    pub(super) async fn show_index(
+        &self,
+        stmt: ShowIndex,
+        query_ctx: QueryContextRef,
+    ) -> Result<Output> {
+        query::sql::show_index(stmt, &self.query_engine, &self.catalog_manager, query_ctx)
+            .await
+            .context(ExecuteStatementSnafu)
+    }
+
     #[tracing::instrument(skip_all)]
     pub async fn show_create_table(
         &self,