add flag --no-skip-on-failture for command build, run, retry, seed

.changes/unreleased/Features-20231107-132842.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: add flag --no-skip-on-failture
+time: 2023-11-07T13:28:42.420727773+01:00
+custom:
+  Author: leo-schick
+  Issue: "2142"

core/dbt/cli/main.py

@@ -186,6 +186,7 @@ def cli(ctx, **kwargs):
 @p.full_refresh
 @p.include_saved_query
 @p.indirect_selection
+@p.no_skip_on_failture
 @p.profile
 @p.profiles_dir
 @p.project_dir
@@ -588,6 +589,7 @@ def parse(ctx, **kwargs):
 @p.deprecated_favor_state
 @p.exclude
 @p.full_refresh
+@p.no_skip_on_failture
 @p.profile
 @p.profiles_dir
 @p.project_dir
@@ -627,6 +629,7 @@ def run(ctx, **kwargs):
 @p.project_dir
 @p.profiles_dir
 @p.vars
+@p.no_skip_on_failture
 @p.profile
 @p.target
 @p.state
@@ -725,6 +728,7 @@ def run_operation(ctx, **kwargs):
 @global_flags
 @p.exclude
 @p.full_refresh
+@p.no_skip_on_failture
 @p.profile
 @p.profiles_dir
 @p.project_dir

core/dbt/cli/params.py

@@ -112,6 +112,13 @@
     help="Stop execution on first failure.",
 )
 
+no_skip_on_failture = click.option(
+    "--no-skip-on-failture",
+    envvar="DBT_NO_SKIP_ON_FAILTURE",
+    help="If specified, dbt will proceed with downstream models even the dependent model failed.",
+    is_flag=True,
+)
+
 favor_state = click.option(
     "--favor-state/--no-favor-state",
     envvar="DBT_FAVOR_STATE",

core/dbt/flags.py

@@ -75,6 +75,7 @@ def get_flag_dict():
         "log_format",
         "version_check",
         "fail_fast",
+        "no_skip_on_failture",
         "send_anonymous_usage_stats",
         "printer_width",
         "indirect_selection",

core/dbt/task/runnable.py

@@ -391,8 +391,10 @@ def _mark_dependent_errors(
     ) -> None:
         if self.graph is None:
             raise DbtInternalError("graph is None in _mark_dependent_errors")
-        for dep_node_id in self.graph.get_dependent_nodes(UniqueId(node_id)):
-            self._skipped_children[dep_node_id] = cause
+        no_skip_on_failture = get_flags().NO_SKIP_ON_FAILTURE
+        if not no_skip_on_failture:
+            for dep_node_id in self.graph.get_dependent_nodes(UniqueId(node_id)):
+                self._skipped_children[dep_node_id] = cause
 
     def populate_adapter_cache(
         self, adapter, required_schemas: Optional[Set[BaseRelation]] = None

core/dbt/utils.py

@@ -350,6 +350,7 @@ def args_to_dict(args):
             "debug",
             "full_refresh",
             "fail_fast",
+            "no_skip_on_failture",
             "warn_error",
             "single_threaded",
             "log_cache_events",