Merge branch 'main' into mday-io/feature/janitor-environment-flag

Author: mday-io

docs/integrations/engines/duckdb.md

@@ -79,6 +79,7 @@ SQLMesh will place models with the explicit catalog "ephemeral", such as `epheme
               type: ducklake
               path: 'catalog.ducklake'
               data_path: data/ducklake
+              override_data_path: true
               encrypted: True
               data_inlining_row_limit: 10
               metadata_schema: main
@@ -105,6 +106,7 @@ SQLMesh will place models with the explicit catalog "ephemeral", such as `epheme
                             type="ducklake",
                             path="catalog.ducklake",
                             data_path="data/ducklake",
+                            override_data_path=False,
                             encrypted=True,
                             data_inlining_row_limit=10,
                             metadata_schema="main",
@@ -120,6 +122,7 @@ SQLMesh will place models with the explicit catalog "ephemeral", such as `epheme
 
 - `path`: Path to the DuckLake catalog file
 - `data_path`: Path where DuckLake data files are stored
+- `override_data_path`: Whether data_override_path option is set
 - `encrypted`: Whether to enable encryption for the catalog (default: `False`)
 - `data_inlining_row_limit`: Maximum number of rows to inline in the catalog (default: `0`)
 - `metadata_schema`: The schema in the catalog server in which to store the DuckLake metadata tables (default: `main`)
@@ -364,6 +367,7 @@ The `filesystems` accepts a list of file systems to register in the DuckDB conne
               type: ducklake
               path: myducklakecatalog.duckdb
               data_path: abfs://MyFabricWorkspace/MyFabricLakehouse.Lakehouse/Files/DuckLake.Files
+              override_data_path: False
         extensions:
           - ducklake
         filesystems:

sqlmesh/core/config/connection.py

@@ -238,6 +238,7 @@ class DuckDBAttachOptions(BaseConfig):
 
     # DuckLake specific options
     data_path: t.Optional[str] = None
+    override_data_path: t.Optional[bool] = False
     encrypted: bool = False
     data_inlining_row_limit: t.Optional[int] = None
     metadata_schema: t.Optional[str] = None
@@ -258,6 +259,8 @@ def to_sql(self, alias: str) -> str:
                 path = f"ducklake:{path}"
             if self.data_path is not None:
                 options.append(f"DATA_PATH '{self.data_path}'")
+                if self.override_data_path:
+                    options.append("OVERRIDE_DATA_PATH true")
             if self.encrypted:
                 options.append("ENCRYPTED")
             if self.data_inlining_row_limit is not None:
@@ -2097,6 +2100,7 @@ class ClickhouseConnectionConfig(ConnectionConfig):
     https_proxy: t.Optional[str] = None
     server_host_name: t.Optional[str] = None
     tls_mode: t.Optional[str] = None
+    secure: bool = False
 
     concurrent_tasks: int = 1
     register_comments: bool = True
@@ -2133,6 +2137,7 @@ def _connection_kwargs_keys(self) -> t.Set[str]:
             "https_proxy",
             "server_host_name",
             "tls_mode",
+            "secure",
         }
         return kwargs
 

sqlmesh/core/dialect.py

@@ -774,7 +774,8 @@ def format_model_expressions(
     if rewrite_casts:
 
         def cast_to_colon(node: exp.Expr) -> exp.Expr:
-            if isinstance(node, exp.Cast) and not any(
+            # Directly check type instead of isinstance to avoid rewriting subclasses of CAST, e.g. JSONCast
+            if type(node) is exp.Cast and not any(
                 # Only convert CAST into :: if it doesn't have additional args set, otherwise this
                 # conversion could alter the semantics (eg. changing SAFE_CAST in BigQuery to CAST)
                 arg

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/databricks.py

@@ -411,3 +411,27 @@ def _build_column_defs(
         return super()._build_column_defs(
             target_columns_to_types, column_descriptions, is_view, materialized
         )
+
+    def columns(
+        self, table_name: TableName, include_pseudo_columns: bool = False
+    ) -> t.Dict[str, exp.DataType]:
+        table = exp.to_table(table_name)
+
+        column_catalog = table.catalog or self.get_current_catalog()
+        query = (
+            exp.select("columns.column_name", "columns.full_data_type")
+            .from_("system.information_schema.columns")
+            .where(
+                exp.and_(
+                    exp.column("table_name").eq(table.name),
+                    exp.column("table_schema").eq(table.db),
+                    exp.column("table_catalog").eq(column_catalog),
+                )
+            )
+            .order_by("ordinal_position ASC")
+        )
+
+        self.cursor.execute(query)
+        result = self.cursor.fetchall()
+
+        return {row[0]: exp.DataType.build(row[1], dialect=self.dialect) for row in result}

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,

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(

tests/core/engine_adapter/test_databricks.py

@@ -526,3 +526,61 @@ def test_drop_data_object_materialized_view_calls_correct_drop(mocker: MockFixtu
     drop_view_mock.assert_called_once_with(
         mv_data_object.to_table(), ignore_if_not_exists=True, materialized=True
     )
+
+
+def test_columns(mocker: MockFixture, make_mocked_engine_adapter: t.Callable):
+    adapter = make_mocked_engine_adapter(DatabricksEngineAdapter, default_catalog="test_catalog")
+
+    # Override/mock get_current_catalog to return default
+    current_catalog_mock = mocker.patch.object(
+        adapter, "get_current_catalog", return_value="test_catalog"
+    )
+    # create long struct columns datatype
+    long_struct_cols = [f"a_{i}:int" for i in range(50)]
+    adapter.cursor.fetchall.return_value = [
+        ("bigint_col", "bigint"),
+        ("binary_col", "binary"),
+        ("boolean_col", "boolean"),
+        ("date_col", "date"),
+        ("decimal_col", "decimal(38,4)"),
+        ("double_col", "double"),
+        ("float_col", "float"),
+        ("int_col", "int"),
+        ("small_int", "smallint"),
+        ("string_col", "string"),
+        ("timestamp_col", "timestamp"),
+        ("timestamp_ntz_col", "timestamp_ntz"),
+        ("tinyint_col", "tinyint"),
+        ("array_col", "array<int>"),
+        ("simple_struct_col", "struct<a:int,b:string>"),
+        ("long_struct_col", f"struct<{','.join(long_struct_cols)}>"),
+    ]
+
+    resp = adapter.columns("test_db.test_table")
+    assert resp == {
+        "bigint_col": exp.DataType.build("bigint", dialect=adapter.dialect),
+        "binary_col": exp.DataType.build("binary", dialect=adapter.dialect),
+        "boolean_col": exp.DataType.build("boolean", dialect=adapter.dialect),
+        "date_col": exp.DataType.build("date", dialect=adapter.dialect),
+        "decimal_col": exp.DataType.build("decimal(38,4)", dialect=adapter.dialect),
+        "double_col": exp.DataType.build("double", dialect=adapter.dialect),
+        "float_col": exp.DataType.build("float", dialect=adapter.dialect),
+        "int_col": exp.DataType.build("int", dialect=adapter.dialect),
+        "small_int": exp.DataType.build("smallint", dialect=adapter.dialect),
+        "string_col": exp.DataType.build("string", dialect=adapter.dialect),
+        "timestamp_col": exp.DataType.build("timestamp", dialect=adapter.dialect),
+        "timestamp_ntz_col": exp.DataType.build("timestamp_ntz", dialect=adapter.dialect),
+        "tinyint_col": exp.DataType.build("tinyint", dialect=adapter.dialect),
+        "array_col": exp.DataType.build("array<int>", dialect=adapter.dialect),
+        "simple_struct_col": exp.DataType.build("struct<a:int,b:string>", dialect=adapter.dialect),
+        "long_struct_col": exp.DataType.build(
+            f"struct<{','.join(long_struct_cols)}>", dialect=adapter.dialect
+        ),
+    }
+
+    adapter.cursor.execute.assert_called_once_with(
+        parse_one(
+            """SELECT columns.column_name, columns.full_data_type FROM system.information_schema.columns WHERE table_name = 'test_table' AND table_schema = 'test_db' AND table_catalog = 'test_catalog' ORDER BY ordinal_position ASC""",
+            dialect="databricks",
+        )
+    )