Fix: Match dbt handling of cluster_by config for view materializations

sqlmesh/dbt/model.py

@@ -2,6 +2,7 @@
 
 import datetime
 import typing as t
+import logging
 
 from sqlglot import exp
 from sqlglot.errors import SqlglotError
@@ -34,6 +35,8 @@
     from sqlmesh.core.audit.definition import ModelAudit
     from sqlmesh.dbt.context import DbtContext
 
+logger = logging.getLogger(__name__)
+
 
 INCREMENTAL_BY_TIME_STRATEGIES = set(["delete+insert", "insert_overwrite", "microbatch"])
 INCREMENTAL_BY_UNIQUE_KEY_STRATEGIES = set(["merge"])
@@ -503,6 +506,7 @@ def to_sqlmesh(
         """Converts the dbt model into a SQLMesh model."""
         model_dialect = self.dialect(context)
         query = d.jinja_query(self.sql_no_config)
+        kind = self.model_kind(context)
 
         optional_kwargs: t.Dict[str, t.Any] = {}
         physical_properties: t.Dict[str, t.Any] = {}
@@ -522,15 +526,20 @@ def to_sqlmesh(
             optional_kwargs["partitioned_by"] = partitioned_by
 
         if self.cluster_by:
-            clustered_by = []
-            for c in self.cluster_by:
-                try:
-                    clustered_by.append(d.parse_one(c, dialect=model_dialect))
-                except SqlglotError as e:
-                    raise ConfigError(
-                        f"Failed to parse model '{self.canonical_name(context)}' cluster_by field '{c}' in '{self.path}': {e}"
-                    ) from e
-            optional_kwargs["clustered_by"] = clustered_by
+            if isinstance(kind, ViewKind):
+                logger.warning(
+                    f"Ignoring cluster_by config for model '{self.name}'; cluster_by is not supported for views."
+                )
+            else:
+                clustered_by = []
+                for c in self.cluster_by:
+                    try:
+                        clustered_by.append(d.parse_one(c, dialect=model_dialect))
+                    except SqlglotError as e:
+                        raise ConfigError(
+                            f"Failed to parse model '{self.canonical_name(context)}' cluster_by field '{c}' in '{self.path}': {e}"
+                        ) from e
+                optional_kwargs["clustered_by"] = clustered_by
 
         model_kwargs = self.sqlmesh_model_kwargs(context)
         if self.sql_header:
@@ -627,7 +636,6 @@ def to_sqlmesh(
             if physical_properties:
                 model_kwargs["physical_properties"] = physical_properties
 
-        kind = self.model_kind(context)
         allow_partials = model_kwargs.pop("allow_partials", None)
         if (
             allow_partials is None

tests/dbt/test_transformation.py

@@ -1837,6 +1837,17 @@ def test_model_cluster_by():
         exp.to_column('"QUX"'),
     ]
 
+    model = ModelConfig(
+        name="model",
+        alias="model",
+        package_name="package",
+        target_schema="test",
+        cluster_by=["Bar", "qux"],
+        sql="SELECT * FROM baz",
+        materialized=Materialization.VIEW.value,
+    )
+    assert model.to_sqlmesh(context).clustered_by == []
+
 
 def test_snowflake_dynamic_table():
     context = DbtContext()