fix: adding in run_id injection

Signed-off-by: Cortland Goffena <30168413+cmgoffena13@users.noreply.github.com>

Author: cmgoffena13

sqlmesh/core/context.py

@@ -118,7 +118,7 @@
     filter_tests_by_patterns,
 )
 from sqlmesh.core.user import User
-from sqlmesh.utils import UniqueKeyDict, Verbosity
+from sqlmesh.utils import CorrelationId, UniqueKeyDict, Verbosity
 from sqlmesh.utils.concurrency import concurrent_apply_to_values
 from sqlmesh.utils.dag import DAG
 from sqlmesh.utils.date import (
@@ -811,6 +811,9 @@ def run(
             engine_type=self.snapshot_evaluator.adapter.dialect,
             state_sync_type=self.state_sync.state_type(),
         )
+        snapshot_evaluator = self.snapshot_evaluator.set_correlation_id(
+            CorrelationId.from_run_id(analytics_run_id)
+        )
         self._load_materializations()
 
         env_check_attempts_num = max(
@@ -863,6 +866,7 @@ def _has_environment_changed() -> bool:
                     select_models=select_models,
                     circuit_breaker=_has_environment_changed,
                     no_auto_upstream=no_auto_upstream,
+                    snapshot_evaluator=snapshot_evaluator,
                 )
                 done = True
             except CircuitBreakerError:
@@ -2586,8 +2590,9 @@ def _run(
         select_models: t.Optional[t.Collection[str]],
         circuit_breaker: t.Optional[t.Callable[[], bool]],
         no_auto_upstream: bool,
+        snapshot_evaluator: t.Optional[SnapshotEvaluator] = None,
     ) -> CompletionStatus:
-        scheduler = self.scheduler(environment=environment)
+        scheduler = self.scheduler(environment=environment, snapshot_evaluator=snapshot_evaluator)
         snapshots = scheduler.snapshots
 
         if select_models is not None:

sqlmesh/utils/__init__.py

@@ -409,6 +409,10 @@ def __str__(self) -> str:
     def from_plan_id(cls, plan_id: str) -> CorrelationId:
         return CorrelationId(JobType.PLAN, plan_id)
 
+    @classmethod
+    def from_run_id(cls, run_id: str) -> CorrelationId:
+        return CorrelationId(JobType.RUN, run_id)
+
 
 def get_source_columns_to_types(
     columns_to_types: t.Dict[str, exp.DataType],