Include config in dbt query payload

Author: georgesittas

sqlmesh/core/model/definition.py

@@ -595,6 +595,7 @@ def _statement_renderer(self, expression: exp.Expression) -> ExpressionRenderer:
                 only_execution_time=False,
                 default_catalog=self.default_catalog,
                 model_fqn=self.fqn,
+                raw_code=self._raw_code,
             )
         return self._statement_renderer_cache[expression_key]
 
@@ -1305,6 +1306,10 @@ def _is_time_column_in_partitioned_by(self) -> bool:
     def violated_rules_for_query(self) -> t.Dict[type[Rule], t.Any]:
         return {}
 
+    @property
+    def _raw_code(self) -> t.Optional[str]:
+        return None
+
 
 class SqlModel(_Model):
     """The model definition which relies on a SQL query to fetch the data.
@@ -1581,6 +1586,7 @@ def _query_renderer(self) -> QueryRenderer:
             default_catalog=self.default_catalog,
             quote_identifiers=not no_quote_identifiers,
             optimize_query=self.optimize_query,
+            raw_code=self._raw_code,
         )
 
     @property
@@ -1606,6 +1612,11 @@ def violated_rules_for_query(self) -> t.Dict[type[Rule], t.Any]:
         self.render_query()
         return self._query_renderer._violated_rules
 
+    @property
+    def _raw_code(self) -> t.Optional[str]:
+        query = self.query
+        return query.name if isinstance(query, d.JinjaQuery) else None
+
 
 class SeedModel(_Model):
     """The model definition which uses a pre-built static dataset to source the data from.

sqlmesh/core/renderer.py

@@ -53,6 +53,7 @@ def __init__(
         model_fqn: t.Optional[str] = None,
         normalize_identifiers: bool = True,
         optimize_query: t.Optional[bool] = True,
+        raw_code: t.Optional[str] = None,
     ):
         self._expression = expression
         self._dialect = dialect
@@ -68,6 +69,7 @@ def __init__(
         self._cache: t.List[t.Optional[exp.Expression]] = []
         self._model_fqn = model_fqn
         self._optimize_query_flag = optimize_query is not False
+        self._raw_code = raw_code
 
     def update_schema(self, schema: t.Dict[str, t.Any]) -> None:
         self.schema = d.normalize_mapping_schema(schema, dialect=self._dialect)
@@ -204,7 +206,7 @@ def _resolve_table(table: str | exp.Table) -> str:
                     "default_catalog": self._default_catalog,
                     "runtime_stage": runtime_stage.value,
                     "resolve_table": _resolve_table,
-                    "raw_code": self._expression.name,
+                    "raw_code": self._raw_code,
                 }
 
                 if this_model:

sqlmesh/dbt/basemodel.py

@@ -168,14 +168,6 @@ def _validate_grants(cls, v: t.Dict[str, str]) -> t.Dict[str, t.List[str]]:
         },
     }
 
-    @property
-    def sql_no_config(self) -> SqlStr:
-        return SqlStr("")
-
-    @property
-    def sql_embedded_config(self) -> SqlStr:
-        return SqlStr("")
-
     @property
     def table_schema(self) -> str:
         """

sqlmesh/dbt/loader.py

@@ -138,7 +138,7 @@ def _to_sqlmesh(config: BMC, context: DbtContext) -> Model:
                 package_models: t.Dict[str, BaseModelConfig] = {**package.models, **package.seeds}
 
                 for model in package_models.values():
-                    if isinstance(model, ModelConfig) and not model.sql_no_config:
+                    if isinstance(model, ModelConfig) and not model.sql.strip():
                         logger.info(f"Skipping empty model '{model.name}' at path '{model.path}'.")
                         continue
 

sqlmesh/dbt/model.py

@@ -27,7 +27,7 @@
 )
 from sqlmesh.core.model.kind import SCDType2ByTimeKind, OnDestructiveChange, OnAdditiveChange
 from sqlmesh.dbt.basemodel import BaseModelConfig, Materialization, SnapshotStrategy
-from sqlmesh.dbt.common import SqlStr, extract_jinja_config, sql_str_validator
+from sqlmesh.dbt.common import SqlStr, sql_str_validator
 from sqlmesh.utils.errors import ConfigError
 from sqlmesh.utils.pydantic import field_validator
 
@@ -138,10 +138,6 @@ class ModelConfig(BaseModelConfig):
     inserts_only: t.Optional[bool] = None
     incremental_predicates: t.Optional[t.List[str]] = None
 
-    # Private fields
-    _sql_embedded_config: t.Optional[SqlStr] = None
-    _sql_no_config: t.Optional[SqlStr] = None
-
     _sql_validator = sql_str_validator
 
     @field_validator(
@@ -432,25 +428,6 @@ def model_kind(self, context: DbtContext) -> ModelKind:
 
         raise ConfigError(f"{materialization.value} materialization not supported.")
 
-    @property
-    def sql_no_config(self) -> SqlStr:
-        if self._sql_no_config is None:
-            self._sql_no_config = SqlStr("")
-            self._extract_sql_config()
-        return self._sql_no_config
-
-    @property
-    def sql_embedded_config(self) -> SqlStr:
-        if self._sql_embedded_config is None:
-            self._sql_embedded_config = SqlStr("")
-            self._extract_sql_config()
-        return self._sql_embedded_config
-
-    def _extract_sql_config(self) -> None:
-        no_config, embedded_config = extract_jinja_config(self.sql)
-        self._sql_no_config = SqlStr(no_config)
-        self._sql_embedded_config = SqlStr(embedded_config)
-
     def _big_query_partition_by_expr(self, context: DbtContext) -> exp.Expression:
         assert isinstance(self.partition_by, dict)
         data_type = self.partition_by["data_type"].lower()
@@ -508,7 +485,7 @@ def to_sqlmesh(
     ) -> Model:
         """Converts the dbt model into a SQLMesh model."""
         model_dialect = self.dialect(context)
-        query = d.jinja_query(self.sql_no_config)
+        query = d.jinja_query(self.sql)
         kind = self.model_kind(context)
 
         optional_kwargs: t.Dict[str, t.Any] = {}

sqlmesh/dbt/test.py

@@ -12,7 +12,6 @@
     Dependencies,
     GeneralConfig,
     SqlStr,
-    extract_jinja_config,
     sql_str_validator,
 )
 from sqlmesh.utils import AttributeDict
@@ -134,9 +133,7 @@ def to_sqlmesh(self, context: DbtContext) -> Audit:
             }
         )
 
-        sql_no_config, _sql_config_only = extract_jinja_config(self.sql)
-        sql_no_config = sql_no_config.replace("**_dbt_generic_test_kwargs", self._kwargs())
-        query = d.jinja_query(sql_no_config)
+        query = d.jinja_query(self.sql.replace("**_dbt_generic_test_kwargs", self._kwargs()))
 
         skip = not self.enabled
         blocking = self.severity == Severity.ERROR

sqlmesh/migrations/v0095_warn_about_dbt_raw_sql_diff.py

@@ -0,0 +1,47 @@
+"""
+Warns dbt users about potential diffs due to inclusion of {{ config(...) }} blocks in model SQL.
+
+Prior to this fix, SQLMesh wasn't including the {{ config(...) }} block in the model's SQL payload
+when processing dbt models. Now these config blocks are properly included in the raw SQL, which
+may cause diffs to appear for existing dbt models even though the actual SQL logic hasn't changed.
+
+This is a one-time diff that will appear after upgrading, and applying a plan will resolve it.
+"""
+
+import json
+
+from sqlglot import exp
+
+from sqlmesh.core.console import get_console
+
+SQLMESH_DBT_PACKAGE = "sqlmesh.dbt"
+
+
+def migrate(state_sync, **kwargs):  # type: ignore
+    engine_adapter = state_sync.engine_adapter
+    schema = state_sync.schema
+    snapshots_table = "_snapshots"
+    if schema:
+        snapshots_table = f"{schema}.{snapshots_table}"
+
+    warning = (
+        "SQLMesh now includes dbt's {{ config(...) }} blocks in the model's raw SQL when "
+        "processing dbt models. This change ensures that all model attributes referenced "
+        "in macros are properly tracked for fingerprinting. As a result, you may see diffs "
+        "for existing dbt models even though the actual SQL logic hasn't changed. This is "
+        "a one-time diff that will be resolved after applying a plan. Run 'sqlmesh diff prod' "
+        "to review any changes, then apply a plan if the diffs look expected."
+    )
+
+    for (snapshot,) in engine_adapter.fetchall(
+        exp.select("snapshot").from_(snapshots_table), quote_identifiers=True
+    ):
+        parsed_snapshot = json.loads(snapshot)
+        node = parsed_snapshot["node"]
+
+        jinja_macros = node.get("jinja_macros") or {}
+        create_builtins_module = jinja_macros.get("create_builtins_module") or ""
+
+        if create_builtins_module == SQLMESH_DBT_PACKAGE:
+            get_console().log_warning(warning)
+            return

tests/core/test_context.py

@@ -1534,9 +1534,24 @@ def test_plan_enable_preview_default(sushi_context: Context, sushi_dbt_context:
     assert sushi_dbt_context._plan_preview_enabled
 
 
-def test_raw_code_missing_from_model_attributes(sushi_test_dbt_context: Context):
-    customers_model = sushi_test_dbt_context.models['"memory"."sushi"."simple_model_a"']
-    assert "raw_code" not in customers_model.jinja_macros.global_objs["model"]  # type: ignore
+@pytest.mark.slow
+def test_raw_code_handling(sushi_test_dbt_context: Context):
+    model = sushi_test_dbt_context.models['"memory"."sushi"."model_with_raw_code"']
+    assert "raw_code" not in model.jinja_macros.global_objs["model"]  # type: ignore
+
+    # logging "pre-hook" (in dbt_projects.yml) + the actual pre-hook in the model file
+    assert len(model.pre_statements) == 2
+
+    original_file_path = model.jinja_macros.global_objs["model"]["original_file_path"]  # type: ignore
+    model_file_path = sushi_test_dbt_context.path / original_file_path
+
+    raw_code_length = len(model_file_path.read_text()) - 1
+
+    hook = model.render_pre_statements()[0]
+    assert (
+        hook.sql()
+        == f'''CREATE TABLE "t" AS SELECT 'Length is {raw_code_length}' AS "length_col"'''
+    )
 
 
 def test_catalog_name_needs_to_be_quoted():

tests/dbt/test_config.py

@@ -276,42 +276,6 @@ def test_singular_test_to_standalone_audit(dbt_dummy_postgres_config: PostgresCo
     assert standalone_audit.dialect == "bigquery"
 
 
-def test_model_config_sql_no_config():
-    assert (
-        ModelConfig(
-            sql="""{{
-  config(
-    materialized='table',
-    incremental_strategy='delete+"insert'
-  )
-}}
-query"""
-        ).sql_no_config.strip()
-        == "query"
-    )
-
-    assert (
-        ModelConfig(
-            sql="""{{
-  config(
-    materialized='table',
-    incremental_strategy='delete+insert',
-    post_hook=" '{{ var('new') }}' "
-  )
-}}
-query"""
-        ).sql_no_config.strip()
-        == "query"
-    )
-
-    assert (
-        ModelConfig(
-            sql="""before {{config(materialized='table', post_hook=" {{ var('new') }} ")}} after"""
-        ).sql_no_config.strip()
-        == "before  after"
-    )
-
-
 @pytest.mark.slow
 def test_variables(assert_exp_eq, sushi_test_project):
     # Case 1: using an undefined variable without a default value
@@ -350,7 +314,6 @@ def test_variables(assert_exp_eq, sushi_test_project):
 
     # Case 3: using a defined variable with a default value
     model_config.sql = "SELECT {{ var('foo', 5) }}"
-    model_config._sql_no_config = None
 
     assert_exp_eq(model_config.to_sqlmesh(**kwargs).render_query(), 'SELECT 6 AS "6"')
 

tests/fixtures/dbt/sushi_test/models/model_with_raw_code.sql

@@ -0,0 +1,11 @@
+{{
+  config(
+    pre_hook=['CREATE TABLE t AS SELECT \'Length is {{ model.raw_code|length }}\' AS length_col']
+  )
+}}
+
+{{ check_model_is_table(model) }}
+{{ check_model_is_table_alt(model) }}
+
+SELECT
+  1 AS c