some cleanup, some bug fixes

core/dbt/artifacts/schemas/base.py

@@ -88,7 +88,7 @@ def inner(cls: Type[VersionedSchema]):
     return inner
 
 
-# This is used in the ArtifactMixin and RemoteResult classes
+# This is used in the ArtifactMixin and RemoteCompileResultMixin classes
 @dataclasses.dataclass
 class VersionedSchema(dbtClassMixin):
     dbt_schema_version: ClassVar[SchemaVersion]

core/dbt/parser/manifest.py

@@ -1071,9 +1071,8 @@ def load_macros(
         # and then throws it away, returning only the
         # manifest
         loader = cls(root_config, projects, macro_hook)
-        macro_manifest = loader.create_macro_manifest()
 
-        return macro_manifest
+        return loader.create_macro_manifest()
 
     # Create tracking event for saving performance info
     def track_project_load(self):

core/dbt/task/run.py

@@ -368,29 +368,29 @@ def run_hooks(self, adapter, hook_type: RunHookType, extra_context) -> None:
                     )
                 )
 
-            with Timer() as timer:
-                if len(sql.strip()) > 0:
-                    response, _ = adapter.execute(sql, auto_begin=False, fetch=False)
-                    status = response._message
-                else:
-                    status = "OK"
-
-            self.ran_hooks.append(hook)
-            hook.update_event_status(finished_at=datetime.utcnow().isoformat())
-            hook.update_event_status(node_status=RunStatus.Success)
-            fire_event(
-                LogHookEndLine(
-                    statement=hook_text,
-                    status=status,
-                    index=idx,
-                    total=num_hooks,
-                    execution_time=timer.elapsed,
-                    node_info=hook.node_info,
+                with Timer() as timer:
+                    if len(sql.strip()) > 0:
+                        response, _ = adapter.execute(sql, auto_begin=False, fetch=False)
+                        status = response._message
+                    else:
+                        status = "OK"
+
+                self.ran_hooks.append(hook)
+                hook.update_event_status(finished_at=datetime.utcnow().isoformat())
+                hook.update_event_status(node_status=RunStatus.Success)
+                fire_event(
+                    LogHookEndLine(
+                        statement=hook_text,
+                        status=status,
+                        index=idx,
+                        total=num_hooks,
+                        execution_time=timer.elapsed,
+                        node_info=hook.node_info,
+                    )
                 )
-            )
-            # `_event_status` dict is only used for logging.  Make sure
-            # it gets deleted when we're done with it
-            hook.clear_event_status()
+                # `_event_status` dict is only used for logging.  Make sure
+                # it gets deleted when we're done with it
+                hook.clear_event_status()
 
         self._total_executed += len(ordered_hooks)
 

core/dbt/task/runnable.py

@@ -48,7 +48,7 @@
 from dbt.parser.manifest import write_manifest
 from dbt.task.base import BaseRunner, ConfiguredTask
 from dbt_common.context import _INVOCATION_CONTEXT_VAR, get_invocation_context
-from dbt_common.events.contextvars import task_contextvars
+from dbt_common.events.contextvars import log_contextvars, task_contextvars
 from dbt_common.events.functions import fire_event, warn_or_error
 from dbt_common.events.types import Formatting
 from dbt_common.exceptions import NotImplementedError
@@ -188,51 +188,52 @@ def get_runner(self, node) -> BaseRunner:
         return cls(self.config, adapter, node, run_count, num_nodes)
 
     def call_runner(self, runner: BaseRunner) -> RunResult:
-        runner.node.update_event_status(
-            started_at=datetime.utcnow().isoformat(), node_status=RunningStatus.Started
-        )
-        fire_event(
-            NodeStart(
-                node_info=runner.node.node_info,
+        with log_contextvars(node_info=runner.node.node_info):
+            runner.node.update_event_status(
+                started_at=datetime.utcnow().isoformat(), node_status=RunningStatus.Started
             )
-        )
-        try:
-            result = runner.run_with_hooks(self.manifest)
-        except Exception as e:
-            thread_exception = e
-        finally:
-            if result is not None:
-                fire_event(
-                    NodeFinished(
-                        node_info=runner.node.node_info,
-                        run_result=result.to_msg_dict(),
-                    )
+            fire_event(
+                NodeStart(
+                    node_info=runner.node.node_info,
                 )
-            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,
+            )
+            try:
+                result = runner.run_with_hooks(self.manifest)
+            except Exception as e:
+                thread_exception = e
+            finally:
+                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,
-                )
+                    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
-        runner.node.clear_event_status()
+            # `_event_status` dict is only used for logging.  Make sure
+            # it gets deleted when we're done with it
+            runner.node.clear_event_status()
 
         fail_fast = get_flags().FAIL_FAST