Fix #11012: Catch DbtRuntimeError for hooks (#11023) (#11045)

(cherry picked from commit f582ac2)

Co-authored-by: Kshitij Aranke <[email protected]>
Co-authored-by: Gerda Shank <[email protected]>

.changes/unreleased/Fixes-20241121-181739.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Catch DbtRuntimeError for hooks
+time: 2024-11-21T18:17:39.753235Z
+custom:
+    Author: aranke
+    Issue: "11012"

core/dbt/task/run.py

@@ -922,7 +922,7 @@ def after_run(self, adapter, results) -> None:
         try:
             with adapter.connection_named("master"):
                 self.safe_run_hooks(adapter, RunHookType.End, extras)
-        except (KeyboardInterrupt, SystemExit):
+        except (KeyboardInterrupt, SystemExit, DbtRuntimeError):
             run_result = self.get_result(
                 results=self.node_results,
                 elapsed_time=time.time() - self.started_at,

tests/functional/adapter/hooks/test_on_run_hooks.py

@@ -2,7 +2,8 @@
 
 from dbt.artifacts.schemas.results import RunStatus
 from dbt.contracts.graph.nodes import HookNode
-from dbt.tests.util import get_artifact, run_dbt_and_capture
+from dbt.tests.util import get_artifact, run_dbt, run_dbt_and_capture
+from dbt_common.exceptions import CompilationError
 
 
 class Test__StartHookFail__FlagIsNone__ModelFail:
@@ -242,3 +243,36 @@ def test_results_in_context_hook_fail(self, project):
         assert "Thread ID: main" in log_output
         assert results[0].thread_id == "main"
         assert "Num Results in context: 2" in log_output  # failed hook and model
+
+
+class Test__HookCompilationError:
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"my_model.sql": "select 1 as id"}
+
+    @pytest.fixture(scope="class")
+    def macros(self):
+        return {
+            "rce.sql": """
+{% macro rce(relation) %}
+    {% if execute %}
+        {{ exceptions.raise_compiler_error("Always raise a compiler error in execute") }}
+    {% endif %}
+{% endmacro %}
+    """
+        }
+
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "on-run-end": ["{{ rce() }}"],
+        }
+
+    def test_results(self, project):
+        with pytest.raises(CompilationError, match="Always raise a compiler error in execute"):
+            run_dbt(["run"], expect_pass=False)
+
+        run_results = get_artifact(project.project_root, "target", "run_results.json")
+        assert [(result["unique_id"], result["status"]) for result in run_results["results"]] == [
+            ("model.test.my_model", RunStatus.Success)
+        ]