PR Feedback 1

Author: VaggelisD

sqlmesh/core/context.py

@@ -448,7 +448,14 @@ def engine_adapter(self) -> EngineAdapter:
     @property
     def snapshot_evaluator(self) -> SnapshotEvaluator:
         if not self._snapshot_evaluator:
-            self._snapshot_evaluator = self._create_snapshot_evaluator(log_level=logging.INFO)
+            self._snapshot_evaluator = SnapshotEvaluator(
+                {
+                    gateway: adapter.with_settings(log_level=logging.INFO)
+                    for gateway, adapter in self.engine_adapters.items()
+                },
+                ddl_concurrent_tasks=self.concurrent_tasks,
+                selected_gateway=self.selected_gateway,
+            )
 
         return self._snapshot_evaluator
 
@@ -1909,7 +1916,7 @@ def _table_diff(
             )
 
         return TableDiff(
-            adapter=adapter.with_settings(logger.getEffectiveLevel()),
+            adapter=adapter.with_settings(log_level=logger.getEffectiveLevel()),
             source=source,
             target=target,
             on=on,
@@ -2960,16 +2967,6 @@ def load_model_tests(
 
         return model_tests
 
-    def _create_snapshot_evaluator(self, **kwargs: t.Any) -> SnapshotEvaluator:
-        return SnapshotEvaluator(
-            {
-                gateway: adapter.with_settings(**kwargs)
-                for gateway, adapter in self.engine_adapters.items()
-            },
-            ddl_concurrent_tasks=self.concurrent_tasks,
-            selected_gateway=self.selected_gateway,
-        )
-
 
 class Context(GenericContext[Config]):
     CONFIG_TYPE = Config

sqlmesh/core/engine_adapter/base.py

@@ -147,13 +147,13 @@ def __init__(
         self._multithreaded = multithreaded
         self.correlation_id = correlation_id
 
-    def with_settings(self, log_level: int = logging.DEBUG, **kwargs: t.Any) -> EngineAdapter:
+    def with_settings(self, **kwargs: t.Any) -> EngineAdapter:
         adapter = self.__class__(
             self._connection_pool,
             dialect=self.dialect,
             sql_gen_kwargs=self._sql_gen_kwargs,
             default_catalog=self._default_catalog,
-            execute_log_level=log_level,
+            execute_log_level=kwargs.pop("log_level", self._execute_log_level),
             register_comments=self._register_comments,
             null_connection=self._extra_config.pop("null_connection", True),
             multithreaded=self._multithreaded,

sqlmesh/core/plan/evaluator.py

@@ -89,8 +89,10 @@ def evaluate(
         circuit_breaker: t.Optional[t.Callable[[], bool]] = None,
     ) -> None:
         self._circuit_breaker = circuit_breaker
+        self.snapshot_evaluator = self.snapshot_evaluator.set_correlation_id(
+            CorrelationId.from_plan_id(plan.plan_id)
+        )
 
-        self.set_correlation_id(CorrelationId.from_plan_id(plan.plan_id))
         self.console.start_plan_evaluation(plan)
         analytics.collector.on_plan_apply_start(
             plan=plan,
@@ -350,13 +352,6 @@ def visit_finalize_environment_stage(
     ) -> None:
         self.state_sync.finalize(plan.environment)
 
-    def set_correlation_id(self, correlation_id: CorrelationId) -> None:
-        for key, adapter in self.snapshot_evaluator.adapters.items():
-            if correlation_id != adapter.correlation_id:
-                self.snapshot_evaluator.adapters[key] = adapter.with_settings(
-                    correlation_id=correlation_id
-                )
-
     def _promote_snapshots(
         self,
         plan: EvaluatablePlan,

sqlmesh/core/snapshot/evaluator.py

@@ -61,7 +61,7 @@
     SnapshotTableCleanupTask,
 )
 from sqlmesh.core.snapshot.definition import parent_snapshots_by_name
-from sqlmesh.utils import random_id
+from sqlmesh.utils import random_id, CorrelationId
 from sqlmesh.utils.concurrency import (
     concurrent_apply_to_snapshots,
     concurrent_apply_to_values,
@@ -1190,6 +1190,18 @@ def _execute_create(
         )
         adapter.execute(snapshot.model.render_post_statements(**create_render_kwargs))
 
+    def set_correlation_id(self, correlation_id: CorrelationId) -> SnapshotEvaluator:
+        return SnapshotEvaluator(
+            {
+                gateway: adapter.with_settings(
+                    log_level=adapter._execute_log_level, correlation_id=correlation_id
+                )
+                for gateway, adapter in self.adapters.items()
+            },
+            self.ddl_concurrent_tasks,
+            self.selected_gateway,
+        )
+
 
 def _evaluation_strategy(snapshot: SnapshotInfoLike, adapter: EngineAdapter) -> EvaluationStrategy:
     klass: t.Type

tests/core/test_table_diff.py

@@ -337,9 +337,9 @@ def test_generated_sql(sushi_context_fixed_date: Context, mocker: MockerFixture)
 
     # make with_settings() return the current instance of engine_adapter so we can still spy on _execute
     mocker.patch.object(
-        engine_adapter, "with_settings", new_callable=lambda: lambda _: engine_adapter
+        engine_adapter, "with_settings", new_callable=lambda: lambda **kwargs: engine_adapter
     )
-    assert engine_adapter.with_settings(1) == engine_adapter
+    assert engine_adapter.with_settings() == engine_adapter
 
     spy_execute = mocker.spy(engine_adapter, "_execute")
     mocker.patch("sqlmesh.core.engine_adapter.base.random_id", return_value="abcdefgh")