Call update_event_status earlier + rename an event (#6572)

* Rename HookFinished -> FinishedRunningStats

* Move update_event_status earlier when node finishes

* Add changelog entry

* Add update_event_status for skip

* Update changelog entry

(cherry picked from commit 86e8722)

.changes/unreleased/Fixes-20230111-134058.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Call update_event_status earlier for node results. Rename event 'HookFinished' -> FinishedRunningStats
+time: 2023-01-11T13:40:58.577722+01:00
+custom:
+  Author: jtcohen6
+  Issue: "6571"

core/dbt/events/types.proto

@@ -747,15 +747,15 @@ message HooksRunningMsg {
 }
 
 // E047
-message HookFinished {
+message FinishedRunningStats {
     string stat_line = 1;
     string execution = 2;
     float execution_time = 3;
 }
 
-message HookFinishedMsg {
+message FinishedRunningStatsMsg {
     EventInfo info = 1;
-    HookFinished data = 2;
+    FinishedRunningStats data = 2;
 }
 
 

core/dbt/events/types.py

@@ -755,7 +755,7 @@ def message(self) -> str:
 
 
 @dataclass
-class HookFinished(InfoLevel, pt.HookFinished):
+class FinishedRunningStats(InfoLevel, pt.FinishedRunningStats):
     def code(self):
         return "E047"
 

core/dbt/task/base.py

@@ -4,6 +4,7 @@
 import traceback
 from abc import ABCMeta, abstractmethod
 from typing import Type, Union, Dict, Any, Optional
+from datetime import datetime
 
 from dbt import tracking
 from dbt import flags
@@ -208,6 +209,9 @@ def run_with_hooks(self, manifest):
             self.before_execute()
 
         result = self.safe_run(manifest)
+        self.node.update_event_status(
+            node_status=result.status, finished_at=datetime.utcnow().isoformat()
+        )
 
         if not self.node.is_ephemeral_model:
             self.after_execute(result)
@@ -448,6 +452,9 @@ def on_skip(self):
                     )
                 )
             else:
+                # 'skipped' nodes should not have a value for 'node_finished_at'
+                # they do have 'node_started_at', which is set in GraphRunnableTask.call_runner
+                self.node.update_event_status(node_status=RunStatus.Skipped)
                 fire_event(
                     SkippingDetails(
                         resource_type=self.node.resource_type,

core/dbt/task/run.py

@@ -32,7 +32,7 @@
     DatabaseErrorRunningHook,
     EmptyLine,
     HooksRunning,
-    HookFinished,
+    FinishedRunningStats,
     LogModelResult,
     LogStartLine,
     LogHookEndLine,
@@ -421,7 +421,9 @@ def print_results_line(self, results, execution_time):
         with TextOnly():
             fire_event(EmptyLine())
         fire_event(
-            HookFinished(stat_line=stat_line, execution=execution, execution_time=execution_time)
+            FinishedRunningStats(
+                stat_line=stat_line, execution=execution, execution_time=execution_time
+            )
         )
 
     def before_run(self, adapter, selected_uids: AbstractSet[str]):

core/dbt/task/runnable.py

@@ -226,10 +226,6 @@ def call_runner(self, runner):
             status: Dict[str, str] = {}
             try:
                 result = runner.run_with_hooks(self.manifest)
-                status = runner.get_result_status(result)
-                runner.node.update_event_status(
-                    node_status=result.status, finished_at=datetime.utcnow().isoformat()
-                )
             finally:
                 finishctx = TimestampNamed("finished_at")
                 with finishctx, DbtModelState(status):

tests/unit/test_events.py

@@ -177,7 +177,7 @@ def test_event_codes(self):
     BuildingCatalog(),
     DatabaseErrorRunningHook(hook_type=""),
     HooksRunning(num_hooks=0, hook_type=""),
-    HookFinished(stat_line="", execution="", execution_time=0),
+    FinishedRunningStats(stat_line="", execution="", execution_time=0),
 
     # I - Project parsing ======================
     ParseCmdOut(msg="testing"),