feat: create database with options (#3751)

* feat: create database with options

* fix: clippy

* fix: clippy

* feat: rebase and add Display test

* feat: sqlness test for creating database with options

* address comments

Signed-off-by: tison <wander4096@gmail.com>

* fixup tests

Signed-off-by: tison <wander4096@gmail.com>

* catch up

Signed-off-by: tison <wander4096@gmail.com>

* DefaultOnNull

Signed-off-by: tison <wander4096@gmail.com>

---------

Signed-off-by: tison <wander4096@gmail.com>
Co-authored-by: tison <wander4096@gmail.com>

src/sql/src/error.rs

@@ -123,6 +123,13 @@ pub enum Error {
     #[snafu(display("Invalid database name: {}", name))]
     InvalidDatabaseName { name: String },
 
+    #[snafu(display("Unrecognized database option key: {}", key))]
+    InvalidDatabaseOption {
+        key: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
     #[snafu(display("Invalid table name: {}", name))]
     InvalidTableName { name: String },
 
@@ -228,6 +235,7 @@ impl ErrorExt for Error {
             InvalidColumnOption { .. }
             | InvalidTableOptionValue { .. }
             | InvalidDatabaseName { .. }
+            | InvalidDatabaseOption { .. }
             | ColumnTypeMismatch { .. }
             | InvalidTableName { .. }
             | InvalidSqlValue { .. }

src/sql/src/parsers/create_parser.rs

@@ -27,8 +27,9 @@ use table::requests::validate_table_option;
 
 use crate::ast::{ColumnDef, Ident, TableConstraint};
 use crate::error::{
-    self, InvalidColumnOptionSnafu, InvalidTableOptionSnafu, InvalidTimeIndexSnafu,
-    MissingTimeIndexSnafu, Result, SyntaxSnafu, UnexpectedSnafu, UnsupportedSnafu,
+    self, InvalidColumnOptionSnafu, InvalidDatabaseOptionSnafu, InvalidTableOptionSnafu,
+    InvalidTimeIndexSnafu, MissingTimeIndexSnafu, Result, SyntaxSnafu, UnexpectedSnafu,
+    UnsupportedSnafu,
 };
 use crate::parser::{ParserContext, FLOW};
 use crate::statements::create::{
@@ -45,6 +46,12 @@ pub const SINK: &str = "SINK";
 pub const EXPIRE: &str = "EXPIRE";
 pub const WHEN: &str = "WHEN";
 
+const DB_OPT_KEY_TTL: &str = "ttl";
+
+fn validate_database_option(key: &str) -> bool {
+    [DB_OPT_KEY_TTL].contains(&key)
+}
+
 /// Parses create [table] statement
 impl<'a> ParserContext<'a> {
     pub(crate) fn parse_create(&mut self) -> Result<Statement> {
@@ -124,9 +131,28 @@ impl<'a> ParserContext<'a> {
             actual: self.peek_token_as_string(),
         })?;
         let database_name = Self::canonicalize_object_name(database_name);
+
+        let options = self
+            .parser
+            .parse_options(Keyword::WITH)
+            .context(SyntaxSnafu)?
+            .into_iter()
+            .map(parse_option_string)
+            .collect::<Result<HashMap<String, String>>>()?;
+
+        for key in options.keys() {
+            ensure!(
+                validate_database_option(key),
+                InvalidDatabaseOptionSnafu {
+                    key: key.to_string()
+                }
+            );
+        }
+
         Ok(Statement::CreateDatabase(CreateDatabase {
             name: database_name,
             if_not_exists,
+            options: options.into(),
         }))
     }
 
@@ -1025,14 +1051,27 @@ mod tests {
         let sql = "CREATE DATABASE `fOo`";
         let result =
             ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
-        let mut stmts = result.unwrap();
-        assert_eq!(
-            stmts.pop().unwrap(),
-            Statement::CreateDatabase(CreateDatabase::new(
-                ObjectName(vec![Ident::with_quote('`', "fOo"),]),
-                false
-            ))
-        );
+        let stmts = result.unwrap();
+        match &stmts.last().unwrap() {
+            Statement::CreateDatabase(c) => {
+                assert_eq!(c.name, ObjectName(vec![Ident::with_quote('`', "fOo")]));
+                assert!(!c.if_not_exists);
+            }
+            _ => unreachable!(),
+        }
+
+        let sql = "CREATE DATABASE prometheus with (ttl='1h');";
+        let result =
+            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default());
+        let stmts = result.unwrap();
+        match &stmts[0] {
+            Statement::CreateDatabase(c) => {
+                assert_eq!(c.name.to_string(), "prometheus");
+                assert!(!c.if_not_exists);
+                assert_eq!(c.options.get("ttl").unwrap(), "1h");
+            }
+            _ => unreachable!(),
+        }
     }
 
     #[test]

src/sql/src/statements/create.rs

@@ -168,14 +168,16 @@ pub struct CreateDatabase {
     pub name: ObjectName,
     /// Create if not exists
     pub if_not_exists: bool,
+    pub options: OptionMap,
 }
 
 impl CreateDatabase {
     /// Creates a statement for `CREATE DATABASE`
-    pub fn new(name: ObjectName, if_not_exists: bool) -> Self {
+    pub fn new(name: ObjectName, if_not_exists: bool, options: OptionMap) -> Self {
         Self {
             name,
             if_not_exists,
+            options,
         }
     }
 }
@@ -186,7 +188,12 @@ impl Display for CreateDatabase {
         if self.if_not_exists {
             write!(f, "IF NOT EXISTS ")?;
         }
-        write!(f, "{}", &self.name)
+        write!(f, "{}", &self.name)?;
+        if !self.options.is_empty() {
+            let options = self.options.kv_pairs();
+            write!(f, "\nWITH(\n{}\n)", format_list_indent!(options))?;
+        }
+        Ok(())
     }
 }
 
@@ -475,6 +482,30 @@ CREATE DATABASE IF NOT EXISTS test"#,
                 unreachable!();
             }
         }
+
+        let sql = r#"CREATE DATABASE IF NOT EXISTS test WITH (ttl='1h');"#;
+        let stmts =
+            ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}, ParseOptions::default())
+                .unwrap();
+        assert_eq!(1, stmts.len());
+        assert_matches!(&stmts[0], Statement::CreateDatabase { .. });
+
+        match &stmts[0] {
+            Statement::CreateDatabase(set) => {
+                let new_sql = format!("\n{}", set);
+                assert_eq!(
+                    r#"
+CREATE DATABASE IF NOT EXISTS test
+WITH(
+  ttl = '1h'
+)"#,
+                    &new_sql
+                );
+            }
+            _ => {
+                unreachable!();
+            }
+        }
     }
 
     #[test]