Feat: Check freshness for mixed external & sqlmesh models

Author: VaggelisD

.circleci/continue_config.yml

@@ -313,10 +313,10 @@ workflows:
                 - athena
                 - fabric
                 - gcp-postgres
-          filters:
-            branches:
-              only:
-                - main
+          # filters:
+          #   branches:
+          #     only:
+          #       - main
       - ui_style
       - ui_test
       - vscode_test

sqlmesh/core/scheduler.py

@@ -373,11 +373,19 @@ def batch_intervals(
                 is_restatement=is_restatement,
             )
 
+            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)
 
@@ -923,6 +931,7 @@ 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.
 
@@ -965,6 +974,7 @@ 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

@@ -2,12 +2,14 @@
 
 import typing as t
 from sqlmesh.utils import UniqueKeyDict, registry_decorator
+from sqlmesh.utils.errors import MissingSourceError
 
 if t.TYPE_CHECKING:
     from sqlmesh.core.context import ExecutionContext
     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):
@@ -42,7 +44,17 @@ class signal(registry_decorator):
 
 
 @signal()
-def freshness(batch: DatetimeRanges, snapshot: Snapshot, context: ExecutionContext) -> bool:
+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:
+    - Any upstream SQLMesh model has available intervals to compute i.e is fresh
+    - Any upstream external model has been altered since the last time the model was evaluated
+    """
     adapter = context.engine_adapter
     if context.is_restatement or not adapter.SUPPORTS_METADATA_TABLE_LAST_MODIFIED_TS:
         return True
@@ -54,24 +66,38 @@ def freshness(batch: DatetimeRanges, snapshot: Snapshot, context: ExecutionConte
         if deployability_index.is_deployable(snapshot)
         else snapshot.dev_last_altered_ts
     )
+
     if not last_altered_ts:
         return True
 
     parent_snapshots = {context.snapshots[p.name] for p in snapshot.parents}
-    if len(parent_snapshots) != len(snapshot.node.depends_on) or not all(
-        p.is_external for p in parent_snapshots
-    ):
+    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
 
-    # Finding new data means that the upstream depedencies have been altered
-    # since the last time the model was evaluated
-    upstream_dep_has_new_data = any(
-        upstream_last_altered_ts > last_altered_ts
-        for upstream_last_altered_ts in adapter.get_table_last_modified_ts(
-            [p.name for p in parent_snapshots]
+    external_parent_snapshots = {p for p in parent_snapshots if p.is_external}
+    upstream_parent_snapshots = parent_snapshots - external_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
+        return True
+
+    if external_parent_snapshots:
+        external_last_altered_timestamps = adapter.get_table_last_modified_ts(
+            [sp.name for sp in external_parent_snapshots]
+        )
+
+        if len(external_last_altered_timestamps) != len(external_parent_snapshots):
+            raise MissingSourceError(
+                f"Expected {len(external_parent_snapshots)} 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
         )
-    )
 
-    # Returning true is a no-op, returning False nullifies the batch so the model will not be evaluated.
-    return upstream_dep_has_new_data
+    return False

sqlmesh/core/snapshot/definition.py

@@ -2451,6 +2451,7 @@ 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 = []
@@ -2467,6 +2468,7 @@ 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))