Merge branch 'main' into state-modified-vars

.changes/unreleased/Features-20240729-173203.yaml

@@ -0,0 +1,7 @@
+kind: Features
+body: Include models that depend on changed vars in state:modified, add state:modified.vars
+  selection method
+time: 2024-07-29T17:32:03.368508-04:00
+custom:
+  Author: michelleark
+  Issue: "4304"

core/dbt/artifacts/resources/v1/components.py

@@ -197,6 +197,7 @@ class ParsedResource(ParsedResourceMandatory):
     unrendered_config_call_dict: Dict[str, Any] = field(default_factory=dict)
     relation_name: Optional[str] = None
     raw_code: str = ""
+    vars: Dict[str, Any] = field(default_factory=dict)
 
     def __post_serialize__(self, dct: Dict, context: Optional[Dict] = None):
         dct = super().__post_serialize__(dct, context)

core/dbt/artifacts/resources/v1/exposure.py

@@ -41,6 +41,7 @@ class Exposure(GraphResource):
     tags: List[str] = field(default_factory=list)
     config: ExposureConfig = field(default_factory=ExposureConfig)
     unrendered_config: Dict[str, Any] = field(default_factory=dict)
+    vars: Dict[str, Any] = field(default_factory=dict)
     url: Optional[str] = None
     depends_on: DependsOn = field(default_factory=DependsOn)
     refs: List[RefArgs] = field(default_factory=list)

core/dbt/artifacts/resources/v1/source_definition.py

@@ -69,6 +69,7 @@ class SourceDefinition(ParsedSourceMandatory):
     config: SourceConfig = field(default_factory=SourceConfig)
     patch_path: Optional[str] = None
     unrendered_config: Dict[str, Any] = field(default_factory=dict)
+    vars: Dict[str, Any] = field(default_factory=dict)
     relation_name: Optional[str] = None
     created_at: float = field(default_factory=lambda: time.time())
     unrendered_database: Optional[str] = None

core/dbt/context/configured.py

@@ -31,23 +31,35 @@ class FQNLookup:
         self.resource_type = NodeType.Model
 
 
+class SchemaYamlVars:
+    def __init__(self):
+        self.env_vars = {}
+        self.vars = {}
+
+
 class ConfiguredVar(Var):
     def __init__(
         self,
         context: Dict[str, Any],
         config: AdapterRequiredConfig,
         project_name: str,
+        schema_yaml_vars: Optional[SchemaYamlVars] = None,
     ):
         super().__init__(context, config.cli_vars)
         self._config = config
         self._project_name = project_name
+        self.schema_yaml_vars = schema_yaml_vars
 
     def __call__(self, var_name, default=Var._VAR_NOTSET):
         my_config = self._config.load_dependencies()[self._project_name]
 
+        var_found = False
+        var_value = None
+
         # cli vars > active project > local project
         if var_name in self._config.cli_vars:
-            return self._config.cli_vars[var_name]
+            var_found = True
+            var_value = self._config.cli_vars[var_name]
 
         adapter_type = self._config.credentials.type
         lookup = FQNLookup(self._project_name)
@@ -58,19 +70,21 @@ class ConfiguredVar(Var):
             all_vars.add(my_config.vars.vars_for(lookup, adapter_type))
         all_vars.add(active_vars)
 
-        if var_name in all_vars:
-            return all_vars[var_name]
+        if not var_found and var_name in all_vars:
+            var_found = True
+            var_value = all_vars[var_name]
 
-        if default is not Var._VAR_NOTSET:
-            return default
+        if not var_found and default is not Var._VAR_NOTSET:
+            var_found = True
+            var_value = default
 
-        return self.get_missing_var(var_name)
+        if not var_found:
+            return self.get_missing_var(var_name)
+        else:
+            if self.schema_yaml_vars:
+                self.schema_yaml_vars.vars[var_name] = var_value
 
-
-class SchemaYamlVars:
-    def __init__(self):
-        self.env_vars = {}
-        self.vars = {}
+            return var_value
 
 
 class SchemaYamlContext(ConfiguredContext):
@@ -82,7 +96,7 @@ class SchemaYamlContext(ConfiguredContext):
 
     @contextproperty()
     def var(self) -> ConfiguredVar:
-        return ConfiguredVar(self._ctx, self.config, self._project_name)
+        return ConfiguredVar(self._ctx, self.config, self._project_name, self.schema_yaml_vars)
 
     @contextmember()
     def env_var(self, var: str, default: Optional[str] = None) -> str:

core/dbt/context/providers.py

@@ -798,6 +798,14 @@ class ParseVar(ModelConfiguredVar):
         # in the parser, just always return None.
         return None
 
+    def __call__(self, var_name: str, default: Any = ModelConfiguredVar._VAR_NOTSET) -> Any:
+        var_value = super().__call__(var_name, default)
+
+        if self._node and hasattr(self._node, "vars"):
+            self._node.vars[var_name] = var_value
+
+        return var_value
+
 
 class RuntimeVar(ModelConfiguredVar):
     pass

core/dbt/contracts/files.py

@@ -216,6 +216,7 @@ class SchemaSourceFile(BaseSourceFile):
     unrendered_configs: Dict[str, Any] = field(default_factory=dict)
     unrendered_databases: Dict[str, Any] = field(default_factory=dict)
     unrendered_schemas: Dict[str, Any] = field(default_factory=dict)
+    vars: Dict[str, Any] = field(default_factory=dict)
     pp_dict: Optional[Dict[str, Any]] = None
     pp_test_index: Optional[Dict[str, Any]] = None
 
@@ -356,6 +357,22 @@ class SchemaSourceFile(BaseSourceFile):
             if not self.unrendered_configs[yaml_key]:
                 del self.unrendered_configs[yaml_key]
 
+    def add_vars(self, vars: Dict[str, Any], yaml_key: str, name: str) -> None:
+        if yaml_key not in self.vars:
+            self.vars[yaml_key] = {}
+
+        if name not in self.vars[yaml_key]:
+            self.vars[yaml_key][name] = vars
+
+    def get_vars(self, yaml_key: str, name: str) -> Dict[str, Any]:
+        if yaml_key not in self.vars:
+            return {}
+
+        if name not in self.vars[yaml_key]:
+            return {}
+
+        return self.vars[yaml_key][name]
+
     def add_env_var(self, var, yaml_key, name):
         if yaml_key not in self.env_vars:
             self.env_vars[yaml_key] = {}

core/dbt/contracts/graph/nodes.py

@@ -369,6 +369,9 @@ class ParsedNode(ParsedResource, NodeInfoMixin, ParsedNodeMandatory, Serializabl
         # This would only apply to seeds
         return True
 
+    def same_vars(self, old) -> bool:
+        return self.vars == old.vars
+
     def same_contents(self, old, adapter_type) -> bool:
         if old is None:
             return False
@@ -376,12 +379,20 @@ class ParsedNode(ParsedResource, NodeInfoMixin, ParsedNodeMandatory, Serializabl
         # Need to ensure that same_contract is called because it
         # could throw an error
         same_contract = self.same_contract(old, adapter_type)
+
+        # Legacy behaviour
+        if not get_flags().state_modified_compare_vars:
+            same_vars = True
+        else:
+            same_vars = self.same_vars(old)
+
         return (
             self.same_body(old)
             and self.same_config(old)
             and self.same_persisted_description(old)
             and self.same_fqn(old)
             and self.same_database_representation(old)
+            and same_vars
             and same_contract
             and True
         )
@@ -1264,6 +1275,9 @@ class SourceDefinition(
             old.unrendered_config,
         )
 
+    def same_vars(self, other: "SourceDefinition") -> bool:
+        return self.vars == other.vars
+
     def same_contents(self, old: Optional["SourceDefinition"]) -> bool:
         # existing when it didn't before is a change!
         if old is None:
@@ -1277,6 +1291,12 @@ class SourceDefinition(
         # freshness changes are changes, I guess
         # metadata/tags changes are not "changes"
         # patching/description changes are not "changes"
+        # Legacy behaviour
+        if not get_flags().state_modified_compare_vars:
+            same_vars = True
+        else:
+            same_vars = self.same_vars(old)
+
         return (
             self.same_database_representation(old)
             and self.same_fqn(old)
@@ -1284,6 +1304,7 @@ class SourceDefinition(
             and self.same_quoting(old)
             and self.same_freshness(old)
             and self.same_external(old)
+            and same_vars
             and True
         )
 
@@ -1380,12 +1401,21 @@ class Exposure(GraphNode, ExposureResource):
             old.unrendered_config,
         )
 
+    def same_vars(self, old: "Exposure") -> bool:
+        return self.vars == old.vars
+
     def same_contents(self, old: Optional["Exposure"]) -> bool:
         # existing when it didn't before is a change!
         # metadata/tags changes are not "changes"
         if old is None:
             return True
 
+        # Legacy behaviour
+        if not get_flags().state_modified_compare_vars:
+            same_vars = True
+        else:
+            same_vars = self.same_vars(old)
+
         return (
             self.same_fqn(old)
             and self.same_exposure_type(old)
@@ -1396,6 +1426,7 @@ class Exposure(GraphNode, ExposureResource):
             and self.same_label(old)
             and self.same_depends_on(old)
             and self.same_config(old)
+            and same_vars
             and True
         )
 
@@ -1647,6 +1678,7 @@ class ParsedNodePatch(ParsedPatch):
     latest_version: Optional[NodeVersion]
     constraints: List[Dict[str, Any]]
     deprecation_date: Optional[datetime]
+    vars: Dict[str, Any]
     time_spine: Optional[TimeSpine] = None
 
 

core/dbt/graph/selector_methods.py

@@ -752,6 +752,7 @@ class StateSelectorMethod(SelectorMethod):
             "modified.relation": self.check_modified_factory("same_database_representation"),
             "modified.macros": self.check_modified_macros,
             "modified.contract": self.check_modified_contract("same_contract", adapter_type),
+            "modified.vars": self.check_modified_factory("same_vars"),
         }
         if selector in state_checks:
             checker = state_checks[selector]

core/dbt/parser/schema_yaml_readers.py

@@ -91,6 +91,9 @@ class ExposureParser(YamlReader):
         unique_id = f"{NodeType.Exposure}.{package_name}.{unparsed.name}"
         path = self.yaml.path.relative_path
 
+        assert isinstance(self.yaml.file, SchemaSourceFile)
+        exposure_vars = self.yaml.file.get_vars(self.key, unparsed.name)
+
         fqn = self.schema_parser.get_fqn_prefix(path)
         fqn.append(unparsed.name)
 
@@ -133,6 +136,7 @@ class ExposureParser(YamlReader):
             maturity=unparsed.maturity,
             config=config,
             unrendered_config=unrendered_config,
+            vars=exposure_vars,
         )
         ctx = generate_parse_exposure(
             parsed,
@@ -144,7 +148,6 @@ class ExposureParser(YamlReader):
         get_rendered(depends_on_jinja, ctx, parsed, capture_macros=True)
         # parsed now has a populated refs/sources/metrics
 
-        assert isinstance(self.yaml.file, SchemaSourceFile)
         if parsed.config.enabled:
             self.manifest.add_exposure(self.yaml.file, parsed)
         else:

core/dbt/parser/schemas.py

@@ -432,10 +432,14 @@ class YamlReader(metaclass=ABCMeta):
 
             if self.schema_yaml_vars.env_vars:
                 self.schema_parser.manifest.env_vars.update(self.schema_yaml_vars.env_vars)
-                for var in self.schema_yaml_vars.env_vars.keys():
-                    schema_file.add_env_var(var, self.key, entry["name"])
+                for env_var in self.schema_yaml_vars.env_vars.keys():
+                    schema_file.add_env_var(env_var, self.key, entry["name"])
                 self.schema_yaml_vars.env_vars = {}
 
+            if self.schema_yaml_vars.vars:
+                schema_file.add_vars(self.schema_yaml_vars.vars, self.key, entry["name"])
+                self.schema_yaml_vars.vars = {}
+
             yield entry
 
     def render_entry(self, dct):
@@ -715,6 +719,9 @@ class NodePatchParser(PatchParser[NodeTarget, ParsedNodePatch], Generic[NodeTarg
         # code consistency.
         deprecation_date: Optional[datetime.datetime] = None
         time_spine: Optional[TimeSpine] = None
+        assert isinstance(self.yaml.file, SchemaSourceFile)
+        source_file: SchemaSourceFile = self.yaml.file
+
         if isinstance(block.target, UnparsedModelUpdate):
             deprecation_date = block.target.deprecation_date
             time_spine = (
@@ -747,9 +754,9 @@ class NodePatchParser(PatchParser[NodeTarget, ParsedNodePatch], Generic[NodeTarg
             constraints=block.target.constraints,
             deprecation_date=deprecation_date,
             time_spine=time_spine,
+            vars=source_file.get_vars(block.target.yaml_key, block.target.name),
         )
-        assert isinstance(self.yaml.file, SchemaSourceFile)
-        source_file: SchemaSourceFile = self.yaml.file
+
         if patch.yaml_key in ["models", "seeds", "snapshots"]:
             unique_id = self.manifest.ref_lookup.get_unique_id(
                 patch.name, self.project.project_name, None
@@ -843,6 +850,8 @@ class NodePatchParser(PatchParser[NodeTarget, ParsedNodePatch], Generic[NodeTarg
         node.description = patch.description
         node.columns = patch.columns
         node.name = patch.name
+        # Prefer node-level vars to vars from patch
+        node.vars = {**patch.vars, **node.vars}
 
         if not isinstance(node, ModelNode):
             for attr in ["latest_version", "access", "version", "constraints"]:
@@ -992,6 +1001,7 @@ class ModelPatchParser(NodePatchParser[UnparsedModelUpdate]):
                     latest_version=latest_version,
                     constraints=unparsed_version.constraints or target.constraints,
                     deprecation_date=unparsed_version.deprecation_date,
+                    vars=source_file.get_vars(block.target.yaml_key, block.target.name),
                 )
                 # Node patched before config because config patching depends on model name,
                 # which may have been updated in the version patch

core/dbt/parser/sources.py

@@ -12,6 +12,7 @@ from dbt.context.context_config import (
     ContextConfigGenerator,
     UnrenderedConfigGenerator,
 )
+from dbt.contracts.files import SchemaSourceFile
 from dbt.contracts.graph.manifest import Manifest, SourceKey
 from dbt.contracts.graph.nodes import (
     GenericTestNode,
@@ -158,6 +159,10 @@ class SourcePatcher:
             rendered=False,
         )
 
+        schema_file = self.manifest.files[target.file_id]
+        assert isinstance(schema_file, SchemaSourceFile)
+        source_vars = schema_file.get_vars("sources", source.name)
+
         if not isinstance(config, SourceConfig):
             raise DbtInternalError(
                 f"Calculated a {type(config)} for a source, but expected a SourceConfig"
@@ -192,6 +197,7 @@ class SourcePatcher:
             tags=tags,
             config=config,
             unrendered_config=unrendered_config,
+            vars=source_vars,
         )
 
         if (

schemas/dbt/manifest/v12.json

@@ -727,6 +727,12 @@
                 "type": "string",
                 "default": ""
               },
+              "vars": {
+                "type": "object",
+                "propertyNames": {
+                  "type": "string"
+                }
+              },
               "root_path": {
                 "anyOf": [
                   {
@@ -1766,6 +1772,12 @@
                 "type": "string",
                 "default": ""
               },
+              "vars": {
+                "type": "object",
+                "propertyNames": {
+                  "type": "string"
+                }
+              },
               "language": {
                 "type": "string",
                 "default": "sql"
@@ -2420,6 +2432,12 @@
                 "type": "string",
                 "default": ""
               },
+              "vars": {
+                "type": "object",
+                "propertyNames": {
+                  "type": "string"
+                }
+              },
               "language": {
                 "type": "string",
                 "default": "sql"
@@ -3211,6 +3229,12 @@
                 "type": "string",
                 "default": ""
               },
+              "vars": {
+                "type": "object",
+                "propertyNames": {
+                  "type": "string"
+                }
+              },
               "language": {
                 "type": "string",
                 "default": "sql"
@@ -4021,6 +4045,12 @@
                 "type": "string",
                 "default": ""
               },
+              "vars": {
+                "type": "object",
+                "propertyNames": {
+                  "type": "string"
+                }
+              },
               "language": {
                 "type": "string",
                 "default": "sql"
@@ -5382,6 +5412,12 @@
                 "type": "string",
                 "default": ""
               },
+              "vars": {
+                "type": "object",
+                "propertyNames": {
+                  "type": "string"
+                }
+              },
               "language": {
                 "type": "string",
                 "default": "sql"
@@ -6036,6 +6072,12 @@
                 "type": "string",
                 "default": ""
               },
+              "vars": {
+                "type": "object",
+                "propertyNames": {
+                  "type": "string"
+                }
+              },
               "language": {
                 "type": "string",
                 "default": "sql"
@@ -7011,6 +7053,12 @@
                 "type": "string",
                 "default": ""
               },
+              "vars": {
+                "type": "object",
+                "propertyNames": {
+                  "type": "string"
+                }
+              },
               "language": {
                 "type": "string",
                 "default": "sql"
@@ -8148,6 +8196,12 @@
               "type": "string"
             }
           },
+          "vars": {
+            "type": "object",
+            "propertyNames": {
+              "type": "string"
+            }
+          },
           "relation_name": {
             "anyOf": [
               {
@@ -8541,6 +8595,12 @@
               "type": "string"
             }
           },
+          "vars": {
+            "type": "object",
+            "propertyNames": {
+              "type": "string"
+            }
+          },
           "url": {
             "anyOf": [
               {
@@ -10618,6 +10678,12 @@
                       "type": "string",
                       "default": ""
                     },
+                    "vars": {
+                      "type": "object",
+                      "propertyNames": {
+                        "type": "string"
+                      }
+                    },
                     "root_path": {
                       "anyOf": [
                         {
@@ -11657,6 +11723,12 @@
                       "type": "string",
                       "default": ""
                     },
+                    "vars": {
+                      "type": "object",
+                      "propertyNames": {
+                        "type": "string"
+                      }
+                    },
                     "language": {
                       "type": "string",
                       "default": "sql"
@@ -12311,6 +12383,12 @@
                       "type": "string",
                       "default": ""
                     },
+                    "vars": {
+                      "type": "object",
+                      "propertyNames": {
+                        "type": "string"
+                      }
+                    },
                     "language": {
                       "type": "string",
                       "default": "sql"
@@ -13102,6 +13180,12 @@
                       "type": "string",
                       "default": ""
                     },
+                    "vars": {
+                      "type": "object",
+                      "propertyNames": {
+                        "type": "string"
+                      }
+                    },
                     "language": {
                       "type": "string",
                       "default": "sql"
@@ -13912,6 +13996,12 @@
                       "type": "string",
                       "default": ""
                     },
+                    "vars": {
+                      "type": "object",
+                      "propertyNames": {
+                        "type": "string"
+                      }
+                    },
                     "language": {
                       "type": "string",
                       "default": "sql"
@@ -15273,6 +15363,12 @@
                       "type": "string",
                       "default": ""
                     },
+                    "vars": {
+                      "type": "object",
+                      "propertyNames": {
+                        "type": "string"
+                      }
+                    },
                     "language": {
                       "type": "string",
                       "default": "sql"
@@ -15927,6 +16023,12 @@
                       "type": "string",
                       "default": ""
                     },
+                    "vars": {
+                      "type": "object",
+                      "propertyNames": {
+                        "type": "string"
+                      }
+                    },
                     "language": {
                       "type": "string",
                       "default": "sql"
@@ -16902,6 +17004,12 @@
                       "type": "string",
                       "default": ""
                     },
+                    "vars": {
+                      "type": "object",
+                      "propertyNames": {
+                        "type": "string"
+                      }
+                    },
                     "language": {
                       "type": "string",
                       "default": "sql"
@@ -18030,6 +18138,12 @@
                         "type": "string"
                       }
                     },
+                    "vars": {
+                      "type": "object",
+                      "propertyNames": {
+                        "type": "string"
+                      }
+                    },
                     "relation_name": {
                       "anyOf": [
                         {
@@ -18221,6 +18335,12 @@
                         "type": "string"
                       }
                     },
+                    "vars": {
+                      "type": "object",
+                      "propertyNames": {
+                        "type": "string"
+                      }
+                    },
                     "url": {
                       "anyOf": [
                         {

tests/functional/artifacts/expected_manifest.py

@@ -364,6 +364,7 @@ def expected_seeded_manifest(project, model_database=None, quote_model=False):
                 "extra_ctes": [],
                 "checksum": checksum_file(model_sql_path),
                 "unrendered_config": unrendered_model_config,
+                "vars": {},
                 "access": "protected",
                 "version": None,
                 "latest_version": None,
@@ -463,6 +464,7 @@ def expected_seeded_manifest(project, model_database=None, quote_model=False):
                 "extra_ctes": [],
                 "checksum": checksum_file(second_model_sql_path),
                 "unrendered_config": unrendered_second_config,
+                "vars": {},
                 "access": "protected",
                 "version": None,
                 "latest_version": None,
@@ -545,6 +547,7 @@ def expected_seeded_manifest(project, model_database=None, quote_model=False):
                 "docs": {"node_color": None, "show": True},
                 "checksum": checksum_file(seed_path),
                 "unrendered_config": unrendered_seed_config,
+                "vars": {},
                 "relation_name": relation_name_node_format.format(
                     project.database, my_schema_name, "seed"
                 ),
@@ -600,6 +603,7 @@ def expected_seeded_manifest(project, model_database=None, quote_model=False):
                 },
                 "checksum": {"name": "none", "checksum": ""},
                 "unrendered_config": unrendered_test_config,
+                "vars": {},
                 "contract": {"checksum": None, "enforced": False, "alias_types": True},
             },
             "snapshot.test.snapshot_seed": {
@@ -647,6 +651,7 @@ def expected_seeded_manifest(project, model_database=None, quote_model=False):
                 "tags": [],
                 "unique_id": "snapshot.test.snapshot_seed",
                 "unrendered_config": unrendered_snapshot_config,
+                "vars": {"alternate_schema": alternate_schema},
             },
             "test.test.test_nothing_model_.5d38568946": {
                 "alias": "test_nothing_model_",
@@ -699,6 +704,7 @@ def expected_seeded_manifest(project, model_database=None, quote_model=False):
                 },
                 "checksum": {"name": "none", "checksum": ""},
                 "unrendered_config": unrendered_test_config,
+                "vars": {},
             },
             "test.test.unique_model_id.67b76558ff": {
                 "alias": "unique_model_id",
@@ -752,6 +758,7 @@ def expected_seeded_manifest(project, model_database=None, quote_model=False):
                 },
                 "checksum": {"name": "none", "checksum": ""},
                 "unrendered_config": unrendered_test_config,
+                "vars": {},
             },
         },
         "sources": {
@@ -810,6 +817,7 @@ def expected_seeded_manifest(project, model_database=None, quote_model=False):
                 "unrendered_config": {},
                 "unrendered_database": None,
                 "unrendered_schema": "{{ var('test_schema') }}",
+                "vars": {"test_schema": ANY},
             },
         },
         "exposures": {
@@ -844,6 +852,7 @@ def expected_seeded_manifest(project, model_database=None, quote_model=False):
                 "unique_id": "exposure.test.notebook_exposure",
                 "url": "http://example.com/notebook/1",
                 "unrendered_config": {},
+                "vars": {},
             },
             "exposure.test.simple_exposure": {
                 "created_at": ANY,
@@ -876,6 +885,7 @@ def expected_seeded_manifest(project, model_database=None, quote_model=False):
                 "meta": {},
                 "tags": [],
                 "unrendered_config": {},
+                "vars": {},
             },
         },
         "metrics": {},
@@ -993,6 +1003,7 @@ def expected_references_manifest(project):
                 "extra_ctes": [],
                 "checksum": checksum_file(ephemeral_copy_path),
                 "unrendered_config": get_unrendered_model_config(materialized="ephemeral"),
+                "vars": {},
                 "access": "protected",
                 "version": None,
                 "latest_version": None,
@@ -1065,6 +1076,7 @@ def expected_references_manifest(project):
                 "unrendered_config": get_unrendered_model_config(
                     materialized="table", group="test_group"
                 ),
+                "vars": {},
                 "access": "protected",
                 "version": None,
                 "latest_version": None,
@@ -1133,6 +1145,7 @@ def expected_references_manifest(project):
                 "extra_ctes": [],
                 "checksum": checksum_file(view_summary_path),
                 "unrendered_config": get_unrendered_model_config(materialized="view"),
+                "vars": {},
                 "access": "protected",
                 "version": None,
                 "latest_version": None,
@@ -1216,6 +1229,7 @@ def expected_references_manifest(project):
                 "unique_id": "seed.test.seed",
                 "checksum": checksum_file(seed_path),
                 "unrendered_config": get_unrendered_seed_config(),
+                "vars": {},
                 "relation_name": '"{0}"."{1}".seed'.format(project.database, my_schema_name),
             },
             "snapshot.test.snapshot_seed": {
@@ -1230,7 +1244,10 @@ def expected_references_manifest(project):
                 "config": get_rendered_snapshot_config(target_schema=alternate_schema),
                 "contract": {"checksum": None, "enforced": False, "alias_types": True},
                 "database": model_database,
-                "depends_on": {"macros": [], "nodes": ["seed.test.seed"]},
+                "depends_on": {
+                    "macros": [],
+                    "nodes": ["seed.test.seed"],
+                },
                 "description": "",
                 "docs": {"node_color": None, "show": True},
                 "extra_ctes": [],
@@ -1258,6 +1275,7 @@ def expected_references_manifest(project):
                 "unrendered_config": get_unrendered_snapshot_config(
                     target_schema=alternate_schema
                 ),
+                "vars": {"alternate_schema": alternate_schema},
             },
         },
         "sources": {
@@ -1314,6 +1332,7 @@ def expected_references_manifest(project):
                 "unrendered_config": {},
                 "unrendered_database": None,
                 "unrendered_schema": "{{ var('test_schema') }}",
+                "vars": {"test_schema": ANY},
             },
         },
         "exposures": {
@@ -1339,6 +1358,7 @@ def expected_references_manifest(project):
                 "package_name": "test",
                 "path": "schema.yml",
                 "refs": [{"name": "view_summary", "package": None, "version": None}],
+                "vars": {},
                 "resource_type": "exposure",
                 "sources": [],
                 "type": "notebook",
@@ -1599,6 +1619,7 @@ def expected_versions_manifest(project):
                     group="test_group",
                     meta={"size": "large", "color": "blue"},
                 ),
+                "vars": {},
                 "access": "protected",
                 "version": 1,
                 "latest_version": 2,
@@ -1670,6 +1691,7 @@ def expected_versions_manifest(project):
                 "unrendered_config": get_unrendered_model_config(
                     materialized="view", group="test_group", meta={"size": "large", "color": "red"}
                 ),
+                "vars": {},
                 "access": "protected",
                 "version": 2,
                 "latest_version": 2,
@@ -1728,6 +1750,7 @@ def expected_versions_manifest(project):
                 "extra_ctes": [],
                 "checksum": checksum_file(ref_versioned_model_path),
                 "unrendered_config": get_unrendered_model_config(),
+                "vars": {},
                 "access": "protected",
                 "version": None,
                 "latest_version": None,
@@ -1785,6 +1808,7 @@ def expected_versions_manifest(project):
                 },
                 "checksum": {"name": "none", "checksum": ""},
                 "unrendered_config": unrendered_test_config,
+                "vars": {},
             },
             "test.test.unique_versioned_model_v1_count.0b4c0b688a": {
                 "alias": "unique_versioned_model_v1_count",
@@ -1838,6 +1862,7 @@ def expected_versions_manifest(project):
                 },
                 "checksum": {"name": "none", "checksum": ""},
                 "unrendered_config": unrendered_test_config,
+                "vars": {},
             },
             "test.test.unique_versioned_model_v2_first_name.998430d28e": {
                 "alias": "unique_versioned_model_v2_first_name",
@@ -1891,6 +1916,7 @@ def expected_versions_manifest(project):
                 },
                 "checksum": {"name": "none", "checksum": ""},
                 "unrendered_config": unrendered_test_config,
+                "vars": {},
             },
         },
         "exposures": {
@@ -1922,6 +1948,7 @@ def expected_versions_manifest(project):
                 "unique_id": "exposure.test.notebook_exposure",
                 "url": None,
                 "unrendered_config": {},
+                "vars": {},
             },
         },
         "metrics": {},

tests/functional/defer_state/test_modified_state.py

@@ -1144,3 +1144,216 @@ class TestChangedSemanticModelContents(BaseModifiedState):
         write_file(modified_semantic_model_schema_yml, "models", "schema.yml")
         results = run_dbt(["list", "-s", "state:modified", "--state", "./state"])
         assert len(results) == 1
+
+
+class TestModifiedVarsLegacy(BaseModifiedState):
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "flags": {
+                "state_modified_compare_vars": False,
+            },
+            "vars": {"my_var": 1},
+        }
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "model_with_var.sql": "select {{ var('my_var') }} as id",
+        }
+
+    def test_changed_vars(self, project):
+        self.run_and_save_state()
+
+        # No var change
+        assert not run_dbt(["list", "-s", "state:modified", "--state", "./state"])
+        assert not run_dbt(["list", "-s", "state:modified.vars", "--state", "./state"])
+
+        # Modify var (my_var: 1 -> 2)
+        update_config_file({"vars": {"my_var": 2}}, "dbt_project.yml")
+
+        # By default, do not detect vars in state:modified to preserve legacy behaviour
+        assert run_dbt(["list", "-s", "state:modified", "--state", "./state"]) == []
+
+        # state:modified.vars is a new selector, opt-in method -> returns results
+        assert run_dbt(["list", "-s", "state:modified.vars", "--state", "./state"]) == [
+            "test.model_with_var"
+        ]
+
+        # Reset dbt_project.yml
+        update_config_file({"vars": {"my_var": 1}}, "dbt_project.yml")
+
+        # Modify var via --var CLI flag
+        assert not run_dbt(
+            ["list", "--vars", '{"my_var": 1}', "-s", "state:modified", "--state", "./state"]
+        )
+        assert (
+            run_dbt(
+                ["list", "--vars", '{"my_var": 2}', "-s", "state:modified", "--state", "./state"]
+            )
+            == []
+        )
+        # state:modified.vars is a new selector, opt-in method -> returns results
+        assert run_dbt(
+            ["list", "--vars", '{"my_var": 2}', "-s", "state:modified.vars", "--state", "./state"]
+        ) == ["test.model_with_var"]
+
+
+class TestModifiedVars(BaseModifiedState):
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "flags": {
+                "state_modified_compare_vars": True,
+            },
+            "vars": {"my_var": 1},
+        }
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "model_with_var.sql": "select {{ var('my_var') }} as id",
+        }
+
+    def test_changed_vars(self, project):
+        self.run_and_save_state()
+
+        # No var change
+        assert not run_dbt(["list", "-s", "state:modified", "--state", "./state"])
+        assert not run_dbt(["list", "-s", "state:modified.vars", "--state", "./state"])
+
+        # Modify var (my_var: 1 -> 2)
+        update_config_file({"vars": {"my_var": 2}}, "dbt_project.yml")
+        assert run_dbt(["list", "-s", "state:modified", "--state", "./state"]) == [
+            "test.model_with_var"
+        ]
+        assert run_dbt(["list", "-s", "state:modified.vars", "--state", "./state"]) == [
+            "test.model_with_var"
+        ]
+
+        # Reset dbt_project.yml
+        update_config_file({"vars": {"my_var": 1}}, "dbt_project.yml")
+
+        # Modify var via --var CLI flag
+        assert not run_dbt(
+            ["list", "--vars", '{"my_var": 1}', "-s", "state:modified", "--state", "./state"]
+        )
+        assert run_dbt(
+            ["list", "--vars", '{"my_var": 2}', "-s", "state:modified", "--state", "./state"]
+        ) == ["test.model_with_var"]
+
+
+macro_with_var_sql = """
+{% macro macro_with_var() %}
+    {{ var('my_var') }}
+{% endmacro %}
+"""
+
+
+class TestModifiedMacroVars(BaseModifiedState):
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "flags": {
+                "state_modified_compare_vars": True,
+            },
+            "vars": {"my_var": 1},
+        }
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {
+            "model_with_macro.sql": "select {{ macro_with_var() }} as id",
+        }
+
+    @pytest.fixture(scope="class")
+    def macros(self):
+        return {
+            "macro_with_var.sql": macro_with_var_sql,
+        }
+
+    def test_changed_vars(self, project):
+        self.run_and_save_state()
+
+        # No var change
+        assert not run_dbt(["list", "-s", "state:modified", "--state", "./state"])
+        assert not run_dbt(["list", "-s", "state:modified.vars", "--state", "./state"])
+
+        # Modify var (my_var: 1 -> 2)
+        update_config_file({"vars": {"my_var": 2}}, "dbt_project.yml")
+        assert run_dbt(["list", "-s", "state:modified", "--state", "./state"]) == [
+            "test.model_with_macro"
+        ]
+        assert run_dbt(["list", "-s", "state:modified.vars", "--state", "./state"]) == [
+            "test.model_with_macro"
+        ]
+
+        # Macros themselves not captured as modified because the var value depends on a node's context
+        assert not run_dbt(["list", "-s", "state:modified.macros", "--state", "./state"])
+
+
+# TODO: test versioned models, tests
+model_with_var_schema_yml = """
+version: 2
+models:
+  - name: model_with_var
+    config:
+        materialized: "{{ var('my_var') }}"
+
+exposures:
+  - name: exposure_name
+    type: dashboard
+    owner:
+      name: "{{ var('my_var') }}"
+
+sources:
+  - name: jaffle_shop
+    database: "{{ var('my_var') }}"
+    schema: jaffle_shop
+    tables:
+      - name: orders
+      - name: customers
+"""
+
+
+class TestModifiedVarsSchemaYml(BaseModifiedState):
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "flags": {
+                "state_modified_compare_vars": True,
+            },
+            "vars": {"my_var": "table"},
+        }
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"model_with_var.sql": "select 1 as id", "schema.yml": model_with_var_schema_yml}
+
+    def test_changed_vars(self, project):
+        self.run_and_save_state()
+
+        # No var change
+        assert not run_dbt(["list", "-s", "state:modified", "--state", "./state"])
+        assert not run_dbt(["list", "-s", "state:modified.vars", "--state", "./state"])
+
+        # Modify var (my_var: table -> view)
+        update_config_file({"vars": {"my_var": "view"}}, "dbt_project.yml")
+        assert sorted(run_dbt(["list", "-s", "state:modified", "--state", "./state"])) == sorted(
+            [
+                "test.model_with_var",
+                "exposure:test.exposure_name",
+                "source:test.jaffle_shop.customers",
+                "source:test.jaffle_shop.orders",
+            ]
+        )
+        assert sorted(
+            run_dbt(["list", "-s", "state:modified.vars", "--state", "./state"])
+        ) == sorted(
+            [
+                "test.model_with_var",
+                "exposure:test.exposure_name",
+                "source:test.jaffle_shop.customers",
+                "source:test.jaffle_shop.orders",
+            ]
+        )

tests/functional/list/test_list.py

@@ -49,7 +49,10 @@ class TestList:
             "json": {
                 "name": "my_snapshot",
                 "package_name": "test",
-                "depends_on": {"nodes": [], "macros": []},
+                "depends_on": {
+                    "nodes": [],
+                    "macros": [],
+                },
                 "tags": [],
                 "config": {
                     "enabled": True,

tests/unit/contracts/graph/test_manifest.py

@@ -97,6 +97,7 @@ REQUIRED_PARSED_NODE_KEYS = frozenset(
         "defer_relation",
         "time_spine",
         "batch_info",
+        "vars",
     }
 )
 

tests/unit/contracts/graph/test_nodes.py

@@ -206,6 +206,7 @@ def basic_compiled_dict():
         },
         "unrendered_config": {},
         "unrendered_config_call_dict": {},
+        "vars": {},
         "config_call_dict": {},
         "access": "protected",
         "constraints": [],
@@ -526,6 +527,7 @@ def basic_compiled_schema_test_dict():
             "severity": "warn",
         },
         "unrendered_config_call_dict": {},
+        "vars": {},
         "config_call_dict": {},
     }
 

tests/unit/contracts/graph/test_nodes_parsed.py

@@ -204,6 +204,7 @@ def base_parsed_model_dict():
         },
         "unrendered_config": {},
         "unrendered_config_call_dict": {},
+        "vars": {},
         "config_call_dict": {},
         "access": AccessType.Protected.value,
         "constraints": [],
@@ -260,6 +261,7 @@ def minimal_parsed_model_dict():
             "checksum": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
         },
         "unrendered_config": {},
+        "vars": {},
     }
 
 
@@ -327,6 +329,7 @@ def complex_parsed_model_dict():
             "post_hook": ['insert into blah(a, b) select "1", 1'],
         },
         "unrendered_config_call_dict": {},
+        "vars": {},
         "config_call_dict": {},
         "access": AccessType.Protected.value,
         "constraints": [],
@@ -537,6 +540,7 @@ def basic_parsed_seed_dict():
         "checksum": {"name": "path", "checksum": "seeds/seed.csv"},
         "unrendered_config": {},
         "unrendered_config_call_dict": {},
+        "vars": {},
         "config_call_dict": {},
     }
 
@@ -643,6 +647,7 @@ def complex_parsed_seed_dict():
             "persist_docs": {"relation": True, "columns": True},
         },
         "unrendered_config_call_dict": {},
+        "vars": {},
         "config_call_dict": {},
     }
 
@@ -843,6 +848,7 @@ def base_parsed_hook_dict():
         },
         "unrendered_config": {},
         "unrendered_config_call_dict": {},
+        "vars": {},
         "config_call_dict": {},
     }
 
@@ -937,6 +943,7 @@ def complex_parsed_hook_dict():
             "materialized": "table",
         },
         "unrendered_config_call_dict": {},
+        "vars": {},
         "config_call_dict": {},
     }
 
@@ -1082,6 +1089,7 @@ def basic_parsed_schema_test_dict():
         },
         "unrendered_config": {},
         "unrendered_config_call_dict": {},
+        "vars": {},
         "config_call_dict": {},
     }
 
@@ -1171,6 +1179,7 @@ def complex_parsed_schema_test_dict():
         },
         "unrendered_config": {"materialized": "table", "severity": "WARN"},
         "unrendered_config_call_dict": {},
+        "vars": {},
         "config_call_dict": {},
     }
 
@@ -1566,6 +1575,7 @@ def basic_timestamp_snapshot_dict():
             "target_schema": "some_snapshot_schema",
         },
         "unrendered_config_call_dict": {},
+        "vars": {},
         "config_call_dict": {},
     }
 
@@ -1672,6 +1682,7 @@ def basic_check_snapshot_dict():
         },
         "unrendered_config_call_dict": {},
         "config_call_dict": {},
+        "vars": {},
     }
 
 
@@ -1881,6 +1892,7 @@ def basic_parsed_source_definition_dict():
             "enabled": True,
         },
         "unrendered_config": {},
+        "vars": {},
     }
 
 
@@ -1913,6 +1925,7 @@ def complex_parsed_source_definition_dict():
         "freshness": {"warn_after": {"period": "hour", "count": 1}, "error_after": {}},
         "loaded_at_field": "loaded_at",
         "unrendered_config": {},
+        "vars": {},
     }
 
 
@@ -2084,6 +2097,7 @@ def basic_parsed_exposure_dict():
             "enabled": True,
         },
         "unrendered_config": {},
+        "vars": {},
     }
 
 
@@ -2139,6 +2153,7 @@ def complex_parsed_exposure_dict():
             "enabled": True,
         },
         "unrendered_config": {},
+        "vars": {},
     }
 
 

tests/unit/parser/test_parser.py

@@ -464,6 +464,7 @@ class SchemaParserSourceTest(SchemaParserTest):
     @mock.patch("dbt.parser.sources.get_adapter")
     def test__parse_basic_source_meta(self, mock_get_adapter):
         block = self.file_block_for(MULTIPLE_TABLE_SOURCE_META, "test_one.yml")
+        self.parser.manifest.files[block.file.file_id] = block.file
         dct = yaml_from_file(block.file)
         self.parser.parse_file(block, dct)
         self.assert_has_manifest_lengths(self.parser.manifest, sources=2)