use all vars when dynamic vars are detected

Author: izeigerman

sqlmesh/dbt/adapter.py

@@ -245,56 +245,6 @@ def _raise_parsetime_adapter_call_error(action: str) -> None:
         raise ParsetimeAdapterCallError(f"Can't {action} at parse time.")
 
 
-class StubParsetimeAdapter(BaseAdapter):
-    """Same as ParsetimeAdapter, but returns stub / empty values instead of raising an error."""
-
-    def get_relation(self, database: str, schema: str, identifier: str) -> t.Optional[BaseRelation]:
-        return None
-
-    def load_relation(self, relation: BaseRelation) -> t.Optional[BaseRelation]:
-        return None
-
-    def list_relations(self, database: t.Optional[str], schema: str) -> t.List[BaseRelation]:
-        return []
-
-    def list_relations_without_caching(self, schema_relation: BaseRelation) -> t.List[BaseRelation]:
-        return []
-
-    def get_columns_in_relation(self, relation: BaseRelation) -> t.List[Column]:
-        return []
-
-    def get_missing_columns(
-        self, from_relation: BaseRelation, to_relation: BaseRelation
-    ) -> t.List[Column]:
-        return []
-
-    def create_schema(self, relation: BaseRelation) -> None:
-        pass
-
-    def drop_schema(self, relation: BaseRelation) -> None:
-        pass
-
-    def drop_relation(self, relation: BaseRelation) -> None:
-        pass
-
-    def rename_relation(self, from_relation: BaseRelation, to_relation: BaseRelation) -> None:
-        pass
-
-    def execute(
-        self, sql: str, auto_begin: bool = False, fetch: bool = False
-    ) -> t.Tuple[AdapterResponse, agate.Table]:
-        from dbt.adapters.base.impl import AdapterResponse
-        from sqlmesh.dbt.util import empty_table
-
-        return AdapterResponse(""), empty_table()
-
-    def resolve_schema(self, relation: BaseRelation) -> t.Optional[str]:
-        return relation.schema
-
-    def resolve_identifier(self, relation: BaseRelation) -> t.Optional[str]:
-        return relation.identifier
-
-
 class RuntimeAdapter(BaseAdapter):
     def __init__(
         self,

sqlmesh/dbt/basemodel.py

@@ -309,15 +309,16 @@ def sqlmesh_model_kwargs(
         self,
         context: DbtContext,
         column_types_override: t.Optional[t.Dict[str, ColumnConfig]] = None,
-        extra_dependencies: t.Optional[Dependencies] = None,
     ) -> t.Dict[str, t.Any]:
         """Get common sqlmesh model parameters"""
         self.remove_tests_with_invalid_refs(context)
         self.check_for_circular_test_refs(context)
 
-        dependencies = self.dependencies
-        if extra_dependencies:
-            dependencies = dependencies.union(extra_dependencies)
+        dependencies = self.dependencies.copy()
+        if dependencies.has_dynamic_var_names:
+            # Include ALL variables as dependencies since we couldn't determine
+            # precisely which variables are referenced in the model
+            dependencies.variables |= set(context.variables)
 
         model_dialect = self.dialect(context)
         model_context = context.context_for_dependencies(
@@ -376,8 +377,3 @@ def _model_jinja_context(
             "config": self.config_attribute_dict,
             **context.jinja_globals,
         }
-
-    def _track_dependencies_on_render(self, input: str, context: DbtContext) -> Dependencies:
-        return context.track_dependencies_on_render(
-            input, self._model_jinja_context(context, self.dependencies), self.package_name
-        )

sqlmesh/dbt/builtin.py

@@ -17,7 +17,7 @@
 from sqlmesh.core.console import get_console
 from sqlmesh.core.engine_adapter import EngineAdapter
 from sqlmesh.core.snapshot.definition import DeployabilityIndex
-from sqlmesh.dbt.adapter import BaseAdapter, ParsetimeAdapter, RuntimeAdapter, StubParsetimeAdapter
+from sqlmesh.dbt.adapter import BaseAdapter, ParsetimeAdapter, RuntimeAdapter
 from sqlmesh.dbt.relation import Policy
 from sqlmesh.dbt.target import TARGET_TYPE_TO_CONFIG_CLASS
 from sqlmesh.dbt.util import DBT_VERSION
@@ -384,15 +384,15 @@ def create_builtin_globals(
             builtin_globals["this"] = this
 
     sources = jinja_globals.pop("sources", None)
-    if sources is not None and "source" not in jinja_globals:
+    if sources is not None:
         builtin_globals["source"] = generate_source(sources, api)
 
     refs = jinja_globals.pop("refs", None)
-    if refs is not None and "ref" not in jinja_globals:
+    if refs is not None:
         builtin_globals["ref"] = generate_ref(refs, api)
 
     variables = jinja_globals.pop("vars", None)
-    if variables is not None and "var" not in jinja_globals:
+    if variables is not None:
         builtin_globals["var"] = Var(variables)
 
     deployability_index = (
@@ -415,7 +415,6 @@ def create_builtin_globals(
         {k: builtin_globals.get(k) for k in ("ref", "source", "config", "var")}
     )
 
-    execute = True
     if engine_adapter is not None:
         builtin_globals["flags"] = Flags(which="run")
         adapter: BaseAdapter = RuntimeAdapter(
@@ -436,11 +435,7 @@ def create_builtin_globals(
         )
     else:
         builtin_globals["flags"] = Flags(which="parse")
-        adapter_class: t.Type[BaseAdapter] = ParsetimeAdapter
-        if jinja_globals.get("use_stub_adapter", False):
-            adapter_class = StubParsetimeAdapter
-            execute = False
-        adapter = adapter_class(
+        adapter = ParsetimeAdapter(
             jinja_macros,
             jinja_globals={**builtin_globals, **jinja_globals},
             project_dialect=project_dialect,
@@ -451,7 +446,7 @@ def create_builtin_globals(
     builtin_globals.update(
         {
             "adapter": adapter,
-            "execute": execute,
+            "execute": True,
             "load_relation": adapter.load_relation,
             "store_result": sql_execution.store_result,
             "load_result": sql_execution.load_result,

sqlmesh/dbt/context.py

@@ -7,7 +7,7 @@
 from dbt.adapters.base import BaseRelation
 
 from sqlmesh.core.config import Config as SQLMeshConfig
-from sqlmesh.dbt.builtin import _relation_info_to_relation, Var
+from sqlmesh.dbt.builtin import _relation_info_to_relation
 from sqlmesh.dbt.common import Dependencies
 from sqlmesh.dbt.manifest import ManifestHelper
 from sqlmesh.dbt.target import TargetConfig
@@ -101,8 +101,6 @@ def add_variables(self, variables: t.Dict[str, t.Any]) -> None:
         self._jinja_environment = None
 
     def set_and_render_variables(self, variables: t.Dict[str, t.Any], package: str) -> None:
-        self.variables = variables
-
         jinja_environment = self.jinja_macros.build_environment(**self.jinja_globals)
 
         def _render_var(value: t.Any) -> t.Any:
@@ -212,38 +210,6 @@ def target(self, value: TargetConfig) -> None:
     def render(self, source: str, **kwargs: t.Any) -> str:
         return self.jinja_environment.from_string(source).render(**kwargs)
 
-    def track_dependencies_on_render(
-        self, input: str, jinja_context: t.Dict[str, t.Any], package_name: t.Optional[str] = None
-    ) -> Dependencies:
-        dependencies_on_render = Dependencies()
-
-        class TrackingVar(Var):
-            def __call__(
-                self, name: str, default: t.Optional[t.Any] = None, **kwargs: t.Any
-            ) -> t.Any:
-                dependencies_on_render.variables.add(name)
-                return super().__call__(name, default, **kwargs)
-
-            def has_var(self, name: str) -> bool:
-                dependencies_on_render.variables.add(name)
-                return super().has_var(name)
-
-        if package_name:
-            top_level_packages = [*self.jinja_macros.top_level_packages, package_name]
-            jinja_macros = self.jinja_macros.copy(update={"top_level_packages": top_level_packages})
-        else:
-            jinja_macros = self.jinja_macros
-
-        jinja_environment = jinja_macros.build_environment(
-            **{
-                **jinja_context,
-                "var": TrackingVar(self.variables),
-                "use_stub_adapter": True,
-            }
-        )
-        jinja_environment.from_string(input).render()
-        return dependencies_on_render
-
     def get_callable_macro(
         self, name: str, package: t.Optional[str] = None
     ) -> t.Optional[t.Callable]:

sqlmesh/dbt/loader.py

@@ -124,8 +124,6 @@ def _to_sqlmesh(config: BMC, context: DbtContext) -> Model:
             )
 
         for project in self._load_projects():
-            context = project.context.copy()
-
             macros_max_mtime = self._macros_max_mtime
             yaml_max_mtimes = self._compute_yaml_max_mtime_per_subfolder(
                 project.context.project_root
@@ -135,12 +133,13 @@ def _to_sqlmesh(config: BMC, context: DbtContext) -> Model:
             logger.debug("Converting models to sqlmesh")
             # Now that config is rendered, create the sqlmesh models
             for package in project.packages.values():
-                context.set_and_render_variables(package.variables, package.name)
+                package_context = project.context.copy()
+                package_context.set_and_render_variables(package.variables, package.name)
                 package_models: t.Dict[str, BaseModelConfig] = {**package.models, **package.seeds}
 
                 for model in package_models.values():
                     sqlmesh_model = cache.get_or_load_models(
-                        model.path, loader=lambda: [_to_sqlmesh(model, context)]
+                        model.path, loader=lambda: [_to_sqlmesh(model, package_context)]
                     )[0]
 
                     models[sqlmesh_model.fqn] = sqlmesh_model
@@ -155,15 +154,14 @@ def _load_audits(
         audits: UniqueKeyDict = UniqueKeyDict("audits")
 
         for project in self._load_projects():
-            context = project.context
-
             logger.debug("Converting audits to sqlmesh")
             for package in project.packages.values():
-                context.set_and_render_variables(package.variables, package.name)
+                package_context = project.context.copy()
+                package_context.set_and_render_variables(package.variables, package.name)
                 for test in package.tests.values():
                     logger.debug("Converting '%s' to sqlmesh format", test.name)
                     try:
-                        audits[test.name] = test.to_sqlmesh(context)
+                        audits[test.name] = test.to_sqlmesh(package_context)
                     except MissingModelError as e:
                         logger.warning(
                             "Skipping audit '%s' because model '%s' is not a valid ref",
@@ -244,9 +242,9 @@ def _load_environment_statements(self, macros: MacroRegistry) -> t.List[Environm
         project_names: t.Set[str] = set()
         dialect = self.config.dialect
         for project in self._load_projects():
-            context = project.context
             for package_name, package in project.packages.items():
-                context.set_and_render_variables(package.variables, package_name)
+                package_context = project.context.copy()
+                package_context.set_and_render_variables(package.variables, package_name)
                 on_run_start: t.List[str] = [
                     on_run_hook.sql
                     for on_run_hook in sorted(package.on_run_start.values(), key=lambda h: h.index)
@@ -261,7 +259,7 @@ def _load_environment_statements(self, macros: MacroRegistry) -> t.List[Environm
                     for hook in [*package.on_run_start.values(), *package.on_run_end.values()]:
                         dependencies = dependencies.union(hook.dependencies)
 
-                    statements_context = context.context_for_dependencies(dependencies)
+                    statements_context = package_context.context_for_dependencies(dependencies)
                     jinja_registry = make_jinja_registry(
                         statements_context.jinja_macros, package_name, set(dependencies.macros)
                     )

sqlmesh/dbt/model.py

@@ -25,8 +25,7 @@
 )
 from sqlmesh.core.model.kind import SCDType2ByTimeKind, OnDestructiveChange, OnAdditiveChange
 from sqlmesh.dbt.basemodel import BaseModelConfig, Materialization, SnapshotStrategy
-from sqlmesh.dbt.column import ColumnConfig
-from sqlmesh.dbt.common import SqlStr, extract_jinja_config, sql_str_validator, Dependencies
+from sqlmesh.dbt.common import SqlStr, extract_jinja_config, sql_str_validator
 from sqlmesh.utils.errors import ConfigError
 from sqlmesh.utils.pydantic import field_validator
 
@@ -437,30 +436,6 @@ def sqlmesh_config_fields(self) -> t.Set[str]:
             "physical_version",
         }
 
-    def sqlmesh_model_kwargs(
-        self,
-        context: DbtContext,
-        column_types_override: t.Optional[t.Dict[str, ColumnConfig]] = None,
-        extra_dependencies: t.Optional[Dependencies] = None,
-    ) -> t.Dict[str, t.Any]:
-        if not self.dependencies.has_dynamic_var_names:
-            return super().sqlmesh_model_kwargs(context, column_types_override, extra_dependencies)
-
-        extra_dependencies = extra_dependencies or Dependencies()
-        extra_dependencies = extra_dependencies.union(
-            self._track_dependencies_on_render(self.sql_no_config, context)
-        )
-        for pre_hook in self.pre_hook:
-            extra_dependencies = extra_dependencies.union(
-                self._track_dependencies_on_render(pre_hook.sql, context)
-            )
-        for post_hook in self.post_hook:
-            extra_dependencies = extra_dependencies.union(
-                self._track_dependencies_on_render(post_hook.sql, context)
-            )
-
-        return super().sqlmesh_model_kwargs(context, column_types_override, extra_dependencies)
-
     def to_sqlmesh(
         self,
         context: DbtContext,

sqlmesh/dbt/project.py

@@ -55,9 +55,6 @@ def load(cls, context: DbtContext, variables: t.Optional[t.Dict[str, t.Any]] = N
             raise ConfigError(f"Could not find {PROJECT_FILENAME} in {context.project_root}")
         project_yaml = load_yaml(project_file_path)
 
-        variable_overrides = variables
-        variables = {**project_yaml.get("vars", {}), **(variables or {})}
-
         project_name = context.render(project_yaml.get("name", ""))
         context.project_name = project_name
         if not context.project_name:
@@ -69,6 +66,7 @@ def load(cls, context: DbtContext, variables: t.Optional[t.Dict[str, t.Any]] = N
         profile = Profile.load(context, context.target_name)
         context.target = profile.target
 
+        variable_overrides = variables or {}
         context.manifest = ManifestHelper(
             project_file_path.parent,
             profile.path.parent,
@@ -101,13 +99,17 @@ def load(cls, context: DbtContext, variables: t.Optional[t.Dict[str, t.Any]] = N
             package = package_loader.load(path.parent)
             packages[package.name] = package
 
+        all_project_variables = {**project_yaml.get("vars", {}), **(variable_overrides or {})}
         for name, package in packages.items():
-            package_vars = variables.get(name)
+            package_vars = all_project_variables.get(name)
 
             if isinstance(package_vars, dict):
                 package.variables.update(package_vars)
 
-            package.variables.update(variables)
+            if name == context.project_name:
+                package.variables.update(all_project_variables)
+            else:
+                package.variables.update(variable_overrides)
 
         return Project(context, profile, packages)
 

tests/dbt/test_config.py

@@ -376,26 +376,10 @@ def test_variables(assert_exp_eq, sushi_test_project):
     expected_customer_variables = {
         "some_var": ["foo", "bar"],
         "some_other_var": 5,
-        "yet_another_var": 1,
+        "yet_another_var": 5,
         "customers:bla": False,
         "customers:customer_id": "customer_id",
         "start": "Jan 1 2022",
-        "top_waiters:limit": 10,
-        "top_waiters:revenue": "revenue",
-        "customers:boo": ["a", "b"],
-        "nested_vars": {
-            "some_nested_var": 2,
-        },
-        "dynamic_test_var": 3,
-        "list_var": [
-            {"name": "item1", "value": 1},
-            {"name": "item2", "value": 2},
-        ],
-        "customers": {
-            "customers:bla": False,
-            "customers:customer_id": "customer_id",
-            "some_var": ["foo", "bar"],
-        },
     }
 
     assert sushi_test_project.packages["sushi"].variables == expected_sushi_variables
@@ -408,7 +392,9 @@ def test_nested_variables(sushi_test_project):
         sql="SELECT {{ var('nested_vars')['some_nested_var'] }}",
         dependencies=Dependencies(variables=["nested_vars"]),
     )
-    sqlmesh_model = model_config.to_sqlmesh(sushi_test_project.context)
+    context = sushi_test_project.context.copy()
+    context.set_and_render_variables(sushi_test_project.packages["sushi"].variables, "sushi")
+    sqlmesh_model = model_config.to_sqlmesh(context)
     assert sqlmesh_model.jinja_macros.global_objs["vars"]["nested_vars"] == {"some_nested_var": 2}