Fix: move WATERMARK logic to risingwave fixes #5989

Author: georgesittas

sqlglot/dialects/risingwave.py

@@ -25,6 +25,20 @@ class Parser(Postgres.Parser):
             "KEY": lambda self: self._parse_encode_property(key=True),
         }
 
+        CONSTRAINT_PARSERS = {
+            **Postgres.Parser.CONSTRAINT_PARSERS,
+            "WATERMARK": lambda self: self.expression(
+                exp.WatermarkColumnConstraint,
+                this=self._match(TokenType.FOR) and self._parse_column(),
+                expression=self._match(TokenType.ALIAS) and self._parse_disjunction(),
+            ),
+        }
+
+        SCHEMA_UNNAMED_CONSTRAINTS = {
+            *Postgres.Parser.SCHEMA_UNNAMED_CONSTRAINTS,
+            "WATERMARK",
+        }
+
         def _parse_table_hints(self) -> t.Optional[t.List[exp.Expression]]:
             # There is no hint in risingwave.
             # Do nothing here to avoid WITH keywords conflict in CREATE SINK statement.

sqlglot/parser.py

@@ -1141,11 +1141,6 @@ class Parser(metaclass=_Parser):
         "TTL": lambda self: self.expression(exp.MergeTreeTTL, expressions=[self._parse_bitwise()]),
         "UNIQUE": lambda self: self._parse_unique(),
         "UPPERCASE": lambda self: self.expression(exp.UppercaseColumnConstraint),
-        "WATERMARK": lambda self: self.expression(
-            exp.WatermarkColumnConstraint,
-            this=self._match(TokenType.FOR) and self._parse_column(),
-            expression=self._match(TokenType.ALIAS) and self._parse_disjunction(),
-        ),
         "WITH": lambda self: self.expression(
             exp.Properties, expressions=self._parse_wrapped_properties()
         ),
@@ -1211,7 +1206,6 @@ def _parse_partitioned_by_bucket_or_truncate(self) -> t.Optional[exp.Expression]
         "PERIOD",
         "PRIMARY KEY",
         "UNIQUE",
-        "WATERMARK",
         "BUCKET",
         "TRUNCATE",
     }

tests/dialects/test_duckdb.py

@@ -284,6 +284,7 @@ def test_duckdb(self):
         self.validate_identity(
             "SELECT LIST_TRANSFORM(LIST_FILTER([0, 1, 2, 3, 4, 5], LAMBDA x : x % 2 = 0), LAMBDA y : y * y)"
         )
+        self.validate_identity("CREATE TABLE notes (watermark TEXT)")
         self.validate_identity("SELECT LIST_TRANSFORM([5, NULL, 6], LAMBDA x : COALESCE(x, 0) + 1)")
         self.validate_identity("SELECT LIST_TRANSFORM(nbr, LAMBDA x : x + 1) FROM article AS a")
         self.validate_identity("SELECT * FROM my_ducklake.demo AT (VERSION => 2)")