Repository: TobikoData/sqlmesh
Pull request #2942 (2024-07-24 22:53:02)

Feat!: Expose Snowflake dynamic tables through the DBT adapter
The initial implementation of Managed Models was for sqlmesh projects only.

This maps `materialized: dynamic_table` for Snowflake DBT projects to a managed model so dynamic tables can be used in DBT projects


Patch:
diff --git a/docs/concepts/macros/macro_variables.md b/docs/concepts/macros/macro_variables.md
index e72ede48..c23310b4 100644
--- a/docs/concepts/macros/macro_variables.md
+++ b/docs/concepts/macros/macro_variables.md
@@ -1,10 +1,20 @@
 # Macro variables
 
-The most common use case for macros is variable substitution. For example, you might have a SQL query that filters by date in the `WHERE` clause.
+Macro variables are placeholders whose values are substituted in when the macro is rendered.
+
+They enable dynamic macro behavior - for example, a date parameter's value might be based on when the macro was run.
+
+!!! note
+
+    This page discusses SQLMesh's built-in macro variables. Learn more about custom, user-defined macro variables on the [SQLMesh macros page](./sqlmesh_macros.md#user-defined-variables).
+
+## Example
+
+Consider a SQL query that filters by date in the `WHERE` clause.
 
 Instead of manually changing the date each time the model is run, you can use a macro variable to make the date dynamic. With the dynamic approach, the date changes automatically based on when the query is run.
 
-Consider this query that filters for rows where column `my_date` is after '2023-01-01':
+This query filters for rows where column `my_date` is after '2023-01-01':
 
 ```sql linenums="1"
 SELECT *
@@ -37,39 +47,43 @@ We describe SQLMesh's predefined variables below; user-defined macro variables a
 ## Predefined Variables
 SQLMesh comes with predefined variables that can be used in your queries. They are automatically set by the SQLMesh runtime.
 
-Most predefined variables are related to time and use a combination of prefixes (start, end, execution) and postfixes (date, ds, ts, epoch, millis). They are described in the next section; [other predefined variables](#runtime-variables) are discussed in the following section.
+Most predefined variables are related to time and use a combination of prefixes (start, end, etc.) and postfixes (date, ds, ts, etc.). They are described in the next section; [other predefined variables](#runtime-variables) are discussed in the following section.
 
 ### Temporal variables
 
-SQLMesh uses the python [datetime module](https://docs.python.org/3/library/datetime.html) for handling dates and times. It uses the standard [Unix epoch](https://en.wikipedia.org/wiki/Unix_time) start of 1970-01-01. *All predefined variables with a time component use the [UTC time zone](https://en.wikipedia.org/wiki/Coordinated_Universal_Time).*
+SQLMesh uses the python [datetime module](https://docs.python.org/3/library/datetime.html) for handling dates and times. It uses the standard [Unix epoch](https://en.wikipedia.org/wiki/Unix_time) start of 1970-01-01.
+
+*All predefined variables with a time component use the [UTC time zone](https://en.wikipedia.org/wiki/Coordinated_Universal_Time).*
 
 Prefixes:
 
-* start - The inclusive starting interval of a model run.
-* end - The inclusive end interval of a model run.
-* execution - The timestamp of when the execution started.
+* start - The inclusive starting interval of a model run
+* end - The inclusive end interval of a model run
+* execution - The timestamp of when the execution started
 
 Postfixes:
 
-* date - A python date object that converts into a native SQL Date.
+* dt - A python datetime object that converts into a native SQL `TIMESTAMP` (or SQL engine equivalent)
+* date - A python date object that converts into a native SQL `DATE`
 * ds - A date string with the format: '%Y-%m-%d'
-* ts - An ISO 8601 datetime formatted string: '%Y-%m-%d %H:%M:%S'.
-* tstz - An ISO 8601 datetime formatted string with timezone: '%Y-%m-%d %H:%M:%S%z'.
-* epoch - An integer representing seconds since Unix epoch.
-* millis - An integer representing milliseconds since Unix epoch.
+* ts - An ISO 8601 datetime formatted string: '%Y-%m-%d %H:%M:%S'
+* tstz - An ISO 8601 datetime formatted string with timezone: '%Y-%m-%d %H:%M:%S%z'
+* hour - An integer representing the hour of the day, with values 0-23
+* epoch - An integer representing seconds since Unix epoch
+* millis - An integer representing milliseconds since Unix epoch
 
 All predefined temporal macro variables:
 
+* dt
+    * @start_dt
+    * @end_dt
+    * @execution_dt
+
 * date
     * @start_date
     * @end_date
     * @execution_date
 
-* datetime
-    * @start_dt
-    * @end_dt
-    * @execution_dt
-
 * ds
     * @start_ds
     * @end_ds
@@ -85,6 +99,11 @@ All predefined temporal macro variables:
     * @end_tstz
     * @execution_tstz
 
+* hour
+    * @start_hour
+    * @end_hour
+    * @execution_hour
+
 * epoch
     * @start_epoch
     * @end_epoch
@@ -99,12 +118,12 @@ All predefined temporal macro variables:
 
 SQLMesh provides two other predefined variables used to modify model behavior based on information available at runtime.
 
-* @runtime_stage - A string value that denotes the current stage of the SQLMesh runtime. It can take one of the following values:
-    * 'loading' - The project is currently being loaded into SQLMesh's runtime context.
+* @runtime_stage - A string value denoting the current stage of the SQLMesh runtime. Typically used in models to conditionally execute pre/post-statements (learn more [here](../models/sql_models.md#optional-prepost-statements)). It returns one of these values:
+    * 'loading' - The project is being loaded into SQLMesh's runtime context.
     * 'creating' - The model tables are being created.
-    * 'evaluating' - The models' logic is being evaluated.
-    * 'testing' - The models' logic is being evaluated in the context of a unit test.
-* @gateway - A string value that represents the name of the selected [gateway](../../guides/connections.md).
+    * 'evaluating' - The model query logic is being evaluated.
+    * 'testing' - The model query logic is being evaluated in the context of a unit test.
+* @gateway - A string value containing the name of the current [gateway](../../guides/connections.md).
 
 ### Audit-only variables
 
diff --git a/docs/concepts/macros/sqlmesh_macros.md b/docs/concepts/macros/sqlmesh_macros.md
index 2396d7bc..bab41a7a 100644
--- a/docs/concepts/macros/sqlmesh_macros.md
+++ b/docs/concepts/macros/sqlmesh_macros.md
@@ -57,17 +57,37 @@ Access global variable values in a model definition using the `@<VAR_NAME>` macr
 
 For example, this SQLMesh configuration key defines six variables of different data types:
 
-```yaml linenums="1"
-variables:
-  int_var: 1
-  float_var: 2.0
-  bool_var: true
-  str_var: "cat"
-  list_var: [1, 2, 3]
-  dict_var:
-    key1: 1
-    key2: 2
-```
+=== "YAML"
+
+    ```yaml linenums="1"
+    variables:
+      int_var: 1
+      float_var: 2.0
+      bool_var: true
+      str_var: "cat"
+      list_var: [1, 2, 3]
+      dict_var:
+        key1: 1
+        key2: 2
+    ```
+
+=== "Python"
+
+    ``` python linenums="1"
+    variables = {
+        "int_var": 1,
+        "float_var": 2.0,
+        "bool_var": True,
+        "str_var": "cat",
+        "list_var": [1, 2, 3],
+        "dict_var": {"key1": 1, "key2": 2},
+    }
+
+    config = Config(
+        variables=variables,
+        ... # other Config arguments
+    )
+    ```
 
 A model definition could access the `int_var` value in a `WHERE` clause like this:
 
@@ -101,13 +121,32 @@ A similar API is available for [Python macro functions](#accessing-global-variab
 
 Like global variables, gateway variables are defined in the project configuration file. However, they are specified in a specific gateway's `variables` key:
 
-```yaml linenums="1"
-gateways:
-  my_gateway:
-    variables:
-      int_var: 1
-    ...
-```
+=== "YAML"
+
+    ```yaml linenums="1"
+    gateways:
+      my_gateway:
+        variables:
+          int_var: 1
+        ...
+    ```
+
+=== "Python"
+
+    ``` python linenums="1"
+    gateway_variables = {
+      "int_var": 1
+    }
+
+    config = Config(
+        gateways={
+          "my_gateway": GatewayConfig(
+            variables=gateway_variables
+            ... # other GatewayConfig arguments
+            ),
+          }
+    )
+    ```
 
 Access them in models using the same methods as [global variables](#global-variables).
 
@@ -1586,11 +1625,15 @@ Typed macros in SQLMesh use Python's type hints. Here's a simple example of a ty
 from sqlmesh import macro
 
 @macro()
-def repeat_string(evaluator, text: str, count: int) -> str:
+def repeat_string(evaluator, text: str, count: int):
     return text * count
 ```
 
-Usage in SQLMesh:
+This macro takes two arguments: `text` of type `str` and `count` of type `int`, and it returns a string.
+
+Without type hints, the inputs are two SQLGlot `exp.Literal` objects you would need to manually convert to Python `str` and `int` types. With type hints, you can work with them as string and integer types directly.
+
+Let's try to use the macro in a SQLMesh model:
 
 ```sql linenums="1"
 SELECT
@@ -1598,7 +1641,44 @@ SELECT
 FROM some_table;
 ```
 
-This macro takes two arguments: `text` of type `str` and `count` of type `int`, and it returns a string. Without type hints, the inputs to the macro would have been two `exp.Literal` objects you would have had to convert to strings and integers manually.
+Unfortunately, this model generates an error when rendered:
+
+```
+Error: Invalid expression / Unexpected token. Line 1, Col: 23.
+  SQLMesh SQLMesh SQLMesh
+```
+
+Why? The macro returned `SQLMesh SQLMesh SQLMesh` as expected, but that string is not valid SQL in the rendered query:
+
+```sql linenums="1" hl_lines="2"
+SELECT
+  SQLMesh SQLMesh SQLMesh as repeated_string ### invalid SQL code
+FROM some_table;
+```
+
+The problem is a mismatch between our macro's Python return type `str` and the type expected by the parsed SQL query.
+
+Recall that SQLMesh macros work by modifying the query's semantic representation. In that representation, a SQLGlot string literal type is expected. SQLMesh will do its best to return the type expected by the query's semantic representation, but that is not possible in all scenarios.
+
+Therefore, we must explicitly convert the output with SQLGlot's `exp.Literal.string()` method:
+
+```python linenums="1" hl_lines="5"
+from sqlmesh import macro
+
+@macro()
+def repeat_string(evaluator, text: str, count: int):
+    return exp.Literal.string(text * count)
+```
+
+Now the query will render with a valid single-quoted string literal:
+
+```sql linenums="1"
+SELECT
+  'SQLMesh SQLMesh SQLMesh ' AS "repeated_string"
+FROM "some_table" AS "some_table"
+```
+
+Typed macros coerce the **inputs** to a macro function, but the macro code is responsible for coercing the **output** to the type expected by the query's semantic representation.
 
 #### Supported Types
 
diff --git a/mkdocs.yml b/mkdocs.yml
index f8b1d6ac..71a958c2 100644
--- a/mkdocs.yml
+++ b/mkdocs.yml
@@ -134,7 +134,7 @@ markdown_extensions:
   - pymdownx.inlinehilite
   - pymdownx.snippets
   - pymdownx.superfences
-  - pymdownx.blocks.tab:
+  - pymdownx.tabbed:
       alternate_style: true
   - admonition
   - pymdownx.details
diff --git a/sqlmesh/core/model/__init__.py b/sqlmesh/core/model/__init__.py
index baac47ae..e6e4ddda 100644
--- a/sqlmesh/core/model/__init__.py
+++ b/sqlmesh/core/model/__init__.py
@@ -32,6 +32,7 @@ from sqlmesh.core.model.kind import (
     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
diff --git a/sqlmesh/dbt/basemodel.py b/sqlmesh/dbt/basemodel.py
index 40a07ea9..b88ca295 100644
--- a/sqlmesh/dbt/basemodel.py
+++ b/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"""
diff --git a/sqlmesh/dbt/model.py b/sqlmesh/dbt/model.py
index 80fffa58..bc9b50fb 100644
--- a/sqlmesh/dbt/model.py
+++ b/sqlmesh/dbt/model.py
@@ -19,6 +19,7 @@ from sqlmesh.core.model import (
     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 @@ class ModelConfig(BaseModelConfig):
             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 @@ class ModelConfig(BaseModelConfig):
             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),


Test patch:
diff --git a/tests/dbt/test_transformation.py b/tests/dbt/test_transformation.py
index 9356e14f..4dfb6805 100644
--- a/tests/dbt/test_transformation.py
+++ b/tests/dbt/test_transformation.py
@@ -21,6 +21,7 @@ from sqlmesh.core.model import (
     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)