Collect selected snapshot triggers

Author: treysp

sqlmesh/core/console.py

@@ -3819,7 +3819,17 @@ def update_snapshot_evaluation_progress(
         audit_only: bool = False,
         snapshot_evaluation_triggers: t.Optional[SnapshotEvaluationTriggers] = None,
     ) -> None:
-        message = f"Evaluating {snapshot.name} | batch={batch_idx} | duration={duration_ms}ms | num_audits_passed={num_audits_passed} | num_audits_failed={num_audits_failed}"
+        message = f"Evaluated {snapshot.name} | batch={batch_idx} | duration={duration_ms}ms | num_audits_passed={num_audits_passed} | num_audits_failed={num_audits_failed}"
+
+        if snapshot_evaluation_triggers:
+            if snapshot_evaluation_triggers.ignore_cron_flag is not None:
+                message += f" | ignore_cron_flag={snapshot_evaluation_triggers.ignore_cron_flag}"
+            if snapshot_evaluation_triggers.cron_ready is not None:
+                message += f" | cron_ready={snapshot_evaluation_triggers.cron_ready}"
+            if snapshot_evaluation_triggers.auto_restatement_triggers:
+                message += f" | auto_restatement_triggers={','.join(trigger.name for trigger in snapshot_evaluation_triggers.auto_restatement_triggers)}"
+            if snapshot_evaluation_triggers.select_snapshot_triggers:
+                message += f" | select_snapshot_triggers={','.join(trigger.name for trigger in snapshot_evaluation_triggers.select_snapshot_triggers)}"
 
         if snapshot_evaluation_triggers:
             if snapshot_evaluation_triggers.auto_restatement_triggers:
@@ -3828,7 +3838,7 @@ def update_snapshot_evaluation_progress(
                 message += f" | select_snapshot_triggers={','.join(trigger.name for trigger in snapshot_evaluation_triggers.select_snapshot_triggers)}"
 
         if audit_only:
-            message = f"Auditing {snapshot.name} duration={duration_ms}ms | num_audits_passed={num_audits_passed} | num_audits_failed={num_audits_failed}"
+            message = f"Audited {snapshot.name} duration={duration_ms}ms | num_audits_passed={num_audits_passed} | num_audits_failed={num_audits_failed}"
 
         self._write(message)
 

sqlmesh/core/context.py

@@ -2898,9 +2898,10 @@ def _select_models_for_run(
             dag.add(fqn, model.depends_on)
         model_selector = self._new_selector(models=models, dag=dag)
         result = set(model_selector.expand_model_selections(select_models))
-        if not no_auto_upstream:
-            result_with_upstream = set(dag.subdag(*result))
-        return result, result_with_upstream - result
+        if no_auto_upstream:
+            return result, set()
+        result_with_upstream = set(dag.subdag(*result))
+        return result_with_upstream, result_with_upstream - result
 
     @cached_property
     def _project_type(self) -> str:

sqlmesh/core/plan/stages.py

@@ -553,7 +553,7 @@ def _missing_intervals(
         snapshots_by_name: t.Dict[str, Snapshot],
         deployability_index: DeployabilityIndex,
     ) -> SnapshotToIntervals:
-        return merged_missing_intervals(
+        missing_intervals, _ = merged_missing_intervals(
             snapshots=snapshots_by_name.values(),
             start=plan.start,
             end=plan.end,
@@ -568,6 +568,7 @@ def _missing_intervals(
             start_override_per_model=plan.start_override_per_model,
             end_override_per_model=plan.end_override_per_model,
         )
+        return missing_intervals
 
     def _get_audit_only_snapshots(
         self, new_snapshots: t.Dict[SnapshotId, Snapshot]

sqlmesh/core/scheduler.py

@@ -147,7 +147,7 @@ def merged_missing_intervals(
         ignore_cron: bool = False,
         end_bounded: bool = False,
         selected_snapshots: t.Optional[t.Set[str]] = None,
-    ) -> SnapshotToIntervals:
+    ) -> t.Tuple[SnapshotToIntervals, t.List[SnapshotId]]:
         """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,
@@ -167,8 +167,11 @@ def merged_missing_intervals(
             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.
             selected_snapshots: A set of snapshot names to run. If not provided, all snapshots will be run.
+
+        Returns:
+            A tuple containing a dict containing all snapshots needing to be run with their associated interval params and a list of snapshots that are ready to run based on their naive cron schedule (ignoring plan/run context and other attributes).
         """
-        snapshots_to_intervals = merged_missing_intervals(
+        snapshots_to_intervals, snapshots_naive_cron_ready = merged_missing_intervals(
             snapshots=self.snapshot_per_version.values(),
             start=start,
             end=end,
@@ -186,7 +189,7 @@ def merged_missing_intervals(
             snapshots_to_intervals = {
                 s: i for s, i in snapshots_to_intervals.items() if s.name in selected_snapshots
             }
-        return snapshots_to_intervals
+        return snapshots_to_intervals, snapshots_naive_cron_ready
 
     def evaluate(
         self,
@@ -755,7 +758,7 @@ def _run_or_audit(
                 {s.name_version: s.next_auto_restatement_ts for s in self.snapshots.values()}
             )
 
-        merged_intervals = self.merged_missing_intervals(
+        merged_intervals, snapshots_naive_cron_ready = self.merged_missing_intervals(
             start,
             end,
             execution_time,
@@ -770,9 +773,7 @@ def _run_or_audit(
         if not merged_intervals:
             return CompletionStatus.NOTHING_TO_DO
 
-        merged_intervals_snapshots = {
-            snapshot.snapshot_id: snapshot for snapshot in merged_intervals.keys()
-        }
+        merged_intervals_snapshots = {snapshot.snapshot_id for snapshot in merged_intervals}
         select_snapshot_triggers: t.Dict[SnapshotId, t.List[SnapshotId]] = {}
         if selected_snapshots and selected_snapshots_auto_upstream:
             # actually selected snapshots are their own triggers
@@ -788,24 +789,25 @@ def _run_or_audit(
                 ]
             }
 
-            # trace upstream by reversing dag of all snapshots to evaluate
-            reversed_intervals_dag = snapshots_to_dag(merged_intervals_snapshots.values()).reversed
-            for s_id in reversed_intervals_dag:
-                if s_id not in select_snapshot_triggers:
-                    triggers = []
-                    for parent_s_id in merged_intervals_snapshots[s_id].parents:
-                        triggers.extend(select_snapshot_triggers[parent_s_id])
+            # trace upstream by walking downstream on reversed dag
+            reversed_dag = snapshots_to_dag(self.snapshots.values()).reversed
+            for s_id in reversed_dag:
+                if s_id in merged_intervals_snapshots:
+                    triggers = select_snapshot_triggers.get(s_id, [])
+                    for parent_s_id in reversed_dag.graph.get(s_id, set()):
+                        triggers.extend(select_snapshot_triggers.get(parent_s_id, []))
                     select_snapshot_triggers[s_id] = list(dict.fromkeys(triggers))
 
         all_snapshot_triggers: t.Dict[SnapshotId, SnapshotEvaluationTriggers] = {
             s_id: SnapshotEvaluationTriggers(
-                ignore_cron=ignore_cron,
+                ignore_cron_flag=ignore_cron,
+                cron_ready=s_id in snapshots_naive_cron_ready,
                 auto_restatement_triggers=auto_restatement_triggers.get(s_id, []),
                 select_snapshot_triggers=select_snapshot_triggers.get(s_id, []),
             )
             for s_id in merged_intervals_snapshots
-            if ignore_cron or s_id in auto_restatement_triggers or s_id in select_snapshot_triggers
         }
+
         errors, _ = self.run_merged_intervals(
             merged_intervals=merged_intervals,
             deployability_index=deployability_index,
@@ -967,7 +969,7 @@ def merged_missing_intervals(
     end_override_per_model: t.Optional[t.Dict[str, datetime]] = None,
     ignore_cron: bool = False,
     end_bounded: bool = False,
-) -> SnapshotToIntervals:
+) -> t.Tuple[SnapshotToIntervals, t.List[SnapshotId]]:
     """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,
@@ -1017,7 +1019,7 @@ def compute_interval_params(
     end_override_per_model: t.Optional[t.Dict[str, datetime]] = None,
     ignore_cron: bool = False,
     end_bounded: bool = False,
-) -> SnapshotToIntervals:
+) -> t.Tuple[SnapshotToIntervals, t.List[SnapshotId]]:
     """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,
@@ -1039,7 +1041,7 @@ def compute_interval_params(
             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.
+        A tuple containing a dict containing all snapshots needing to be run with their associated interval params and a list of snapshots that are ready to run based on their naive cron schedule (ignoring plan/run context and other attributes).
     """
     snapshot_merged_intervals = {}
 
@@ -1067,7 +1069,11 @@ def compute_interval_params(
             contiguous_batch.append((next_batch[0][0], next_batch[-1][-1]))
         snapshot_merged_intervals[snapshot] = contiguous_batch
 
-    return snapshot_merged_intervals
+    snapshots_naive_cron_ready = [
+        snap.snapshot_id for snap in missing_intervals(snapshots, execution_time=execution_time)
+    ]
+
+    return snapshot_merged_intervals, snapshots_naive_cron_ready
 
 
 def interval_diff(

sqlmesh/core/snapshot/definition.py

@@ -328,11 +328,10 @@ def table_name_for_environment(
 
 
 class SnapshotEvaluationTriggers(PydanticModel):
-    ignore_cron: bool
+    ignore_cron_flag: t.Optional[bool] = None
+    cron_ready: t.Optional[bool] = None
     auto_restatement_triggers: t.List[SnapshotId] = []
     select_snapshot_triggers: t.List[SnapshotId] = []
-    directly_modified_triggers: t.List[SnapshotId] = []
-    manual_restatement_triggers: t.List[SnapshotId] = []
 
 
 class SnapshotInfoMixin(ModelKindMixin):