Repository: tobymao/sqlglot
Issue category: Feature Request

Support Snowflake WITH TAG syntax
**Is your feature request related to a problem? Please describe.**
This library does not support snowflake's WITH TAG syntax for setting tags while creating an object.

For example,
```
sqlglot.parse_one("CREATE VIEW my_view WITH TAG (my_tag = 'tag')", dialect="snowflake")
'CREATE VIEW my_view WITH TAG (my_tag = 'tag')' contains unsupported syntax. Falling back to parsing as a 'Command'.
> Command(this=CREATE, expression=VIEW my_view WITH TAG (my_tag = 'tag'))
```
Note that multiple tags can be specified in the same statement.

**Describe the solution you'd like**
Parse the above statement into something like:
```
Create(
  this=Table(
    this=Identifier(this=my_view, quoted=False)),
  kind=VIEW,
  actions=[tag=[
        Paren(
          this=EQ(
            this=Column(
              this=Identifier(this=my_tag, quoted=False)),
            expression=Literal(this=tag, is_string=True)))])
  ]
)
```
This was pulled from an `ALTER VIEW ... SET TAG` statement, idk if this is actually the correct Create object. My primary desire is to avoid the Command parsing fallback.

**Additional context**
Snowflake CREATE TABLE reference: https://docs.snowflake.com/en/sql-reference/sql/create-table
Example CREATE WAREHOUSE statement: https://docs.snowflake.com/en/user-guide/object-tagging

There are similar `WITH ...` syntaxes in snowflake, e.g. `WITH ROW ACCESS POLICY` and `WITH AGGREGATION POLICY`, or `WITH MASKING POLICY` and `WITH PROJECTION POLICY` for columns. This feature request is not specifically asking for support for that syntax.


Functions to edit:
sqlglot/dialects/snowflake.py:Snowflake
sqlglot/expressions.py:DictProperty
sqlglot/expressions.py:DictSubProperty
sqlglot/generator.py:Generator.dictproperty_sql
sqlglot/generator.py:Generator.dictsubproperty_sql
sqlglot/parser.py:Parser
sqlglot/parser.py:Parser._parse_dict_property
Functions added: sqlglot/dialects/snowflake.py:Snowflake
sqlglot/parser.py:Parser
sqlglot/parser.py:Parser._parse_key_value_list

Patch:
diff --git a/sqlglot/dialects/snowflake.py b/sqlglot/dialects/snowflake.py
index 1d2b246e5d..350f7773b5 100644
--- a/sqlglot/dialects/snowflake.py
+++ b/sqlglot/dialects/snowflake.py
@@ -465,6 +465,12 @@ class Parser(parser.Parser):
         PROPERTY_PARSERS = {
             **parser.Parser.PROPERTY_PARSERS,
             "LOCATION": lambda self: self._parse_location_property(),
+            "TAG": lambda self: self._parse_dict_property(
+                this="TAG",
+                has_kind=False,
+                separator=(TokenType.COMMA, ","),
+                delimiter=(TokenType.EQ, "="),
+            ),
         }
 
         TYPE_CONVERTERS = {
diff --git a/sqlglot/expressions.py b/sqlglot/expressions.py
index a69e8362f9..de88fef6b8 100644
--- a/sqlglot/expressions.py
+++ b/sqlglot/expressions.py
@@ -2778,10 +2778,11 @@ class ClusteredByProperty(Property):
 
 
 class DictProperty(Property):
-    arg_types = {"this": True, "kind": True, "settings": False}
+    arg_types = {"this": True, "kind": False, "settings": False, "separator": False}
 
 
 class DictSubProperty(Property):
+    arg_types = {"this": True, "value": True, "delimiter": False}
     pass
 
 
diff --git a/sqlglot/generator.py b/sqlglot/generator.py
index ae9dc35e49..c20e5ed222 100644
--- a/sqlglot/generator.py
+++ b/sqlglot/generator.py
@@ -3720,9 +3720,16 @@ def tonumber_sql(self, expression: exp.ToNumber) -> str:
     def dictproperty_sql(self, expression: exp.DictProperty) -> str:
         this = self.sql(expression, "this")
         kind = self.sql(expression, "kind")
-        settings_sql = self.expressions(expression, key="settings", sep=" ")
-        args = f"({self.sep('')}{settings_sql}{self.seg(')', sep='')}" if settings_sql else "()"
-        return f"{this}({kind}{args})"
+        separator = self.sql(expression, "separator")
+        settings_sql = self.expressions(expression, key="settings", sep=f"{separator} ")
+        if kind:
+            settings_section = (
+                f"({self.sep('')}{settings_sql}{self.seg(')', sep='')}" if settings_sql else "()"
+            )
+            args = f"{kind}{settings_section}"
+        else:
+            args = f"{self.sep('')}{settings_sql}{self.sep(sep='')}"
+        return f"{this} ({args})"
 
     def dictrange_sql(self, expression: exp.DictRange) -> str:
         this = self.sql(expression, "this")
@@ -3731,7 +3738,8 @@ def dictrange_sql(self, expression: exp.DictRange) -> str:
         return f"{this}(MIN {min} MAX {max})"
 
     def dictsubproperty_sql(self, expression: exp.DictSubProperty) -> str:
-        return f"{self.sql(expression, 'this')} {self.sql(expression, 'value')}"
+        delimiter = self.sql(expression, "delimiter") or " "
+        return f"{self.sql(expression, 'this')}{delimiter}{self.sql(expression, 'value')}"
 
     def duplicatekeyproperty_sql(self, expression: exp.DuplicateKeyProperty) -> str:
         return f"DUPLICATE KEY ({self.expressions(expression, flat=True)})"
diff --git a/sqlglot/parser.py b/sqlglot/parser.py
index cbbeffe5e1..860425825c 100644
--- a/sqlglot/parser.py
+++ b/sqlglot/parser.py
@@ -936,7 +936,7 @@ class Parser(metaclass=_Parser):
         "INPUT": lambda self: self.expression(exp.InputModelProperty, this=self._parse_schema()),
         "JOURNAL": lambda self, **kwargs: self._parse_journal(**kwargs),
         "LANGUAGE": lambda self: self._parse_property_assignment(exp.LanguageProperty),
-        "LAYOUT": lambda self: self._parse_dict_property(this="LAYOUT"),
+        "LAYOUT": lambda self: self._parse_dict_property(this="LAYOUT", has_kind=True),
         "LIFETIME": lambda self: self._parse_dict_range(this="LIFETIME"),
         "LIKE": lambda self: self._parse_create_like(),
         "LOCATION": lambda self: self._parse_property_assignment(exp.LocationProperty),
@@ -973,7 +973,7 @@ class Parser(metaclass=_Parser):
         "SETTINGS": lambda self: self._parse_settings_property(),
         "SHARING": lambda self: self._parse_property_assignment(exp.SharingProperty),
         "SORTKEY": lambda self: self._parse_sortkey(),
-        "SOURCE": lambda self: self._parse_dict_property(this="SOURCE"),
+        "SOURCE": lambda self: self._parse_dict_property(this="SOURCE", has_kind=True),
         "STABLE": lambda self: self.expression(
             exp.StabilityProperty, this=exp.Literal.string("STABLE")
         ),
@@ -7108,21 +7108,39 @@ def _parse_as_command(self, start: Token) -> exp.Command:
         self._warn_unsupported()
         return exp.Command(this=text[:size], expression=text[size:])
 
-    def _parse_dict_property(self, this: str) -> exp.DictProperty:
+    def _parse_dict_property(
+        self,
+        this: str,
+        has_kind: bool,
+        separator: t.Optional[t.Tuple[TokenType, str]] = None,
+        delimiter: t.Optional[t.Tuple[TokenType, str]] = None,
+    ) -> exp.DictProperty:
+        """
+        Parses a dictionary property, which is a key-value pair enclosed in parentheses.
+
+        Args:
+            this: The name of the property.
+            has_kind: Whether the property is labeled, e.g. is of the form `PROPERTY (KIND (k1=v1, k2=v2, ...))`.
+            separator: The separator token between key-value pairs, and its string representation.
+            delimiter: The delimiter token between key and value, and its string representation.
+        """
+        separator_token = None
+        separator_str = None
+        if separator:
+            separator_token, separator_str = separator
+
         settings = []
 
         self._match_l_paren()
-        kind = self._parse_id_var()
 
-        if self._match(TokenType.L_PAREN):
-            while True:
-                key = self._parse_id_var()
-                value = self._parse_primary()
-
-                if not key and value is None:
-                    break
-                settings.append(self.expression(exp.DictSubProperty, this=key, value=value))
-            self._match(TokenType.R_PAREN)
+        if has_kind:
+            kind = self._parse_id_var()
+            if self._match(TokenType.L_PAREN):
+                settings = self._parse_key_value_list(this, separator_token, delimiter)
+                self._match(TokenType.R_PAREN)
+        else:
+            kind = None
+            settings = self._parse_key_value_list(this, separator_token, delimiter)
 
         self._match_r_paren()
 
@@ -7131,8 +7149,39 @@ def _parse_dict_property(self, this: str) -> exp.DictProperty:
             this=this,
             kind=kind.this if kind else None,
             settings=settings,
+            separator=separator_str,
         )
 
+    def _parse_key_value_list(
+        self,
+        this: str,
+        separator_token: t.Optional[TokenType],
+        delimiter: t.Optional[t.Tuple[TokenType, str]],
+    ) -> t.List[exp.DictSubProperty]:
+        delimiter_token = None
+        delimiter_str = None
+        if delimiter:
+            delimiter_token, delimiter_str = delimiter
+
+        lst = []
+
+        while True:
+            key = self._parse_id_var()
+            if delimiter_token:
+                self._match(delimiter_token)
+            value = self._parse_primary()
+
+            if not key and value is None:
+                break
+            lst.append(
+                self.expression(exp.DictSubProperty, this=key, value=value, delimiter=delimiter_str)
+            )
+
+            if separator_token:
+                self._match(separator_token)
+
+        return lst
+
     def _parse_dict_range(self, this: str) -> exp.DictRange:
         self._match_l_paren()
         has_min = self._match_text_seq("MIN")

Test patch: diff --git a/tests/dialects/test_clickhouse.py b/tests/dialects/test_clickhouse.py
index 19b3ce3934..65b12bbd58 100644
--- a/tests/dialects/test_clickhouse.py
+++ b/tests/dialects/test_clickhouse.py
@@ -974,11 +974,11 @@ def test_ddl(self):
   amount Float64
 )
 PRIMARY KEY (id)
-SOURCE(CLICKHOUSE(
+SOURCE (CLICKHOUSE(
   TABLE 'discounts'
 ))
 LIFETIME(MIN 1 MAX 1000)
-LAYOUT(RANGE_HASHED(
+LAYOUT (RANGE_HASHED(
   range_lookup_strategy 'max'
 ))
 RANGE(MIN discount_start_date MAX discount_end_date)""",
@@ -1004,10 +1004,10 @@ def test_ddl(self):
   cca2 String DEFAULT '??'
 )
 PRIMARY KEY (prefix)
-SOURCE(CLICKHOUSE(
+SOURCE (CLICKHOUSE(
   TABLE 'my_ip_addresses'
 ))
-LAYOUT(IP_TRIE())
+LAYOUT (IP_TRIE())
 LIFETIME(MIN 0 MAX 3600)""",
             },
             pretty=True,
@@ -1030,10 +1030,10 @@ def test_ddl(self):
   name String
 )
 PRIMARY KEY (key)
-SOURCE(CLICKHOUSE(
+SOURCE (CLICKHOUSE(
   TABLE 'polygons_test_table'
 ))
-LAYOUT(POLYGON(
+LAYOUT (POLYGON(
   STORE_POLYGON_KEY_COLUMN 1
 ))
 LIFETIME(MIN 0 MAX 0)""",
diff --git a/tests/dialects/test_snowflake.py b/tests/dialects/test_snowflake.py
index 157947df46..8058bcf6de 100644
--- a/tests/dialects/test_snowflake.py
+++ b/tests/dialects/test_snowflake.py
@@ -1479,13 +1479,20 @@ def test_ddl(self):
                 "snowflake": "CREATE OR REPLACE TRANSIENT TABLE a (id INT)",
             },
         )
-
         self.validate_all(
             "CREATE TABLE a (b INT)",
             read={"teradata": "CREATE MULTISET TABLE a (b INT)"},
             write={"snowflake": "CREATE TABLE a (b INT)"},
         )
 
+        self.validate_identity("CREATE TABLE a TAG (key1='value_1', key2='value_2')")
+        self.validate_all(
+            "CREATE TABLE a TAG (key1='value_1')",
+            read={
+                "snowflake": "CREATE TABLE a WITH TAG (key1='value_1')",
+            },
+        )
+
         for action in ("SET", "DROP"):
             with self.subTest(f"ALTER COLUMN {action} NOT NULL"):
                 self.validate_all(