Repository: tobymao/sqlglot

Issue:
Snowflake changes syntax can't be parsed
**Before you file an issue**
- Make sure you specify the "read" dialect eg. `parse_one(sql, read="spark")`
Parsing the Snowflake dialect
- Make sure you specify the "write" dialect eg. `ast.sql(dialect="duckdb")`
- Check if the issue still exists on main
sqlglot==25.4.0
sqlglotrs==0.2.8

**Fully reproducible code snippet**
Please include a fully reproducible code snippet or the input sql, dialect, and expected output.

``` python
from sqlglot import parse_one
print(parse_one("""
        select 1 from
        some_table changes (information => default) at (timestamp => to_timestamp_tz(''2024-07-01 00:00:00+00:00'')) end (timestamp => to_timestamp_tz(''2024-07-01 14:28:59.999999+00:00''))
""", dialect="snowflake"))
```

<img width="1195" alt="image" src="https://github.com/tobymao/sqlglot/assets/110614540/f0fae2af-3255-4141-9652-16144f5992f7">


**Official Documentation**
Please include links to official SQL documentation related to your issue.

https://docs.snowflake.com/en/sql-reference/constructs/changes


Discussion:

Patch:
diff --git a/sqlglot/dialects/snowflake.py b/sqlglot/dialects/snowflake.py
index 1eab756b..a3e84f47 100644
--- a/sqlglot/dialects/snowflake.py
+++ b/sqlglot/dialects/snowflake.py
@@ -504,10 +504,11 @@ class Snowflake(Dialect):
 
             return lateral
 
-        def _parse_at_before(self, table: exp.Table) -> exp.Table:
+        def _parse_historical_data(self) -> t.Optional[exp.HistoricalData]:
             # https://docs.snowflake.com/en/sql-reference/constructs/at-before
             index = self._index
-            if self._match_texts(("AT", "BEFORE")):
+            historical_data = None
+            if self._match_texts(self.HISTORICAL_DATA_PREFIX):
                 this = self._prev.text.upper()
                 kind = (
                     self._match(TokenType.L_PAREN)
@@ -518,14 +519,27 @@ class Snowflake(Dialect):
 
                 if expression:
                     self._match_r_paren()
-                    when = self.expression(
+                    historical_data = self.expression(
                         exp.HistoricalData, this=this, kind=kind, expression=expression
                     )
-                    table.set("when", when)
                 else:
                     self._retreat(index)
 
-            return table
+            return historical_data
+
+        def _parse_changes(self) -> t.Optional[exp.Changes]:
+            if not self._match_text_seq("CHANGES", "(", "INFORMATION", "=>"):
+                return None
+
+            information = self._parse_var(any_token=True)
+            self._match_r_paren()
+
+            return self.expression(
+                exp.Changes,
+                information=information,
+                at_before=self._parse_historical_data(),
+                end=self._parse_historical_data(),
+            )
 
         def _parse_table_parts(
             self, schema: bool = False, is_db_reference: bool = False, wildcard: bool = False
@@ -559,7 +573,15 @@ class Snowflake(Dialect):
             else:
                 table = super()._parse_table_parts(schema=schema, is_db_reference=is_db_reference)
 
-            return self._parse_at_before(table)
+            changes = self._parse_changes()
+            if changes:
+                table.set("changes", changes)
+
+            at_before = self._parse_historical_data()
+            if at_before:
+                table.set("when", at_before)
+
+            return table
 
         def _parse_id_var(
             self,
diff --git a/sqlglot/expressions.py b/sqlglot/expressions.py
index 764810dc..79511da4 100644
--- a/sqlglot/expressions.py
+++ b/sqlglot/expressions.py
@@ -2002,6 +2002,10 @@ class Check(Expression):
     pass
 
 
+class Changes(Expression):
+    arg_types = {"information": True, "at_before": False, "end": False}
+
+
 # https://docs.snowflake.com/en/sql-reference/constructs/connect-by
 class Connect(Expression):
     arg_types = {"start": False, "connect": True, "nocycle": False}
@@ -3034,6 +3038,7 @@ class Table(Expression):
         "when": False,
         "only": False,
         "partition": False,
+        "changes": False,
     }
 
     @property
diff --git a/sqlglot/generator.py b/sqlglot/generator.py
index 6a2589f3..40ad4680 100644
--- a/sqlglot/generator.py
+++ b/sqlglot/generator.py
@@ -1749,7 +1749,10 @@ class Generator(metaclass=_Generator):
         if when:
             table = f"{table} {when}"
 
-        return f"{only}{table}{partition}{version}{file_format}{alias}{hints}{pivots}{joins}{laterals}{ordinality}"
+        changes = self.sql(expression, "changes")
+        changes = f" {changes}" if changes else ""
+
+        return f"{only}{table}{changes}{partition}{version}{file_format}{alias}{hints}{pivots}{joins}{laterals}{ordinality}"
 
     def tablesample_sql(
         self,
@@ -4028,3 +4031,13 @@ class Generator(metaclass=_Generator):
             expression.args.get("format"),
             expression.args.get("zone"),
         )
+
+    def changes_sql(self, expression: exp.Changes) -> str:
+        information = self.sql(expression, "information")
+        information = f"INFORMATION => {information}"
+        at_before = self.sql(expression, "at_before")
+        at_before = f"{self.seg('')}{at_before}" if at_before else ""
+        end = self.sql(expression, "end")
+        end = f"{self.seg('')}{end}" if end else ""
+
+        return f"CHANGES ({information}){at_before}{end}"
diff --git a/sqlglot/parser.py b/sqlglot/parser.py
index d5fe792b..d655b849 100644
--- a/sqlglot/parser.py
+++ b/sqlglot/parser.py
@@ -1132,6 +1132,7 @@ class Parser(metaclass=_Parser):
     INSERT_ALTERNATIVES = {"ABORT", "FAIL", "IGNORE", "REPLACE", "ROLLBACK"}
 
     CLONE_KEYWORDS = {"CLONE", "COPY"}
+    HISTORICAL_DATA_PREFIX = {"AT", "BEFORE", "END"}
     HISTORICAL_DATA_KIND = {"TIMESTAMP", "OFFSET", "STATEMENT", "STREAM"}
 
     OPCLASS_FOLLOW_KEYWORDS = {"ASC", "DESC", "NULLS", "WITH"}


Test patch:
diff --git a/tests/dialects/test_snowflake.py b/tests/dialects/test_snowflake.py
index 80f8faaa..88b2148d 100644
--- a/tests/dialects/test_snowflake.py
+++ b/tests/dialects/test_snowflake.py
@@ -190,18 +190,6 @@ WHERE
             """SELECT PARSE_JSON('{"food":{"fruit":"banana"}}'):food.fruit::VARCHAR""",
             """SELECT CAST(GET_PATH(PARSE_JSON('{"food":{"fruit":"banana"}}'), 'food.fruit') AS VARCHAR)""",
         )
-        self.validate_identity(
-            "SELECT * FROM foo at",
-            "SELECT * FROM foo AS at",
-        )
-        self.validate_identity(
-            "SELECT * FROM foo before",
-            "SELECT * FROM foo AS before",
-        )
-        self.validate_identity(
-            "SELECT * FROM foo at (col)",
-            "SELECT * FROM foo AS at(col)",
-        )
         self.validate_identity(
             "SELECT * FROM unnest(x) with ordinality",
             "SELECT * FROM TABLE(FLATTEN(INPUT => x)) AS _u(seq, key, path, index, value, this)",
@@ -1204,6 +1192,17 @@ WHERE
             "SELECT oldt.*, newt.* FROM my_table BEFORE (STATEMENT => '8e5d0ca9-005e-44e6-b858-a8f5b37c5726') AS oldt FULL OUTER JOIN my_table AT (STATEMENT => '8e5d0ca9-005e-44e6-b858-a8f5b37c5726') AS newt ON oldt.id = newt.id WHERE oldt.id IS NULL OR newt.id IS NULL",
         )
 
+        # Make sure that the historical data keywords can still be used as aliases
+        for historical_data_prefix in ("AT", "BEFORE", "END", "CHANGES"):
+            for schema_suffix in ("", "(col)"):
+                with self.subTest(
+                    f"Testing historical data prefix alias: {historical_data_prefix}{schema_suffix}"
+                ):
+                    self.validate_identity(
+                        f"SELECT * FROM foo {historical_data_prefix}{schema_suffix}",
+                        f"SELECT * FROM foo AS {historical_data_prefix}{schema_suffix}",
+                    )
+
     def test_ddl(self):
         for constraint_prefix in ("WITH ", ""):
             with self.subTest(f"Constraint prefix: {constraint_prefix}"):
@@ -2033,3 +2032,15 @@ SINGLE = TRUE""",
 
         self.validate_identity("ALTER TABLE foo UNSET TAG a, b, c")
         self.validate_identity("ALTER TABLE foo UNSET DATA_RETENTION_TIME_IN_DAYS, CHANGE_TRACKING")
+
+    def test_from_changes(self):
+        self.validate_identity(
+            """SELECT C1 FROM t1 CHANGES (INFORMATION => APPEND_ONLY) AT (STREAM => 's1') END (TIMESTAMP => $ts2)"""
+        )
+        self.validate_identity(
+            """SELECT C1 FROM t1 CHANGES (INFORMATION => APPEND_ONLY) BEFORE (STATEMENT => 'STMT_ID') END (TIMESTAMP => $ts2)"""
+        )
+        self.validate_identity(
+            """SELECT 1 FROM some_table CHANGES (INFORMATION => APPEND_ONLY) AT (TIMESTAMP => TO_TIMESTAMP_TZ('2024-07-01 00:00:00+00:00')) END (TIMESTAMP => TO_TIMESTAMP_TZ('2024-07-01 14:28:59.999999+00:00'))""",
+            """SELECT 1 FROM some_table CHANGES (INFORMATION => APPEND_ONLY) AT (TIMESTAMP => CAST('2024-07-01 00:00:00+00:00' AS TIMESTAMPTZ)) END (TIMESTAMP => CAST('2024-07-01 14:28:59.999999+00:00' AS TIMESTAMPTZ))""",
+        )