refactor impl

Author: geooo109

sqlmesh/dbt/model.py

@@ -157,20 +157,14 @@ class ModelConfig(BaseModelConfig):
 
     @field_validator(
         "unique_key",
+        "cluster_by",
         "tags",
         mode="before",
     )
     @classmethod
     def _validate_list(cls, v: t.Union[str, t.List[str]]) -> t.List[str]:
         return ensure_list(v)
 
-    @field_validator("cluster_by", mode="before")
-    @classmethod
-    def _validate_cluster_by(cls, v: t.Union[str, t.List[str]]) -> t.Union[str, t.List[str]]:
-        if isinstance(v, str):
-            return [c.strip() for c in v.split(",")]
-        return ensure_list(v)
-
     @field_validator("check_cols", mode="before")
     @classmethod
     def _validate_check_cols(cls, v: t.Union[str, t.List[str]]) -> t.Union[str, t.List[str]]:
@@ -607,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

@@ -2320,6 +2320,46 @@ 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.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()