Repository: tobymao/sqlglot

Issue:
TokenError: Error tokenizing 'select /*+ ORDERED */* from dual'
The tokenizer fails to parse a hint followed by a `*` if it is not separated by a space:

    /*+ ... */* from dual

```
import sqlglot

sqlglot.parse_one(
    '''
       select /*+ ORDERED */* from dual   
    '''
)
```

    sqlglot.errors.TokenError: Error tokenizing 'lect /*+ ORDERED */* from dual

This is valid syntax.

If I add a space between inside of `*/*` such that it is `*/ *`, it works fine.

```
sqlglot.parse_one(
    '''
       select /*+ ORDERED */ * from dual   
    '''
)
```

It also works when changing `*/*` to `/*1`.

```
sqlglot.parse_one(
    '''
       select /*+ ORDERED */1 from dual   
    '''
)

```
Note that this only works for hints, not comments, so this is fine:

```
sqlglot.parse_one(
    '''
       select /*+ ORDERED */1 from dual   
    '''
)
```

My guess is that the `*/*` is confusing the tokenizer for the start of a comment/hint like `/*`.

Thank you.

Discussion:

Solution patch:
diff --git a/README.md b/README.md
index 747887ea1a..1bccf10715 100644
--- a/README.md
+++ b/README.md
@@ -1,6 +1,6 @@
 ![SQLGlot logo](sqlglot.png)
 
-SQLGlot is a no-dependency SQL parser, transpiler, optimizer, and engine. It can be used to format SQL or translate between [23 different dialects](https://github.com/tobymao/sqlglot/blob/main/sqlglot/dialects/__init__.py) like [DuckDB](https://duckdb.org/), [Presto](https://prestodb.io/) / [Trino](https://trino.io/), [Spark](https://spark.apache.org/) / [Databricks](https://www.databricks.com/), [Snowflake](https://www.snowflake.com/en/), and [BigQuery](https://cloud.google.com/bigquery/). It aims to read a wide variety of SQL inputs and output syntactically and semantically correct SQL in the targeted dialects.
+SQLGlot is a no-dependency SQL parser, transpiler, optimizer, and engine. It can be used to format SQL or translate between [24 different dialects](https://github.com/tobymao/sqlglot/blob/main/sqlglot/dialects/__init__.py) like [DuckDB](https://duckdb.org/), [Presto](https://prestodb.io/) / [Trino](https://trino.io/), [Spark](https://spark.apache.org/) / [Databricks](https://www.databricks.com/), [Snowflake](https://www.snowflake.com/en/), and [BigQuery](https://cloud.google.com/bigquery/). It aims to read a wide variety of SQL inputs and output syntactically and semantically correct SQL in the targeted dialects.
 
 It is a very comprehensive generic SQL parser with a robust [test suite](https://github.com/tobymao/sqlglot/blob/main/tests/). It is also quite [performant](#benchmarks), while being written purely in Python.
 
diff --git a/sqlglot/dialects/oracle.py b/sqlglot/dialects/oracle.py
index 1025a075a6..cda13e7897 100644
--- a/sqlglot/dialects/oracle.py
+++ b/sqlglot/dialects/oracle.py
@@ -15,7 +15,6 @@
 from sqlglot.helper import seq_get
 from sqlglot.parser import OPTIONS_TYPE, build_coalesce
 from sqlglot.tokens import TokenType
-from sqlglot.errors import ParseError
 
 if t.TYPE_CHECKING:
     from sqlglot._typing import E
@@ -207,35 +206,6 @@ def _parse_json_array(self, expr_type: t.Type[E], **kwargs) -> E:
                 **kwargs,
             )
 
-        def _parse_hint(self) -> t.Optional[exp.Hint]:
-            start_index = self._index
-            should_fallback_to_string = False
-
-            if not self._match(TokenType.HINT):
-                return None
-
-            hints = []
-
-            try:
-                for hint in iter(
-                    lambda: self._parse_csv(
-                        lambda: self._parse_hint_function_call() or self._parse_var(upper=True),
-                    ),
-                    [],
-                ):
-                    hints.extend(hint)
-            except ParseError:
-                should_fallback_to_string = True
-
-            if not self._match_pair(TokenType.STAR, TokenType.SLASH):
-                should_fallback_to_string = True
-
-            if should_fallback_to_string:
-                self._retreat(start_index)
-                return self._parse_hint_fallback_to_string()
-
-            return self.expression(exp.Hint, expressions=hints)
-
         def _parse_hint_function_call(self) -> t.Optional[exp.Expression]:
             if not self._curr or not self._next or self._next.token_type != TokenType.L_PAREN:
                 return None
@@ -258,20 +228,6 @@ def _parse_hint_args(self):
 
             return args
 
-        def _parse_hint_fallback_to_string(self) -> t.Optional[exp.Hint]:
-            if self._match(TokenType.HINT):
-                start = self._curr
-                while self._curr and not self._match_pair(TokenType.STAR, TokenType.SLASH):
-                    self._advance()
-
-                if not self._curr:
-                    self.raise_error("Expected */ after HINT")
-
-                end = self._tokens[self._index - 3]
-                return exp.Hint(expressions=[self._find_sql(start, end)])
-
-            return None
-
         def _parse_query_restrictions(self) -> t.Optional[exp.Expression]:
             kind = self._parse_var_from_options(self.QUERY_RESTRICTIONS, raise_unmatched=False)
 
diff --git a/sqlglot/dialects/risingwave.py b/sqlglot/dialects/risingwave.py
index 503907eb13..97a68a59a6 100644
--- a/sqlglot/dialects/risingwave.py
+++ b/sqlglot/dialects/risingwave.py
@@ -1,7 +1,6 @@
 from __future__ import annotations
 from sqlglot.dialects.postgres import Postgres
 from sqlglot.tokens import TokenType
-from collections import defaultdict
 import typing as t
 
 from sqlglot import exp
@@ -16,11 +15,13 @@ class Tokenizer(Postgres.Tokenizer):
         }
 
     class Parser(Postgres.Parser):
+        WRAPPED_TRANSFORM_COLUMN_CONSTRAINT = False
+
         PROPERTY_PARSERS = {
             **Postgres.Parser.PROPERTY_PARSERS,
-            "ENCODE": lambda self: self._parse_encode_property(is_key=False),
-            "KEY": lambda self: self._parse_key_encode_property(),
+            "ENCODE": lambda self: self._parse_encode_property(),
             "INCLUDE": lambda self: self._parse_include_property(),
+            "KEY": lambda self: self._parse_encode_property(key=True),
         }
 
         def _parse_table_hints(self) -> t.Optional[t.List[exp.Expression]]:
@@ -28,443 +29,46 @@ def _parse_table_hints(self) -> t.Optional[t.List[exp.Expression]]:
             # Do nothing here to avoid WITH keywords conflict in CREATE SINK statement.
             return None
 
-        # @FIXME: this method look too complicated.
         def _parse_include_property(self) -> t.Optional[exp.Expression]:
-            column_type = self._parse_var_or_string()
-            column_alias = None
-            inner_field = None
-            header_inner_expect_type = None
-            if not column_type:
-                return None
-            inner_field = self._parse_string() or self._parse_var()
-            if inner_field:
-                if self._match_set(self.TYPE_TOKENS, advance=False):
-                    header_inner_expect_type = self._parse_var_or_string()
-
-            include_property = self.expression(exp.IncludeProperty, column_type=column_type)
-
-            if self._match(TokenType.ALIAS):
-                column_alias = self._parse_var_or_string()
-            if column_alias:
-                include_property.set("column_alias", column_alias)
-            if inner_field:
-                include_property.set("inner_field", inner_field)
-            if header_inner_expect_type:
-                include_property.set("header_inner_expect_type", header_inner_expect_type)
-
-            return include_property
-
-        def _parse_key_encode_property(self) -> t.Optional[exp.Expression]:
-            index = self._index - 1
-            if not self._match_texts("ENCODE"):
-                self._retreat(index)
-                return None
-            return self._parse_encode_property(is_key=True)
-
-        def _parse_encode_property(self, is_key: bool) -> t.Optional[exp.Expression]:
-            encode_var = self._parse_var_or_string()
-            if not encode_var:
-                return None
-            encode_property: exp.Expression = self.expression(
-                exp.EncodeProperty, this=encode_var, is_key=is_key
-            )
-            if self._match(TokenType.L_PAREN, advance=False):
-                expressions = exp.Properties(expressions=self._parse_wrapped_properties())
-                encode_property.set("expressions", expressions)
-            return encode_property
-
-        # @FIXME: refactor this method.
-        def _parse_create(self) -> exp.Create | exp.Command:
-            # Note: this can't be None because we've matched a statement parser
-            start = self._prev
-            comments = self._prev_comments
-
-            replace = (
-                start.token_type == TokenType.REPLACE
-                or self._match_pair(TokenType.OR, TokenType.REPLACE)
-                or self._match_pair(TokenType.OR, TokenType.ALTER)
-            )
-            refresh = self._match_pair(TokenType.OR, TokenType.REFRESH)
-
-            unique = self._match(TokenType.UNIQUE)
-
-            if self._match_text_seq("CLUSTERED", "COLUMNSTORE"):
-                clustered = True
-            elif self._match_text_seq("NONCLUSTERED", "COLUMNSTORE") or self._match_text_seq(
-                "COLUMNSTORE"
-            ):
-                clustered = False
-            else:
-                clustered = None
-
-            if self._match_pair(TokenType.TABLE, TokenType.FUNCTION, advance=False):
-                self._advance()
-
-            properties = None
-            create_token = self._match_set(self.CREATABLES) and self._prev
-            if not create_token:
-                # exp.Properties.Location.POST_CREATE
-                properties = self._parse_properties()
-                create_token = self._match_set(self.CREATABLES) and self._prev
-
-                if not properties or not create_token:
-                    return self._parse_as_command(start)
-
-            concurrently = self._match_text_seq("CONCURRENTLY")
-            exists = self._parse_exists(not_=True)
-            this = None
-            expression: t.Optional[exp.Expression] = None
-            indexes = None
-            no_schema_binding = None
-            begin = None
-            end = None
-            clone = None
-
-            def extend_props(temp_props: t.Optional[exp.Properties]) -> None:
-                nonlocal properties
-                if properties and temp_props:
-                    properties.expressions.extend(temp_props.expressions)
-                elif temp_props:
-                    properties = temp_props
-
-            if create_token.token_type in (TokenType.FUNCTION, TokenType.PROCEDURE):
-                this = self._parse_user_defined_function(kind=create_token.token_type)
-
-                # exp.Properties.Location.POST_SCHEMA ("schema" here is the UDF's type signature)
-                extend_props(self._parse_properties())
+            header: t.Optional[exp.Expression] = None
+            coldef: t.Optional[exp.Expression] = None
 
-                expression = self._match(TokenType.ALIAS) and self._parse_heredoc()
-                extend_props(self._parse_properties())
+            this = self._parse_var_or_string()
 
-                if not expression:
-                    if self._match(TokenType.COMMAND):
-                        expression = self._parse_as_command(self._prev)
-                    else:
-                        begin = self._match(TokenType.BEGIN)
-                        return_ = self._match_text_seq("RETURN")
-
-                        if self._match(TokenType.STRING, advance=False):
-                            # Takes care of BigQuery's JavaScript UDF definitions that end in an OPTIONS property
-                            # # https://cloud.google.com/bigquery/docs/reference/standard-sql/data-definition-language#create_function_statement
-                            expression = self._parse_string()
-                            extend_props(self._parse_properties())
-                        else:
-                            expression = self._parse_user_defined_function_expression()
-
-                        end = self._match_text_seq("END")
-
-                        if return_:
-                            expression = self.expression(exp.Return, this=expression)
-            elif create_token.token_type == TokenType.INDEX:
-                # Postgres allows anonymous indexes, eg. CREATE INDEX IF NOT EXISTS ON t(c)
-                if not self._match(TokenType.ON):
-                    index = self._parse_id_var()
-                    anonymous = False
-                else:
-                    index = None
-                    anonymous = True
-
-                this = self._parse_index(index=index, anonymous=anonymous)
-            elif create_token.token_type in self.DB_CREATABLES:
-                table_parts = self._parse_table_parts(
-                    schema=True, is_db_reference=create_token.token_type == TokenType.SCHEMA
-                )
-
-                # exp.Properties.Location.POST_NAME
-                self._match(TokenType.COMMA)
-                extend_props(self._parse_properties(before=True))
-
-                this = self._parse_schema(this=table_parts)
-
-                # exp.Properties.Location.POST_SCHEMA and POST_WITH
-                extend_props(self._parse_properties())
-
-                self._match(TokenType.ALIAS)
-                if not self._match_set(self.DDL_SELECT_TOKENS, advance=False):
-                    # exp.Properties.Location.POST_ALIAS
-                    extend_props(self._parse_properties())
-
-                if create_token.token_type == TokenType.SEQUENCE:
-                    expression = self._parse_types()
-                    extend_props(self._parse_properties())
-                else:
-                    expression = self._parse_ddl_select()
-
-                if create_token.token_type == TokenType.TABLE:
-                    # exp.Properties.Location.POST_EXPRESSION
-                    extend_props(self._parse_properties())
-
-                    indexes = []
-                    while True:
-                        index = self._parse_index()
-                        # exp.Properties.Location.POST_INDEX
-                        extend_props(self._parse_properties())
-                        if not index:
-                            break
-                        else:
-                            self._match(TokenType.COMMA)
-                            indexes.append(index)
-                elif create_token.token_type == TokenType.VIEW:
-                    if self._match_text_seq("WITH", "NO", "SCHEMA", "BINDING"):
-                        no_schema_binding = True
-                # Newly added to support Risingwave sink.
-                elif create_token.token_type == TokenType.SINK:
-                    extend_props(self._parse_properties())
-                elif create_token.token_type == TokenType.SOURCE:
-                    extend_props(self._parse_properties())
-
-                shallow = self._match_text_seq("SHALLOW")
-
-                if self._match_texts(self.CLONE_KEYWORDS):
-                    copy = self._prev.text.lower() == "copy"
-                    clone = self.expression(
-                        exp.Clone, this=self._parse_table(schema=True), shallow=shallow, copy=copy
-                    )
-
-            if self._curr and not self._match_set(
-                (TokenType.R_PAREN, TokenType.COMMA), advance=False
-            ):
-                return self._parse_as_command(start)
-
-            create_kind_text = create_token.text.upper()
-            return self.expression(
-                exp.Create,
-                comments=comments,
-                this=this,
-                kind=self.dialect.CREATABLE_KIND_MAPPING.get(create_kind_text) or create_kind_text,
-                replace=replace,
-                refresh=refresh,
-                unique=unique,
-                expression=expression,
-                exists=exists,
-                properties=properties,
-                indexes=indexes,
-                no_schema_binding=no_schema_binding,
-                begin=begin,
-                end=end,
-                clone=clone,
-                concurrently=concurrently,
-                clustered=clustered,
-            )
-
-        def _parse_watermark_field(self) -> t.Optional[exp.Expression]:
-            if not self._match_texts("WATERMARK"):
-                return None
-            if not self._match(TokenType.FOR):
-                return None
-            column: t.Optional[exp.Expression] = self._parse_var_or_string()
-            if not column:
-                return None
             if not self._match(TokenType.ALIAS):
-                return None
-            expr: t.Optional[exp.Expression] = self._parse_expression()
-            if not expr:
-                return None
-            return self.expression(exp.Watermark, column=column, expression=expr)
+                header = self._parse_field()
+                if header:
+                    coldef = self.expression(exp.ColumnDef, this=header, kind=self._parse_types())
 
-        def _parse_field_def(self) -> t.Optional[exp.Expression]:
-            # First parse if it is watermark field before parsing super class field def.
-            return self._parse_watermark_field() or super()._parse_field_def()
+            self._match(TokenType.ALIAS)
+            alias = self._parse_id_var(tokens=self.ALIAS_TOKENS)
 
-        def _parse_column_def(self, this: t.Optional[exp.Expression]) -> t.Optional[exp.Expression]:
-            # column defs are not really columns, they're identifiers
-            if isinstance(this, exp.Column):
-                this = this.this
+            return self.expression(exp.IncludeProperty, this=this, alias=alias, column_def=coldef)
 
-            kind = self._parse_types(schema=True)
-            constraints: t.List[exp.Expression] = []
+        def _parse_encode_property(self, key: t.Optional[bool] = None) -> exp.EncodeProperty:
+            self._match_text_seq("ENCODE")
+            this = self._parse_var_or_string()
 
-            if (
-                kind
-                and self._match(TokenType.ALIAS, advance=False)
-                and not self._match_pair(TokenType.ALIAS, TokenType.L_PAREN, advance=False)
-            ):
-                # Deal with CREATE SOURCE statement, where we may have col_name data_type [ AS generation_expression ] and generation_expression may not have parentheses.
-                self._match(TokenType.ALIAS)
-                constraints.append(
-                    self.expression(
-                        exp.ColumnConstraint,
-                        kind=exp.TransformColumnConstraint(this=self._parse_expression()),
-                    )
+            if self._match(TokenType.L_PAREN, advance=False):
+                properties = self.expression(
+                    exp.Properties, expressions=self._parse_wrapped_properties()
                 )
+            else:
+                properties = None
 
-                while True:
-                    constraint = self._parse_column_constraint()
-                    if not constraint:
-                        break
-                    constraints.append(constraint)
-
-                return self.expression(exp.ColumnDef, this=this, kind=kind, constraints=constraints)
-
-            return super()._parse_column_def(this)
+            return self.expression(exp.EncodeProperty, this=this, properties=properties, key=key)
 
     class Generator(Postgres.Generator):
         LOCKING_READS_SUPPORTED = False
 
+        TRANSFORMS = {
+            **Postgres.Generator.TRANSFORMS,
+            exp.FileFormatProperty: lambda self, e: f"FORMAT {self.sql(e, 'this')}",
+        }
+
         PROPERTIES_LOCATION = {
             **Postgres.Generator.PROPERTIES_LOCATION,
             exp.FileFormatProperty: exp.Properties.Location.POST_EXPRESSION,
-            exp.EncodeProperty: exp.Properties.Location.POST_EXPRESSION,
-            exp.IncludeProperty: exp.Properties.Location.POST_SCHEMA,
         }
 
-        def _encode_property_sql(self, expression: exp.Expression) -> str:
-            is_key = expression.args.get("is_key")
-            property_sql_str: str = "KEY ENCODE" if is_key else "ENCODE"
-
-            property_value = self.sql(expression, "this")
-            property_sql_str = f"{property_sql_str} {property_value}"
-
-            encode_expr = expression.args.get("expressions")
-            if encode_expr:
-                property_sql_str = f"{property_sql_str} {self.properties(encode_expr)}"
-
-            return property_sql_str
-
-        def watermark_sql(self, expression: exp.Expression) -> str:
-            column = expression.args.get("column")
-            expression_str = expression.args.get("expression")
-            return f"WATERMARK FOR {column} AS {expression_str}"
-
-        def _include_property_sql(self, expression: exp.Expression) -> str:
-            column_type = self.sql(expression, "column_type")
-            property_sql_str = f"INCLUDE {column_type}"
-            inner_field = self.sql(expression, "inner_field")
-            if inner_field:
-                property_sql_str = f"{property_sql_str} {inner_field}"
-            column_alias = self.sql(expression, "column_alias")
-            header_inner_expect_type = self.sql(expression, "header_inner_expect_type")
-            if header_inner_expect_type:
-                property_sql_str = f"{property_sql_str} {header_inner_expect_type}"
-            if column_alias:
-                property_sql_str = f"{property_sql_str} AS {column_alias}"
-            return property_sql_str
-
-        def property_sql(self, expression: exp.Property) -> str:
-            property_cls = expression.__class__
-            if property_cls == exp.EncodeProperty:
-                return self._encode_property_sql(expression)
-
-            if property_cls == exp.FileFormatProperty:
-                return f"FORMAT {self.sql(expression, 'this')}"
-
-            if property_cls == exp.IncludeProperty:
-                return self._include_property_sql(expression)
-
-            return super().property_sql(expression)
-
-        # @FIXME: refactor this method.
-        def create_sql(self, expression: exp.Create) -> str:
-            kind = self.sql(expression, "kind")
-            kind = self.dialect.INVERSE_CREATABLE_KIND_MAPPING.get(kind) or kind
-            properties = expression.args.get("properties")
-            properties_locs = self.locate_properties(properties) if properties else defaultdict()
-
-            this = self.createable_sql(expression, properties_locs)
-
-            properties_sql = ""
-            if properties_locs.get(exp.Properties.Location.POST_SCHEMA) or properties_locs.get(
-                exp.Properties.Location.POST_WITH
-            ):
-                properties_sql = self.sql(
-                    exp.Properties(
-                        expressions=[
-                            *properties_locs[exp.Properties.Location.POST_SCHEMA],
-                            *properties_locs[exp.Properties.Location.POST_WITH],
-                        ]
-                    )
-                )
-
-                if properties_locs.get(exp.Properties.Location.POST_SCHEMA):
-                    properties_sql = self.sep() + properties_sql
-                elif not self.pretty:
-                    # Standalone POST_WITH properties need a leading whitespace in non-pretty mode
-                    properties_sql = f" {properties_sql}"
-
-            begin = " BEGIN" if expression.args.get("begin") else ""
-            end = " END" if expression.args.get("end") else ""
-
-            expression_sql = self.sql(expression, "expression")
-            if expression_sql:
-                expression_sql = f"{begin}{self.sep()}{expression_sql}{end}"
-
-                if self.CREATE_FUNCTION_RETURN_AS or not isinstance(
-                    expression.expression, exp.Return
-                ):
-                    postalias_props_sql = ""
-                    if properties_locs.get(exp.Properties.Location.POST_ALIAS):
-                        postalias_props_sql = self.properties(
-                            exp.Properties(
-                                expressions=properties_locs[exp.Properties.Location.POST_ALIAS]
-                            ),
-                            wrapped=False,
-                        )
-                    postalias_props_sql = f" {postalias_props_sql}" if postalias_props_sql else ""
-                    expression_sql = f" AS{postalias_props_sql}{expression_sql}"
-
-            postindex_props_sql = ""
-            if properties_locs.get(exp.Properties.Location.POST_INDEX):
-                postindex_props_sql = self.properties(
-                    exp.Properties(expressions=properties_locs[exp.Properties.Location.POST_INDEX]),
-                    wrapped=False,
-                    prefix=" ",
-                )
-
-            indexes = self.expressions(expression, key="indexes", indent=False, sep=" ")
-            indexes = f" {indexes}" if indexes else ""
-            index_sql = indexes + postindex_props_sql
-
-            replace = " OR REPLACE" if expression.args.get("replace") else ""
-            refresh = " OR REFRESH" if expression.args.get("refresh") else ""
-            unique = " UNIQUE" if expression.args.get("unique") else ""
-
-            clustered = expression.args.get("clustered")
-            if clustered is None:
-                clustered_sql = ""
-            elif clustered:
-                clustered_sql = " CLUSTERED COLUMNSTORE"
-            else:
-                clustered_sql = " NONCLUSTERED COLUMNSTORE"
-
-            postcreate_props_sql = ""
-            if properties_locs.get(exp.Properties.Location.POST_CREATE):
-                postcreate_props_sql = self.properties(
-                    exp.Properties(
-                        expressions=properties_locs[exp.Properties.Location.POST_CREATE]
-                    ),
-                    sep=" ",
-                    prefix=" ",
-                    wrapped=False,
-                )
-
-            modifiers = "".join((clustered_sql, replace, refresh, unique, postcreate_props_sql))
-
-            postexpression_props_sql = ""
-            if properties_locs.get(exp.Properties.Location.POST_EXPRESSION):
-                postexpression_props_sql = self.properties(
-                    exp.Properties(
-                        expressions=properties_locs[exp.Properties.Location.POST_EXPRESSION]
-                    ),
-                    sep=" ",
-                    prefix=" ",
-                    wrapped=False,
-                )
-
-            if kind == "SINK" or kind == "SOURCE" and self.pretty:
-                postexpression_props_sql = self.sep() + postexpression_props_sql.strip()
-
-            concurrently = " CONCURRENTLY" if expression.args.get("concurrently") else ""
-            exists_sql = " IF NOT EXISTS" if expression.args.get("exists") else ""
-            no_schema_binding = (
-                " WITH NO SCHEMA BINDING" if expression.args.get("no_schema_binding") else ""
-            )
-
-            clone = self.sql(expression, "clone")
-            clone = f" {clone}" if clone else ""
-
-            if kind == "SINK":
-                # Tailored for risingwave sink.
-                expression_sql = f"CREATE SINK{concurrently}{exists_sql} {this}{expression_sql}{properties_sql}{postexpression_props_sql}{index_sql}{no_schema_binding}{clone}"
-            else:
-                expression_sql = f"CREATE{modifiers} {kind}{concurrently}{exists_sql} {this}{properties_sql}{expression_sql}{postexpression_props_sql}{index_sql}{no_schema_binding}{clone}"
-            return self.prepend_ctes(expression, expression_sql)
+        EXPRESSION_PRECEDES_PROPERTIES_CREATABLES = {"SINK"}
diff --git a/sqlglot/expressions.py b/sqlglot/expressions.py
index 6c99846a7e..14fdfe14e1 100644
--- a/sqlglot/expressions.py
+++ b/sqlglot/expressions.py
@@ -1789,10 +1789,6 @@ class ColumnConstraintKind(Expression):
     pass
 
 
-class Watermark(Expression):
-    arg_types = {"column": True, "expression": True}
-
-
 class AutoIncrementColumnConstraint(ColumnConstraintKind):
     pass
 
@@ -1938,6 +1934,11 @@ class UppercaseColumnConstraint(ColumnConstraintKind):
     arg_types: t.Dict[str, t.Any] = {}
 
 
+# https://docs.risingwave.com/processing/watermarks#syntax
+class WatermarkColumnConstraint(Expression):
+    arg_types = {"this": True, "expression": True}
+
+
 class PathColumnConstraint(ColumnConstraintKind):
     pass
 
@@ -3029,16 +3030,11 @@ class WithProcedureOptions(Property):
 
 
 class EncodeProperty(Property):
-    arg_types = {"this": True, "is_key": True, "expressions": False}
+    arg_types = {"this": True, "properties": False, "key": False}
 
 
 class IncludeProperty(Property):
-    arg_types = {
-        "column_type": True,
-        "inner_field": False,
-        "header_inner_expect_type": False,
-        "column_alias": False,
-    }
+    arg_types = {"this": True, "alias": False, "column_def": False}
 
 
 class Properties(Expression):
diff --git a/sqlglot/generator.py b/sqlglot/generator.py
index 83222cdccc..478871db8b 100644
--- a/sqlglot/generator.py
+++ b/sqlglot/generator.py
@@ -496,6 +496,8 @@ class Generator(metaclass=_Generator):
     PARAMETER_TOKEN = "@"
     NAMED_PLACEHOLDER_TOKEN = ":"
 
+    EXPRESSION_PRECEDES_PROPERTIES_CREATABLES: t.Set[str] = set()
+
     PROPERTIES_LOCATION = {
         exp.AllowedValuesProperty: exp.Properties.Location.POST_SCHEMA,
         exp.AlgorithmProperty: exp.Properties.Location.POST_CREATE,
@@ -520,6 +522,7 @@ class Generator(metaclass=_Generator):
         exp.DistKeyProperty: exp.Properties.Location.POST_SCHEMA,
         exp.DistStyleProperty: exp.Properties.Location.POST_SCHEMA,
         exp.EmptyProperty: exp.Properties.Location.POST_SCHEMA,
+        exp.EncodeProperty: exp.Properties.Location.POST_EXPRESSION,
         exp.EngineProperty: exp.Properties.Location.POST_SCHEMA,
         exp.ExecuteAsProperty: exp.Properties.Location.POST_SCHEMA,
         exp.ExternalProperty: exp.Properties.Location.POST_CREATE,
@@ -530,6 +533,7 @@ class Generator(metaclass=_Generator):
         exp.HeapProperty: exp.Properties.Location.POST_WITH,
         exp.InheritsProperty: exp.Properties.Location.POST_SCHEMA,
         exp.IcebergProperty: exp.Properties.Location.POST_CREATE,
+        exp.IncludeProperty: exp.Properties.Location.POST_SCHEMA,
         exp.InputModelProperty: exp.Properties.Location.POST_SCHEMA,
         exp.IsolatedLoadingProperty: exp.Properties.Location.POST_NAME,
         exp.JournalProperty: exp.Properties.Location.POST_NAME,
@@ -1154,7 +1158,12 @@ def create_sql(self, expression: exp.Create) -> str:
         clone = self.sql(expression, "clone")
         clone = f" {clone}" if clone else ""
 
-        expression_sql = f"CREATE{modifiers} {kind}{concurrently}{exists_sql} {this}{properties_sql}{expression_sql}{postexpression_props_sql}{index_sql}{no_schema_binding}{clone}"
+        if kind in self.EXPRESSION_PRECEDES_PROPERTIES_CREATABLES:
+            properties_expression = f"{expression_sql}{properties_sql}"
+        else:
+            properties_expression = f"{properties_sql}{expression_sql}"
+
+        expression_sql = f"CREATE{modifiers} {kind}{concurrently}{exists_sql} {this}{properties_expression}{postexpression_props_sql}{index_sql}{no_schema_binding}{clone}"
         return self.prepend_ctes(expression, expression_sql)
 
     def sequenceproperties_sql(self, expression: exp.SequenceProperties) -> str:
@@ -4556,3 +4565,32 @@ def featuresattime_sql(self, expression: exp.FeaturesAtTime) -> str:
             expression.args.get("num_rows"),
             expression.args.get("ignore_feature_nulls"),
         )
+
+    def watermarkcolumnconstraint_sql(self, expression: exp.WatermarkColumnConstraint) -> str:
+        return (
+            f"WATERMARK FOR {self.sql(expression, 'this')} AS {self.sql(expression, 'expression')}"
+        )
+
+    def encodeproperty_sql(self, expression: exp.EncodeProperty) -> str:
+        encode = "KEY ENCODE" if expression.args.get("key") else "ENCODE"
+        encode = f"{encode} {self.sql(expression, 'this')}"
+
+        properties = expression.args.get("properties")
+        if properties:
+            encode = f"{encode} {self.properties(properties)}"
+
+        return encode
+
+    def includeproperty_sql(self, expression: exp.IncludeProperty) -> str:
+        this = self.sql(expression, "this")
+        include = f"INCLUDE {this}"
+
+        column_def = self.sql(expression, "column_def")
+        if column_def:
+            include = f"{include} {column_def}"
+
+        alias = self.sql(expression, "alias")
+        if alias:
+            include = f"{include} AS {alias}"
+
+        return include
diff --git a/sqlglot/parser.py b/sqlglot/parser.py
index 3d12af5981..dde52bf219 100644
--- a/sqlglot/parser.py
+++ b/sqlglot/parser.py
@@ -758,6 +758,7 @@ class Parser(metaclass=_Parser):
         exp.From: lambda self: self._parse_from(joins=True),
         exp.Group: lambda self: self._parse_group(),
         exp.Having: lambda self: self._parse_having(),
+        exp.Hint: lambda self: self._parse_hint_body(),
         exp.Identifier: lambda self: self._parse_id_var(),
         exp.Join: lambda self: self._parse_join(),
         exp.Lambda: lambda self: self._parse_lambda(),
@@ -1057,6 +1058,11 @@ 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()
         ),
@@ -1091,6 +1097,7 @@ class Parser(metaclass=_Parser):
         "PERIOD",
         "PRIMARY KEY",
         "UNIQUE",
+        "WATERMARK",
     }
 
     NO_PAREN_FUNCTION_PARSERS = {
@@ -1360,6 +1367,9 @@ class Parser(metaclass=_Parser):
     # Whether a PARTITION clause can follow a table reference
     SUPPORTS_PARTITION_SELECTION = False
 
+    # Whether the `name AS expr` schema/column constraint requires parentheses around `expr`
+    WRAPPED_TRANSFORM_COLUMN_CONSTRAINT = True
+
     __slots__ = (
         "error_level",
         "error_message_context",
@@ -1913,6 +1923,8 @@ def extend_props(temp_props: t.Optional[exp.Properties]) -> None:
             elif create_token.token_type == TokenType.VIEW:
                 if self._match_text_seq("WITH", "NO", "SCHEMA", "BINDING"):
                     no_schema_binding = True
+            elif create_token.token_type in (TokenType.SINK, TokenType.SOURCE):
+                extend_props(self._parse_properties())
 
             shallow = self._match_text_seq("SHALLOW")
 
@@ -3228,21 +3240,42 @@ def _parse_query_modifiers(
 
         return this
 
-    def _parse_hint(self) -> t.Optional[exp.Hint]:
-        if self._match(TokenType.HINT):
-            hints = []
+    def _parse_hint_fallback_to_string(self) -> t.Optional[exp.Hint]:
+        start = self._curr
+        while self._curr:
+            self._advance()
+
+        end = self._tokens[self._index - 1]
+        return exp.Hint(expressions=[self._find_sql(start, end)])
+
+    def _parse_hint_function_call(self) -> t.Optional[exp.Expression]:
+        return self._parse_function_call()
+
+    def _parse_hint_body(self) -> t.Optional[exp.Hint]:
+        start_index = self._index
+        should_fallback_to_string = False
+
+        hints = []
+        try:
             for hint in iter(
                 lambda: self._parse_csv(
-                    lambda: self._parse_function() or self._parse_var(upper=True)
+                    lambda: self._parse_hint_function_call() or self._parse_var(upper=True),
                 ),
                 [],
             ):
                 hints.extend(hint)
+        except ParseError:
+            should_fallback_to_string = True
 
-            if not self._match_pair(TokenType.STAR, TokenType.SLASH):
-                self.raise_error("Expected */ after HINT")
+        if should_fallback_to_string or self._curr:
+            self._retreat(start_index)
+            return self._parse_hint_fallback_to_string()
 
-            return self.expression(exp.Hint, expressions=hints)
+        return self.expression(exp.Hint, expressions=hints)
+
+    def _parse_hint(self) -> t.Optional[exp.Hint]:
+        if self._match(TokenType.HINT) and self._prev_comments:
+            return exp.maybe_parse(self._prev_comments[0], into=exp.Hint, dialect=self.dialect)
 
         return None
 
@@ -5477,12 +5510,19 @@ def _parse_column_def(self, this: t.Optional[exp.Expression]) -> t.Optional[exp.
                 not_null=self._match_pair(TokenType.NOT, TokenType.NULL),
             )
             constraints.append(self.expression(exp.ColumnConstraint, kind=constraint_kind))
-        elif kind and self._match_pair(TokenType.ALIAS, TokenType.L_PAREN, advance=False):
-            self._match(TokenType.ALIAS)
+        elif (
+            kind
+            and self._match(TokenType.ALIAS, advance=False)
+            and (
+                not self.WRAPPED_TRANSFORM_COLUMN_CONSTRAINT
+                or (self._next and self._next.token_type == TokenType.L_PAREN)
+            )
+        ):
+            self._advance()
             constraints.append(
                 self.expression(
                     exp.ColumnConstraint,
-                    kind=exp.TransformColumnConstraint(this=self._parse_field()),
+                    kind=exp.TransformColumnConstraint(this=self._parse_disjunction()),
                 )
             )
 
diff --git a/sqlglot/tokens.py b/sqlglot/tokens.py
index ad4c0fe158..65d1ec0bb4 100644
--- a/sqlglot/tokens.py
+++ b/sqlglot/tokens.py
@@ -511,6 +511,8 @@ def _quotes_to_format(
             ),
             "{#": "#}",  # Ensure Jinja comments are tokenized correctly in all dialects
         }
+        if klass.HINT_START in klass.KEYWORDS:
+            klass._COMMENTS[klass.HINT_START] = "*/"
 
         klass._KEYWORD_TRIE = new_trie(
             key.upper()
@@ -548,6 +550,10 @@ def _quotes_to_format(
                 heredoc_tag_is_identifier=klass.HEREDOC_TAG_IS_IDENTIFIER,
                 string_escapes_allowed_in_raw_strings=klass.STRING_ESCAPES_ALLOWED_IN_RAW_STRINGS,
                 nested_comments=klass.NESTED_COMMENTS,
+                hint_start=klass.HINT_START,
+                tokens_preceding_hint={
+                    _TOKEN_TYPE_TO_INDEX[v] for v in klass.TOKENS_PRECEDING_HINT
+                },
             )
             token_types = RsTokenTypeSettings(
                 bit_string=_TOKEN_TYPE_TO_INDEX[TokenType.BIT_STRING],
@@ -563,6 +569,7 @@ def _quotes_to_format(
                 string=_TOKEN_TYPE_TO_INDEX[TokenType.STRING],
                 var=_TOKEN_TYPE_TO_INDEX[TokenType.VAR],
                 heredoc_string_alternative=_TOKEN_TYPE_TO_INDEX[klass.HEREDOC_STRING_ALTERNATIVE],
+                hint=_TOKEN_TYPE_TO_INDEX[TokenType.HINT],
             )
             klass._RS_TOKENIZER = RsTokenizer(settings, token_types)
         else:
@@ -633,6 +640,10 @@ class Tokenizer(metaclass=_Tokenizer):
 
     NESTED_COMMENTS = True
 
+    HINT_START = "/*+"
+
+    TOKENS_PRECEDING_HINT = {TokenType.SELECT, TokenType.INSERT, TokenType.UPDATE, TokenType.DELETE}
+
     # Autofilled
     _COMMENTS: t.Dict[str, str] = {}
     _FORMAT_STRINGS: t.Dict[str, t.Tuple[str, TokenType]] = {}
@@ -648,7 +659,7 @@ class Tokenizer(metaclass=_Tokenizer):
         **{f"{prefix}%}}": TokenType.BLOCK_END for prefix in ("", "+", "-")},
         **{f"{{{{{postfix}": TokenType.BLOCK_START for postfix in ("+", "-")},
         **{f"{prefix}}}}}": TokenType.BLOCK_END for prefix in ("+", "-")},
-        "/*+": TokenType.HINT,
+        HINT_START: TokenType.HINT,
         "==": TokenType.EQ,
         "::": TokenType.DCOLON,
         "||": TokenType.DPIPE,
@@ -1232,6 +1243,13 @@ def _scan_comment(self, comment_start: str) -> bool:
                 self._advance(alnum=True)
             self._comments.append(self._text[comment_start_size:])
 
+        if (
+            comment_start == self.HINT_START
+            and self.tokens
+            and self.tokens[-1].token_type in self.TOKENS_PRECEDING_HINT
+        ):
+            self._add(TokenType.HINT)
+
         # Leading comment is attached to the succeeding token, whilst trailing comment to the preceding.
         # Multiple consecutive comments are preserved by appending them to the current comments list.
         if comment_start_line == self._prev_token_line:
diff --git a/sqlglotrs/src/settings.rs b/sqlglotrs/src/settings.rs
index 3068fd2813..7bc4882e3b 100644
--- a/sqlglotrs/src/settings.rs
+++ b/sqlglotrs/src/settings.rs
@@ -19,6 +19,7 @@ pub struct TokenTypeSettings {
     pub string: TokenType,
     pub var: TokenType,
     pub heredoc_string_alternative: TokenType,
+    pub hint: TokenType,
 }
 
 #[pymethods]
@@ -38,6 +39,7 @@ impl TokenTypeSettings {
         string: TokenType,
         var: TokenType,
         heredoc_string_alternative: TokenType,
+        hint: TokenType,
     ) -> Self {
         TokenTypeSettings {
             bit_string,
@@ -53,6 +55,7 @@ impl TokenTypeSettings {
             string,
             var,
             heredoc_string_alternative,
+            hint,
         }
     }
 }
@@ -75,9 +78,11 @@ pub struct TokenizerSettings {
     pub var_single_tokens: HashSet<char>,
     pub commands: HashSet<TokenType>,
     pub command_prefix_tokens: HashSet<TokenType>,
+    pub tokens_preceding_hint: HashSet<TokenType>,
     pub heredoc_tag_is_identifier: bool,
     pub string_escapes_allowed_in_raw_strings: bool,
     pub nested_comments: bool,
+    pub hint_start: String,
 }
 
 #[pymethods]
@@ -99,9 +104,11 @@ impl TokenizerSettings {
         var_single_tokens: HashSet<String>,
         commands: HashSet<TokenType>,
         command_prefix_tokens: HashSet<TokenType>,
+        tokens_preceding_hint: HashSet<TokenType>,
         heredoc_tag_is_identifier: bool,
         string_escapes_allowed_in_raw_strings: bool,
         nested_comments: bool,
+        hint_start: String,
     ) -> Self {
         let to_char = |v: &String| {
             if v.len() == 1 {
@@ -150,9 +157,11 @@ impl TokenizerSettings {
             var_single_tokens: var_single_tokens_native,
             commands,
             command_prefix_tokens,
+            tokens_preceding_hint,
             heredoc_tag_is_identifier,
             string_escapes_allowed_in_raw_strings,
             nested_comments,
+            hint_start,
         }
     }
 }
diff --git a/sqlglotrs/src/tokenizer.rs b/sqlglotrs/src/tokenizer.rs
index beff96e2fc..8228b5a8c5 100644
--- a/sqlglotrs/src/tokenizer.rs
+++ b/sqlglotrs/src/tokenizer.rs
@@ -395,6 +395,12 @@ impl<'a> TokenizerState<'a> {
                 .push(self.text()[comment_start_size..].to_string());
         }
 
+        if comment_start == self.settings.hint_start
+            && self.tokens.last().is_some()
+            && self.settings.tokens_preceding_hint.contains(&self.tokens.last().unwrap().token_type) {
+            self.add(self.token_types.hint, None)?;
+        }
+
         // Leading comment is attached to the succeeding token, whilst trailing comment to the preceding.
         // Multiple consecutive comments are preserved by appending them to the current comments list.
         if Some(comment_start_line) == self.previous_token_line {


Test patch:
diff --git a/tests/dialects/test_oracle.py b/tests/dialects/test_oracle.py
index 0784810a8e..2bb7a2d2f9 100644
--- a/tests/dialects/test_oracle.py
+++ b/tests/dialects/test_oracle.py
@@ -287,6 +287,17 @@ def test_oracle(self):
                 "clickhouse": "TRIM(BOTH 'h' FROM 'Hello World')",
             },
         )
+        self.validate_identity(
+            "SELECT /*+ ORDERED */* FROM tbl", "SELECT /*+ ORDERED */ * FROM tbl"
+        )
+        self.validate_identity(
+            "SELECT /* test */ /*+ ORDERED */* FROM tbl",
+            "/* test */ SELECT /*+ ORDERED */ * FROM tbl",
+        )
+        self.validate_identity(
+            "SELECT /*+ ORDERED */*/* test */ FROM tbl",
+            "SELECT /*+ ORDERED */ * /* test */ FROM tbl",
+        )
 
     def test_join_marker(self):
         self.validate_identity("SELECT e1.x, e2.x FROM e e1, e e2 WHERE e1.y (+) = e2.y")
diff --git a/tests/dialects/test_postgres.py b/tests/dialects/test_postgres.py
index ffe08c67eb..8f84d9ff59 100644
--- a/tests/dialects/test_postgres.py
+++ b/tests/dialects/test_postgres.py
@@ -127,10 +127,6 @@ def test_postgres(self):
             "pg_catalog.PG_TABLE_IS_VISIBLE(c.oid) "
             "ORDER BY 2, 3"
         )
-        self.validate_identity(
-            "/*+ some comment*/ SELECT b.foo, b.bar FROM baz AS b",
-            "/* + some comment */ SELECT b.foo, b.bar FROM baz AS b",
-        )
         self.validate_identity(
             "SELECT ARRAY[1, 2, 3] <@ ARRAY[1, 2]",
             "SELECT ARRAY[1, 2] @> ARRAY[1, 2, 3]",
@@ -819,6 +815,11 @@ def test_postgres(self):
             },
         )
 
+        self.validate_identity(
+            "/*+ some comment*/ SELECT b.foo, b.bar FROM baz AS b",
+            "/* + some comment */ SELECT b.foo, b.bar FROM baz AS b",
+        )
+
     def test_ddl(self):
         # Checks that user-defined types are parsed into DataType instead of Identifier
         self.parse_one("CREATE TABLE t (a udt)").this.expressions[0].args["kind"].assert_is(
diff --git a/tests/dialects/test_risingwave.py b/tests/dialects/test_risingwave.py
index f645d570b8..80c5265cb6 100644
--- a/tests/dialects/test_risingwave.py
+++ b/tests/dialects/test_risingwave.py
@@ -13,7 +13,7 @@ def test_risingwave(self):
             },
         )
         self.validate_identity(
-            "CREATE SOURCE from_kafka (*, gen_i32_field INT AS int32_field + 2, gen_i64_field INT AS int64_field + 2, WATERMARK FOR time_col AS time_col - INTERVAL '5' SECOND) INCLUDE header foo varchar AS myheader INCLUDE key AS mykey WITH (connector='kafka', topic='my_topic') FORMAT PLAIN ENCODE PROTOBUF (A=1, B=2) KEY ENCODE PROTOBUF (A=3, B=4)"
+            "CREATE SOURCE from_kafka (*, gen_i32_field INT AS int32_field + 2, gen_i64_field INT AS int64_field + 2, WATERMARK FOR time_col AS time_col - INTERVAL '5 SECOND') INCLUDE header foo VARCHAR AS myheader INCLUDE key AS mykey WITH (connector='kafka', topic='my_topic') FORMAT PLAIN ENCODE PROTOBUF (A=1, B=2) KEY ENCODE PROTOBUF (A=3, B=4)"
         )
         self.validate_identity(
             "CREATE SINK my_sink AS SELECT * FROM A WITH (connector='kafka', topic='my_topic') FORMAT PLAIN ENCODE PROTOBUF (A=1, B=2) KEY ENCODE PROTOBUF (A=3, B=4)"
@@ -21,6 +21,3 @@ def test_risingwave(self):
         self.validate_identity(
             "WITH t1 AS MATERIALIZED (SELECT 1), t2 AS NOT MATERIALIZED (SELECT 2) SELECT * FROM t1, t2"
         )
-        self.validate_identity(
-            """LAST_VALUE("col1") OVER (ORDER BY "col2" RANGE BETWEEN INTERVAL '1 DAY' PRECEDING AND '1 month' FOLLOWING)"""
-        )