Add group info to RunResultError log line

core/dbt/events/core_types.proto

@@ -1845,10 +1845,21 @@ message StatsLineMsg {
     StatsLine data = 2;
 }
 
+message Group {
+    string name = 1;
+    string resource_type = 2;
+    string package_name = 3;
+    string path = 4;
+    string original_file_path = 5;
+    string unique_id = 6;
+    map<string, string> owner = 7;
+}
+
 // Z024
 message RunResultError {
     string msg = 1;
     NodeInfo node_info = 2;
+    Group group = 3;
 }
 
 message RunResultErrorMsg {

core/dbt/task/printer.py

@@ -1,6 +1,7 @@
-from typing import Dict
+from typing import Dict, Optional
 
 from dbt.artifacts.schemas.results import NodeStatus
+from dbt.contracts.graph.nodes import Group
 from dbt.events.types import (
     CheckNodeTestFailure,
     EndOfRunSummary,
@@ -68,7 +69,9 @@ def print_run_status_line(results) -> None:
     fire_event(StatsLine(stats=stats))
 
 
-def print_run_result_error(result, newline: bool = True, is_warning: bool = False) -> None:
+def print_run_result_error(
+    result, newline: bool = True, is_warning: bool = False, group: Optional[Group] = None
+) -> None:
     # set node_info for logging events
     node_info = None
     if hasattr(result, "node") and result.node:
@@ -119,10 +122,13 @@ def print_run_result_error(result, newline: bool = True, is_warning: bool = Fals
     elif result.message is not None:
         if newline:
             fire_event(Formatting(""))
-        fire_event(RunResultError(msg=result.message, node_info=node_info))
+        group_dict = group.to_dict() if group else None
+        fire_event(RunResultError(msg=result.message, node_info=node_info, group=group_dict))
 
 
-def print_run_end_messages(results, keyboard_interrupt: bool = False) -> None:
+def print_run_end_messages(
+    results, keyboard_interrupt: bool = False, groups: Optional[Dict[str, Group]] = None
+) -> None:
     errors, warnings = [], []
     for r in results:
         if r.status in (NodeStatus.RuntimeErr, NodeStatus.Error, NodeStatus.Fail):
@@ -144,9 +150,11 @@ def print_run_end_messages(results, keyboard_interrupt: bool = False) -> None:
     )
 
     for error in errors:
-        print_run_result_error(error, is_warning=False)
+        group = groups.get(error.node.unique_id) if groups else None
+        print_run_result_error(error, is_warning=False, group=group)
 
     for warning in warnings:
-        print_run_result_error(warning, is_warning=True)
+        group = groups.get(warning.node.unique_id) if groups else None
+        print_run_result_error(warning, is_warning=True, group=group)
 
     print_run_status_line(results)

core/dbt/task/run.py

@@ -475,6 +475,17 @@ def get_node_selector(self) -> ResourceTypeSelector:
     def get_runner_type(self, _):
         return ModelRunner
 
+    def get_groups_for_nodes(self, nodes):
+        node_to_group_name_map = {i: k for k, v in self.manifest.group_map.items() for i in v}
+        group_name_to_group_map = {v.name: v for v in self.manifest.groups.values()}
+
+        return {
+            node.unique_id: group_name_to_group_map[node_to_group_name_map[node.unique_id]]
+            for node in nodes
+        }
+
     def task_end_messages(self, results) -> None:
+        groups = self.get_groups_for_nodes([r.node for r in results])
+
         if results:
-            print_run_end_messages(results)
+            print_run_end_messages(results, groups=groups)

tests/functional/logging/test_logging.py

@@ -103,10 +103,28 @@ def test_invalid_event_value(project, logs_dir):
     assert str(excinfo.value) == "[InvalidOptionYAML]: Unable to parse dict {'option_name': 1}"
 
 
+groups_yml = """
+groups:
+  - name: my_group
+    owner:
+      name: my_name
+      email: [email protected]
+      slack: my_slack
+
+models:
+  - name: my_model
+    group: my_group
+    access: public
+"""
+
+
 class TestNodeInfo:
     @pytest.fixture(scope="class")
     def models(self):
-        return {"my_model.sql": "select not_found as id"}
+        return {
+            "my_model.sql": "select not_found as id",
+            "groups.yml": groups_yml,
+        }
 
     def test_node_info_on_results(self, project, logs_dir):
         results = run_dbt(["--log-format=json", "run"], expect_pass=False)