diff --git a/.changes/unreleased/Features-20231107-132842.yaml b/.changes/unreleased/Features-20231107-132842.yaml
new file mode 100644
index 00000000000..d048f8a5c02
--- /dev/null
+++ b/.changes/unreleased/Features-20231107-132842.yaml
@@ -0,0 +1,6 @@
+kind: Features
+body: add flag --no-skip-on-failure
+time: 2023-11-07T13:28:42.420727773+01:00
+custom:
+  Author: leo-schick
+  Issue: "2142"
diff --git a/core/dbt/cli/main.py b/core/dbt/cli/main.py
index deff7d8d341..da9ca5d8a85 100644
--- a/core/dbt/cli/main.py
+++ b/core/dbt/cli/main.py
@@ -174,6 +174,7 @@ def cli(ctx, **kwargs):
 @p.export_saved_queries
 @p.full_refresh
 @p.deprecated_include_saved_query
+@p.no_skip_on_failure
 @p.profiles_dir
 @p.project_dir
 @p.resource_type
@@ -541,6 +542,7 @@ def parse(ctx, **kwargs):
 @global_flags
 @p.exclude
 @p.full_refresh
+@p.no_skip_on_failure
 @p.profiles_dir
 @p.project_dir
 @p.empty
@@ -578,6 +580,7 @@ def run(ctx, **kwargs):
 @p.profiles_dir
 @p.vars
 @p.target_path
+@p.no_skip_on_failure
 @p.threads
 @p.full_refresh
 @requires.postflight
@@ -606,6 +609,7 @@ def retry(ctx, **kwargs):
 @global_flags
 @p.exclude
 @p.full_refresh
+@p.no_skip_on_failure
 @p.profiles_dir
 @p.project_dir
 @p.resource_type
@@ -674,6 +678,7 @@ def run_operation(ctx, **kwargs):
 @global_flags
 @p.exclude
 @p.full_refresh
+@p.no_skip_on_failure
 @p.profiles_dir
 @p.project_dir
 @p.select
diff --git a/core/dbt/cli/params.py b/core/dbt/cli/params.py
index b2716728ce6..481c4e62699 100644
--- a/core/dbt/cli/params.py
+++ b/core/dbt/cli/params.py
@@ -120,6 +120,13 @@
     help="Stop execution on first failure.",
 )
 
+no_skip_on_failure = click.option(
+    "--no-skip-on-failure",
+    envvar="DBT_NO_SKIP_ON_FAILURE",
+    help="Proceed with downstream nodes even if an upstream node fails.",
+    is_flag=True,
+)
+
 favor_state = click.option(
     "--favor-state/--no-favor-state",
     envvar="DBT_FAVOR_STATE",
diff --git a/core/dbt/flags.py b/core/dbt/flags.py
index 7deb8966013..1efd468e0b9 100644
--- a/core/dbt/flags.py
+++ b/core/dbt/flags.py
@@ -75,6 +75,7 @@ def get_flag_dict():
         "log_format",
         "version_check",
         "fail_fast",
+        "no_skip_on_failure",
         "send_anonymous_usage_stats",
         "printer_width",
         "indirect_selection",
diff --git a/core/dbt/task/runnable.py b/core/dbt/task/runnable.py
index 746c08bf656..5aef459d4ed 100644
--- a/core/dbt/task/runnable.py
+++ b/core/dbt/task/runnable.py
@@ -450,8 +450,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_failure = get_flags().NO_SKIP_ON_FAILURE
+        if not no_skip_on_failure:
+            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
diff --git a/core/dbt/utils.py b/core/dbt/utils.py
index 8f7509a5dec..a04420a878d 100644
--- a/core/dbt/utils.py
+++ b/core/dbt/utils.py
@@ -350,6 +350,7 @@ def args_to_dict(args):
             "debug",
             "full_refresh",
             "fail_fast",
+            "no_skip_on_failure",
             "warn_error",
             "single_threaded",
             "log_cache_events",