Feat!: Expose Snowflake dynamic tables through the DBT adapter (#2942)

Author: erindru

sqlmesh/core/model/__init__.py

@@ -32,6 +32,7 @@
     SeedKind as SeedKind,
     TimeColumn as TimeColumn,
     ViewKind as ViewKind,
+    ManagedKind as ManagedKind,
     model_kind_validator as model_kind_validator,
 )
 from sqlmesh.core.model.meta import ModelMeta as ModelMeta

sqlmesh/dbt/basemodel.py

@@ -46,6 +46,9 @@ class Materialization(str, Enum):
     EPHEMERAL = "ephemeral"
     SNAPSHOT = "snapshot"
 
+    # Snowflake, https://docs.getdbt.com/reference/resource-configs/snowflake-configs#dynamic-tables
+    DYNAMIC_TABLE = "dynamic_table"
+
 
 class SnapshotStrategy(str, Enum):
     """DBT snapshot strategies"""

sqlmesh/dbt/model.py

@@ -19,6 +19,7 @@
     ModelKind,
     SCDType2ByColumnKind,
     ViewKind,
+    ManagedKind,
     create_sql_model,
 )
 from sqlmesh.core.model.kind import SCDType2ByTimeKind, OnDestructiveChange
@@ -104,6 +105,9 @@ class ModelConfig(BaseModelConfig):
 
     # snowflake
     snowflake_warehouse: t.Optional[str] = None
+    # note: for Snowflake dynamic tables, in the DBT adapter we only support properties that DBT supports
+    # which are defined here: https://docs.getdbt.com/reference/resource-configs/snowflake-configs#dynamic-tables
+    target_lag: t.Optional[str] = None
 
     # Private fields
     _sql_embedded_config: t.Optional[SqlStr] = None
@@ -298,6 +302,10 @@ def model_kind(self, context: DbtContext) -> ModelKind:
             return SCDType2ByTimeKind(
                 updated_at_name=self.updated_at, updated_at_as_valid_from=True, **shared_kwargs
             )
+
+        if materialization == Materialization.DYNAMIC_TABLE:
+            return ManagedKind()
+
         raise ConfigError(f"{materialization.value} materialization not supported.")
 
     @property
@@ -399,8 +407,20 @@ def to_sqlmesh(self, context: DbtContext) -> Model:
             if physical_properties:
                 model_kwargs["physical_properties"] = physical_properties
 
-        if context.target.dialect == "snowflake" and self.snowflake_warehouse is not None:
-            model_kwargs["session_properties"] = {"warehouse": self.snowflake_warehouse}
+        if context.target.dialect == "snowflake":
+            if self.snowflake_warehouse is not None:
+                model_kwargs["session_properties"] = {"warehouse": self.snowflake_warehouse}
+
+            if self.model_materialization == Materialization.DYNAMIC_TABLE:
+                if not self.snowflake_warehouse:
+                    raise ConfigError("`snowflake_warehouse` must be set for dynamic tables")
+                if not self.target_lag:
+                    raise ConfigError("`target_lag` must be set for dynamic tables")
+
+                model_kwargs["physical_properties"] = {
+                    "warehouse": self.snowflake_warehouse,
+                    "target_lag": self.target_lag,
+                }
 
         return create_sql_model(
             self.canonical_name(context),

tests/dbt/test_transformation.py

@@ -21,6 +21,7 @@
     IncrementalByTimeRangeKind,
     IncrementalByUniqueKeyKind,
     IncrementalUnmanagedKind,
+    ManagedKind,
     SqlModel,
     ViewKind,
 )
@@ -264,6 +265,13 @@ def test_model_kind():
         insert_overwrite=True, disable_restatement=True
     )
 
+    assert (
+        ModelConfig(materialized=Materialization.DYNAMIC_TABLE, target_lag="1 hour").model_kind(
+            context
+        )
+        == ManagedKind()
+    )
+
     with pytest.raises(ConfigError):
         ModelConfig(
             materialized=Materialization.INCREMENTAL,
@@ -1120,3 +1128,39 @@ def test_model_cluster_by():
         materialized=Materialization.TABLE.value,
     )
     assert model.to_sqlmesh(context).clustered_by == ["BAR", "QUX"]
+
+
+def test_snowflake_dynamic_table():
+    context = DbtContext()
+    context._target = SnowflakeConfig(
+        name="target",
+        schema="test",
+        database="test",
+        account="account",
+        user="user",
+        password="password",
+    )
+
+    model = ModelConfig(
+        name="model",
+        alias="model",
+        package_name="package",
+        target_schema="test",
+        sql="SELECT * FROM baz",
+        materialized=Materialization.DYNAMIC_TABLE.value,
+        target_lag="1 hour",
+        snowflake_warehouse="SMALL",
+    )
+
+    as_sqlmesh = model.to_sqlmesh(context)
+    assert as_sqlmesh.kind == ManagedKind()
+    assert as_sqlmesh.physical_properties == {
+        "target_lag": exp.Literal.string("1 hour"),
+        "warehouse": exp.Literal.string("SMALL"),
+    }
+
+    # both target_lag and snowflake_warehouse are required properties
+    # https://docs.getdbt.com/reference/resource-configs/snowflake-configs#dynamic-tables
+    for required_property in ["target_lag", "snowflake_warehouse"]:
+        with pytest.raises(ConfigError, match=r".*must be set for dynamic tables"):
+            model.copy(update={required_property: None}).to_sqlmesh(context)