Fix #11012: Catch DbtRuntimeError for hooks

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)
+        ]