Snowflake: parse CREATE SCHEMA ... [WITH] TAG (...)

src/ast/mod.rs

@@ -4820,6 +4820,9 @@ pub enum Statement {
         clone: Option<ObjectName>,
         /// Optional schema comment (Snowflake `COMMENT = '...'`).
         comment: Option<CommentDef>,
+        /// Snowflake inline `[ WITH ] TAG ( <t> = '<v>' [, ...] )` clause;
+        /// `None` when absent.
+        with_tags: Option<Vec<Tag>>,
     },
     /// ```sql
     /// CREATE DATABASE
@@ -7350,6 +7353,7 @@ impl fmt::Display for Statement {
                 default_collate_spec,
                 clone,
                 comment,
+                with_tags,
             } => {
                 write!(
                     f,
@@ -7380,6 +7384,10 @@ impl fmt::Display for Statement {
                     write!(f, " CLONE {clone}")?;
                 }
 
+                if let Some(tags) = with_tags {
+                    write!(f, " WITH TAG ({})", display_comma_separated(tags))?;
+                }
+
                 if let Some(comment) = comment {
                     match comment {
                         CommentDef::WithEq(c) => write!(f, " COMMENT = '{c}'")?,

src/parser/mod.rs

@@ -6043,6 +6043,21 @@ impl<'a> Parser<'a> {
         let with_managed_access =
             self.parse_keywords(&[Keyword::WITH, Keyword::MANAGED, Keyword::ACCESS]);
 
+        // Snowflake inline `[ WITH ] TAG ( <t> = '<v>', ... )` clause. The
+        // optional `WITH` shares its keyword with the Trino option list below,
+        // so intercept `WITH TAG` (and the bare `TAG`) here; `parse_keywords`
+        // backtracks when `TAG` does not follow, leaving `WITH (k='v')` intact.
+        let with_tags = if self.parse_keywords(&[Keyword::WITH, Keyword::TAG])
+            || self.parse_keyword(Keyword::TAG)
+        {
+            self.expect_token(&Token::LParen)?;
+            let tags = self.parse_comma_separated(Parser::parse_tag)?;
+            self.expect_token(&Token::RParen)?;
+            Some(tags)
+        } else {
+            None
+        };
+
         let with = if !with_managed_access && self.peek_keyword(Keyword::WITH) {
             Some(self.parse_options(Keyword::WITH)?)
         } else {
@@ -6074,6 +6089,7 @@ impl<'a> Parser<'a> {
             default_collate_spec,
             clone,
             comment,
+            with_tags,
         })
     }
 

tests/sqlparser_snowflake.rs

@@ -559,6 +559,35 @@ fn test_snowflake_create_transient_schema() {
     }
 }
 
+#[test]
+fn test_snowflake_create_schema_with_tag() {
+    // `WITH TAG` roundtrips and populates `with_tags`; the bare `TAG (...)`
+    // form (optional `WITH`) normalises to the `WITH TAG` rendering.
+    let sql = "CREATE SCHEMA my_schema WITH TAG (cost_center='sales', env='prod')";
+    match snowflake_and_generic().verified_stmt(sql) {
+        Statement::CreateSchema {
+            schema_name,
+            with_tags,
+            ..
+        } => {
+            assert_eq!("my_schema", schema_name.to_string());
+            let tags = with_tags.expect("with_tags populated");
+            assert_eq!(tags.len(), 2);
+            assert_eq!(tags[0].to_string(), "cost_center='sales'");
+            assert_eq!(tags[1].to_string(), "env='prod'");
+        }
+        _ => unreachable!(),
+    }
+
+    snowflake_and_generic().one_statement_parses_to(
+        "CREATE SCHEMA my_schema TAG (env = 'prod')",
+        "CREATE SCHEMA my_schema WITH TAG (env='prod')",
+    );
+
+    // `WITH MANAGED ACCESS` and the Trino `WITH (k='v')` option list still parse.
+    snowflake_and_generic().verified_stmt("CREATE SCHEMA my_schema WITH MANAGED ACCESS");
+}
+
 #[test]
 fn test_snowflake_create_table_column_comment() {
     let sql = "CREATE TABLE my_table (a STRING COMMENT 'some comment')";