Repository: TobikoData/sqlmesh
Pull request #3405 (2024-11-20 23:56:29)

chore!: refactor compute interval params to do more at runtime
Prior to this PR, `compute_interval_params` was written to determine the absolute set of intervals to be processed. This meant taking into account depends on past limitations and signals. In addition it would then batch these intervals based on the model's batch size.

This PR changes this function's behavior to instead return a superset of what may actually need to run just based on the intervals that are missing. When actually running the final set of intervals are calculated based on depends on past, signals, batch size, ect. Therefore what is actually run will be a subset of what is provided.

The overall functionality of SQLMesh is unchanged in this PR just this internal detail has been changed.

Two key other changes in the PR: 
* Prior to this PR we had `Intervals` defined as tuple of datetimes in some places and tuples of ints in other places. Now `Intervals` is consistently ints and the functions consistently exchange ints. I couldn't see a good reason why this different existed.
* Airflow was "insulated" from these changes by making of copy of existing functionality into the Airflow codebase and then also moving over the datetime tuple to there too. 


Patch:
diff --git a/docs/guides/signals.md b/docs/guides/signals.md
index 0cad6be5..3466d7b2 100644
--- a/docs/guides/signals.md
+++ b/docs/guides/signals.md
@@ -46,7 +46,9 @@ The method returns `True` (indicating that all intervals are ready for evaluatio
 ```python linenums="1"
 import random
 import typing as t
-from sqlmesh.core.scheduler import signal_factory, Batch, Signal
+from sqlmesh.core.scheduler import signal_factory, Signal
+from sqlmesh.utils.date import DatetimeRanges
+
 
 class RandomSignal(Signal):
     def __init__(
@@ -55,7 +57,7 @@ class RandomSignal(Signal):
     ):
         self.signal_metadata = signal_metadata
 
-    def check_intervals(self, batch: Batch) -> t.Union[bool, Batch]:
+    def check_intervals(self, batch: DatetimeRanges) -> t.Union[bool, DatetimeRanges]:
         threshold = self.signal_metadata["threshold"]
         return random.random() > threshold
 ```
@@ -69,7 +71,9 @@ We can now add a factory function that returns the `RandomSignal` sub-class. Not
 ```python linenums="1" hl_lines="8-10"
 import random
 import typing as t
-from sqlmesh.core.scheduler import signal_factory, Batch, Signal
+from sqlmesh.core.scheduler import signal_factory, Signal
+from sqlmesh.utils.date import DatetimeRanges
+
 
 class RandomSignal(Signal):
     def __init__(
@@ -78,10 +82,11 @@ class RandomSignal(Signal):
     ):
         self.signal_metadata = signal_metadata
 
-    def check_intervals(self, batch: Batch) -> t.Union[bool, Batch]:
+    def check_intervals(self, batch: DatetimeRanges) -> t.Union[bool, DatetimeRanges]:
         threshold = self.signal_metadata["threshold"]
         return random.random() > threshold
 
+
 @signal_factory
 def my_signal_factory(signal_metadata: t.Dict[str, t.Union[str, int, float, bool]]) -> Signal:
     return RandomSignal(signal_metadata)
@@ -122,13 +127,13 @@ In this example, there are two signals.
 import typing as t
 from datetime import datetime
 
-from sqlmesh.core.scheduler import signal_factory, Batch, Signal
-from sqlmesh.utils.date import to_datetime
+from sqlmesh.core.scheduler import signal_factory, Signal
+from sqlmesh.utils.date import DatetimeRanges, to_datetime
 
 
 class AlwaysReady(Signal):
     # signal that indicates every interval is always ready
-    def check_intervals(self, batch: Batch) -> t.Union[bool, Batch]:
+    def check_intervals(self, batch: DatetimeRanges) -> t.Union[bool, DatetimeRanges]:
         return True
 
 
@@ -137,7 +142,7 @@ class OneweekAgo(Signal):
         self.dt = dt
 
     # signal that returns only intervals that are <= 1 week ago
-    def check_intervals(self, batch: Batch) -> t.Union[bool, Batch]:
+    def check_intervals(self, batch: DatetimeRanges) -> t.Union[bool, DatetimeRanges]:
         return [
             (start, end)
             for start, end in batch
diff --git a/sqlmesh/core/scheduler.py b/sqlmesh/core/scheduler.py
index 982cdbd6..f9913526 100644
--- a/sqlmesh/core/scheduler.py
+++ b/sqlmesh/core/scheduler.py
@@ -4,13 +4,13 @@ import abc
 import logging
 import traceback
 import typing as t
-from datetime import datetime
 
 from sqlglot import exp
 
 from sqlmesh.core import constants as c
 from sqlmesh.core.console import Console, get_console
 from sqlmesh.core.environment import EnvironmentNamingInfo
+from sqlmesh.core.node import IntervalUnit
 from sqlmesh.core.notification_target import (
     NotificationEvent,
     NotificationTargetManager,
@@ -21,9 +21,10 @@ from sqlmesh.core.snapshot import (
     SnapshotEvaluator,
     earliest_start_date,
     missing_intervals,
+    Intervals,
 )
-from sqlmesh.core.snapshot.definition import Interval as SnapshotInterval
-from sqlmesh.core.snapshot.definition import Intervals, SnapshotId, merge_intervals
+from sqlmesh.core.snapshot.definition import Interval, expand_range
+from sqlmesh.core.snapshot.definition import SnapshotId, merge_intervals
 from sqlmesh.core.state_sync import StateSync
 from sqlmesh.utils import format_exception
 from sqlmesh.utils.concurrency import concurrent_apply_to_dag, NodeExecutionFailedError
@@ -35,13 +36,12 @@ from sqlmesh.utils.date import (
     to_datetime,
     to_timestamp,
     validate_date_range,
+    DatetimeRanges,
 )
 from sqlmesh.utils.errors import AuditError, CircuitBreakerError, SQLMeshError
 
 logger = logging.getLogger(__name__)
-Interval = t.Tuple[datetime, datetime]
-Batch = t.List[Interval]
-SnapshotToBatches = t.Dict[Snapshot, Batch]
+SnapshotToIntervals = t.Dict[Snapshot, Intervals]
 # we store snapshot name instead of snapshots/snapshotids because pydantic
 # is extremely slow to hash. snapshot names should be unique within a dag run
 SchedulingUnit = t.Tuple[str, t.Tuple[Interval, int]]
@@ -49,7 +49,7 @@ SchedulingUnit = t.Tuple[str, t.Tuple[Interval, int]]
 
 class Signal(abc.ABC):
     @abc.abstractmethod
-    def check_intervals(self, batch: Batch) -> t.Union[bool, Batch]:
+    def check_intervals(self, batch: DatetimeRanges) -> t.Union[bool, DatetimeRanges]:
         """Returns which intervals are ready from a list of scheduled intervals.
 
         When SQLMesh wishes to execute a batch of intervals, say between `a` and `d`, then
@@ -148,32 +148,29 @@ class Scheduler:
         )
         self.signal_factory = signal_factory or _registered_signal_factory
 
-    def batches(
+    def merged_missing_intervals(
         self,
         start: t.Optional[TimeLike] = None,
         end: t.Optional[TimeLike] = None,
         execution_time: t.Optional[TimeLike] = None,
         deployability_index: t.Optional[DeployabilityIndex] = None,
-        restatements: t.Optional[t.Dict[SnapshotId, SnapshotInterval]] = None,
+        restatements: t.Optional[t.Dict[SnapshotId, Interval]] = None,
         interval_end_per_model: t.Optional[t.Dict[str, int]] = None,
         ignore_cron: bool = False,
         end_bounded: bool = False,
         selected_snapshots: t.Optional[t.Set[str]] = None,
-    ) -> SnapshotToBatches:
-        """Find the optimal date interval paramaters based on what needs processing and maximal batch size.
+    ) -> SnapshotToIntervals:
+        """Find the largest contiguous date interval parameters based only on what is missing.
 
         For each node name, find all dependencies and look for a stored snapshot from the metastore. If a snapshot is found,
         calculate the missing intervals that need to be processed given the passed in start and end intervals.
 
-        If a snapshot's node specifies a batch size, consecutive intervals are merged into batches of a size that is less than
-        or equal to the configured one. If no batch size is specified, then it uses the intervals that correspond to the node's cron expression.
-        For example, if a node is supposed to run daily and has 70 days to backfill with a batch size set to 30, there would be 2 jobs
-        with 30 days and 1 job with 10.
+        This is a superset of what may actually get processed at runtime based on things like batch size, signal readiness, etc.
 
         Args:
             start: The start of the run. Defaults to the min node start date.
             end: The end of the run. Defaults to now.
-            execution_time: The date/time time reference to use for execution time. Defaults to now.
+            execution_time: The date/time reference to use for execution time. Defaults to now.
             deployability_index: Determines snapshots that are deployable in the context of this evaluation.
             restatements: A set of snapshot names being restated.
             interval_end_per_model: The mapping from model FQNs to target end dates.
@@ -201,7 +198,6 @@ class Scheduler:
             interval_end_per_model=interval_end_per_model,
             ignore_cron=ignore_cron,
             end_bounded=end_bounded,
-            signal_factory=self.signal_factory,
         )
 
     def evaluate(
@@ -285,7 +281,7 @@ class Scheduler:
         start: t.Optional[TimeLike] = None,
         end: t.Optional[TimeLike] = None,
         execution_time: t.Optional[TimeLike] = None,
-        restatements: t.Optional[t.Dict[SnapshotId, SnapshotInterval]] = None,
+        restatements: t.Optional[t.Dict[SnapshotId, Interval]] = None,
         interval_end_per_model: t.Optional[t.Dict[str, int]] = None,
         ignore_cron: bool = False,
         end_bounded: bool = False,
@@ -333,7 +329,7 @@ class Scheduler:
             else DeployabilityIndex.all_deployable()
         )
         execution_time = execution_time or now()
-        batches = self.batches(
+        merged_intervals = self.merged_missing_intervals(
             start,
             end,
             execution_time,
@@ -344,20 +340,17 @@ class Scheduler:
             end_bounded=end_bounded,
             selected_snapshots=selected_snapshots,
         )
-        if not batches:
+        if not merged_intervals:
             return True
 
-        self.console.start_evaluation_progress(
-            {snapshot: len(intervals) for snapshot, intervals in batches.items()},
-            environment_naming_info,
-            self.default_catalog,
-        )
-
-        errors, skipped_intervals = self.run_batches(
-            batches=batches,
+        errors, skipped_intervals = self.run_merged_intervals(
+            merged_intervals=merged_intervals,
             deployability_index=deployability_index,
+            environment_naming_info=environment_naming_info,
             execution_time=execution_time,
             circuit_breaker=circuit_breaker,
+            start=start,
+            end=end,
         )
 
         self.console.stop_evaluation_progress(success=not errors)
@@ -380,26 +373,107 @@ class Scheduler:
 
         return not errors
 
-    def run_batches(
+    def batch_intervals(
         self,
-        batches: SnapshotToBatches,
+        merged_intervals: SnapshotToIntervals,
+        start: t.Optional[TimeLike] = None,
+        end: t.Optional[TimeLike] = None,
+        execution_time: t.Optional[TimeLike] = None,
+    ) -> t.Dict[Snapshot, Intervals]:
+        def expand_range_as_interval(
+            start_ts: int, end_ts: int, interval_unit: IntervalUnit
+        ) -> t.List[Interval]:
+            values = expand_range(start_ts, end_ts, interval_unit)
+            return [(values[i], values[i + 1]) for i in range(len(values) - 1)]
+
+        dag = DAG[str]()
+
+        for snapshot in merged_intervals:
+            dag.add(snapshot.name, [p.name for p in snapshot.parents])
+
+        snapshot_intervals = {
+            snapshot: [
+                i
+                for interval in intervals
+                for i in expand_range_as_interval(*interval, snapshot.node.interval_unit)
+            ]
+            for snapshot, intervals in merged_intervals.items()
+        }
+        snapshot_batches = {}
+        all_unready_intervals: t.Dict[str, set[Interval]] = {}
+        for snapshot, intervals in snapshot_intervals.items():
+            if self.signal_factory and snapshot.is_model:
+                unready = set(intervals)
+
+                for signal in snapshot.model.render_signals(
+                    start=start, end=end, execution_time=execution_time
+                ):
+                    intervals = _check_ready_intervals(
+                        signal=self.signal_factory(signal),
+                        intervals=intervals,
+                    )
+                unready -= set(intervals)
+            else:
+                unready = set()
+
+            for parent in snapshot.parents:
+                if parent.name in all_unready_intervals:
+                    unready.update(all_unready_intervals[parent.name])
+
+            all_unready_intervals[snapshot.name] = unready
+
+            batches = []
+            batch_size = snapshot.node.batch_size
+            next_batch: t.List[t.Tuple[int, int]] = []
+
+            for interval in interval_diff(
+                intervals, merge_intervals(unready), uninterrupted=snapshot.depends_on_past
+            ):
+                if (batch_size and len(next_batch) >= batch_size) or (
+                    next_batch and interval[0] != next_batch[-1][-1]
+                ):
+                    batches.append((next_batch[0][0], next_batch[-1][-1]))
+                    next_batch = []
+                next_batch.append(interval)
+            if next_batch:
+                batches.append((next_batch[0][0], next_batch[-1][-1]))
+            snapshot_batches[snapshot] = batches
+        return snapshot_batches
+
+    def run_merged_intervals(
+        self,
+        *,
+        merged_intervals: SnapshotToIntervals,
         deployability_index: DeployabilityIndex,
+        environment_naming_info: EnvironmentNamingInfo,
         execution_time: t.Optional[TimeLike] = None,
         circuit_breaker: t.Optional[t.Callable[[], bool]] = None,
+        start: t.Optional[TimeLike] = None,
+        end: t.Optional[TimeLike] = None,
     ) -> t.Tuple[t.List[NodeExecutionFailedError[SchedulingUnit]], t.List[SchedulingUnit]]:
         """Runs precomputed batches of missing intervals.
 
         Args:
-            batches: The batches of snapshots and intervals to evaluate.
+            merged_intervals: The snapshots and contiguous interval ranges to evaluate.
             deployability_index: Determines snapshots that are deployable in the context of this evaluation.
-            execution_time: The date/time time reference to use for execution time.
+            environment_naming_info: The environment naming info the user is targeting when applying their change.
+            execution_time: The date/time reference to use for execution time.
             circuit_breaker: An optional handler which checks if the run should be aborted.
+            start: The start of the run.
+            end: The end of the run.
 
         Returns:
             A tuple of errors and skipped intervals.
         """
-        execution_time = execution_time or now()
-        dag = self._dag(batches)
+        batched_intervals = self.batch_intervals(merged_intervals, start, end, execution_time)
+
+        self.console.start_evaluation_progress(
+            {snapshot: len(intervals) for snapshot, intervals in batched_intervals.items()},
+            environment_naming_info,
+            self.default_catalog,
+        )
+
+        dag = self._dag(batched_intervals)
 
         snapshots_by_name = {snapshot.name: snapshot for snapshot in self.snapshots.values()}
 
@@ -438,7 +512,7 @@ class Scheduler:
         finally:
             self.state_sync.recycle()
 
-    def _dag(self, batches: SnapshotToBatches) -> DAG[SchedulingUnit]:
+    def _dag(self, batches: SnapshotToIntervals) -> DAG[SchedulingUnit]:
         """Builds a DAG of snapshot intervals to be evaluated.
 
         Args:
@@ -453,7 +527,7 @@ class Scheduler:
         }
 
         dag = DAG[SchedulingUnit]()
-        terminal_node = ((to_datetime(0), to_datetime(0)), -1)
+        terminal_node = ((to_timestamp(0), to_timestamp(0)), -1)
 
         for snapshot, intervals in batches.items():
             if not intervals:
@@ -503,29 +577,24 @@ def compute_interval_params(
     end: TimeLike,
     deployability_index: t.Optional[DeployabilityIndex] = None,
     execution_time: t.Optional[TimeLike] = None,
-    restatements: t.Optional[t.Dict[SnapshotId, SnapshotInterval]] = None,
+    restatements: t.Optional[t.Dict[SnapshotId, Interval]] = None,
     interval_end_per_model: t.Optional[t.Dict[str, int]] = None,
     ignore_cron: bool = False,
     end_bounded: bool = False,
-    signal_factory: t.Optional[SignalFactory] = None,
-) -> SnapshotToBatches:
-    """Find the optimal date interval paramaters based on what needs processing and maximal batch size.
+) -> SnapshotToIntervals:
+    """Find the largest contiguous date interval parameters based only on what is missing.
 
     For each node name, find all dependencies and look for a stored snapshot from the metastore. If a snapshot is found,
     calculate the missing intervals that need to be processed given the passed in start and end intervals.
 
-    If a snapshot's node specifies a batch size, consecutive intervals are merged into batches of a size that is less than
-    or equal to the configured one. If no batch size is specified, then it uses the intervals that correspond to the node's cron expression.
-    For example, if a node is supposed to run daily and has 70 days to backfill with a batch size set to 30, there would be 2 jobs
-    with 30 days and 1 job with 10.
+    This is a superset of what may actually get processed at runtime based on things like batch size, signal readiness, etc.
 
     Args:
         snapshots: A set of target snapshots for which intervals should be computed.
-        intervals: A list of all snapshot intervals that should be considered.
         start: Start of the interval.
         end: End of the interval.
         deployability_index: Determines snapshots that are deployable in the context of this evaluation.
-        execution_time: The date/time time reference to use for execution time.
+        execution_time: The date/time reference to use for execution time.
         restatements: A dict of snapshot names being restated and their intervals.
         interval_end_per_model: The mapping from model FQNs to target end dates.
         ignore_cron: Whether to ignore the node's cron schedule.
@@ -535,20 +604,10 @@ def compute_interval_params(
     Returns:
         A dict containing all snapshots needing to be run with their associated interval params.
     """
-    snapshot_batches = {}
-
-    snapshots_by_name = {snapshot.name: snapshot for snapshot in snapshots}
-
-    dag = DAG[str]()
-
-    for snapshot in snapshots_by_name.values():
-        dag.add(snapshot.name, [p.name for p in snapshot.parents])
-
-    all_unready_intervals: t.Dict[str, set[SnapshotInterval]] = {}
+    snapshot_merged_intervals = {}
 
     for snapshot, intervals in missing_intervals(
-        # need to iterate through missing intervals in dag order to propogate unready intervals
-        [snapshots_by_name[fqn] for fqn in dag if fqn in snapshots_by_name],
+        snapshots,
         start=start,
         end=end,
         execution_time=execution_time,
@@ -558,44 +617,19 @@ def compute_interval_params(
         ignore_cron=ignore_cron,
         end_bounded=end_bounded,
     ).items():
-        if signal_factory and snapshot.is_model:
-            unready = set(intervals)
-
-            for signal in snapshot.model.render_signals(
-                start=start, end=end, execution_time=execution_time
-            ):
-                intervals = _check_ready_intervals(
-                    signal=signal_factory(signal),
-                    intervals=intervals,
-                )
-            unready -= set(intervals)
-        else:
-            unready = set()
+        contiguous_batch = []
+        next_batch: Intervals = []
 
-        for parent in snapshot.parents:
-            if parent.name in all_unready_intervals:
-                unready.update(all_unready_intervals[parent.name])
-
-        all_unready_intervals[snapshot.name] = unready
-
-        batches = []
-        batch_size = snapshot.node.batch_size
-        next_batch: t.List[t.Tuple[int, int]] = []
-
-        for interval in interval_diff(
-            intervals, merge_intervals(unready), uninterrupted=snapshot.depends_on_past
-        ):
-            if (batch_size and len(next_batch) >= batch_size) or (
-                next_batch and interval[0] != next_batch[-1][-1]
-            ):
-                batches.append((next_batch[0][0], next_batch[-1][-1]))
+        for interval in intervals:
+            if next_batch and interval[0] != next_batch[-1][-1]:
+                contiguous_batch.append((next_batch[0][0], next_batch[-1][-1]))
                 next_batch = []
             next_batch.append(interval)
         if next_batch:
-            batches.append((next_batch[0][0], next_batch[-1][-1]))
-        snapshot_batches[snapshot] = [(to_datetime(s), to_datetime(e)) for s, e in batches]
+            contiguous_batch.append((next_batch[0][0], next_batch[-1][-1]))
+        snapshot_merged_intervals[snapshot] = contiguous_batch
 
-    return snapshot_batches
+    return snapshot_merged_intervals
 
 
 def interval_diff(
@@ -659,7 +693,7 @@ def _contiguous_intervals(
 ) -> t.List[Intervals]:
     """Given a list of intervals with gaps, returns a list of sequences of contiguous intervals."""
     contiguous_intervals = []
-    current_batch: t.List[SnapshotInterval] = []
+    current_batch: t.List[Interval] = []
     for interval in intervals:
         if len(current_batch) == 0 or interval[0] == current_batch[-1][-1]:
             current_batch.append(interval)
diff --git a/sqlmesh/core/snapshot/definition.py b/sqlmesh/core/snapshot/definition.py
index 9c17b6ef..fdddb339 100644
--- a/sqlmesh/core/snapshot/definition.py
+++ b/sqlmesh/core/snapshot/definition.py
@@ -1642,6 +1642,25 @@ def missing_intervals(
     return missing
 
 
+@lru_cache(maxsize=None)
+def expand_range(start_ts: int, end_ts: int, interval_unit: IntervalUnit) -> t.List[int]:
+    croniter = interval_unit.croniter(start_ts)
+    timestamps = [start_ts]
+
+    while True:
+        ts = to_timestamp(croniter.get_next(estimate=True))
+
+        if ts > end_ts:
+            if len(timestamps) > 1:
+                timestamps[-1] = end_ts
+            else:
+                timestamps.append(end_ts)
+            break
+
+        timestamps.append(ts)
+    return timestamps
+
+
 @lru_cache(maxsize=None)
 def compute_missing_intervals(
     interval_unit: IntervalUnit,
@@ -1667,21 +1686,7 @@ def compute_missing_intervals(
     if start_ts == end_ts:
         return []
 
-    croniter = interval_unit.croniter(start_ts)
-    timestamps = [start_ts]
-
-    while True:
-        ts = to_timestamp(croniter.get_next(estimate=True))
-
-        if ts > end_ts:
-            if len(timestamps) > 1:
-                timestamps[-1] = end_ts
-            else:
-                timestamps.append(end_ts)
-            break
-
-        timestamps.append(ts)
-
+    timestamps = expand_range(start_ts, end_ts, interval_unit)
     missing = set()
 
     for current_ts, next_ts in zip(timestamps, timestamps[1:]):
diff --git a/sqlmesh/schedulers/airflow/common.py b/sqlmesh/schedulers/airflow/common.py
index 3e781ae5..01fc40a1 100644
--- a/sqlmesh/schedulers/airflow/common.py
+++ b/sqlmesh/schedulers/airflow/common.py
@@ -6,7 +6,6 @@ from sqlmesh.core import constants as c
 from sqlmesh.core.environment import Environment
 from sqlmesh.core.notification_target import NotificationTarget
 from sqlmesh.core.plan.definition import EvaluatablePlan
-from sqlmesh.core.scheduler import Interval
 from sqlmesh.core.snapshot import (
     DeployabilityIndex,
     Snapshot,
@@ -17,7 +16,7 @@ from sqlmesh.core.snapshot import (
 )
 from sqlmesh.core.user import User
 from sqlmesh.utils import sanitize_name
-from sqlmesh.utils.date import TimeLike
+from sqlmesh.utils.date import TimeLike, DatetimeRanges
 from sqlmesh.utils.pydantic import PydanticModel
 
 JANITOR_DAG_ID = "sqlmesh_janitor_dag"
@@ -35,6 +34,8 @@ AIRFLOW_LOCAL_URL = "http://localhost:8080/"
 
 SQLMESH_API_BASE_PATH: str = f"{c.SQLMESH}/api/v1"
 
+SnapshotToDatetimeRanges = t.Dict[Snapshot, DatetimeRanges]
+
 
 class PlanApplicationRequest(PydanticModel):
     plan: EvaluatablePlan
@@ -46,7 +47,7 @@ class PlanApplicationRequest(PydanticModel):
 
 class BackfillIntervalsPerSnapshot(PydanticModel):
     snapshot_id: SnapshotId
-    intervals: t.List[Interval]
+    intervals: DatetimeRanges
     before_promote: bool = True
 
 
diff --git a/sqlmesh/schedulers/airflow/plan.py b/sqlmesh/schedulers/airflow/plan.py
index dcfdfc4e..f0e6d274 100644
--- a/sqlmesh/schedulers/airflow/plan.py
+++ b/sqlmesh/schedulers/airflow/plan.py
@@ -5,15 +5,16 @@ import typing as t
 import pandas as pd
 from sqlglot import exp
 
-from sqlmesh.core import scheduler
+from sqlmesh import Snapshot
 from sqlmesh.core.engine_adapter import EngineAdapter
 from sqlmesh.core.environment import Environment
 from sqlmesh.core.plan import update_intervals_for_new_snapshots
-from sqlmesh.core.snapshot import DeployabilityIndex, SnapshotTableInfo
+from sqlmesh.core.snapshot import DeployabilityIndex, SnapshotTableInfo, SnapshotId
+from sqlmesh.core.snapshot.definition import Interval as SnapshotInterval, missing_intervals
 from sqlmesh.core.state_sync import EngineAdapterStateSync, StateSync
 from sqlmesh.core.state_sync.base import DelegatingStateSync
 from sqlmesh.schedulers.airflow import common
-from sqlmesh.utils.date import now, to_timestamp
+from sqlmesh.utils.date import now, to_timestamp, TimeLike, to_datetime
 from sqlmesh.utils.errors import SQLMeshError
 
 
@@ -80,6 +81,74 @@ class PlanDagState:
         )
 
 
+def airflow_compute_interval_params(
+    snapshots: t.Collection[Snapshot],
+    *,
+    start: TimeLike,
+    end: TimeLike,
+    deployability_index: t.Optional[DeployabilityIndex] = None,
+    execution_time: t.Optional[TimeLike] = None,
+    restatements: t.Optional[t.Dict[SnapshotId, SnapshotInterval]] = None,
+    interval_end_per_model: t.Optional[t.Dict[str, int]] = None,
+    ignore_cron: bool = False,
+    end_bounded: bool = False,
+) -> common.SnapshotToDatetimeRanges:
+    """Find the optimal date interval parameters based on what needs processing and maximal batch size.
+
+    For each node name, find all dependencies and look for a stored snapshot from the metastore. If a snapshot is found,
+    calculate the missing intervals that need to be processed given the passed in start and end intervals.
+
+    If a snapshot's node specifies a batch size, consecutive intervals are merged into batches of a size that is less than
+    or equal to the configured one. If no batch size is specified, then it uses the intervals that correspond to the node's cron expression.
+    For example, if a node is supposed to run daily and has 70 days to backfill with a batch size set to 30, there would be 2 jobs
+    with 30 days and 1 job with 10.
+
+    Args:
+        snapshots: A set of target snapshots for which intervals should be computed.
+        start: Start of the interval.
+        end: End of the interval.
+        deployability_index: Determines snapshots that are deployable in the context of this evaluation.
+        execution_time: The date/time reference to use for execution time.
+        restatements: A dict of snapshot names being restated and their intervals.
+        interval_end_per_model: The mapping from model FQNs to target end dates.
+        ignore_cron: Whether to ignore the node's cron schedule.
+        end_bounded: If set to true, the returned intervals will be bounded by the target end date, disregarding lookback,
+            allow_partials, and other attributes that could cause the intervals to exceed the target end date.
+
+    Returns:
+        A dict containing all snapshots needing to be run with their associated interval params.
+    """
+    snapshot_batches = {}
+
+    for snapshot, intervals in missing_intervals(
+        snapshots,
+        start=start,
+        end=end,
+        execution_time=execution_time,
+        restatements=restatements,
+        deployability_index=deployability_index,
+        interval_end_per_model=interval_end_per_model,
+        ignore_cron=ignore_cron,
+        end_bounded=end_bounded,
+    ).items():
+        batches = []
+        batch_size = snapshot.node.batch_size
+        next_batch: t.List[t.Tuple[int, int]] = []
+
+        for interval in intervals:
+            if (batch_size and len(next_batch) >= batch_size) or (
+                next_batch and interval[0] != next_batch[-1][-1]
+            ):
+                batches.append((next_batch[0][0], next_batch[-1][-1]))
+                next_batch = []
+            next_batch.append(interval)
+        if next_batch:
+            batches.append((next_batch[0][0], next_batch[-1][-1]))
+        snapshot_batches[snapshot] = [(to_datetime(s), to_datetime(e)) for s, e in batches]
+
+    return snapshot_batches
+
+
 def create_plan_dag_spec(
     request: common.PlanApplicationRequest, state_sync: StateSync
 ) -> common.PlanDagSpec:
@@ -127,7 +196,7 @@ def create_plan_dag_spec(
     )
 
     if not plan.skip_backfill:
-        backfill_batches = scheduler.compute_interval_params(
+        backfill_batches = airflow_compute_interval_params(
             [s for s in all_snapshots.values() if plan.is_selected_for_backfill(s.name)],
             start=plan.environment.start_at,
             end=end,
@@ -136,7 +205,6 @@ def create_plan_dag_spec(
             restatements=restatements,
             interval_end_per_model=plan.interval_end_per_model,
             end_bounded=plan.end_bounded,
-            signal_factory=None,
         )
     else:
         backfill_batches = {}
diff --git a/sqlmesh/utils/date.py b/sqlmesh/utils/date.py
index d204c3b2..a2b7cce4 100644
--- a/sqlmesh/utils/date.py
+++ b/sqlmesh/utils/date.py
@@ -19,10 +19,10 @@ from sqlmesh.utils import ttl_cache
 
 UTC = timezone.utc
 TimeLike = t.Union[date, datetime, str, int, float]
+DatetimeRange = t.Tuple[datetime, datetime]
+DatetimeRanges = t.List[DatetimeRange]
 DATE_INT_FMT = "%Y%m%d"
 
-if t.TYPE_CHECKING:
-    from sqlmesh.core.scheduler import Interval
 
 warnings.filterwarnings(
     "ignore",
@@ -284,7 +284,7 @@ def is_date(obj: TimeLike) -> bool:
         return False
 
 
-def make_inclusive(start: TimeLike, end: TimeLike) -> Interval:
+def make_inclusive(start: TimeLike, end: TimeLike) -> DatetimeRange:
     """Adjust start and end times to to become inclusive datetimes.
 
     SQLMesh treats start and end times as inclusive so that filters can be written as
diff --git a/web/server/api/endpoints/plan.py b/web/server/api/endpoints/plan.py
index 569c8905..6bccb381 100644
--- a/web/server/api/endpoints/plan.py
+++ b/web/server/api/endpoints/plan.py
@@ -131,7 +131,8 @@ def _get_plan_changes(context: Context, plan: Plan) -> models.PlanChanges:
 
 def _get_plan_backfills(context: Context, plan: Plan) -> t.Dict[str, t.Any]:
     """Get plan backfills"""
-    batches = context.scheduler().batches()
+    merged_intervals = context.scheduler().merged_missing_intervals()
+    batches = context.scheduler().batch_intervals(merged_intervals)
     tasks = {snapshot.name: len(intervals) for snapshot, intervals in batches.items()}
     snapshots = plan.context_diff.snapshots
     default_catalog = context.default_catalog


Test patch:
diff --git a/tests/core/test_scheduler.py b/tests/core/test_scheduler.py
index 84e5dd4b..a9d2730b 100644
--- a/tests/core/test_scheduler.py
+++ b/tests/core/test_scheduler.py
@@ -21,8 +21,8 @@ from sqlmesh.core.scheduler import (
     interval_diff,
     compute_interval_params,
     signal_factory,
-    Batch,
     Signal,
+    SnapshotToIntervals,
 )
 from sqlmesh.core.snapshot import (
     Snapshot,
@@ -30,7 +30,7 @@ from sqlmesh.core.snapshot import (
     SnapshotChangeCategory,
     DeployabilityIndex,
 )
-from sqlmesh.utils.date import to_datetime
+from sqlmesh.utils.date import to_datetime, to_timestamp, DatetimeRanges, TimeLike
 from sqlmesh.utils.errors import CircuitBreakerError, AuditError
 
 
@@ -59,18 +59,30 @@ def test_interval_params(scheduler: Scheduler, sushi_context_fixed_date: Context
 
     assert compute_interval_params([orders, waiter_revenue], start=start_ds, end=end_ds) == {
         orders: [
-            (to_datetime(start_ds), to_datetime("2022-01-31")),
-            (to_datetime("2022-01-31"), to_datetime("2022-02-06")),
+            (to_timestamp(start_ds), to_timestamp("2022-02-06")),
         ],
         waiter_revenue: [
-            (to_datetime(start_ds), to_datetime("2022-01-11")),
-            (to_datetime("2022-01-11"), to_datetime("2022-01-21")),
-            (to_datetime("2022-01-21"), to_datetime("2022-01-31")),
-            (to_datetime("2022-01-31"), to_datetime("2022-02-06")),
+            (to_timestamp(start_ds), to_timestamp("2022-02-06")),
         ],
     }
 
 
+@pytest.fixture
+def get_batched_missing_intervals() -> (
+    t.Callable[[Scheduler, TimeLike, TimeLike, t.Optional[TimeLike]], SnapshotToIntervals]
+):
+    def _get_batched_missing_intervals(
+        scheduler: Scheduler,
+        start: TimeLike,
+        end: TimeLike,
+        execution_time: t.Optional[TimeLike] = None,
+    ) -> SnapshotToIntervals:
+        merged_intervals = scheduler.merged_missing_intervals(start, end, execution_time)
+        return scheduler.batch_intervals(merged_intervals, start, end, execution_time)
+
+    return _get_batched_missing_intervals
+
+
 def test_interval_params_nonconsecutive(scheduler: Scheduler, orders: Snapshot):
     start_ds = "2022-01-01"
     end_ds = "2022-02-05"
@@ -79,8 +91,8 @@ def test_interval_params_nonconsecutive(scheduler: Scheduler, orders: Snapshot):
 
     assert compute_interval_params([orders], start=start_ds, end=end_ds) == {
         orders: [
-            (to_datetime(start_ds), to_datetime("2022-01-10")),
-            (to_datetime("2022-01-16"), to_datetime("2022-02-06")),
+            (to_timestamp(start_ds), to_timestamp("2022-01-10")),
+            (to_timestamp("2022-01-16"), to_timestamp("2022-02-06")),
         ]
     }
 
@@ -96,7 +108,7 @@ def test_interval_params_missing(scheduler: Scheduler, sushi_context_fixed_date:
     assert compute_interval_params(
         sushi_context_fixed_date.snapshots.values(), start=start_ds, end=end_ds
     )[waiters] == [
-        (to_datetime(start_ds), to_datetime("2022-03-02")),
+        (to_timestamp(start_ds), to_timestamp("2022-03-02")),
     ]
 
 
@@ -118,7 +130,9 @@ def test_run(sushi_context_fixed_date: Context, scheduler: Scheduler):
     ) == (0, "Hotate", 5.99)
 
 
-def test_incremental_by_unique_key_kind_dag(mocker: MockerFixture, make_snapshot):
+def test_incremental_by_unique_key_kind_dag(
+    mocker: MockerFixture, make_snapshot, get_batched_missing_intervals
+):
     """
     Test that when given a week of data that it batches dates together.
     """
@@ -144,18 +158,20 @@ def test_incremental_by_unique_key_kind_dag(mocker: MockerFixture, make_snapshot
         max_workers=2,
         default_catalog=None,
     )
-    batches = scheduler.batches(start, end, end)
+    batches = get_batched_missing_intervals(scheduler, start, end, end)
     dag = scheduler._dag(batches)
     assert dag.graph == {
         (
             unique_by_key_snapshot.name,
-            ((to_datetime("2023-01-01"), to_datetime("2023-01-07")), 0),
+            ((to_timestamp("2023-01-01"), to_timestamp("2023-01-07")), 0),
         ): set(),
     }
     mock_state_sync.refresh_snapshot_intervals.assert_called_once()
 
 
-def test_incremental_time_self_reference_dag(mocker: MockerFixture, make_snapshot):
+def test_incremental_time_self_reference_dag(
+    mocker: MockerFixture, make_snapshot, get_batched_missing_intervals
+):
     """
     Test that we always process a day at a time and all future days rely on the previous day
     """
@@ -183,62 +199,62 @@ def test_incremental_time_self_reference_dag(mocker: MockerFixture, make_snapsho
         max_workers=2,
         default_catalog=None,
     )
-    batches = scheduler.batches(start, end, end)
+    batches = get_batched_missing_intervals(scheduler, start, end, end)
     dag = scheduler._dag(batches)
 
     assert dag.graph == {
         # Only run one day at a time and each day relies on the previous days
         (
             incremental_self_snapshot.name,
-            ((to_datetime("2023-01-01"), to_datetime("2023-01-02")), 0),
+            ((to_timestamp("2023-01-01"), to_timestamp("2023-01-02")), 0),
         ): set(),
         (
             incremental_self_snapshot.name,
-            ((to_datetime("2023-01-03"), to_datetime("2023-01-04")), 1),
+            ((to_timestamp("2023-01-03"), to_timestamp("2023-01-04")), 1),
         ): {
             (
                 incremental_self_snapshot.name,
-                ((to_datetime("2023-01-01"), to_datetime("2023-01-02")), 0),
+                ((to_timestamp("2023-01-01"), to_timestamp("2023-01-02")), 0),
             )
         },
         (
             incremental_self_snapshot.name,
-            ((to_datetime("2023-01-04"), to_datetime("2023-01-05")), 2),
+            ((to_timestamp("2023-01-04"), to_timestamp("2023-01-05")), 2),
         ): {
             (
                 incremental_self_snapshot.name,
-                ((to_datetime("2023-01-03"), to_datetime("2023-01-04")), 1),
+                ((to_timestamp("2023-01-03"), to_timestamp("2023-01-04")), 1),
             ),
         },
         (
             incremental_self_snapshot.name,
-            ((to_datetime("2023-01-06"), to_datetime("2023-01-07")), 3),
+            ((to_timestamp("2023-01-06"), to_timestamp("2023-01-07")), 3),
         ): {
             (
                 incremental_self_snapshot.name,
-                ((to_datetime("2023-01-04"), to_datetime("2023-01-05")), 2),
+                ((to_timestamp("2023-01-04"), to_timestamp("2023-01-05")), 2),
             ),
         },
         (
             incremental_self_snapshot.name,
-            ((to_datetime(0), to_datetime(0)), -1),
+            ((to_timestamp(0), to_timestamp(0)), -1),
         ): set(
             [
                 (
                     incremental_self_snapshot.name,
-                    ((to_datetime("2023-01-01"), to_datetime("2023-01-02")), 0),
+                    ((to_timestamp("2023-01-01"), to_timestamp("2023-01-02")), 0),
                 ),
                 (
                     incremental_self_snapshot.name,
-                    ((to_datetime("2023-01-03"), to_datetime("2023-01-04")), 1),
+                    ((to_timestamp("2023-01-03"), to_timestamp("2023-01-04")), 1),
                 ),
                 (
                     incremental_self_snapshot.name,
-                    ((to_datetime("2023-01-04"), to_datetime("2023-01-05")), 2),
+                    ((to_timestamp("2023-01-04"), to_timestamp("2023-01-05")), 2),
                 ),
                 (
                     incremental_self_snapshot.name,
-                    ((to_datetime("2023-01-06"), to_datetime("2023-01-07")), 3),
+                    ((to_timestamp("2023-01-06"), to_timestamp("2023-01-07")), 3),
                 ),
             ]
         ),
@@ -254,14 +270,14 @@ def test_incremental_time_self_reference_dag(mocker: MockerFixture, make_snapsho
             {
                 (
                     '"test_model"',
-                    ((to_datetime("2023-01-01"), to_datetime("2023-01-03")), 0),
+                    ((to_timestamp("2023-01-01"), to_timestamp("2023-01-03")), 0),
                 ): set(),
                 (
                     '"test_model"',
-                    ((to_datetime("2023-01-03"), to_datetime("2023-01-05")), 1),
+                    ((to_timestamp("2023-01-03"), to_timestamp("2023-01-05")), 1),
                 ): set(),
-                ('"test_model"', ((to_datetime("2023-01-05"), to_datetime("2023-01-07")), 2)): {
-                    ('"test_model"', ((to_datetime("2023-01-01"), to_datetime("2023-01-03")), 0)),
+                ('"test_model"', ((to_timestamp("2023-01-05"), to_timestamp("2023-01-07")), 2)): {
+                    ('"test_model"', ((to_timestamp("2023-01-01"), to_timestamp("2023-01-03")), 0)),
                 },
             },
         ),
@@ -271,24 +287,24 @@ def test_incremental_time_self_reference_dag(mocker: MockerFixture, make_snapsho
             {
                 (
                     '"test_model"',
-                    ((to_datetime("2023-01-01"), to_datetime("2023-01-02")), 0),
+                    ((to_timestamp("2023-01-01"), to_timestamp("2023-01-02")), 0),
                 ): set(),
                 (
                     '"test_model"',
-                    ((to_datetime("2023-01-02"), to_datetime("2023-01-03")), 1),
+                    ((to_timestamp("2023-01-02"), to_timestamp("2023-01-03")), 1),
                 ): set(),
                 (
                     '"test_model"',
-                    ((to_datetime("2023-01-03"), to_datetime("2023-01-04")), 2),
+                    ((to_timestamp("2023-01-03"), to_timestamp("2023-01-04")), 2),
                 ): set(),
-                ('"test_model"', ((to_datetime("2023-01-04"), to_datetime("2023-01-05")), 3)): {
-                    ('"test_model"', ((to_datetime("2023-01-01"), to_datetime("2023-01-02")), 0)),
+                ('"test_model"', ((to_timestamp("2023-01-04"), to_timestamp("2023-01-05")), 3)): {
+                    ('"test_model"', ((to_timestamp("2023-01-01"), to_timestamp("2023-01-02")), 0)),
                 },
-                ('"test_model"', ((to_datetime("2023-01-05"), to_datetime("2023-01-06")), 4)): {
-                    ('"test_model"', ((to_datetime("2023-01-02"), to_datetime("2023-01-03")), 1)),
+                ('"test_model"', ((to_timestamp("2023-01-05"), to_timestamp("2023-01-06")), 4)): {
+                    ('"test_model"', ((to_timestamp("2023-01-02"), to_timestamp("2023-01-03")), 1)),
                 },
-                ('"test_model"', ((to_datetime("2023-01-06"), to_datetime("2023-01-07")), 5)): {
-                    ('"test_model"', ((to_datetime("2023-01-03"), to_datetime("2023-01-04")), 2)),
+                ('"test_model"', ((to_timestamp("2023-01-06"), to_timestamp("2023-01-07")), 5)): {
+                    ('"test_model"', ((to_timestamp("2023-01-03"), to_timestamp("2023-01-04")), 2)),
                 },
             },
         ),
@@ -298,27 +314,27 @@ def test_incremental_time_self_reference_dag(mocker: MockerFixture, make_snapsho
             {
                 (
                     '"test_model"',
-                    ((to_datetime("2023-01-01"), to_datetime("2023-01-02")), 0),
+                    ((to_timestamp("2023-01-01"), to_timestamp("2023-01-02")), 0),
                 ): set(),
                 (
                     '"test_model"',
-                    ((to_datetime("2023-01-02"), to_datetime("2023-01-03")), 1),
+                    ((to_timestamp("2023-01-02"), to_timestamp("2023-01-03")), 1),
                 ): set(),
                 (
                     '"test_model"',
-                    ((to_datetime("2023-01-03"), to_datetime("2023-01-04")), 2),
+                    ((to_timestamp("2023-01-03"), to_timestamp("2023-01-04")), 2),
                 ): set(),
                 (
                     '"test_model"',
-                    ((to_datetime("2023-01-04"), to_datetime("2023-01-05")), 3),
+                    ((to_timestamp("2023-01-04"), to_timestamp("2023-01-05")), 3),
                 ): set(),
                 (
                     '"test_model"',
-                    ((to_datetime("2023-01-05"), to_datetime("2023-01-06")), 4),
+                    ((to_timestamp("2023-01-05"), to_timestamp("2023-01-06")), 4),
                 ): set(),
                 (
                     '"test_model"',
-                    ((to_datetime("2023-01-06"), to_datetime("2023-01-07")), 5),
+                    ((to_timestamp("2023-01-06"), to_timestamp("2023-01-07")), 5),
                 ): set(),
             },
         ),
@@ -328,7 +344,7 @@ def test_incremental_time_self_reference_dag(mocker: MockerFixture, make_snapsho
             {
                 (
                     '"test_model"',
-                    ((to_datetime("2023-01-01"), to_datetime("2023-01-07")), 0),
+                    ((to_timestamp("2023-01-01"), to_timestamp("2023-01-07")), 0),
                 ): set(),
             },
         ),
@@ -338,7 +354,7 @@ def test_incremental_time_self_reference_dag(mocker: MockerFixture, make_snapsho
             {
                 (
                     '"test_model"',
-                    ((to_datetime("2023-01-01"), to_datetime("2023-01-07")), 0),
+                    ((to_timestamp("2023-01-01"), to_timestamp("2023-01-07")), 0),
                 ): set(),
             },
         ),
@@ -347,6 +363,7 @@ def test_incremental_time_self_reference_dag(mocker: MockerFixture, make_snapsho
 def test_incremental_batch_concurrency(
     mocker: MockerFixture,
     make_snapshot,
+    get_batched_missing_intervals,
     batch_size: int,
     batch_concurrency: int,
     expected_graph: t.Dict[str, t.Any],
@@ -375,7 +392,7 @@ def test_incremental_batch_concurrency(
         default_catalog=None,
     )
 
-    batches = scheduler.batches(start, end, end)
+    batches = get_batched_missing_intervals(scheduler, start, end, end)
     dag = scheduler._dag(batches)
     graph = {k: v for k, v in dag.graph.items() if k[1][1] != -1}  # exclude the terminal node.}
     assert graph == expected_graph
@@ -392,7 +409,9 @@ def test_circuit_breaker(scheduler: Scheduler):
         )
 
 
-def test_intervals_with_end_date_on_model(mocker: MockerFixture, make_snapshot):
+def test_intervals_with_end_date_on_model(
+    mocker: MockerFixture, make_snapshot, get_batched_missing_intervals
+):
     snapshot: Snapshot = make_snapshot(
         SqlModel(
             name="name",
@@ -415,27 +434,35 @@ def test_intervals_with_end_date_on_model(mocker: MockerFixture, make_snapshot):
 
     # generate for 1 year to show that the returned batches should only cover
     # the range defined on the model itself
-    batches = scheduler.batches(start="2023-01-01", end="2024-01-01")[snapshot]
+    batches = get_batched_missing_intervals(scheduler, start="2023-01-01", end="2024-01-01")[
+        snapshot
+    ]
 
     assert len(batches) == 31  # days in Jan 2023
-    assert batches[0] == (to_datetime("2023-01-01"), to_datetime("2023-01-02"))
-    assert batches[-1] == (to_datetime("2023-01-31"), to_datetime("2023-02-01"))
+    assert batches[0] == (to_timestamp("2023-01-01"), to_timestamp("2023-01-02"))
+    assert batches[-1] == (to_timestamp("2023-01-31"), to_timestamp("2023-02-01"))
 
     # generate for less than 1 month to ensure that the scheduler end date
     # takes precedence over the model end date
-    batches = scheduler.batches(start="2023-01-01", end="2023-01-10")[snapshot]
+    batches = get_batched_missing_intervals(scheduler, start="2023-01-01", end="2023-01-10")[
+        snapshot
+    ]
 
     assert len(batches) == 10
-    assert batches[0] == (to_datetime("2023-01-01"), to_datetime("2023-01-02"))
-    assert batches[-1] == (to_datetime("2023-01-10"), to_datetime("2023-01-11"))
+    assert batches[0] == (to_timestamp("2023-01-01"), to_timestamp("2023-01-02"))
+    assert batches[-1] == (to_timestamp("2023-01-10"), to_timestamp("2023-01-11"))
 
     # generate for the last day of range
-    batches = scheduler.batches(start="2023-01-31", end="2023-01-31")[snapshot]
+    batches = get_batched_missing_intervals(scheduler, start="2023-01-31", end="2023-01-31")[
+        snapshot
+    ]
     assert len(batches) == 1
-    assert batches[0] == (to_datetime("2023-01-31"), to_datetime("2023-02-01"))
+    assert batches[0] == (to_timestamp("2023-01-31"), to_timestamp("2023-02-01"))
 
     # generate for future days to ensure no future batches are loaded
-    snapshot_to_batches = scheduler.batches(start="2023-02-01", end="2023-02-28")
+    snapshot_to_batches = get_batched_missing_intervals(
+        scheduler, start="2023-02-01", end="2023-02-28"
+    )
     assert len(snapshot_to_batches) == 0
 
 
@@ -493,7 +520,8 @@ def test_contiguous_intervals():
 
 
 def test_check_ready_intervals(mocker: MockerFixture):
-    from sqlmesh.core.scheduler import _check_ready_intervals, Interval
+    from sqlmesh.core.scheduler import _check_ready_intervals
+    from sqlmesh.core.snapshot.definition import Interval
 
     def const_signal(const):
         signal_mock = mocker.Mock()
@@ -629,7 +657,7 @@ def test_audit_failure_notifications(
 
 def test_signal_factory(mocker: MockerFixture, make_snapshot):
     class AlwaysReadySignal(Signal):
-        def check_intervals(self, batch: Batch):
+        def check_intervals(self, batch: DatetimeRanges):
             return True
 
     signal_factory_invoked = 0
@@ -662,8 +690,17 @@ def test_signal_factory(mocker: MockerFixture, make_snapshot):
         state_sync=mocker.MagicMock(),
         max_workers=2,
         default_catalog=None,
+        console=mocker.MagicMock(),
+    )
+    merged_intervals = scheduler.merged_missing_intervals(start, end, end)
+    assert len(merged_intervals) == 1
+    scheduler.run_merged_intervals(
+        merged_intervals=merged_intervals,
+        deployability_index=DeployabilityIndex.all_deployable(),
+        environment_naming_info=EnvironmentNamingInfo(),
+        start=start,
+        end=end,
     )
-    scheduler.batches(start, end, end)
 
     assert signal_factory_invoked > 0
 
@@ -697,12 +734,12 @@ def test_interval_diff():
     assert interval_diff([(1, 2), (2, 3)], [(2, 3)], uninterrupted=True) == [(1, 2)]
 
 
-def test_signal_intervals(mocker: MockerFixture, make_snapshot):
+def test_signal_intervals(mocker: MockerFixture, make_snapshot, get_batched_missing_intervals):
     class TestSignal(Signal):
         def __init__(self, signal: t.Dict):
             self.name = signal["kind"]
 
-        def check_intervals(self, batch: Batch):
+        def check_intervals(self, batch: DatetimeRanges):
             if self.name == "a":
                 return [batch[0], batch[1]]
             if self.name == "b":
@@ -747,16 +784,21 @@ def test_signal_intervals(mocker: MockerFixture, make_snapshot):
         nodes={a.name: a.model, b.name: b.model, c.name: c.model},
     )
 
-    batches = compute_interval_params(
-        [c, a, d, b],
-        start="2023-01-01",
-        end="2023-01-03",
+    snapshot_evaluator = SnapshotEvaluator(adapter=mocker.MagicMock(), ddl_concurrent_tasks=1)
+    scheduler = Scheduler(
+        snapshots=[a, b, c, d],
+        snapshot_evaluator=snapshot_evaluator,
+        state_sync=mocker.MagicMock(),
+        max_workers=2,
+        default_catalog=None,
         signal_factory=lambda signal: TestSignal(signal),
     )
 
+    batches = get_batched_missing_intervals(scheduler, "2023-01-01", "2023-01-03", None)
+
     assert batches == {
-        a: [(to_datetime("2023-01-01"), to_datetime("2023-01-03"))],
-        b: [(to_datetime("2023-01-01 23:00:00"), to_datetime("2023-01-04"))],
-        c: [(to_datetime("2023-01-02"), to_datetime("2023-01-03"))],
+        a: [(to_timestamp("2023-01-01"), to_timestamp("2023-01-03"))],
+        b: [(to_timestamp("2023-01-01 23:00:00"), to_timestamp("2023-01-04"))],
+        c: [(to_timestamp("2023-01-02"), to_timestamp("2023-01-03"))],
         d: [],
     }