fix: robust cluster_by config parsing (#5478)

Author: geooo109

sqlmesh/dbt/model.py

@@ -601,7 +601,13 @@ def to_sqlmesh(
                 clustered_by = []
                 for c in self.cluster_by:
                     try:
-                        clustered_by.append(d.parse_one(c, dialect=model_dialect))
+                        cluster_expr = exp.maybe_parse(
+                            c, into=exp.Cluster, prefix="CLUSTER BY", dialect=model_dialect
+                        )
+                        for expr in cluster_expr.expressions:
+                            clustered_by.append(
+                                expr.this if isinstance(expr, exp.Ordered) else expr
+                            )
                     except SqlglotError as e:
                         raise ConfigError(
                             f"Failed to parse model '{self.canonical_name(context)}' cluster_by field '{c}' in '{self.path}': {e}"

tests/dbt/test_transformation.py

@@ -2306,6 +2306,60 @@ def test_model_cluster_by():
     )
     assert model.to_sqlmesh(context).clustered_by == []
 
+    model = ModelConfig(
+        name="model",
+        alias="model",
+        package_name="package",
+        target_schema="test",
+        cluster_by="Bar, qux",
+        sql="SELECT * FROM baz",
+        materialized=Materialization.TABLE.value,
+    )
+    assert model.to_sqlmesh(context).clustered_by == [
+        exp.to_column('"BAR"'),
+        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.TABLE.value,
+    )
+    assert model.to_sqlmesh(context).clustered_by == [
+        exp.to_column('"Bar,qux"'),
+    ]
+
+    model = ModelConfig(
+        name="model",
+        alias="model",
+        package_name="package",
+        target_schema="test",
+        cluster_by='"Bar,qux"',
+        sql="SELECT * FROM baz",
+        materialized=Materialization.TABLE.value,
+    )
+    assert model.to_sqlmesh(context).clustered_by == [
+        exp.to_column('"Bar,qux"'),
+    ]
+
+    model = ModelConfig(
+        name="model",
+        alias="model",
+        package_name="package",
+        target_schema="test",
+        cluster_by=["to_date(Bar),qux"],
+        sql="SELECT * FROM baz",
+        materialized=Materialization.TABLE.value,
+    )
+    assert model.to_sqlmesh(context).clustered_by == [
+        exp.TsOrDsToDate(this=exp.to_column('"BAR"')),
+        exp.to_column('"QUX"'),
+    ]
+
 
 def test_snowflake_dynamic_table():
     context = DbtContext()