Feat: allow parameterizing model kinds using macros (#3401)

Author: georgesittas

sqlmesh/core/dialect.py

@@ -552,11 +552,15 @@ def parse(self: Parser) -> t.Optional[exp.Expression]:
             elif key == "columns":
                 value = self._parse_schema()
             elif key == "kind":
-                id_var = self._parse_id_var(any_token=True)
-                if not id_var:
-                    value = None
+                if self._match(TokenType.PARAMETER):
+                    field = _parse_macro(self)
                 else:
-                    kind = ModelKindName[id_var.name.upper()]
+                    field = self._parse_id_var(any_token=True)
+
+                if not field or isinstance(field, (MacroVar, MacroFunc)):
+                    value = field
+                else:
+                    kind = ModelKindName[field.name.upper()]
 
                     if kind in (
                         ModelKindName.INCREMENTAL_BY_TIME_RANGE,
@@ -573,11 +577,7 @@ def parse(self: Parser) -> t.Optional[exp.Expression]:
                     else:
                         props = None
 
-                    value = self.expression(
-                        ModelKind,
-                        this=kind.value,
-                        expressions=props,
-                    )
+                    value = self.expression(ModelKind, this=kind.value, expressions=props)
             elif key == "expression":
                 value = self._parse_conjunction()
             else:

tests/core/test_model.py

@@ -38,6 +38,7 @@
     SqlModel,
     TimeColumn,
     ExternalKind,
+    ViewKind,
     create_external_model,
     create_seed_model,
     create_sql_model,
@@ -6091,3 +6092,48 @@ def test_cluster_with_complex_expression():
 
     model = load_sql_based_model(expressions)
     assert [expr.sql("snowflake") for expr in model.clustered_by] == ['(TO_DATE("CLUSTER_COL"))']
+
+
+def test_parametric_model_kind(tmp_path: Path):
+    init_example_project(tmp_path, dialect="duckdb")
+
+    test_sql_file = tmp_path / "models/test_model.sql"
+    test_sql_file.write_text(
+        """
+        MODEL (
+          name test_schema.test_model,
+          kind @IF(@gateway = 'main', VIEW, FULL)
+        );
+
+        SELECT
+          1 AS c
+        """
+    )
+
+    db_path = str(tmp_path / "db.db")
+    config = Config(
+        gateways={
+            "main": GatewayConfig(connection=DuckDBConnectionConfig(database=db_path)),
+            "other": GatewayConfig(connection=DuckDBConnectionConfig(database=db_path)),
+        },
+        model_defaults=ModelDefaultsConfig(dialect="duckdb"),
+    )
+
+    context = Context(paths=tmp_path, config=config)
+    plan = context.plan(no_prompts=True, auto_apply=True, no_diff=True)
+
+    assert len(plan.context_diff.new_snapshots) == 4
+    assert isinstance(context.get_model("test_schema.test_model").kind, ViewKind)
+
+    context = Context(paths=tmp_path, config=config, gateway="other")
+    plan = context.plan(no_prompts=True, auto_apply=True, no_diff=True)
+    diff = plan.context_diff
+
+    assert len(diff.new_snapshots) == 1
+    assert len(diff.modified_snapshots) == 1
+
+    new_snapshot, old_snapshot = next(iter(diff.modified_snapshots.values()))
+    assert isinstance(t.cast(SqlModel, new_snapshot.node).kind, FullKind)
+    assert isinstance(t.cast(SqlModel, old_snapshot.node).kind, ViewKind)
+
+    assert isinstance(context.get_model("test_schema.test_model").kind, FullKind)