Refactor parent_intervals to use snapshot_batches

Author: VaggelisD

sqlmesh/core/context.py

@@ -278,7 +278,7 @@ def __init__(
         default_dialect: t.Optional[str] = None,
         default_catalog: t.Optional[str] = None,
         is_restatement: t.Optional[bool] = None,
-        parent_intervals: t.Optional[t.List[Intervals]] = None,
+        parent_intervals: t.Optional[Intervals] = None,
         variables: t.Optional[t.Dict[str, t.Any]] = None,
         blueprint_variables: t.Optional[t.Dict[str, t.Any]] = None,
     ):
@@ -320,7 +320,7 @@ def is_restatement(self) -> t.Optional[bool]:
         return self._is_restatement
 
     @property
-    def parent_intervals(self) -> t.Optional[t.List[Intervals]]:
+    def parent_intervals(self) -> t.Optional[Intervals]:
         return self._parent_intervals
 
     def var(self, var_name: str, default: t.Optional[t.Any] = None) -> t.Optional[t.Any]:

sqlmesh/core/scheduler.py

@@ -352,7 +352,7 @@ def batch_intervals(
             )
             for snapshot, intervals in merged_intervals.items()
         }
-        snapshot_batches = {}
+        snapshot_batches: t.Dict[Snapshot, Intervals] = {}
         all_unready_intervals: t.Dict[str, set[Interval]] = {}
         for snapshot_id in dag:
             if snapshot_id not in snapshot_intervals:
@@ -364,12 +364,13 @@ def batch_intervals(
 
             adapter = self.snapshot_evaluator.get_adapter(snapshot.model_gateway)
 
-            parent_intervals = []
-            for parent in snapshot.parents:
-                if parent.snapshot_id not in snapshot_intervals:
+            parent_intervals: Intervals = []
+            for parent_id in snapshot.parents:
+                parent_snapshot, _ = snapshot_intervals.get(parent_id, (None, []))
+                if not parent_snapshot or parent_snapshot.is_external:
                     continue
-                _, p_intervals = snapshot_intervals[parent.snapshot_id]
-                parent_intervals.append(p_intervals)
+
+                parent_intervals.extend(snapshot_batches[parent_snapshot])
 
             context = ExecutionContext(
                 adapter,