feat: impl show table status (#4303)

* feat: impl show table status

* chore: style and comment

* test: revert lz4 compression

src/catalog/src/information_schema/tables.rs

@@ -64,7 +64,7 @@ pub const TABLE_COMMENT: &str = "table_comment";
 pub const MAX_INDEX_LENGTH: &str = "max_index_length";
 pub const TEMPORARY: &str = "temporary";
 const TABLE_ID: &str = "table_id";
-const ENGINE: &str = "engine";
+pub const ENGINE: &str = "engine";
 const INIT_CAPACITY: usize = 42;
 
 pub(super) struct InformationSchemaTables {

src/common/wal/src/config.rs

@@ -207,7 +207,7 @@ mod tests {
         let datanode_wal_config: DatanodeWalConfig = toml::from_str(toml_str).unwrap();
         let expected = DatanodeKafkaConfig {
             broker_endpoints: vec!["127.0.0.1:9092".to_string()],
-            compression: Compression::Lz4,
+            compression: Compression::NoCompression,
             max_batch_bytes: ReadableSize::mb(1),
             consumer_wait_timeout: Duration::from_millis(100),
             backoff: BackoffConfig {
@@ -229,7 +229,7 @@ mod tests {
             num_partitions: 1,
             replication_factor: 1,
             create_topic_timeout: Duration::from_secs(30),
-            compression: Compression::Lz4,
+            compression: Compression::NoCompression,
             max_batch_bytes: ReadableSize::mb(1),
             consumer_wait_timeout: Duration::from_millis(100),
             backoff: BackoffConfig {

src/common/wal/src/config/kafka/datanode.rs

@@ -46,7 +46,7 @@ impl Default for DatanodeKafkaConfig {
     fn default() -> Self {
         Self {
             broker_endpoints: vec![BROKER_ENDPOINT.to_string()],
-            compression: Compression::Lz4,
+            compression: Compression::NoCompression,
             // Warning: Kafka has a default limit of 1MB per message in a topic.
             max_batch_bytes: ReadableSize::mb(1),
             consumer_wait_timeout: Duration::from_millis(100),

src/common/wal/src/config/kafka/standalone.rs

@@ -67,7 +67,7 @@ impl Default for StandaloneKafkaConfig {
             num_partitions: 1,
             replication_factor,
             create_topic_timeout: Duration::from_secs(30),
-            compression: Compression::Lz4,
+            compression: Compression::NoCompression,
             // Warning: Kafka has a default limit of 1MB per message in a topic.
             max_batch_bytes: ReadableSize::mb(1),
             consumer_wait_timeout: Duration::from_millis(100),

src/frontend/src/instance.rs

@@ -520,6 +520,9 @@ pub fn check_permission(
         Statement::ShowTables(stmt) => {
             validate_db_permission!(stmt, query_ctx);
         }
+        Statement::ShowTableStatus(stmt) => {
+            validate_db_permission!(stmt, query_ctx);
+        }
         Statement::ShowColumns(stmt) => {
             validate_db_permission!(stmt, query_ctx);
         }

src/operator/src/statement.rs

@@ -126,6 +126,8 @@ impl StatementExecutor {
 
             Statement::ShowTables(stmt) => self.show_tables(stmt, query_ctx).await,
 
+            Statement::ShowTableStatus(stmt) => self.show_table_status(stmt, query_ctx).await,
+
             Statement::ShowCollation(kind) => self.show_collation(kind, query_ctx).await,
 
             Statement::ShowCharset(kind) => self.show_charset(kind, query_ctx).await,

src/operator/src/statement/show.rs

@@ -22,7 +22,7 @@ use snafu::ResultExt;
 use sql::ast::Ident;
 use sql::statements::create::Partitions;
 use sql::statements::show::{
-    ShowColumns, ShowDatabases, ShowIndex, ShowKind, ShowTables, ShowVariables,
+    ShowColumns, ShowDatabases, ShowIndex, ShowKind, ShowTableStatus, ShowTables, ShowVariables,
 };
 use sqlparser::ast::ObjectName;
 use table::metadata::TableType;
@@ -55,6 +55,17 @@ impl StatementExecutor {
             .context(ExecuteStatementSnafu)
     }
 
+    #[tracing::instrument(skip_all)]
+    pub(super) async fn show_table_status(
+        &self,
+        stmt: ShowTableStatus,
+        query_ctx: QueryContextRef,
+    ) -> Result<Output> {
+        query::sql::show_table_status(stmt, &self.query_engine, &self.catalog_manager, query_ctx)
+            .await
+            .context(ExecuteStatementSnafu)
+    }
+
     #[tracing::instrument(skip_all)]
     pub(super) async fn show_columns(
         &self,

src/query/src/sql.rs

@@ -54,7 +54,7 @@ use sql::ast::Ident;
 use sql::parser::ParserContext;
 use sql::statements::create::{CreateFlow, Partitions};
 use sql::statements::show::{
-    ShowColumns, ShowDatabases, ShowIndex, ShowKind, ShowTables, ShowVariables,
+    ShowColumns, ShowDatabases, ShowIndex, ShowKind, ShowTableStatus, ShowTables, ShowVariables,
 };
 use sqlparser::ast::ObjectName;
 use table::requests::{FILE_TABLE_LOCATION_KEY, FILE_TABLE_PATTERN_KEY};
@@ -445,6 +445,7 @@ pub async fn show_index(
     .await
 }
 
+/// Execute [`ShowTables`] statement and return the [`Output`] if success.
 pub async fn show_tables(
     stmt: ShowTables,
     query_engine: &QueryEngineRef,
@@ -489,6 +490,63 @@ pub async fn show_tables(
     .await
 }
 
+/// Execute [`ShowTableStatus`] statement and return the [`Output`] if success.
+pub async fn show_table_status(
+    stmt: ShowTableStatus,
+    query_engine: &QueryEngineRef,
+    catalog_manager: &CatalogManagerRef,
+    query_ctx: QueryContextRef,
+) -> Result<Output> {
+    let schema_name = if let Some(database) = stmt.database {
+        database
+    } else {
+        query_ctx.current_schema()
+    };
+
+    // Refer to https://dev.mysql.com/doc/refman/8.4/en/show-table-status.html
+    let projects = vec![
+        (tables::TABLE_NAME, "Name"),
+        (tables::ENGINE, "Engine"),
+        (tables::VERSION, "Version"),
+        (tables::ROW_FORMAT, "Row_format"),
+        (tables::TABLE_ROWS, "Rows"),
+        (tables::AVG_ROW_LENGTH, "Avg_row_length"),
+        (tables::DATA_LENGTH, "Data_length"),
+        (tables::MAX_DATA_LENGTH, "Max_data_length"),
+        (tables::INDEX_LENGTH, "Index_length"),
+        (tables::DATA_FREE, "Data_free"),
+        (tables::AUTO_INCREMENT, "Auto_increment"),
+        (tables::CREATE_TIME, "Create_time"),
+        (tables::UPDATE_TIME, "Update_time"),
+        (tables::CHECK_TIME, "Check_time"),
+        (tables::TABLE_COLLATION, "Collation"),
+        (tables::CHECKSUM, "Checksum"),
+        (tables::CREATE_OPTIONS, "Create_options"),
+        (tables::TABLE_COMMENT, "Comment"),
+    ];
+
+    let filters = vec![
+        col(tables::TABLE_SCHEMA).eq(lit(schema_name.clone())),
+        col(tables::TABLE_CATALOG).eq(lit(query_ctx.current_catalog())),
+    ];
+    let like_field = Some(tables::TABLE_NAME);
+    let sort = vec![col(tables::TABLE_NAME).sort(true, true)];
+
+    query_from_information_schema_table(
+        query_engine,
+        catalog_manager,
+        query_ctx,
+        TABLES,
+        vec![],
+        projects,
+        filters,
+        like_field,
+        sort,
+        stmt.kind,
+    )
+    .await
+}
+
 /// Execute `SHOW COLLATION` statement and returns the `Output` if success.
 pub async fn show_collations(
     kind: ShowKind,

src/sql/src/parsers/show_parser.rs

@@ -22,7 +22,7 @@ use crate::error::{
 use crate::parser::ParserContext;
 use crate::statements::show::{
     ShowColumns, ShowCreateFlow, ShowCreateTable, ShowDatabases, ShowIndex, ShowKind, ShowStatus,
-    ShowTables, ShowVariables,
+    ShowTableStatus, ShowTables, ShowVariables,
 };
 use crate::statements::statement::Statement;
 
@@ -36,6 +36,14 @@ impl<'a> ParserContext<'a> {
         } else if self.matches_keyword(Keyword::TABLES) {
             self.parser.next_token();
             self.parse_show_tables(false)
+        } else if self.matches_keyword(Keyword::TABLE) {
+            self.parser.next_token();
+            if self.matches_keyword(Keyword::STATUS) {
+                self.parser.next_token();
+                self.parse_show_table_status()
+            } else {
+                self.unsupported(self.peek_token_as_string())
+            }
         } else if self.matches_keyword(Keyword::CHARSET) {
             self.parser.next_token();
             Ok(Statement::ShowCharset(self.parse_show_kind()?))
@@ -342,6 +350,32 @@ impl<'a> ParserContext<'a> {
         }))
     }
 
+    fn parse_show_table_status(&mut self) -> Result<Statement> {
+        let database = match self.parser.peek_token().token {
+            Token::EOF | Token::SemiColon => {
+                return Ok(Statement::ShowTableStatus(ShowTableStatus {
+                    kind: ShowKind::All,
+                    database: None,
+                }));
+            }
+
+            // SHOW TABLE STATUS [in | FROM] [DATABASE]
+            Token::Word(w) => match w.keyword {
+                Keyword::IN | Keyword::FROM => self.parse_db_name()?,
+
+                _ => None,
+            },
+            _ => None,
+        };
+
+        let kind = self.parse_show_kind()?;
+
+        Ok(Statement::ShowTableStatus(ShowTableStatus {
+            kind,
+            database,
+        }))
+    }
+
     /// Parses `SHOW DATABASES` statement.
     pub fn parse_show_databases(&mut self, full: bool) -> Result<Statement> {
         let tok = self.parser.next_token().token;
@@ -835,4 +869,41 @@ mod tests {
             Statement::ShowCharset(ShowKind::Like(_))
         ));
     }
+
+    fn parse_show_table_status(sql: &str) -> ShowTableStatus {
+        let result =
+            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
+        let mut stmts = result.unwrap();
+        assert_eq!(1, stmts.len());
+
+        match stmts.remove(0) {
+            Statement::ShowTableStatus(stmt) => stmt,
+            _ => panic!("Failed to parse show table status"),
+        }
+    }
+
+    #[test]
+    pub fn test_show_table_status() {
+        let sql = "SHOW TABLE STATUS";
+        let stmt = parse_show_table_status(sql);
+        assert!(stmt.database.is_none());
+        assert_eq!(sql, stmt.to_string());
+
+        let sql = "SHOW TABLE STATUS IN test";
+        let stmt = parse_show_table_status(sql);
+        assert_eq!("test", stmt.database.as_ref().unwrap());
+        assert_eq!(sql, stmt.to_string());
+
+        let sql = "SHOW TABLE STATUS LIKE '%monitor'";
+        let stmt = parse_show_table_status(sql);
+        assert!(stmt.database.is_none());
+        assert!(matches!(stmt.kind, ShowKind::Like(_)));
+        assert_eq!(sql, stmt.to_string());
+
+        let sql = "SHOW TABLE STATUS IN test WHERE Name = 'monitor'";
+        let stmt = parse_show_table_status(sql);
+        assert_eq!("test", stmt.database.as_ref().unwrap());
+        assert!(matches!(stmt.kind, ShowKind::Where(_)));
+        assert_eq!(sql, stmt.to_string());
+    }
 }