feat: support for show full tables (GreptimeTeam#2410)

* feat: added show tables command

* fix(tests): fixed parser and statement unit tests

* chore: implemeted display trait for table type

* fix: handled no tabletype and error for usopprted command in show databse

* chore: removed full as a show kind, instead as a show option

* chore(tests): fixed failing test and added more tests for show full

* chore: refactored table types to use filters

* fix: changed table_type to tables

src/datatypes/src/vectors/helper.rs

@@ -350,6 +350,15 @@ impl Helper {
         let result = compute::filter(&array, &filter).context(error::ArrowComputeSnafu)?;
         Helper::try_into_vector(result)
     }
+
+    pub fn like_utf8_filter(names: Vec<String>, s: &str) -> Result<(VectorRef, BooleanVector)> {
+        let array = StringArray::from(names);
+        let filter = comparison::like_utf8_scalar(&array, s).context(error::ArrowComputeSnafu)?;
+        let result = compute::filter(&array, &filter).context(error::ArrowComputeSnafu)?;
+        let vector = Helper::try_into_vector(result)?;
+
+        Ok((vector, BooleanVector::from(filter)))
+    }
 }
 
 #[cfg(test)]
@@ -463,6 +472,37 @@ mod tests {
         assert_vector(vec!["greptime", "hello", "public", "world"], &ret);
     }
 
+    #[test]
+    fn test_like_utf8_filter() {
+        fn assert_vector(expected: Vec<&str>, actual: &VectorRef) {
+            let actual = actual.as_any().downcast_ref::<StringVector>().unwrap();
+            assert_eq!(*actual, StringVector::from(expected));
+        }
+
+        fn assert_filter(array: Vec<String>, s: &str, expected_filter: &BooleanVector) {
+            let array = StringArray::from(array);
+            let actual_filter = comparison::like_utf8_scalar(&array, s).unwrap();
+            assert_eq!(BooleanVector::from(actual_filter), *expected_filter);
+        }
+
+        let names: Vec<String> = vec!["greptime", "timeseries", "cloud", "database"]
+            .into_iter()
+            .map(|x| x.to_string())
+            .collect();
+
+        let (table, filter) = Helper::like_utf8_filter(names.clone(), "%ti%").unwrap();
+        assert_vector(vec!["greptime", "timeseries"], &table);
+        assert_filter(names.clone(), "%ti%", &filter);
+
+        let (tables, filter) = Helper::like_utf8_filter(names.clone(), "%lou").unwrap();
+        assert_vector(vec![], &tables);
+        assert_filter(names.clone(), "%lou", &filter);
+
+        let (tables, filter) = Helper::like_utf8_filter(names.clone(), "%d%").unwrap();
+        assert_vector(vec!["cloud", "database"], &tables);
+        assert_filter(names.clone(), "%d%", &filter);
+    }
+
     fn check_try_into_vector(array: impl Array + 'static) {
         let array: ArrayRef = Arc::new(array);
         let vector = Helper::try_into_vector(array.clone()).unwrap();

src/query/src/sql.rs

@@ -144,42 +144,86 @@ pub async fn show_tables(
     catalog_manager: CatalogManagerRef,
     query_ctx: QueryContextRef,
 ) -> Result<Output> {
-    let schema = if let Some(database) = stmt.database {
+    let schema_name = if let Some(database) = stmt.database {
         database
     } else {
         query_ctx.current_schema().to_owned()
     };
     // TODO(sunng87): move this function into query_ctx
     let mut tables = catalog_manager
-        .table_names(query_ctx.current_catalog(), &schema)
+        .table_names(query_ctx.current_catalog(), &schema_name)
         .await
         .context(error::CatalogSnafu)?;
 
     // TODO(dennis): Specify the order of the results in schema provider API
     tables.sort();
-    let schema = Arc::new(Schema::new(vec![ColumnSchema::new(
+
+    let table_types: Option<Arc<dyn Vector>> = {
+        if stmt.full {
+            Some(
+                get_table_types(
+                    &tables,
+                    catalog_manager.clone(),
+                    query_ctx.clone(),
+                    &schema_name,
+                )
+                .await?,
+            )
+        } else {
+            None
+        }
+    };
+
+    let mut column_schema = vec![ColumnSchema::new(
         TABLES_COLUMN,
         ConcreteDataType::string_datatype(),
         false,
-    )]));
+    )];
+    if table_types.is_some() {
+        column_schema.push(ColumnSchema::new(
+            "Table_type",
+            ConcreteDataType::string_datatype(),
+            false,
+        ));
+    }
+
+    let schema = Arc::new(Schema::new(column_schema));
+
     match stmt.kind {
         ShowKind::All => {
             let tables = Arc::new(StringVector::from(tables)) as _;
-            let records = RecordBatches::try_from_columns(schema, vec![tables])
+            let mut columns = vec![tables];
+            if let Some(table_types) = table_types {
+                columns.push(table_types)
+            }
+
+            let records = RecordBatches::try_from_columns(schema, columns)
                 .context(error::CreateRecordBatchSnafu)?;
             Ok(Output::RecordBatches(records))
         }
         ShowKind::Where(filter) => {
-            let columns = vec![Arc::new(StringVector::from(tables)) as _];
+            let mut columns = vec![Arc::new(StringVector::from(tables)) as _];
+            if let Some(table_types) = table_types {
+                columns.push(table_types)
+            }
             let record_batch =
                 RecordBatch::new(schema, columns).context(error::CreateRecordBatchSnafu)?;
             let result = execute_show_with_filter(record_batch, Some(filter)).await?;
             Ok(result)
         }
         ShowKind::Like(ident) => {
-            let tables =
-                Helper::like_utf8(tables, &ident.value).context(error::VectorComputationSnafu)?;
-            let records = RecordBatches::try_from_columns(schema, vec![tables])
+            let (tables, filter) = Helper::like_utf8_filter(tables, &ident.value)
+                .context(error::VectorComputationSnafu)?;
+            let mut columns = vec![tables];
+
+            if let Some(table_types) = table_types {
+                let table_types = table_types
+                    .filter(&filter)
+                    .context(error::VectorComputationSnafu)?;
+                columns.push(table_types)
+            }
+
+            let records = RecordBatches::try_from_columns(schema, columns)
                 .context(error::CreateRecordBatchSnafu)?;
             Ok(Output::RecordBatches(records))
         }
@@ -452,6 +496,25 @@ fn parse_file_table_format(options: &HashMap<String, String>) -> Result<Box<dyn
     )
 }
 
+async fn get_table_types(
+    tables: &[String],
+    catalog_manager: CatalogManagerRef,
+    query_ctx: QueryContextRef,
+    schema_name: &str,
+) -> Result<Arc<dyn Vector>> {
+    let mut table_types = Vec::with_capacity(tables.len());
+    for table_name in tables {
+        if let Some(table) = catalog_manager
+            .table(query_ctx.current_catalog(), schema_name, table_name)
+            .await
+            .context(error::CatalogSnafu)?
+        {
+            table_types.push(table.table_type().to_string());
+        }
+    }
+    Ok(Arc::new(StringVector::from(table_types)) as _)
+}
+
 #[cfg(test)]
 mod test {
     use std::sync::Arc;

src/sql/src/parsers/show_parser.rs

@@ -30,13 +30,19 @@ impl<'a> ParserContext<'a> {
             self.parse_show_databases()
         } else if self.matches_keyword(Keyword::TABLES) {
             let _ = self.parser.next_token();
-            self.parse_show_tables()
+            self.parse_show_tables(false)
         } else if self.consume_token("CREATE") {
             if self.consume_token("TABLE") {
                 self.parse_show_create_table()
             } else {
                 self.unsupported(self.peek_token_as_string())
             }
+        } else if self.consume_token("FULL") {
+            if self.consume_token("TABLES") {
+                self.parse_show_tables(true)
+            } else {
+                self.unsupported(self.peek_token_as_string())
+            }
         } else {
             self.unsupported(self.peek_token_as_string())
         }
@@ -61,12 +67,13 @@ impl<'a> ParserContext<'a> {
         Ok(Statement::ShowCreateTable(ShowCreateTable { table_name }))
     }
 
-    fn parse_show_tables(&mut self) -> Result<Statement> {
+    fn parse_show_tables(&mut self, full: bool) -> Result<Statement> {
         let database = match self.parser.peek_token().token {
             Token::EOF | Token::SemiColon => {
                 return Ok(Statement::ShowTables(ShowTables {
                     kind: ShowKind::All,
                     database: None,
+                    full,
                 }));
             }
 
@@ -126,7 +133,11 @@ impl<'a> ParserContext<'a> {
             _ => return self.unsupported(self.peek_token_as_string()),
         };
 
-        Ok(Statement::ShowTables(ShowTables { kind, database }))
+        Ok(Statement::ShowTables(ShowTables {
+            kind,
+            database,
+            full,
+        }))
     }
 
     /// Parses `SHOW DATABASES` statement.
@@ -234,6 +245,7 @@ mod tests {
             Statement::ShowTables(ShowTables {
                 kind: ShowKind::All,
                 database: None,
+                full: false
             })
         );
     }
@@ -253,6 +265,7 @@ mod tests {
                     quote_style: None,
                 }),
                 database: None,
+                full: false
             })
         );
 
@@ -269,6 +282,7 @@ mod tests {
                     quote_style: None,
                 }),
                 database: Some(_),
+                full: false
             })
         );
     }
@@ -285,6 +299,7 @@ mod tests {
             Statement::ShowTables(ShowTables {
                 kind: ShowKind::Where(sqlparser::ast::Expr::Like { .. }),
                 database: None,
+                full: false
             })
         );
 
@@ -298,6 +313,76 @@ mod tests {
             Statement::ShowTables(ShowTables {
                 kind: ShowKind::Where(sqlparser::ast::Expr::Like { .. }),
                 database: Some(_),
+                full: false
+            })
+        );
+    }
+
+    #[test]
+    pub fn test_show_full_tables() {
+        let sql = "SHOW FULL TABLES";
+        let stmts = ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}).unwrap();
+        assert_eq!(1, stmts.len());
+        assert_matches!(&stmts[0], Statement::ShowTables { .. });
+        match &stmts[0] {
+            Statement::ShowTables(show) => {
+                assert!(show.full);
+            }
+            _ => {
+                unreachable!();
+            }
+        }
+    }
+
+    #[test]
+    pub fn test_show_full_tables_where() {
+        let sql = "SHOW FULL TABLES IN test_db WHERE Tables LIKE test_table";
+        let stmts = ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}).unwrap();
+        assert_eq!(1, stmts.len());
+
+        assert_matches!(
+            &stmts[0],
+            Statement::ShowTables(ShowTables {
+                kind: ShowKind::Where(sqlparser::ast::Expr::Like { .. }),
+                database: Some(_),
+                full: true
+            })
+        );
+    }
+
+    #[test]
+    pub fn test_show_full_tables_like() {
+        let sql = "SHOW FULL TABLES LIKE test_table";
+        let result = ParserContext::create_with_dialect(sql, &GreptimeDbDialect {});
+        let stmts = result.unwrap();
+        assert_eq!(1, stmts.len());
+
+        assert_matches!(
+            &stmts[0],
+            Statement::ShowTables(ShowTables {
+                kind: ShowKind::Like(sqlparser::ast::Ident {
+                    value: _,
+                    quote_style: None,
+                }),
+                database: None,
+                full: true
+            })
+        );
+
+        let sql = "SHOW FULL TABLES in test_db LIKE test_table";
+        let result = ParserContext::create_with_dialect(sql, &GreptimeDbDialect {});
+        let stmts = result.unwrap();
+        assert_eq!(1, stmts.len());
+
+        assert_matches!(
+            &stmts[0],
+            Statement::ShowTables(ShowTables {
+                kind: ShowKind::Like(sqlparser::ast::Ident {
+                    value: _,
+                    quote_style: None,
+                }),
+                database: Some(_),
+                full: true
             })
         );
     }

src/sql/src/statements/show.rs

@@ -54,6 +54,7 @@ impl ShowDatabases {
 pub struct ShowTables {
     pub kind: ShowKind,
     pub database: Option<String>,
+    pub full: bool,
 }
 
 /// SQL structure for `SHOW CREATE TABLE`.

src/table/src/metadata.rs

@@ -80,6 +80,16 @@ pub enum TableType {
     Temporary,
 }
 
+impl std::fmt::Display for TableType {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            TableType::Base => f.write_str("BASE TABLE"),
+            TableType::Temporary => f.write_str("TEMPORARY"),
+            TableType::View => f.write_str("VIEW"),
+        }
+    }
+}
+
 /// Identifier of the table.
 #[derive(Serialize, Deserialize, Clone, Debug, Eq, PartialEq, Default)]
 pub struct TableIdent {