feat: show database options (#4174)

* test: test create table with database ttl

* feat: show database options

* fix: comment

* chore: apply suggestion

Co-authored-by: Jeremyhi <jiachun_feng@proton.me>

* chore: fix CR comments and refactor

* chore: style

Co-authored-by: Weny Xu <wenymedia@gmail.com>

---------

Co-authored-by: Jeremyhi <jiachun_feng@proton.me>
Co-authored-by: Weny Xu <wenymedia@gmail.com>

src/auth/tests/mod.rs

@@ -52,7 +52,10 @@ fn test_permission_checker() {
 
     let sql_result = checker.check_permission(
         None,
-        PermissionReq::SqlStatement(&Statement::ShowDatabases(ShowDatabases::new(ShowKind::All))),
+        PermissionReq::SqlStatement(&Statement::ShowDatabases(ShowDatabases::new(
+            ShowKind::All,
+            false,
+        ))),
     );
     assert_matches!(sql_result, Ok(PermissionResp::Reject));
 

src/catalog/src/information_schema/schemata.rs

@@ -17,6 +17,7 @@ use std::sync::{Arc, Weak};
 use arrow_schema::SchemaRef as ArrowSchemaRef;
 use common_catalog::consts::INFORMATION_SCHEMA_SCHEMATA_TABLE_ID;
 use common_error::ext::BoxedError;
+use common_meta::key::schema_name::SchemaNameKey;
 use common_recordbatch::adapter::RecordBatchStreamAdapter;
 use common_recordbatch::{RecordBatch, SendableRecordBatchStream};
 use datafusion::execution::TaskContext;
@@ -32,15 +33,18 @@ use store_api::storage::{ScanRequest, TableId};
 
 use super::SCHEMATA;
 use crate::error::{
-    CreateRecordBatchSnafu, InternalSnafu, Result, UpgradeWeakCatalogManagerRefSnafu,
+    CreateRecordBatchSnafu, InternalSnafu, Result, SchemaNotFoundSnafu, TableMetadataManagerSnafu,
+    UpgradeWeakCatalogManagerRefSnafu,
 };
-use crate::information_schema::{InformationTable, Predicates};
+use crate::information_schema::{utils, InformationTable, Predicates};
 use crate::CatalogManager;
 
 pub const CATALOG_NAME: &str = "catalog_name";
 pub const SCHEMA_NAME: &str = "schema_name";
 const DEFAULT_CHARACTER_SET_NAME: &str = "default_character_set_name";
 const DEFAULT_COLLATION_NAME: &str = "default_collation_name";
+/// The database options
+pub const SCHEMA_OPTS: &str = "options";
 const INIT_CAPACITY: usize = 42;
 
 /// The `information_schema.schemata` table implementation.
@@ -74,6 +78,7 @@ impl InformationSchemaSchemata {
                 false,
             ),
             ColumnSchema::new("sql_path", ConcreteDataType::string_datatype(), true),
+            ColumnSchema::new(SCHEMA_OPTS, ConcreteDataType::string_datatype(), true),
         ]))
     }
 
@@ -133,6 +138,7 @@ struct InformationSchemaSchemataBuilder {
     charset_names: StringVectorBuilder,
     collation_names: StringVectorBuilder,
     sql_paths: StringVectorBuilder,
+    schema_options: StringVectorBuilder,
 }
 
 impl InformationSchemaSchemataBuilder {
@@ -150,6 +156,7 @@ impl InformationSchemaSchemataBuilder {
             charset_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
             collation_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
             sql_paths: StringVectorBuilder::with_capacity(INIT_CAPACITY),
+            schema_options: StringVectorBuilder::with_capacity(INIT_CAPACITY),
         }
     }
 
@@ -160,21 +167,50 @@ impl InformationSchemaSchemataBuilder {
             .catalog_manager
             .upgrade()
             .context(UpgradeWeakCatalogManagerRefSnafu)?;
+        let table_metadata_manager = utils::table_meta_manager(&self.catalog_manager)?;
         let predicates = Predicates::from_scan_request(&request);
 
         for schema_name in catalog_manager.schema_names(&catalog_name).await? {
-            self.add_schema(&predicates, &catalog_name, &schema_name);
+            let opts = if let Some(table_metadata_manager) = &table_metadata_manager {
+                let schema_opts = table_metadata_manager
+                    .schema_manager()
+                    .get(SchemaNameKey::new(&catalog_name, &schema_name))
+                    .await
+                    .context(TableMetadataManagerSnafu)?
+                    .context(SchemaNotFoundSnafu {
+                        catalog: &catalog_name,
+                        schema: &schema_name,
+                    })?;
+
+                Some(format!("{schema_opts}"))
+            } else {
+                None
+            };
+
+            self.add_schema(
+                &predicates,
+                &catalog_name,
+                &schema_name,
+                opts.as_deref().unwrap_or(""),
+            );
         }
 
         self.finish()
     }
 
-    fn add_schema(&mut self, predicates: &Predicates, catalog_name: &str, schema_name: &str) {
+    fn add_schema(
+        &mut self,
+        predicates: &Predicates,
+        catalog_name: &str,
+        schema_name: &str,
+        schema_options: &str,
+    ) {
         let row = [
             (CATALOG_NAME, &Value::from(catalog_name)),
             (SCHEMA_NAME, &Value::from(schema_name)),
             (DEFAULT_CHARACTER_SET_NAME, &Value::from("utf8")),
             (DEFAULT_COLLATION_NAME, &Value::from("utf8_bin")),
+            (SCHEMA_OPTS, &Value::from(schema_options)),
         ];
 
         if !predicates.eval(&row) {
@@ -186,6 +222,7 @@ impl InformationSchemaSchemataBuilder {
         self.charset_names.push(Some("utf8"));
         self.collation_names.push(Some("utf8_bin"));
         self.sql_paths.push(None);
+        self.schema_options.push(Some(schema_options));
     }
 
     fn finish(&mut self) -> Result<RecordBatch> {
@@ -195,6 +232,7 @@ impl InformationSchemaSchemataBuilder {
             Arc::new(self.charset_names.finish()),
             Arc::new(self.collation_names.finish()),
             Arc::new(self.sql_paths.finish()),
+            Arc::new(self.schema_options.finish()),
         ];
         RecordBatch::new(self.schema.clone(), columns).context(CreateRecordBatchSnafu)
     }

src/catalog/src/information_schema/utils.rs

@@ -15,6 +15,7 @@
 use std::sync::{Arc, Weak};
 
 use common_config::Mode;
+use common_meta::key::TableMetadataManagerRef;
 use meta_client::client::MetaClient;
 use snafu::OptionExt;
 
@@ -51,3 +52,17 @@ pub fn meta_client(catalog_manager: &Weak<dyn CatalogManager>) -> Result<Option<
 
     Ok(meta_client)
 }
+
+/// Try to get the `[TableMetadataManagerRef]` from `[CatalogManager]` weak reference.
+pub fn table_meta_manager(
+    catalog_manager: &Weak<dyn CatalogManager>,
+) -> Result<Option<TableMetadataManagerRef>> {
+    let catalog_manager = catalog_manager
+        .upgrade()
+        .context(UpgradeWeakCatalogManagerRefSnafu)?;
+
+    Ok(catalog_manager
+        .as_any()
+        .downcast_ref::<KvBackendCatalogManager>()
+        .map(|manager| manager.table_metadata_manager_ref().clone()))
+}

src/common/meta/src/key/schema_name.rs

@@ -57,6 +57,17 @@ pub struct SchemaNameValue {
     pub ttl: Option<Duration>,
 }
 
+impl Display for SchemaNameValue {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        if let Some(ttl) = self.ttl {
+            let ttl = humantime::format_duration(ttl);
+            write!(f, "ttl='{ttl}'")?;
+        }
+
+        Ok(())
+    }
+}
+
 impl TryFrom<&HashMap<String, String>> for SchemaNameValue {
     type Error = Error;
 
@@ -233,6 +244,17 @@ mod tests {
     use super::*;
     use crate::kv_backend::memory::MemoryKvBackend;
 
+    #[test]
+    fn test_display_schema_value() {
+        let schema_value = SchemaNameValue { ttl: None };
+        assert_eq!("", schema_value.to_string());
+
+        let schema_value = SchemaNameValue {
+            ttl: Some(Duration::from_secs(9)),
+        };
+        assert_eq!("ttl='9s'", schema_value.to_string());
+    }
+
     #[test]
     fn test_serialization() {
         let key = SchemaNameKey::new("my-catalog", "my-schema");

src/query/src/sql.rs

@@ -66,6 +66,7 @@ use crate::planner::DfLogicalPlanner;
 use crate::QueryEngineRef;
 
 const SCHEMAS_COLUMN: &str = "Database";
+const OPTIONS_COLUMN: &str = "Options";
 const TABLES_COLUMN: &str = "Tables";
 const FIELD_COLUMN: &str = "Field";
 const TABLE_TYPE_COLUMN: &str = "Table_type";
@@ -155,7 +156,14 @@ pub async fn show_databases(
     catalog_manager: &CatalogManagerRef,
     query_ctx: QueryContextRef,
 ) -> Result<Output> {
-    let projects = vec![(schemata::SCHEMA_NAME, SCHEMAS_COLUMN)];
+    let projects = if stmt.full {
+        vec![
+            (schemata::SCHEMA_NAME, SCHEMAS_COLUMN),
+            (schemata::SCHEMA_OPTS, OPTIONS_COLUMN),
+        ]
+    } else {
+        vec![(schemata::SCHEMA_NAME, SCHEMAS_COLUMN)]
+    };
 
     let filters = vec![col(schemata::CATALOG_NAME).eq(lit(query_ctx.current_catalog()))];
     let like_field = Some(schemata::SCHEMA_NAME);

src/sql/src/parsers/show_parser.rs

@@ -32,7 +32,7 @@ impl<'a> ParserContext<'a> {
     /// todo(hl) support `show settings`/`show create`/`show users` etc.
     pub(crate) fn parse_show(&mut self) -> Result<Statement> {
         if self.consume_token("DATABASES") || self.consume_token("SCHEMAS") {
-            self.parse_show_databases()
+            self.parse_show_databases(false)
         } else if self.matches_keyword(Keyword::TABLES) {
             self.parser.next_token();
             self.parse_show_tables(false)
@@ -75,6 +75,8 @@ impl<'a> ParserContext<'a> {
             } else if self.consume_token("COLUMNS") || self.consume_token("FIELDS") {
                 // SHOW {COLUMNS | FIELDS}
                 self.parse_show_columns(true)
+            } else if self.consume_token("DATABASES") || self.consume_token("SCHEMAS") {
+                self.parse_show_databases(true)
             } else {
                 self.unsupported(self.peek_token_as_string())
             }
@@ -341,12 +343,13 @@ impl<'a> ParserContext<'a> {
     }
 
     /// Parses `SHOW DATABASES` statement.
-    pub fn parse_show_databases(&mut self) -> Result<Statement> {
+    pub fn parse_show_databases(&mut self, full: bool) -> Result<Statement> {
         let tok = self.parser.next_token().token;
         match &tok {
-            Token::EOF | Token::SemiColon => {
-                Ok(Statement::ShowDatabases(ShowDatabases::new(ShowKind::All)))
-            }
+            Token::EOF | Token::SemiColon => Ok(Statement::ShowDatabases(ShowDatabases::new(
+                ShowKind::All,
+                full,
+            ))),
             Token::Word(w) => match w.keyword {
                 Keyword::LIKE => Ok(Statement::ShowDatabases(ShowDatabases::new(
                     ShowKind::Like(self.parse_identifier().with_context(|_| {
@@ -356,6 +359,7 @@ impl<'a> ParserContext<'a> {
                             actual: tok.to_string(),
                         }
                     })?),
+                    full,
                 ))),
                 Keyword::WHERE => Ok(Statement::ShowDatabases(ShowDatabases::new(
                     ShowKind::Where(self.parser.parse_expr().with_context(|_| {
@@ -365,6 +369,7 @@ impl<'a> ParserContext<'a> {
                             actual: self.peek_token_as_string(),
                         }
                     })?),
+                    full,
                 ))),
                 _ => self.unsupported(self.peek_token_as_string()),
             },
@@ -395,7 +400,39 @@ mod tests {
         assert_matches!(
             &stmts[0],
             Statement::ShowDatabases(ShowDatabases {
-                kind: ShowKind::All
+                kind: ShowKind::All,
+                full: false,
+            })
+        );
+    }
+
+    #[test]
+    pub fn test_show_full_databases() {
+        let sql = "SHOW FULL DATABASES";
+        let result =
+            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
+        let stmts = result.unwrap();
+        assert_eq!(1, stmts.len());
+
+        assert_matches!(
+            &stmts[0],
+            Statement::ShowDatabases(ShowDatabases {
+                kind: ShowKind::All,
+                full: true,
+            })
+        );
+
+        let sql = "SHOW FULL DATABASES LIKE 'test%'";
+        let result =
+            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
+        let stmts = result.unwrap();
+        assert_eq!(1, stmts.len());
+
+        assert_matches!(
+            &stmts[0],
+            Statement::ShowDatabases(ShowDatabases {
+                kind: ShowKind::Like(_),
+                full: true,
             })
         );
     }
@@ -414,7 +451,8 @@ mod tests {
                 kind: ShowKind::Like(sqlparser::ast::Ident {
                     value: _,
                     quote_style: None,
-                })
+                }),
+                ..
             })
         );
     }
@@ -434,7 +472,8 @@ mod tests {
                     left: _,
                     right: _,
                     op: sqlparser::ast::BinaryOperator::Or,
-                })
+                }),
+                ..
             })
         );
     }

src/sql/src/statements/show.rs

@@ -40,6 +40,7 @@ impl Display for ShowKind {
 #[derive(Debug, Clone, PartialEq, Eq, Visit, VisitMut)]
 pub struct ShowDatabases {
     pub kind: ShowKind,
+    pub full: bool,
 }
 
 /// The SQL `SHOW COLUMNS` statement
@@ -85,15 +86,20 @@ impl Display for ShowIndex {
 
 impl ShowDatabases {
     /// Creates a statement for `SHOW DATABASES`
-    pub fn new(kind: ShowKind) -> Self {
-        ShowDatabases { kind }
+    pub fn new(kind: ShowKind, full: bool) -> Self {
+        ShowDatabases { kind, full }
     }
 }
 
 impl Display for ShowDatabases {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         let kind = &self.kind;
-        write!(f, r#"SHOW DATABASES {kind}"#)
+
+        if self.full {
+            write!(f, r#"SHOW FULL DATABASES {kind}"#)
+        } else {
+            write!(f, r#"SHOW DATABASES {kind}"#)
+        }
     }
 }
 

tests/cases/standalone/common/create/create_database.result

@@ -10,10 +10,6 @@ create database 'illegal-database';
 
 Affected Rows: 1
 
-create database mydb with (ttl = '1h');
-
-Affected Rows: 1
-
 show databases;
 
 +--------------------+
@@ -22,15 +18,10 @@ show databases;
 | greptime_private   |
 | illegal-database   |
 | information_schema |
-| mydb               |
 | public             |
 +--------------------+
 
 drop database 'illegal-database';
 
 Affected Rows: 0
 
-drop database mydb;
-
-Affected Rows: 0
-