Merge branch 'main' into fix/local-only-format

Author: j-hartshorn

docs/reference/model_configuration.md

@@ -178,6 +178,7 @@ The SQLMesh project-level `model_defaults` key supports the following options, d
 - kind
 - dialect
 - cron
+- cron_tz
 - owner
 - start
 - table_format

sqlmesh/core/config/model.py

@@ -14,7 +14,7 @@
     OnAdditiveChange,
 )
 from sqlmesh.core.model.meta import FunctionCall
-from sqlmesh.core.node import IntervalUnit
+from sqlmesh.core.node import IntervalUnit, cron_tz_validator
 from sqlmesh.utils.date import TimeLike
 from sqlmesh.utils.pydantic import field_validator
 
@@ -27,6 +27,7 @@ class ModelDefaultsConfig(BaseConfig):
         dialect: The SQL dialect that the model's query is written in.
         cron: A cron string specifying how often the model should be refreshed, leveraging the
             [croniter](https://github.com/kiorky/croniter) library.
+        cron_tz: The timezone for the cron expression, defaults to UTC. [IANA time zones](https://docs.python.org/3/library/zoneinfo.html).
         owner: The owner of the model.
         start: The earliest date that the model will be backfilled for. If this is None,
             then the date is inferred by taking the most recent start date of its ancestors.
@@ -55,6 +56,7 @@ class ModelDefaultsConfig(BaseConfig):
     kind: t.Optional[ModelKind] = None
     dialect: t.Optional[str] = None
     cron: t.Optional[str] = None
+    cron_tz: t.Any = None
     owner: t.Optional[str] = None
     start: t.Optional[TimeLike] = None
     table_format: t.Optional[str] = None
@@ -78,6 +80,7 @@ class ModelDefaultsConfig(BaseConfig):
     _model_kind_validator = model_kind_validator
     _on_destructive_change_validator = on_destructive_change_validator
     _on_additive_change_validator = on_additive_change_validator
+    _cron_tz_validator = cron_tz_validator
 
     @field_validator("audits", mode="before")
     def _audits_validator(cls, v: t.Any) -> t.Any:

sqlmesh/core/engine_adapter/clickhouse.py

@@ -716,8 +716,12 @@ def use_server_nulls_for_unmatched_after_join(
         return query
 
     def _build_settings_property(
-        self, key: str, value: exp.Expr | str | int | float
+        self, settings: t.Mapping[str, exp.Expr | str | int | float]
     ) -> exp.SettingsProperty:
+        # ClickHouse requires every key=value pair to live under a single
+        # SETTINGS clause (`SETTINGS a = 1, b = 2`). Emitting one
+        # SettingsProperty per pair produces repeated SETTINGS keywords and a
+        # syntax error at execution time.
         return exp.SettingsProperty(
             expressions=[
                 exp.EQ(
@@ -726,6 +730,7 @@ def _build_settings_property(
                     if isinstance(value, exp.Expr)
                     else exp.Literal(this=value, is_string=isinstance(value, str)),
                 )
+                for key, value in settings.items()
             ]
         )
 
@@ -827,9 +832,7 @@ def _build_table_properties_exp(
             properties.append(exp.EmptyProperty())
 
         if table_properties_copy:
-            properties.extend(
-                [self._build_settings_property(k, v) for k, v in table_properties_copy.items()]
-            )
+            properties.append(self._build_settings_property(table_properties_copy))
 
         if table_description:
             properties.append(
@@ -858,9 +861,7 @@ def _build_view_properties_exp(
             properties.append(exp.OnCluster(this=exp.to_identifier(self.cluster)))
 
         if view_properties_copy:
-            properties.extend(
-                [self._build_settings_property(k, v) for k, v in view_properties_copy.items()]
-            )
+            properties.append(self._build_settings_property(view_properties_copy))
 
         if table_description:
             properties.append(

sqlmesh/core/engine_adapter/redshift.py

@@ -4,6 +4,7 @@
 import typing as t
 
 from sqlglot import exp
+from sqlglot.helper import ensure_list
 
 from sqlmesh.core.dialect import to_schema
 from sqlmesh.core.engine_adapter.base import MERGE_SOURCE_ALIAS, MERGE_TARGET_ALIAS
@@ -30,6 +31,7 @@
 
     from sqlmesh.core._typing import SchemaName, TableName
     from sqlmesh.core.engine_adapter.base import QueryOrDF, Query
+    from sqlmesh.core.node import IntervalUnit
 
 logger = logging.getLogger(__name__)
 
@@ -249,6 +251,63 @@ def create_view(
             **create_kwargs,
         )
 
+    def _build_table_properties_exp(
+        self,
+        catalog_name: t.Optional[str] = None,
+        table_format: t.Optional[str] = None,
+        storage_format: t.Optional[str] = None,
+        partitioned_by: t.Optional[t.List[exp.Expr]] = None,
+        partition_interval_unit: t.Optional[IntervalUnit] = None,
+        clustered_by: t.Optional[t.List[exp.Expr]] = None,
+        table_properties: t.Optional[t.Dict[str, exp.Expr]] = None,
+        target_columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None,
+        table_description: t.Optional[str] = None,
+        table_kind: t.Optional[str] = None,
+        **kwargs: t.Any,
+    ) -> t.Optional[exp.Properties]:
+        properties: t.List[exp.Expr] = []
+
+        if table_description:
+            properties.append(
+                exp.SchemaCommentProperty(
+                    this=exp.Literal.string(self._truncate_table_comment(table_description))
+                )
+            )
+
+        def _to_identifier_if_string(expression: exp.Expr) -> exp.Expr:
+            if isinstance(expression, exp.Literal) and expression.is_string:
+                return exp.to_identifier(expression.this)
+            return expression.copy()
+
+        if table_properties:
+            table_properties = {k.upper(): v for k, v in table_properties.items()}
+
+            table_type = self._pop_creatable_type_from_properties(table_properties)
+            properties.extend(ensure_list(table_type))
+
+            diststyle = table_properties.get("DISTSTYLE")
+            if diststyle:
+                properties.append(exp.DistStyleProperty(this=exp.var(diststyle.name.upper())))
+
+            distkey = table_properties.get("DISTKEY")
+            if distkey:
+                properties.append(exp.DistKeyProperty(this=_to_identifier_if_string(distkey)))
+
+            sortkey = table_properties.get("SORTKEY")
+            if sortkey:
+                sortkey_expressions = sortkey.expressions if sortkey.expressions else [sortkey]
+                properties.append(
+                    exp.SortKeyProperty(
+                        this=[
+                            _to_identifier_if_string(expression)
+                            for expression in sortkey_expressions
+                        ],
+                        compound=False,
+                    )
+                )
+
+        return exp.Properties(expressions=properties) if properties else None
+
     def replace_query(
         self,
         table_name: TableName,

sqlmesh/core/node.py

@@ -260,6 +260,30 @@ def dbt_fqn(self) -> t.Optional[str]:
 }
 
 
+def _cron_tz_validator(cls: t.Type, v: t.Any) -> t.Optional[zoneinfo.ZoneInfo]:
+    if not v or v == "UTC":
+        return None
+
+    v = str_or_exp_to_str(v)
+
+    try:
+        return zoneinfo.ZoneInfo(v)
+    except Exception as e:
+        available_timezones = zoneinfo.available_timezones()
+
+        if available_timezones:
+            raise ConfigError(f"{e}. {v} must be in {available_timezones}.")
+        else:
+            raise ConfigError(
+                f"{e}. IANA time zone data is not available on your system. `pip install tzdata` to leverage cron time zones or remove this field which will default to UTC."
+            )
+
+    return None
+
+
+cron_tz_validator = field_validator("cron_tz", mode="before")(_cron_tz_validator)
+
+
 class _Node(DbtInfoMixin, PydanticModel):
     """
     Node is the core abstraction for entity that can be executed within the scheduler.
@@ -302,6 +326,8 @@ class _Node(DbtInfoMixin, PydanticModel):
     _croniter: t.Optional[CroniterCache] = None
     __inferred_interval_unit: t.Optional[IntervalUnit] = None
 
+    _cron_tz_validator = cron_tz_validator
+
     def __str__(self) -> str:
         path = f": {self._path.name}" if self._path else ""
         return f"{self.__class__.__name__}<{self.name}{path}>"
@@ -328,27 +354,6 @@ def _name_validator(cls, v: t.Any) -> t.Optional[str]:
             return v.meta["sql"]
         return str(v)
 
-    @field_validator("cron_tz", mode="before")
-    def _cron_tz_validator(cls, v: t.Any) -> t.Optional[zoneinfo.ZoneInfo]:
-        if not v or v == "UTC":
-            return None
-
-        v = str_or_exp_to_str(v)
-
-        try:
-            return zoneinfo.ZoneInfo(v)
-        except Exception as e:
-            available_timezones = zoneinfo.available_timezones()
-
-            if available_timezones:
-                raise ConfigError(f"{e}. {v} must be in {available_timezones}.")
-            else:
-                raise ConfigError(
-                    f"{e}. IANA time zone data is not available on your system. `pip install tzdata` to leverage cron time zones or remove this field which will default to UTC."
-                )
-
-        return None
-
     @field_validator("start", "end", mode="before")
     @classmethod
     def _date_validator(cls, v: t.Any) -> t.Optional[TimeLike]:

sqlmesh/core/test/definition.py

@@ -308,7 +308,6 @@ def _to_hashable(x: t.Any) -> t.Any:
                 expected,
                 actual,
                 check_dtype=False,
-                check_datetimelike_compat=True,
                 check_like=True,  # Ignore column order
             )
         except AssertionError as e:

tests/core/engine_adapter/test_clickhouse.py

@@ -316,13 +316,32 @@ def build_properties_sql(storage_format="", order_by="", primary_key="", propert
         == "ENGINE=MergeTree ORDER BY (a, b + 1) PRIMARY KEY (a, b)"
     )
 
+    # Multiple physical_properties must be combined into a single comma-separated
+    # SETTINGS clause. ClickHouse rejects repeated SETTINGS keywords with a syntax
+    # error (see https://github.com/SQLMesh/sqlmesh/issues/5803).
     assert (
         build_properties_sql(
             order_by="ORDER_BY = (a, b + 1),",
             primary_key="PRIMARY_KEY = (a, b),",
             properties="PROP1 = 1, PROP2 = '2'",
         )
-        == "ENGINE=MergeTree ORDER BY (a, b + 1) PRIMARY KEY (a, b) SETTINGS prop1 = 1 SETTINGS prop2 = '2'"
+        == "ENGINE=MergeTree ORDER BY (a, b + 1) PRIMARY KEY (a, b) SETTINGS prop1 = 1, prop2 = '2'"
+    )
+
+    # Regression test for #5803: three or more SETTINGS entries also combine.
+    assert (
+        build_properties_sql(
+            order_by="ORDER_BY = (orders_id),",
+            properties=(
+                "min_age_to_force_merge_seconds = 3600, "
+                "min_age_to_force_merge_on_partition_only = 1, "
+                "index_granularity = 8192"
+            ),
+        )
+        == "ENGINE=MergeTree ORDER BY (orders_id) "
+        "SETTINGS min_age_to_force_merge_seconds = 3600, "
+        "min_age_to_force_merge_on_partition_only = 1, "
+        "index_granularity = 8192"
     )
 
     assert (
@@ -345,6 +364,20 @@ def build_properties_sql(storage_format="", order_by="", primary_key="", propert
     )
 
 
+def test_view_properties_combine_settings(adapter: ClickhouseEngineAdapter):
+    # View properties hit the same SettingsProperty code path as table
+    # properties (#5803): multiple entries must collapse into one SETTINGS
+    # clause rather than emit repeated SETTINGS keywords.
+    view_properties_exp = adapter._build_view_properties_exp(
+        view_properties={
+            "prop1": exp.Literal.number(1),
+            "prop2": exp.Literal.string("2"),
+        }
+    )
+    assert view_properties_exp is not None
+    assert view_properties_exp.sql("clickhouse") == "SETTINGS prop1 = 1, prop2 = '2'"
+
+
 def test_partitioned_by_expr(make_mocked_engine_adapter: t.Callable):
     # user doesn't specify, unknown time column type
     model = load_sql_based_model(