Handle exceptions during node execution more elegantly. (#9585)

* Handle exceptions during node execution more elegantly. * Add changelog entry. * Fix import * Add task documentation. * Change event type for noting thread exceptions.
dbt-labs · Mar 1, 2024 · 6fd0a94 · 6fd0a94
1 parent e4fe839
commit 6fd0a94
Show file tree

Hide file tree

Showing 3 changed files with 81 additions and 7 deletions.
diff --git a/.changes/unreleased/Fixes-20240216-145632.yaml b/.changes/unreleased/Fixes-20240216-145632.yaml
@@ -0,0 +1,6 @@
+kind: Fixes
+body: Tighten exception handling to avoid worker thread hangs.
+time: 2024-02-16T14:56:32.858967-05:00
+custom:
+  Author: peterallenwebb
+  Issue: "9583"
diff --git a/core/dbt/task/README.md b/core/dbt/task/README.md
@@ -1 +1,43 @@
 # Task README
+
+### Task Hierarchy
+```
+BaseTask
+ ┣ CleanTask
+ ┣ ConfiguredTask
+ ┃ ┣ GraphRunnableTask
+ ┃ ┃  ┣ CloneTask
+ ┃ ┃  ┣ CompileTask
+ ┃ ┃  ┃ ┣ GenerateTask
+ ┃ ┃  ┃ ┣ RunTask
+ ┃ ┃  ┃ ┃ ┣ BuildTask
+ ┃ ┃  ┃ ┃ ┣ FreshnessTask
+ ┃ ┃  ┃ ┃ ┣ SeedTask
+ ┃ ┃  ┃ ┃ ┣ SnapshotTask
+ ┃ ┃  ┃ ┃ ┗ TestTask 
+ ┃ ┃  ┃ ┗ ShowTask
+ ┃ ┃  ┗ ListTask
+ ┃ ┣ RetryTask 
+ ┃ ┣ RunOperationTask
+ ┃ ┗ ServeTask
+ ┣ DebugTask
+ ┣ DepsTask
+ ┗ InitTask
+```
+
+### Runner Hierarchy
+```
+BaseRunner
+ ┣ CloneRunner
+ ┣ CompileRunner
+ ┃ ┣ GenericSqlRunner
+ ┃ ┃  ┣ SqlCompileRunner
+ ┃ ┃  ┗ SqlExecuteRunner
+ ┃ ┣ ModelRunner
+ ┃ ┃ ┣ SeedRunner
+ ┃ ┃ ┗ SnapshotRunner
+ ┃ ┣ ShowRunner
+ ┃ ┗ TestRunner
+ ┣ FreshnessRunner
+ ┗ SavedQueryRunner
+```
diff --git a/core/dbt/task/runnable.py b/core/dbt/task/runnable.py
@@ -32,6 +32,7 @@
     ConcurrencyLine,
     EndRunResult,
     NothingToDo,
+    GenericExceptionOnRun,
 )
 from dbt.exceptions import (
     DbtInternalError,
@@ -221,19 +222,44 @@ def call_runner(self, runner: BaseRunner) -> RunResult:
                     )
                 )
             status: Dict[str, str] = {}
+            result = None
+            thread_exception = None
             try:
                 result = runner.run_with_hooks(self.manifest)
-            except Exception as exc:
-                raise DbtInternalError(f"Unable to execute node: {exc}")
+            except Exception as e:
+                thread_exception = e
             finally:
                 finishctx = TimestampNamed("finished_at")
                 with finishctx, DbtModelState(status):
-                    fire_event(
-                        NodeFinished(
-                            node_info=runner.node.node_info,
-                            run_result=result.to_msg_dict(),
+                    if result is not None:
+                        fire_event(
+                            NodeFinished(
+                                node_info=runner.node.node_info,
+                                run_result=result.to_msg_dict(),
+                            )
                         )
-                    )
+                    else:
+                        msg = f"Exception on worker thread. {thread_exception}"
+
+                        fire_event(
+                            GenericExceptionOnRun(
+                                unique_id=runner.node.unique_id,
+                                exc=str(thread_exception),
+                                node_info=runner.node.node_info,
+                            )
+                        )
+
+                        result = RunResult(
+                            status=RunStatus.Error,  # type: ignore
+                            timing=[],
+                            thread_id="",
+                            execution_time=0.0,
+                            adapter_response={},
+                            message=msg,
+                            failures=None,
+                            node=runner.node,
+                        )
+
             # `_event_status` dict is only used for logging.  Make sure
             # it gets deleted when we're done with it, except for unit tests
             if not runner.node.resource_type == NodeType.Unit: