Snowflake create database (#1939)

src/ast/helpers/mod.rs

@@ -16,5 +16,6 @@
 // under the License.
 pub mod attached_token;
 pub mod key_value_options;
+pub mod stmt_create_database;
 pub mod stmt_create_table;
 pub mod stmt_data_loading;

src/ast/helpers/stmt_create_database.rs

@@ -0,0 +1,324 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#[cfg(not(feature = "std"))]
+use alloc::{boxed::Box, format, string::String, vec, vec::Vec};
+
+#[cfg(feature = "serde")]
+use serde::{Deserialize, Serialize};
+
+#[cfg(feature = "visitor")]
+use sqlparser_derive::{Visit, VisitMut};
+
+use crate::ast::{
+    CatalogSyncNamespaceMode, ContactEntry, ObjectName, Statement, StorageSerializationPolicy, Tag,
+};
+use crate::parser::ParserError;
+
+/// Builder for create database statement variant ([1]).
+///
+/// This structure helps building and accessing a create database with more ease, without needing to:
+/// - Match the enum itself a lot of times; or
+/// - Moving a lot of variables around the code.
+///
+/// # Example
+/// ```rust
+/// use sqlparser::ast::helpers::stmt_create_database::CreateDatabaseBuilder;
+/// use sqlparser::ast::{ColumnDef, Ident, ObjectName};
+/// let builder = CreateDatabaseBuilder::new(ObjectName::from(vec![Ident::new("database_name")]))
+///    .if_not_exists(true);
+/// // You can access internal elements with ease
+/// assert!(builder.if_not_exists);
+/// // Convert to a statement
+/// assert_eq!(
+///    builder.build().to_string(),
+///    "CREATE DATABASE IF NOT EXISTS database_name"
+/// )
+/// ```
+///
+/// [1]: Statement::CreateDatabase
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[cfg_attr(feature = "visitor", derive(Visit, VisitMut))]
+pub struct CreateDatabaseBuilder {
+    pub db_name: ObjectName,
+    pub if_not_exists: bool,
+    pub location: Option<String>,
+    pub managed_location: Option<String>,
+    pub or_replace: bool,
+    pub transient: bool,
+    pub clone: Option<ObjectName>,
+    pub data_retention_time_in_days: Option<u64>,
+    pub max_data_extension_time_in_days: Option<u64>,
+    pub external_volume: Option<String>,
+    pub catalog: Option<String>,
+    pub replace_invalid_characters: Option<bool>,
+    pub default_ddl_collation: Option<String>,
+    pub storage_serialization_policy: Option<StorageSerializationPolicy>,
+    pub comment: Option<String>,
+    pub catalog_sync: Option<String>,
+    pub catalog_sync_namespace_mode: Option<CatalogSyncNamespaceMode>,
+    pub catalog_sync_namespace_flatten_delimiter: Option<String>,
+    pub with_tags: Option<Vec<Tag>>,
+    pub with_contacts: Option<Vec<ContactEntry>>,
+}
+
+impl CreateDatabaseBuilder {
+    pub fn new(name: ObjectName) -> Self {
+        Self {
+            db_name: name,
+            if_not_exists: false,
+            location: None,
+            managed_location: None,
+            or_replace: false,
+            transient: false,
+            clone: None,
+            data_retention_time_in_days: None,
+            max_data_extension_time_in_days: None,
+            external_volume: None,
+            catalog: None,
+            replace_invalid_characters: None,
+            default_ddl_collation: None,
+            storage_serialization_policy: None,
+            comment: None,
+            catalog_sync: None,
+            catalog_sync_namespace_mode: None,
+            catalog_sync_namespace_flatten_delimiter: None,
+            with_tags: None,
+            with_contacts: None,
+        }
+    }
+
+    pub fn location(mut self, location: Option<String>) -> Self {
+        self.location = location;
+        self
+    }
+
+    pub fn managed_location(mut self, managed_location: Option<String>) -> Self {
+        self.managed_location = managed_location;
+        self
+    }
+
+    pub fn or_replace(mut self, or_replace: bool) -> Self {
+        self.or_replace = or_replace;
+        self
+    }
+
+    pub fn transient(mut self, transient: bool) -> Self {
+        self.transient = transient;
+        self
+    }
+
+    pub fn if_not_exists(mut self, if_not_exists: bool) -> Self {
+        self.if_not_exists = if_not_exists;
+        self
+    }
+
+    pub fn clone_clause(mut self, clone: Option<ObjectName>) -> Self {
+        self.clone = clone;
+        self
+    }
+
+    pub fn data_retention_time_in_days(mut self, data_retention_time_in_days: Option<u64>) -> Self {
+        self.data_retention_time_in_days = data_retention_time_in_days;
+        self
+    }
+
+    pub fn max_data_extension_time_in_days(
+        mut self,
+        max_data_extension_time_in_days: Option<u64>,
+    ) -> Self {
+        self.max_data_extension_time_in_days = max_data_extension_time_in_days;
+        self
+    }
+
+    pub fn external_volume(mut self, external_volume: Option<String>) -> Self {
+        self.external_volume = external_volume;
+        self
+    }
+
+    pub fn catalog(mut self, catalog: Option<String>) -> Self {
+        self.catalog = catalog;
+        self
+    }
+
+    pub fn replace_invalid_characters(mut self, replace_invalid_characters: Option<bool>) -> Self {
+        self.replace_invalid_characters = replace_invalid_characters;
+        self
+    }
+
+    pub fn default_ddl_collation(mut self, default_ddl_collation: Option<String>) -> Self {
+        self.default_ddl_collation = default_ddl_collation;
+        self
+    }
+
+    pub fn storage_serialization_policy(
+        mut self,
+        storage_serialization_policy: Option<StorageSerializationPolicy>,
+    ) -> Self {
+        self.storage_serialization_policy = storage_serialization_policy;
+        self
+    }
+
+    pub fn comment(mut self, comment: Option<String>) -> Self {
+        self.comment = comment;
+        self
+    }
+
+    pub fn catalog_sync(mut self, catalog_sync: Option<String>) -> Self {
+        self.catalog_sync = catalog_sync;
+        self
+    }
+
+    pub fn catalog_sync_namespace_mode(
+        mut self,
+        catalog_sync_namespace_mode: Option<CatalogSyncNamespaceMode>,
+    ) -> Self {
+        self.catalog_sync_namespace_mode = catalog_sync_namespace_mode;
+        self
+    }
+
+    pub fn catalog_sync_namespace_flatten_delimiter(
+        mut self,
+        catalog_sync_namespace_flatten_delimiter: Option<String>,
+    ) -> Self {
+        self.catalog_sync_namespace_flatten_delimiter = catalog_sync_namespace_flatten_delimiter;
+        self
+    }
+
+    pub fn with_tags(mut self, with_tags: Option<Vec<Tag>>) -> Self {
+        self.with_tags = with_tags;
+        self
+    }
+
+    pub fn with_contacts(mut self, with_contacts: Option<Vec<ContactEntry>>) -> Self {
+        self.with_contacts = with_contacts;
+        self
+    }
+
+    pub fn build(self) -> Statement {
+        Statement::CreateDatabase {
+            db_name: self.db_name,
+            if_not_exists: self.if_not_exists,
+            managed_location: self.managed_location,
+            location: self.location,
+            or_replace: self.or_replace,
+            transient: self.transient,
+            clone: self.clone,
+            data_retention_time_in_days: self.data_retention_time_in_days,
+            max_data_extension_time_in_days: self.max_data_extension_time_in_days,
+            external_volume: self.external_volume,
+            catalog: self.catalog,
+            replace_invalid_characters: self.replace_invalid_characters,
+            default_ddl_collation: self.default_ddl_collation,
+            storage_serialization_policy: self.storage_serialization_policy,
+            comment: self.comment,
+            catalog_sync: self.catalog_sync,
+            catalog_sync_namespace_mode: self.catalog_sync_namespace_mode,
+            catalog_sync_namespace_flatten_delimiter: self.catalog_sync_namespace_flatten_delimiter,
+            with_tags: self.with_tags,
+            with_contacts: self.with_contacts,
+        }
+    }
+}
+
+impl TryFrom<Statement> for CreateDatabaseBuilder {
+    type Error = ParserError;
+
+    fn try_from(stmt: Statement) -> Result<Self, Self::Error> {
+        match stmt {
+            Statement::CreateDatabase {
+                db_name,
+                if_not_exists,
+                location,
+                managed_location,
+                or_replace,
+                transient,
+                clone,
+                data_retention_time_in_days,
+                max_data_extension_time_in_days,
+                external_volume,
+                catalog,
+                replace_invalid_characters,
+                default_ddl_collation,
+                storage_serialization_policy,
+                comment,
+                catalog_sync,
+                catalog_sync_namespace_mode,
+                catalog_sync_namespace_flatten_delimiter,
+                with_tags,
+                with_contacts,
+            } => Ok(Self {
+                db_name,
+                if_not_exists,
+                location,
+                managed_location,
+                or_replace,
+                transient,
+                clone,
+                data_retention_time_in_days,
+                max_data_extension_time_in_days,
+                external_volume,
+                catalog,
+                replace_invalid_characters,
+                default_ddl_collation,
+                storage_serialization_policy,
+                comment,
+                catalog_sync,
+                catalog_sync_namespace_mode,
+                catalog_sync_namespace_flatten_delimiter,
+                with_tags,
+                with_contacts,
+            }),
+            _ => Err(ParserError::ParserError(format!(
+                "Expected create database statement, but received: {stmt}"
+            ))),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::ast::helpers::stmt_create_database::CreateDatabaseBuilder;
+    use crate::ast::{Ident, ObjectName, Statement};
+    use crate::parser::ParserError;
+
+    #[test]
+    pub fn test_from_valid_statement() {
+        let builder = CreateDatabaseBuilder::new(ObjectName::from(vec![Ident::new("db_name")]));
+
+        let stmt = builder.clone().build();
+
+        assert_eq!(builder, CreateDatabaseBuilder::try_from(stmt).unwrap());
+    }
+
+    #[test]
+    pub fn test_from_invalid_statement() {
+        let stmt = Statement::Commit {
+            chain: false,
+            end: false,
+            modifier: None,
+        };
+
+        assert_eq!(
+            CreateDatabaseBuilder::try_from(stmt).unwrap_err(),
+            ParserError::ParserError(
+                "Expected create database statement, but received: COMMIT".to_owned()
+            )
+        );
+    }
+}

src/ast/mod.rs

@@ -3867,19 +3867,29 @@ pub enum Statement {
     /// ```sql
     /// CREATE DATABASE
     /// ```
+    /// See:
+    /// <https://docs.snowflake.com/en/sql-reference/sql/create-database>
     CreateDatabase {
         db_name: ObjectName,
         if_not_exists: bool,
         location: Option<String>,
         managed_location: Option<String>,
-        /// Clones a database
+        or_replace: bool,
-        ///
+        transient: bool,
-        /// ```sql
-        /// CREATE DATABASE mydb CLONE otherdb
-        /// ```
-        ///
-        /// [Snowflake](https://docs.snowflake.com/en/sql-reference/sql/create-clone#databases-schemas)
         clone: Option<ObjectName>,
+        data_retention_time_in_days: Option<u64>,
+        max_data_extension_time_in_days: Option<u64>,
+        external_volume: Option<String>,
+        catalog: Option<String>,
+        replace_invalid_characters: Option<bool>,
+        default_ddl_collation: Option<String>,
+        storage_serialization_policy: Option<StorageSerializationPolicy>,
+        comment: Option<String>,
+        catalog_sync: Option<String>,
+        catalog_sync_namespace_mode: Option<CatalogSyncNamespaceMode>,
+        catalog_sync_namespace_flatten_delimiter: Option<String>,
+        with_tags: Option<Vec<Tag>>,
+        with_contacts: Option<Vec<ContactEntry>>,
     },
     /// ```sql
     /// CREATE FUNCTION
@@ -4836,13 +4846,32 @@ impl fmt::Display for Statement {
                 if_not_exists,
                 location,
                 managed_location,
+                or_replace,
+                transient,
                 clone,
+                data_retention_time_in_days,
+                max_data_extension_time_in_days,
+                external_volume,
+                catalog,
+                replace_invalid_characters,
+                default_ddl_collation,
+                storage_serialization_policy,
+                comment,
+                catalog_sync,
+                catalog_sync_namespace_mode,
+                catalog_sync_namespace_flatten_delimiter,
+                with_tags,
+                with_contacts,
             } => {
-                write!(f, "CREATE DATABASE")?;
+                write!(
-                if *if_not_exists {
+                    f,
-                    write!(f, " IF NOT EXISTS")?;
+                    "CREATE {or_replace}{transient}DATABASE {if_not_exists}{name}",
-                }
+                    or_replace = if *or_replace { "OR REPLACE " } else { "" },
-                write!(f, " {db_name}")?;
+                    transient = if *transient { "TRANSIENT " } else { "" },
+                    if_not_exists = if *if_not_exists { "IF NOT EXISTS " } else { "" },
+                    name = db_name,
+                )?;
+
                 if let Some(l) = location {
                     write!(f, " LOCATION '{l}'")?;
                 }
@@ -4852,6 +4881,60 @@ impl fmt::Display for Statement {
                 if let Some(clone) = clone {
                     write!(f, " CLONE {clone}")?;
                 }
+
+                if let Some(value) = data_retention_time_in_days {
+                    write!(f, " DATA_RETENTION_TIME_IN_DAYS = {value}")?;
+                }
+
+                if let Some(value) = max_data_extension_time_in_days {
+                    write!(f, " MAX_DATA_EXTENSION_TIME_IN_DAYS = {value}")?;
+                }
+
+                if let Some(vol) = external_volume {
+                    write!(f, " EXTERNAL_VOLUME = '{vol}'")?;
+                }
+
+                if let Some(cat) = catalog {
+                    write!(f, " CATALOG = '{cat}'")?;
+                }
+
+                if let Some(true) = replace_invalid_characters {
+                    write!(f, " REPLACE_INVALID_CHARACTERS = TRUE")?;
+                } else if let Some(false) = replace_invalid_characters {
+                    write!(f, " REPLACE_INVALID_CHARACTERS = FALSE")?;
+                }
+
+                if let Some(collation) = default_ddl_collation {
+                    write!(f, " DEFAULT_DDL_COLLATION = '{collation}'")?;
+                }
+
+                if let Some(policy) = storage_serialization_policy {
+                    write!(f, " STORAGE_SERIALIZATION_POLICY = {policy}")?;
+                }
+
+                if let Some(comment) = comment {
+                    write!(f, " COMMENT = '{comment}'")?;
+                }
+
+                if let Some(sync) = catalog_sync {
+                    write!(f, " CATALOG_SYNC = '{sync}'")?;
+                }
+
+                if let Some(mode) = catalog_sync_namespace_mode {
+                    write!(f, " CATALOG_SYNC_NAMESPACE_MODE = {mode}")?;
+                }
+
+                if let Some(delim) = catalog_sync_namespace_flatten_delimiter {
+                    write!(f, " CATALOG_SYNC_NAMESPACE_FLATTEN_DELIMITER = '{delim}'")?;
+                }
+
+                if let Some(tags) = with_tags {
+                    write!(f, " WITH TAG ({})", display_comma_separated(tags))?;
+                }
+
+                if let Some(contacts) = with_contacts {
+                    write!(f, " WITH CONTACT ({})", display_comma_separated(contacts))?;
+                }
                 Ok(())
             }
             Statement::CreateFunction(create_function) => create_function.fmt(f),
@@ -9682,6 +9765,23 @@ impl Display for Tag {
     }
 }
 
+/// Snowflake `WITH CONTACT ( purpose = contact [ , purpose = contact ...] )`
+///
+/// <https://docs.snowflake.com/en/sql-reference/sql/create-database>
+#[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[cfg_attr(feature = "visitor", derive(Visit, VisitMut))]
+pub struct ContactEntry {
+    pub purpose: String,
+    pub contact: String,
+}
+
+impl Display for ContactEntry {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{} = {}", self.purpose, self.contact)
+    }
+}
+
 /// Helper to indicate if a comment includes the `=` in the display form
 #[derive(Debug, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
@@ -10134,6 +10234,29 @@ impl Display for StorageSerializationPolicy {
     }
 }
 
+/// Snowflake CatalogSyncNamespaceMode
+/// ```sql
+/// [ CATALOG_SYNC_NAMESPACE_MODE = { NEST | FLATTEN } ]
+/// ```
+///
+/// <https://docs.snowflake.com/en/sql-reference/sql/create-database>
+#[derive(Debug, Copy, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[cfg_attr(feature = "visitor", derive(Visit, VisitMut))]
+pub enum CatalogSyncNamespaceMode {
+    Nest,
+    Flatten,
+}
+
+impl Display for CatalogSyncNamespaceMode {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            CatalogSyncNamespaceMode::Nest => write!(f, "NEST"),
+            CatalogSyncNamespaceMode::Flatten => write!(f, "FLATTEN"),
+        }
+    }
+}
+
 /// Variants of the Snowflake `COPY INTO` statement
 #[derive(Debug, Copy, Clone, PartialEq, PartialOrd, Eq, Ord, Hash)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]

src/dialect/snowflake.rs

@@ -20,15 +20,17 @@ use crate::alloc::string::ToString;
 use crate::ast::helpers::key_value_options::{
     KeyValueOption, KeyValueOptionType, KeyValueOptions, KeyValueOptionsDelimiter,
 };
+use crate::ast::helpers::stmt_create_database::CreateDatabaseBuilder;
 use crate::ast::helpers::stmt_create_table::CreateTableBuilder;
 use crate::ast::helpers::stmt_data_loading::{
     FileStagingCommand, StageLoadSelectItem, StageLoadSelectItemKind, StageParamsObject,
 };
 use crate::ast::{
-    ColumnOption, ColumnPolicy, ColumnPolicyProperty, CopyIntoSnowflakeKind, DollarQuotedString,
+    CatalogSyncNamespaceMode, ColumnOption, ColumnPolicy, ColumnPolicyProperty, ContactEntry,
-    Ident, IdentityParameters, IdentityProperty, IdentityPropertyFormatKind, IdentityPropertyKind,
+    CopyIntoSnowflakeKind, DollarQuotedString, Ident, IdentityParameters, IdentityProperty,
-    IdentityPropertyOrder, ObjectName, ObjectNamePart, RowAccessPolicy, ShowObjects, SqlOption,
+    IdentityPropertyFormatKind, IdentityPropertyKind, IdentityPropertyOrder, ObjectName,
-    Statement, TagsColumnOption, WrappedCollection,
+    ObjectNamePart, RowAccessPolicy, ShowObjects, SqlOption, Statement, StorageSerializationPolicy,
+    TagsColumnOption, WrappedCollection,
 };
 use crate::dialect::{Dialect, Precedence};
 use crate::keywords::Keyword;
@@ -44,7 +46,6 @@ use alloc::vec::Vec;
 use alloc::{format, vec};
 
 use super::keywords::RESERVED_FOR_IDENTIFIER;
-use sqlparser::ast::StorageSerializationPolicy;
 
 const RESERVED_KEYWORDS_FOR_SELECT_ITEM_OPERATOR: [Keyword; 1] = [Keyword::CONNECT_BY_ROOT];
 
@@ -260,6 +261,8 @@ impl Dialect for SnowflakeDialect {
                 return Some(parse_create_table(
                     or_replace, global, temporary, volatile, transient, iceberg, parser,
                 ));
+            } else if parser.parse_keyword(Keyword::DATABASE) {
+                return Some(parse_create_database(or_replace, transient, parser));
             } else {
                 // need to go back with the cursor
                 let mut back = 1;
@@ -678,29 +681,11 @@ pub fn parse_create_table(
                 }
                 Keyword::ENABLE_SCHEMA_EVOLUTION => {
                     parser.expect_token(&Token::Eq)?;
-                    let enable_schema_evolution =
+                    builder = builder.enable_schema_evolution(Some(parser.parse_boolean_string()?));
-                        match parser.parse_one_of_keywords(&[Keyword::TRUE, Keyword::FALSE]) {
-                            Some(Keyword::TRUE) => true,
-                            Some(Keyword::FALSE) => false,
-                            _ => {
-                                return parser.expected("TRUE or FALSE", next_token);
-                            }
-                        };
-
-                    builder = builder.enable_schema_evolution(Some(enable_schema_evolution));
                 }
                 Keyword::CHANGE_TRACKING => {
                     parser.expect_token(&Token::Eq)?;
-                    let change_tracking =
+                    builder = builder.change_tracking(Some(parser.parse_boolean_string()?));
-                        match parser.parse_one_of_keywords(&[Keyword::TRUE, Keyword::FALSE]) {
-                            Some(Keyword::TRUE) => true,
-                            Some(Keyword::FALSE) => false,
-                            _ => {
-                                return parser.expected("TRUE or FALSE", next_token);
-                            }
-                        };
-
-                    builder = builder.change_tracking(Some(change_tracking));
                 }
                 Keyword::DATA_RETENTION_TIME_IN_DAYS => {
                     parser.expect_token(&Token::Eq)?;
@@ -830,6 +815,115 @@ pub fn parse_create_table(
     Ok(builder.build())
 }
 
+/// Parse snowflake create database statement.
+/// <https://docs.snowflake.com/en/sql-reference/sql/create-database>
+pub fn parse_create_database(
+    or_replace: bool,
+    transient: bool,
+    parser: &mut Parser,
+) -> Result<Statement, ParserError> {
+    let if_not_exists = parser.parse_keywords(&[Keyword::IF, Keyword::NOT, Keyword::EXISTS]);
+    let name = parser.parse_object_name(false)?;
+
+    let mut builder = CreateDatabaseBuilder::new(name)
+        .or_replace(or_replace)
+        .transient(transient)
+        .if_not_exists(if_not_exists);
+
+    loop {
+        let next_token = parser.next_token();
+        match &next_token.token {
+            Token::Word(word) => match word.keyword {
+                Keyword::CLONE => {
+                    builder = builder.clone_clause(Some(parser.parse_object_name(false)?));
+                }
+                Keyword::DATA_RETENTION_TIME_IN_DAYS => {
+                    parser.expect_token(&Token::Eq)?;
+                    builder =
+                        builder.data_retention_time_in_days(Some(parser.parse_literal_uint()?));
+                }
+                Keyword::MAX_DATA_EXTENSION_TIME_IN_DAYS => {
+                    parser.expect_token(&Token::Eq)?;
+                    builder =
+                        builder.max_data_extension_time_in_days(Some(parser.parse_literal_uint()?));
+                }
+                Keyword::EXTERNAL_VOLUME => {
+                    parser.expect_token(&Token::Eq)?;
+                    builder = builder.external_volume(Some(parser.parse_literal_string()?));
+                }
+                Keyword::CATALOG => {
+                    parser.expect_token(&Token::Eq)?;
+                    builder = builder.catalog(Some(parser.parse_literal_string()?));
+                }
+                Keyword::REPLACE_INVALID_CHARACTERS => {
+                    parser.expect_token(&Token::Eq)?;
+                    builder =
+                        builder.replace_invalid_characters(Some(parser.parse_boolean_string()?));
+                }
+                Keyword::DEFAULT_DDL_COLLATION => {
+                    parser.expect_token(&Token::Eq)?;
+                    builder = builder.default_ddl_collation(Some(parser.parse_literal_string()?));
+                }
+                Keyword::STORAGE_SERIALIZATION_POLICY => {
+                    parser.expect_token(&Token::Eq)?;
+                    let policy = parse_storage_serialization_policy(parser)?;
+                    builder = builder.storage_serialization_policy(Some(policy));
+                }
+                Keyword::COMMENT => {
+                    parser.expect_token(&Token::Eq)?;
+                    builder = builder.comment(Some(parser.parse_literal_string()?));
+                }
+                Keyword::CATALOG_SYNC => {
+                    parser.expect_token(&Token::Eq)?;
+                    builder = builder.catalog_sync(Some(parser.parse_literal_string()?));
+                }
+                Keyword::CATALOG_SYNC_NAMESPACE_FLATTEN_DELIMITER => {
+                    parser.expect_token(&Token::Eq)?;
+                    builder = builder.catalog_sync_namespace_flatten_delimiter(Some(
+                        parser.parse_literal_string()?,
+                    ));
+                }
+                Keyword::CATALOG_SYNC_NAMESPACE_MODE => {
+                    parser.expect_token(&Token::Eq)?;
+                    let mode =
+                        match parser.parse_one_of_keywords(&[Keyword::NEST, Keyword::FLATTEN]) {
+                            Some(Keyword::NEST) => CatalogSyncNamespaceMode::Nest,
+                            Some(Keyword::FLATTEN) => CatalogSyncNamespaceMode::Flatten,
+                            _ => {
+                                return parser.expected("NEST or FLATTEN", next_token);
+                            }
+                        };
+                    builder = builder.catalog_sync_namespace_mode(Some(mode));
+                }
+                Keyword::WITH => {
+                    if parser.parse_keyword(Keyword::TAG) {
+                        parser.expect_token(&Token::LParen)?;
+                        let tags = parser.parse_comma_separated(Parser::parse_tag)?;
+                        parser.expect_token(&Token::RParen)?;
+                        builder = builder.with_tags(Some(tags));
+                    } else if parser.parse_keyword(Keyword::CONTACT) {
+                        parser.expect_token(&Token::LParen)?;
+                        let contacts = parser.parse_comma_separated(|p| {
+                            let purpose = p.parse_identifier()?.value;
+                            p.expect_token(&Token::Eq)?;
+                            let contact = p.parse_identifier()?.value;
+                            Ok(ContactEntry { purpose, contact })
+                        })?;
+                        parser.expect_token(&Token::RParen)?;
+                        builder = builder.with_contacts(Some(contacts));
+                    } else {
+                        return parser.expected("TAG or CONTACT", next_token);
+                    }
+                }
+                _ => return parser.expected("end of statement", next_token),
+            },
+            Token::SemiColon | Token::EOF => break,
+            _ => return parser.expected("end of statement", next_token),
+        }
+    }
+    Ok(builder.build())
+}
+
 pub fn parse_storage_serialization_policy(
     parser: &mut Parser,
 ) -> Result<StorageSerializationPolicy, ParserError> {

src/keywords.rs

@@ -166,6 +166,8 @@ define_keywords!(
     CAST,
     CATALOG,
     CATALOG_SYNC,
+    CATALOG_SYNC_NAMESPACE_FLATTEN_DELIMITER,
+    CATALOG_SYNC_NAMESPACE_MODE,
     CATCH,
     CEIL,
     CEILING,
@@ -213,6 +215,7 @@ define_keywords!(
     CONNECTOR,
     CONNECT_BY_ROOT,
     CONSTRAINT,
+    CONTACT,
     CONTAINS,
     CONTINUE,
     CONVERT,
@@ -366,6 +369,7 @@ define_keywords!(
     FIRST,
     FIRST_VALUE,
     FIXEDSTRING,
+    FLATTEN,
     FLOAT,
     FLOAT32,
     FLOAT4,
@@ -584,6 +588,7 @@ define_keywords!(
     NATURAL,
     NCHAR,
     NCLOB,
+    NEST,
     NESTED,
     NETWORK,
     NEW,
@@ -756,6 +761,7 @@ define_keywords!(
     REPAIR,
     REPEATABLE,
     REPLACE,
+    REPLACE_INVALID_CHARACTERS,
     REPLICA,
     REPLICATE,
     REPLICATION,

src/parser/mod.rs

@@ -5054,7 +5054,22 @@ impl<'a> Parser<'a> {
             if_not_exists: ine,
             location,
             managed_location,
+            or_replace: false,
+            transient: false,
             clone,
+            data_retention_time_in_days: None,
+            max_data_extension_time_in_days: None,
+            external_volume: None,
+            catalog: None,
+            replace_invalid_characters: None,
+            default_ddl_collation: None,
+            storage_serialization_policy: None,
+            comment: None,
+            catalog_sync: None,
+            catalog_sync_namespace_mode: None,
+            catalog_sync_namespace_flatten_delimiter: None,
+            with_tags: None,
+            with_contacts: None,
         })
     }
 
@@ -9763,6 +9778,15 @@ impl<'a> Parser<'a> {
         }
     }
 
+    /// Parse a boolean string
+    pub(crate) fn parse_boolean_string(&mut self) -> Result<bool, ParserError> {
+        match self.parse_one_of_keywords(&[Keyword::TRUE, Keyword::FALSE]) {
+            Some(Keyword::TRUE) => Ok(true),
+            Some(Keyword::FALSE) => Ok(false),
+            _ => self.expected("TRUE or FALSE", self.peek_token()),
+        }
+    }
+
     /// Parse a literal unicode normalization clause
     pub fn parse_unicode_is_normalized(&mut self, expr: Expr) -> Result<Expr, ParserError> {
         let neg = self.parse_keyword(Keyword::NOT);

tests/sqlparser_common.rs

@@ -7950,6 +7950,7 @@ fn parse_create_database() {
             location,
             managed_location,
             clone,
+            ..
         } => {
             assert_eq!("mydb", db_name.to_string());
             assert!(!if_not_exists);
@@ -7967,6 +7968,7 @@ fn parse_create_database() {
             location,
             managed_location,
             clone,
+            ..
         } => {
             assert_eq!("mydb", db_name.to_string());
             assert!(!if_not_exists);
@@ -7991,6 +7993,7 @@ fn parse_create_database_ine() {
             location,
             managed_location,
             clone,
+            ..
         } => {
             assert_eq!("mydb", db_name.to_string());
             assert!(if_not_exists);

tests/sqlparser_snowflake.rs

@@ -4516,3 +4516,54 @@ fn test_snowflake_identifier_function() {
     snowflake().verified_stmt("GRANT ROLE IDENTIFIER('AAA') TO USER IDENTIFIER('AAA')");
     snowflake().verified_stmt("REVOKE ROLE IDENTIFIER('AAA') FROM USER IDENTIFIER('AAA')");
 }
+
+#[test]
+fn test_create_database() {
+    snowflake().verified_stmt("CREATE DATABASE my_db");
+    snowflake().verified_stmt("CREATE OR REPLACE DATABASE my_db");
+    snowflake().verified_stmt("CREATE TRANSIENT DATABASE IF NOT EXISTS my_db");
+    snowflake().verified_stmt("CREATE DATABASE my_db CLONE src_db");
+    snowflake().verified_stmt(
+        "CREATE OR REPLACE DATABASE my_db CLONE src_db DATA_RETENTION_TIME_IN_DAYS = 1",
+    );
+    snowflake().one_statement_parses_to(
+        r#"
+        CREATE OR REPLACE TRANSIENT DATABASE IF NOT EXISTS my_db
+        CLONE src_db
+        DATA_RETENTION_TIME_IN_DAYS = 1
+        MAX_DATA_EXTENSION_TIME_IN_DAYS = 5
+        EXTERNAL_VOLUME = 'volume1'
+        CATALOG = 'my_catalog'
+        REPLACE_INVALID_CHARACTERS = TRUE
+        DEFAULT_DDL_COLLATION = 'en-ci'
+        STORAGE_SERIALIZATION_POLICY = COMPATIBLE
+        COMMENT = 'This is my database'
+        CATALOG_SYNC = 'sync_integration'
+        CATALOG_SYNC_NAMESPACE_MODE = NEST
+        CATALOG_SYNC_NAMESPACE_FLATTEN_DELIMITER = '/'
+        WITH TAG (env = 'prod', team = 'data')
+        WITH CONTACT (owner = 'admin', dpo = 'compliance')
+    "#,
+        "CREATE OR REPLACE TRANSIENT DATABASE IF NOT EXISTS \
+        my_db CLONE src_db DATA_RETENTION_TIME_IN_DAYS = 1 MAX_DATA_EXTENSION_TIME_IN_DAYS = 5 \
+        EXTERNAL_VOLUME = 'volume1' CATALOG = 'my_catalog' \
+        REPLACE_INVALID_CHARACTERS = TRUE DEFAULT_DDL_COLLATION = 'en-ci' \
+        STORAGE_SERIALIZATION_POLICY = COMPATIBLE COMMENT = 'This is my database' \
+        CATALOG_SYNC = 'sync_integration' CATALOG_SYNC_NAMESPACE_MODE = NEST \
+        CATALOG_SYNC_NAMESPACE_FLATTEN_DELIMITER = '/' \
+        WITH TAG (env='prod', team='data') \
+        WITH CONTACT (owner = admin, dpo = compliance)",
+    );
+
+    let err = snowflake()
+        .parse_sql_statements("CREATE DATABASE")
+        .unwrap_err()
+        .to_string();
+    assert!(err.contains("Expected"), "Unexpected error: {err}");
+
+    let err = snowflake()
+        .parse_sql_statements("CREATE DATABASE my_db CLONE")
+        .unwrap_err()
+        .to_string();
+    assert!(err.contains("Expected"), "Unexpected error: {err}");
+}