Merge branch 'main' into state-modified-source-unrendered-database

.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"

.changes/unreleased/Features-20240918-162959.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Create 'skip_nodes_if_on_run_start_fails' behavior change flag
+time: 2024-09-18T16:29:59.268422+01:00
+custom:
+    Author: aranke
+    Issue: "7387"

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 @@ def __init__(self, package_name: str):
         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 @@ def __call__(self, var_name, default=Var._VAR_NOTSET):
             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
-
-        return self.get_missing_var(var_name)
+        if not var_found and default is not Var._VAR_NOTSET:
+            var_found = True
+            var_value = default
 
+        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 @@ def __init__(self, config, project_name: str, schema_yaml_vars: Optional[SchemaY
 
     @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

@@ -790,6 +790,14 @@ def get_missing_var(self, var_name):
         # 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

@@ -215,6 +215,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
 
@@ -355,6 +356,22 @@ def delete_from_unrendered_configs(self, yaml_key, name):
             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,19 +369,30 @@ def same_contract(self, old, adapter_type=None) -> bool:
         # 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
 
         # 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
         )
@@ -1255,6 +1266,9 @@ def same_config(self, old: "SourceDefinition") -> bool:
             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:
@@ -1268,13 +1282,20 @@ def same_contents(self, old: Optional["SourceDefinition"]) -> bool:
         # 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)
             and self.same_config(old)
             and self.same_quoting(old)
             and self.same_freshness(old)
             and self.same_external(old)
+            and same_vars
             and True
         )
 
@@ -1371,12 +1392,21 @@ def same_config(self, old: "Exposure") -> bool:
             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)
@@ -1387,6 +1417,7 @@ def same_contents(self, old: Optional["Exposure"]) -> bool:
             and self.same_label(old)
             and self.same_depends_on(old)
             and self.same_config(old)
+            and same_vars
             and True
         )
 
@@ -1638,6 +1669,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
 
 
@@ -1673,6 +1705,7 @@ class ParsedMacroPatch(ParsedPatch):
 ResultNode = Union[
     ManifestNode,
     SourceDefinition,
+    HookNode,
 ]
 
 # All nodes that can be in the DAG

core/dbt/contracts/project.py

@@ -341,15 +341,19 @@ class ProjectFlags(ExtensibleDbtClassMixin):
     require_explicit_package_overrides_for_builtin_materializations: bool = True
     require_resource_names_without_spaces: bool = False
     source_freshness_run_project_hooks: bool = False
+    skip_nodes_if_on_run_start_fails: bool = False
     state_modified_compare_more_unrendered_values: bool = False
+    state_modified_compare_vars: bool = False
 
     @property
     def project_only_flags(self) -> Dict[str, Any]:
         return {
             "require_explicit_package_overrides_for_builtin_materializations": self.require_explicit_package_overrides_for_builtin_materializations,
             "require_resource_names_without_spaces": self.require_resource_names_without_spaces,
             "source_freshness_run_project_hooks": self.source_freshness_run_project_hooks,
+            "skip_nodes_if_on_run_start_fails": self.skip_nodes_if_on_run_start_fails,
             "state_modified_compare_more_unrendered_values": self.state_modified_compare_more_unrendered_values,
+            "state_modified_compare_vars": self.state_modified_compare_vars,
         }
 
 

core/dbt/events/types.py

@@ -1516,10 +1516,20 @@ def code(self) -> str:
         return "Q033"
 
     def message(self) -> str:
-        msg = f"OK hook: {self.statement}"
+        if self.status == "success":
+            info = "OK"
+            status = green(info)
+        elif self.status == "skipped":
+            info = "SKIP"
+            status = yellow(info)
+        else:
+            info = "ERROR"
+            status = red(info)
+        msg = f"{info} hook: {self.statement}"
+
         return format_fancy_output_line(
             msg=msg,
-            status=green(self.status),
+            status=status,
             index=self.index,
             total=self.total,
             execution_time=self.execution_time,

core/dbt/graph/selector_methods.py

@@ -752,6 +752,7 @@ def search(self, included_nodes: Set[UniqueId], selector: str) -> Iterator[Uniqu
             "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 @@ def parse_exposure(self, unparsed: UnparsedExposure) -> None:
         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 @@ def parse_exposure(self, unparsed: UnparsedExposure) -> None:
             maturity=unparsed.maturity,
             config=config,
             unrendered_config=unrendered_config,
+            vars=exposure_vars,
         )
         ctx = generate_parse_exposure(
             parsed,
@@ -144,7 +148,6 @@ def parse_exposure(self, unparsed: UnparsedExposure) -> None:
         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: