diff --git a/pyproject.toml b/pyproject.toml
index ca7527868d..42ac250e5b 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -23,7 +23,7 @@ dependencies = [
     "requests",
     "rich[jupyter]",
     "ruamel.yaml",
-    "sqlglot~=30.8.0",
+    "sqlglot>=30.14.0,<31.0.0",
     "tenacity",
     "time-machine",
     "json-stream"
diff --git a/sqlmesh/core/audit/definition.py b/sqlmesh/core/audit/definition.py
index 4c90151ee4..43acf8b36d 100644
--- a/sqlmesh/core/audit/definition.py
+++ b/sqlmesh/core/audit/definition.py
@@ -7,14 +7,13 @@
 
 from pydantic import Field
 from sqlglot import exp
-from sqlglot.optimizer.simplify import gen
-
 from sqlmesh.core import dialect as d
 from sqlmesh.core.macros import MacroRegistry, macro
 from sqlmesh.core.model.common import (
     bool_validator,
     default_catalog_validator,
     depends_on_validator,
+    gen_for_jinja,
     sort_python_env,
     sorted_python_env_payloads,
 )
@@ -452,8 +451,8 @@ def load_audit(
     extra_kwargs: t.Dict[str, t.Any] = {}
     if is_standalone:
         jinja_macro_refrences, referenced_variables = extract_macro_references_and_variables(
-            *(gen(s) for s in statements),
-            gen(query),
+            *(gen_for_jinja(s) for s in statements),
+            gen_for_jinja(query),
         )
         jinja_macros = (jinja_macros or JinjaMacroRegistry()).trim(jinja_macro_refrences)
         for jinja_macro in jinja_macros.root_macros.values():
diff --git a/sqlmesh/core/dialect.py b/sqlmesh/core/dialect.py
index e4ab522198..a5b89f6c33 100644
--- a/sqlmesh/core/dialect.py
+++ b/sqlmesh/core/dialect.py
@@ -315,16 +315,26 @@ def _parse_with(self: Parser, skip_with_token: bool = False) -> t.Optional[exp.E
 
 
 def _parse_join(
-    self: Parser, skip_join_token: bool = False, parse_bracket: bool = False
+    self: Parser,
+    skip_join_token: bool = False,
+    parse_bracket: bool = False,
+    alias_tokens: t.Optional[t.Collection[TokenType]] = None,
 ) -> t.Optional[exp.Expr]:
     index = self._index
     method, side, kind = self._parse_join_parts()
     macro = _parse_matching_macro(self, "JOIN")
     if not macro:
         self._retreat(index)
-        return self.__parse_join(skip_join_token=skip_join_token, parse_bracket=parse_bracket)  # type: ignore
+        parse_join = self.__parse_join  # type: ignore
+        if "alias_tokens" in parse_join.__code__.co_varnames:
+            return parse_join(
+                skip_join_token=skip_join_token,
+                parse_bracket=parse_bracket,
+                alias_tokens=alias_tokens,
+            )
+        return parse_join(skip_join_token=skip_join_token, parse_bracket=parse_bracket)
 
-    join = self.__parse_join(skip_join_token=True)  # type: ignore
+    join = self.__parse_join(skip_join_token=True, alias_tokens=alias_tokens)  # type: ignore
     if method:
         join.set("method", method.text)
     if side:
@@ -575,7 +585,11 @@ def altercolumn_sql(self: Generator, expression: exp.AlterColumn) -> str:
     # sqlglot's generator returns as soon as it renders the type, so the nullability parsed
     # above has to be appended here
     allow_null = expression.args.get("allow_null")
-    if expression.args.get("dtype") and allow_null is not None:
+    if (
+        expression.args.get("dtype")
+        and allow_null is not None
+        and not hasattr(self, "_alter_column_null_constraint_sql")
+    ):
         sql = f"{sql} NULL" if allow_null else f"{sql} NOT NULL"
 
     return sql
@@ -800,8 +814,14 @@ def _whens_sql(self: Generator, expression: exp.Whens) -> str:
     return self.wrap(self.expressions(expression, sep=" ", indent=False))
 
 
-def _parse_interval_span(self: Parser, this: exp.Expr) -> exp.Interval:
-    interval = self.__parse_interval_span(this)  # type: ignore
+def _parse_interval_span(
+    self: Parser, this: exp.Expr, parse_function_unit: bool = True
+) -> exp.Interval:
+    parse_interval_span = self.__parse_interval_span  # type: ignore
+    if "parse_function_unit" in parse_interval_span.__code__.co_varnames:
+        interval = parse_interval_span(this, parse_function_unit=parse_function_unit)
+    else:
+        interval = parse_interval_span(this)
     # Without this, @unit in `INTERVAL @value @unit` is misread as an alias.
     if not interval.args.get("unit") and self._match(TokenType.PARAMETER):
         macro = _parse_macro(self)
@@ -1235,6 +1255,10 @@ def extend_sqlglot() -> None:
     # DuckDB's prefix absolute power operator `@` clashes with the macro syntax
     DuckDB.Parser.NO_PAREN_FUNCTION_PARSERS.pop("@", None)
 
+    # SQLGlot 30.17 stopped marking DuckDB divisions as safe. Keep the behavior
+    # used by earlier SQLGlot releases, where division by zero returns NULL.
+    DuckDB.SAFE_DIVISION = True
+
 
 def select_from_values(
     values: t.List[t.Tuple[t.Any, ...]],
diff --git a/sqlmesh/core/engine_adapter/snowflake.py b/sqlmesh/core/engine_adapter/snowflake.py
index d589b5d15b..a82fbfafeb 100644
--- a/sqlmesh/core/engine_adapter/snowflake.py
+++ b/sqlmesh/core/engine_adapter/snowflake.py
@@ -86,6 +86,15 @@ class SnowflakeEngineAdapter(
     CURRENT_USER_OR_ROLE_EXPRESSION: exp.Expr = exp.func("CURRENT_ROLE")
     USE_CATALOG_IN_GRANTS = True
 
+    def _build_clustered_by_exp(
+        self,
+        clustered_by: t.List[exp.Expr],
+        **kwargs: t.Any,
+    ) -> t.Optional[exp.Cluster]:
+        # Snowflake requires the clustering key to be parenthesized. SQLGlot 30.17
+        # no longer adds parentheses when Cluster contains bare expressions.
+        return exp.Cluster(expressions=[exp.Tuple(expressions=[c.copy() for c in clustered_by])])
+
     @contextlib.contextmanager
     def session(self, properties: SessionProperties) -> t.Iterator[None]:
         warehouse = properties.get("warehouse")
diff --git a/sqlmesh/core/engine_adapter/starrocks.py b/sqlmesh/core/engine_adapter/starrocks.py
index 05120db0e3..fb4f08daa5 100644
--- a/sqlmesh/core/engine_adapter/starrocks.py
+++ b/sqlmesh/core/engine_adapter/starrocks.py
@@ -3227,7 +3227,7 @@ def _build_order_by_property(
         self,
         table_properties: t.Dict[str, t.Any],
         clustered_by: t.Optional[t.List[exp.Expr]],
-    ) -> t.Optional[exp.Cluster]:
+    ) -> t.Optional[exp.Order]:
         """
         Build ORDER BY (clustering) property.
 
@@ -3264,7 +3264,7 @@ def _build_order_by_property(
                 clustered_by = list(normalized)
 
         if clustered_by:
-            result = exp.Cluster(expressions=clustered_by)
+            result = exp.Order(expressions=[exp.Tuple(expressions=clustered_by)])
             return result
         else:  # noqa: RET505
             return None
diff --git a/sqlmesh/core/loader.py b/sqlmesh/core/loader.py
index cb951b4f9e..77b181201f 100644
--- a/sqlmesh/core/loader.py
+++ b/sqlmesh/core/loader.py
@@ -683,6 +683,8 @@ def _load_python_models(
                             virtual_environment_mode=self.config.virtual_environment_mode,
                         ):
                             if model.enabled:
+                                if model.fqn in models:
+                                    raise ValueError(f"Duplicate name: '{model.fqn}'.")
                                 models[model.fqn] = model
                 except Exception as ex:
                     raise ConfigError(self._failed_to_load_model_error(path, ex), path)
diff --git a/sqlmesh/core/model/cache.py b/sqlmesh/core/model/cache.py
index 1f038c5d79..2b9ab2013a 100644
--- a/sqlmesh/core/model/cache.py
+++ b/sqlmesh/core/model/cache.py
@@ -10,6 +10,7 @@
 from sqlglot.schema import MappingSchema
 
 from sqlmesh.core import constants as c
+from sqlmesh.core.model.common import gen_for_jinja
 from sqlmesh.core.model.definition import ExternalModel, Model, SqlModel, _Model
 from sqlmesh.utils.cache import FileCache
 from sqlmesh.utils.hashing import crc32
@@ -152,7 +153,7 @@ def _put(self, name: str, model: SqlModel) -> None:
     @staticmethod
     def _entry_name(model: SqlModel) -> str:
         hash_data = _mapping_schema_hash_data(model.mapping_schema)
-        hash_data.append(gen(model.query, comments=True))
+        hash_data.append(gen_for_jinja(model.query, comments=True))
         hash_data.append(str([gen(d) for d in model.macro_definitions]))
         hash_data.append(str([(k, v) for k, v in model.sorted_python_env]))
         hash_data.extend(model.jinja_macros.data_hash_values)
diff --git a/sqlmesh/core/model/common.py b/sqlmesh/core/model/common.py
index f03cf49753..70f910a92a 100644
--- a/sqlmesh/core/model/common.py
+++ b/sqlmesh/core/model/common.py
@@ -7,6 +7,7 @@
 from difflib import get_close_matches
 from sqlglot import exp
 from sqlglot.helper import ensure_list
+from sqlglot.optimizer.simplify import gen
 
 from sqlmesh.core import constants as c
 from sqlmesh.core import dialect as d
@@ -36,6 +37,13 @@
     MacroCallable = t.Union[Executable, registry_decorator]
 
 
+def gen_for_jinja(expression: exp.Expr, **kwargs: t.Any) -> str:
+    """Generate source text without escaping quotes inside Jinja expressions."""
+    if isinstance(expression, d.Jinja):
+        return expression.this.this
+    return gen(expression, **kwargs)
+
+
 def make_python_env(
     expressions: t.Union[
         exp.Expr,
diff --git a/sqlmesh/core/model/definition.py b/sqlmesh/core/model/definition.py
index c3569f9bd3..fe56c1e050 100644
--- a/sqlmesh/core/model/definition.py
+++ b/sqlmesh/core/model/definition.py
@@ -24,6 +24,7 @@
 from sqlmesh.core.macros import MacroRegistry, macro
 from sqlmesh.core.model.common import (
     ParsableSql,
+    gen_for_jinja,
     make_python_env,
     parse_dependencies,
     parse_strings_with_macro_refs,
@@ -835,7 +836,10 @@ def set_time_format(self, default_time_format: str = c.DEFAULT_TIME_COLUMN_FORMA
             # Transpile the time column format into the generic dialect
             formatted_time = format_time(
                 self.time_column.format,
-                d.Dialect.get_or_raise(self.dialect).TIME_MAPPING,
+                {
+                    token: value.removesuffix("strict")
+                    for token, value in d.Dialect.get_or_raise(self.dialect).TIME_MAPPING.items()
+                },
             )
             assert formatted_time is not None
             self.time_column.format = formatted_time
@@ -2646,7 +2650,7 @@ def _create_model(
         statements.append(kwargs["kind"].merge_filter)
 
     jinja_macro_references, referenced_variables = extract_macro_references_and_variables(
-        *(gen(e if isinstance(e, exp.Expr) else e[0]) for e in statements)
+        *(gen_for_jinja(e if isinstance(e, exp.Expr) else e[0]) for e in statements)
     )
 
     if jinja_macros:
diff --git a/tests/core/test_dialect.py b/tests/core/test_dialect.py
index 142b40b31f..351196927e 100644
--- a/tests/core/test_dialect.py
+++ b/tests/core/test_dialect.py
@@ -1027,6 +1027,20 @@ def test_tsql_alter_column_nullability():
     )
 
 
+def test_sqlglot_parser_signature_compatibility():
+    # SQLGlot passes UPDATE-specific alias tokens to _parse_join. The SQLMesh
+    # override must preserve those tokens when delegating to SQLGlot.
+    assert (
+        parse_one("UPDATE target JOIN source SET target.x = source.x", read="tsql").sql(
+            dialect="tsql"
+        )
+        == "UPDATE target, source SET target.x = source.x"
+    )
+
+    # SQLGlot 30.17 passes parse_function_unit to _parse_interval_span.
+    assert parse_one("INTERVAL '1' DAY").sql() == "INTERVAL '1' DAY"
+
+
 def test_model_name_cannot_be_string():
     with pytest.raises(ParseError) as parse_error:
         parse(
@@ -1231,3 +1245,12 @@ def test_extend_sqlglot_is_idempotent():
     assert parse_one("SELECT CAST(1 AS INT)").sql() == "SELECT CAST(1 AS INT)"
     # The class-level registries must not grow on repeated calls.
     assert Generator.UNWRAPPED_INTERVAL_VALUES == before
+
+
+def test_extend_sqlglot_supports_collated_types():
+    # SQLGlot calls Parser._parse_types with this keyword while parsing
+    # collated types. Keep the SQLMesh override compatible with that API.
+    assert (
+        parse_one("SELECT CAST('x' AS VARCHAR COLLATE utf8)", "spark").sql("spark")
+        == "SELECT CAST('x' AS STRING COLLATE utf8)"
+    )
diff --git a/tests/core/test_model.py b/tests/core/test_model.py
index 1f3cde265b..1a6d3db8d3 100644
--- a/tests/core/test_model.py
+++ b/tests/core/test_model.py
@@ -2945,6 +2945,26 @@ def test_parse(assert_exp_eq):
     )
 
 
+def test_parse_jinja_query_with_quoted_macro_argument():
+    expressions = d.parse(
+        """
+        MODEL (
+            name sushi.items,
+            kind FULL,
+        );
+
+        JINJA_QUERY_BEGIN;
+        SELECT {{ alias(identity(x), 'flag') }};
+        JINJA_END;
+        """
+    )
+
+    model = load_sql_based_model(expressions)
+
+    assert isinstance(model.query, d.JinjaQuery)
+    assert "{{ alias(identity(x), 'flag') }}" in model.query.this.this
+
+
 def test_dialect_pattern():
     def make_test_sql(text: str) -> str:
         return f"""