From a0674db8400673d7dd4ebf7597dd82b3349337ce Mon Sep 17 00:00:00 2001
From: Chenyu Li <chenyu.li@dbtlabs.com>
Date: Fri, 18 Oct 2024 15:07:03 -0700
Subject: [PATCH] exclude hook results from results in on-run-end context
 (#10885)

* exclude hook results from results in on-run-end context

* changelog

* preserve previous behavior
---
 .../unreleased/Fixes-20241018-135810.yaml     |  6 ++
 core/dbt/task/run.py                          |  4 +-
 core/dbt/task/runnable.py                     |  1 -
 .../adapter/hooks/test_on_run_hooks.py        | 74 +++++++++++++++++++
 4 files changed, 83 insertions(+), 2 deletions(-)
 create mode 100644 .changes/unreleased/Fixes-20241018-135810.yaml

diff --git a/.changes/unreleased/Fixes-20241018-135810.yaml b/.changes/unreleased/Fixes-20241018-135810.yaml
new file mode 100644
index 00000000000..c205e15bb09
--- /dev/null
+++ b/.changes/unreleased/Fixes-20241018-135810.yaml
@@ -0,0 +1,6 @@
+kind: Fixes
+body: Exclude hook result from results in on-run-end context
+time: 2024-10-18T13:58:10.396884-07:00
+custom:
+  Author: ChenyuLInx
+  Issue: "7387"
diff --git a/core/dbt/task/run.py b/core/dbt/task/run.py
index f159861e1ae..7a321e69d30 100644
--- a/core/dbt/task/run.py
+++ b/core/dbt/task/run.py
@@ -774,7 +774,9 @@ def after_run(self, adapter, results) -> None:
 
         extras = {
             "schemas": list({s for _, s in database_schema_set}),
-            "results": results,
+            "results": [
+                r for r in results if r.thread_id != "main" or r.status == RunStatus.Error
+            ],  # exclude that didn't fail to preserve backwards compatibility
             "database_schemas": list(database_schema_set),
         }
         with adapter.connection_named("master"):
diff --git a/core/dbt/task/runnable.py b/core/dbt/task/runnable.py
index a37308f81ea..7ad9d87e10d 100644
--- a/core/dbt/task/runnable.py
+++ b/core/dbt/task/runnable.py
@@ -512,7 +512,6 @@ def execute_with_hooks(self, selected_uids: AbstractSet[str]):
         self.started_at = time.time()
         try:
             before_run_status = self.before_run(adapter, selected_uids)
-
             if before_run_status == RunStatus.Success or (
                 not get_flags().skip_nodes_if_on_run_start_fails
             ):
diff --git a/tests/functional/adapter/hooks/test_on_run_hooks.py b/tests/functional/adapter/hooks/test_on_run_hooks.py
index b9239b93b4a..b85784be3cf 100644
--- a/tests/functional/adapter/hooks/test_on_run_hooks.py
+++ b/tests/functional/adapter/hooks/test_on_run_hooks.py
@@ -168,3 +168,77 @@ def test_results(self, project):
 
         run_results = get_artifact(project.project_root, "target", "run_results.json")
         assert run_results["results"] == []
+
+
+class Test__HookContext__HookSuccess:
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "on-run-start": [
+                "select 1 as id",  # success
+                "select 1 as id",  # success
+            ],
+            "on-run-end": [
+                '{{ log("Num Results in context: " ~ results|length)}}'
+                "{{ output_thread_ids(results) }}",
+            ],
+        }
+
+    @pytest.fixture(scope="class")
+    def macros(self):
+        return {
+            "log.sql": """
+{% macro output_thread_ids(results) %}
+    {% for result in results %}
+        {{ log("Thread ID: " ~ result.thread_id) }}
+    {% endfor %}
+{% endmacro %}
+"""
+        }
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"my_model.sql": "select 1"}
+
+    def test_results_in_context_success(self, project):
+        results, log_output = run_dbt_and_capture(["--debug", "run"])
+        assert "Thread ID: " in log_output
+        assert "Thread ID: main" not in log_output
+        assert results[0].thread_id == "main"  # hook still exists in run results
+        assert "Num Results in context: 1" in log_output  # only model given hook was successful
+
+
+class Test__HookContext__HookFail:
+    @pytest.fixture(scope="class")
+    def project_config_update(self):
+        return {
+            "on-run-start": [
+                "select a as id",  # fail
+            ],
+            "on-run-end": [
+                '{{ log("Num Results in context: " ~ results|length)}}'
+                "{{ output_thread_ids(results) }}",
+            ],
+        }
+
+    @pytest.fixture(scope="class")
+    def macros(self):
+        return {
+            "log.sql": """
+{% macro output_thread_ids(results) %}
+    {% for result in results %}
+        {{ log("Thread ID: " ~ result.thread_id) }}
+    {% endfor %}
+{% endmacro %}
+"""
+        }
+
+    @pytest.fixture(scope="class")
+    def models(self):
+        return {"my_model.sql": "select 1"}
+
+    def test_results_in_context_hook_fail(self, project):
+        results, log_output = run_dbt_and_capture(["--debug", "run"], expect_pass=False)
+        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