Remove create_scheduler refactor

Author: VaggelisD

sqlmesh/core/context.py

@@ -535,11 +535,9 @@ def scheduler(self, environment: t.Optional[str] = None) -> Scheduler:
         if not snapshots:
             raise ConfigError("No models were found")
 
-        return self.create_scheduler(snapshots, self.snapshot_evaluator)
+        return self.create_scheduler(snapshots)
 
-    def create_scheduler(
-        self, snapshots: t.Iterable[Snapshot], snapshot_evaluator: SnapshotEvaluator
-    ) -> Scheduler:
+    def create_scheduler(self, snapshots: t.Iterable[Snapshot]) -> Scheduler:
         """Creates the built-in scheduler.
 
         Args:
@@ -550,7 +548,7 @@ def create_scheduler(
         """
         return Scheduler(
             snapshots,
-            snapshot_evaluator,
+            self.snapshot_evaluator,
             self.state_sync,
             default_catalog=self.default_catalog,
             max_workers=self.concurrent_tasks,

sqlmesh/core/plan/evaluator.py

@@ -71,7 +71,7 @@ def __init__(
         self,
         state_sync: StateSync,
         snapshot_evaluator: SnapshotEvaluator,
-        create_scheduler: t.Callable[[t.Iterable[Snapshot], SnapshotEvaluator], Scheduler],
+        create_scheduler: t.Callable[[t.Iterable[Snapshot]], Scheduler],
         default_catalog: t.Optional[str],
         console: t.Optional[Console] = None,
     ):
@@ -228,7 +228,7 @@ def visit_backfill_stage(self, stage: stages.BackfillStage, plan: EvaluatablePla
             self.console.log_success("SKIP: No model batches to execute")
             return
 
-        scheduler = self.create_scheduler(stage.all_snapshots.values(), self.snapshot_evaluator)
+        scheduler = self.create_scheduler(stage.all_snapshots.values())
         errors, _ = scheduler.run_merged_intervals(
             merged_intervals=stage.snapshot_to_intervals,
             deployability_index=stage.deployability_index,
@@ -249,7 +249,7 @@ def visit_audit_only_run_stage(
             return
 
         # If there are any snapshots to be audited, we'll reuse the scheduler's internals to audit them
-        scheduler = self.create_scheduler(audit_snapshots, self.snapshot_evaluator)
+        scheduler = self.create_scheduler(audit_snapshots)
         completion_status = scheduler.audit(
             plan.environment,
             plan.start,