Refactor parent signals

Author: VaggelisD

sqlmesh/core/context.py

@@ -153,6 +153,8 @@
     )
     from sqlmesh.core.snapshot import Node
 
+    from sqlmesh.core.snapshot.definition import Intervals
+
     ModelOrSnapshot = t.Union[str, Model, Snapshot]
     NodeOrSnapshot = t.Union[str, Model, StandaloneAudit, Snapshot]
 
@@ -275,6 +277,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,
         variables: t.Optional[t.Dict[str, t.Any]] = None,
         blueprint_variables: t.Optional[t.Dict[str, t.Any]] = None,
     ):
@@ -286,6 +289,7 @@ def __init__(
         self._variables = variables or {}
         self._blueprint_variables = blueprint_variables or {}
         self._is_restatement = is_restatement
+        self._parent_intervals = parent_intervals
 
     @property
     def default_dialect(self) -> t.Optional[str]:
@@ -314,6 +318,10 @@ def gateway(self) -> t.Optional[str]:
     def is_restatement(self) -> t.Optional[bool]:
         return self._is_restatement
 
+    @property
+    def parent_intervals(self) -> t.Optional[t.List[Intervals]]:
+        return self._parent_intervals
+
     def var(self, var_name: str, default: t.Optional[t.Any] = None) -> t.Optional[t.Any]:
         """Returns a variable value."""
         return self._variables.get(var_name.lower(), default)

sqlmesh/core/scheduler.py

@@ -364,28 +364,28 @@ 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:
+                    continue
+                _, p_intervals = snapshot_intervals[parent.snapshot_id]
+                parent_intervals.append(p_intervals)
+
             context = ExecutionContext(
                 adapter,
                 self.snapshots_by_name,
                 deployability_index,
                 default_dialect=adapter.dialect,
                 default_catalog=self.default_catalog,
                 is_restatement=is_restatement,
+                parent_intervals=parent_intervals,
             )
 
-            parent_intervals = []
-            for parent in snapshot.parents:
-                if parent.snapshot_id not in snapshot_intervals:
-                    continue
-                _, p_intervals = snapshot_intervals[parent.snapshot_id]
-                parent_intervals.append(p_intervals)
-
             intervals = self._check_ready_intervals(
                 snapshot,
                 intervals,
                 context,
                 environment_naming_info,
-                parent_intervals=parent_intervals,
             )
             unready -= set(intervals)
 
@@ -931,7 +931,6 @@ def _check_ready_intervals(
         intervals: Intervals,
         context: ExecutionContext,
         environment_naming_info: EnvironmentNamingInfo,
-        parent_intervals: t.Optional[t.List[Intervals]] = None,
     ) -> Intervals:
         """Checks if the intervals are ready for evaluation for the given snapshot.
 
@@ -974,7 +973,6 @@ def _check_ready_intervals(
                     dialect=snapshot.model.dialect,
                     path=snapshot.model._path,
                     snapshot=snapshot,
-                    parent_intervals=parent_intervals,
                     kwargs=kwargs,
                 )
             except SQLMeshError as e:

sqlmesh/core/signal.py

@@ -9,7 +9,6 @@
     from sqlmesh.core.snapshot.definition import Snapshot
     from sqlmesh.utils.date import DatetimeRanges
     from sqlmesh.core.snapshot.definition import DeployabilityIndex
-    from sqlmesh.core.snapshot.definition import Intervals
 
 
 class signal(registry_decorator):
@@ -48,7 +47,6 @@ def freshness(
     batch: DatetimeRanges,
     snapshot: Snapshot,
     context: ExecutionContext,
-    parent_intervals: t.Optional[t.List[Intervals]] = None,
 ) -> bool:
     """
     Implements model freshness as a signal, i.e it considers this model to be fresh if:
@@ -71,33 +69,30 @@ def freshness(
         return True
 
     parent_snapshots = {context.snapshots[p.name] for p in snapshot.parents}
-    if len(parent_snapshots) != len(snapshot.node.depends_on):
-        # The mismatch can happen if e.g an external model is not registered in the project
-        return True
 
-    external_parent_snapshots = {p for p in parent_snapshots if p.is_external}
-    upstream_parent_snapshots = parent_snapshots - external_parent_snapshots
+    upstream_parent_snapshots = {p for p in parent_snapshots if not p.is_external}
+    external_parents = snapshot.node.depends_on - {p.name for p in upstream_parent_snapshots}
 
-    if upstream_parent_snapshots and parent_intervals:
-        # At least one upstream sqlmesh model has intervals to compute (i.e is not fresh),
-        # so the current model should be considered fresh
+    if context.parent_intervals:
+        # At least one upstream sqlmesh model has intervals to compute (i.e is fresh),
+        # so the current model is considered fresh too
         return True
 
-    if external_parent_snapshots:
+    if external_parents:
         external_last_altered_timestamps = adapter.get_table_last_modified_ts(
-            [sp.name for sp in external_parent_snapshots]
+            list(external_parents)
         )
 
-        if len(external_last_altered_timestamps) != len(external_parent_snapshots):
+        if len(external_last_altered_timestamps) != len(external_parents):
             raise MissingSourceError(
-                f"Expected {len(external_parent_snapshots)} sources to be present, but got {len(external_last_altered_timestamps)}."
+                f"Expected {len(external_parents)} sources to be present, but got {len(external_last_altered_timestamps)}."
             )
 
         # Finding new data means that the upstream depedencies have been altered
         # since the last time the model was evaluated
         return any(
-            upstream_last_altered_ts > last_altered_ts
-            for upstream_last_altered_ts in external_last_altered_timestamps
+            external_last_altered_ts > last_altered_ts
+            for external_last_altered_ts in external_last_altered_timestamps
         )
 
     return False

sqlmesh/core/snapshot/definition.py

@@ -2451,7 +2451,6 @@ def check_ready_intervals(
     dialect: DialectType = None,
     path: t.Optional[Path] = None,
     snapshot: t.Optional[Snapshot] = None,
-    parent_intervals: t.Optional[t.List[Intervals]] = None,
     kwargs: t.Optional[t.Dict] = None,
 ) -> Intervals:
     checked_intervals: Intervals = []
@@ -2468,7 +2467,6 @@ def check_ready_intervals(
                 provided_kwargs=(kwargs or {}),
                 context=context,
                 snapshot=snapshot,
-                parent_intervals=parent_intervals,
             )
         except Exception as ex:
             raise SignalEvalError(format_evaluated_code_exception(ex, python_env))