Fix: Use the default start in plan when there are paused forward-only snapshots (#1548)

Author: izeigerman

sqlmesh/core/plan/definition.py

@@ -107,12 +107,17 @@ def __init__(
         self.categorizer_config = categorizer_config or CategorizerConfig()
         self.auto_categorization_enabled = auto_categorization_enabled
         self.include_unmodified = include_unmodified
+
+        self.__snapshot_mapping: t.Optional[t.Dict[str, Snapshot]] = None
+        self.__dag: t.Optional[DAG[str]] = None
+
+        self._start = start
+        if not self._start and is_dev and (forward_only or self._has_paused_forward_only()):
+            self._start = default_start or yesterday_ds()
+
+        self._end = end if end or not is_dev else (default_end or now())
         self._restate_models = set(restate_models or [])
         self._effective_from: t.Optional[TimeLike] = None
-        self._start = (
-            start if start or not (is_dev and forward_only) else (default_start or yesterday_ds())
-        )
-        self._end = end if end or not is_dev else (default_end or now())
         self._execution_time = execution_time or now()
         self._apply = apply
         self.__missing_intervals: t.Optional[t.Dict[t.Tuple[str, str], Intervals]] = None
@@ -215,7 +220,15 @@ def snapshots(self) -> t.List[Snapshot]:
     @property
     def _snapshot_mapping(self) -> t.Dict[str, Snapshot]:
         """Gets a mapping of snapshot name to snapshot."""
-        return self.__snapshot_mapping
+        return self.__snapshot_mapping or self.context_diff.snapshots
+
+    @property
+    def _dag(self) -> DAG[str]:
+        if self.__dag is None:
+            self.__dag = DAG()
+            for name, snapshot in self._snapshot_mapping.items():
+                self.__dag.add(name, snapshot.node.depends_on)
+        return self.__dag
 
     @property
     def new_snapshots(self) -> t.List[Snapshot]:
@@ -649,7 +662,7 @@ def _ensure_new_env_with_changes(self) -> None:
     def _refresh_dag_and_ignored_snapshots(self) -> None:
         self._restatements = {}
         (
-            self._dag,
+            self.__dag,
             self._snapshots,
             self._new_snapshots,
             self.__snapshot_mapping,
@@ -715,6 +728,12 @@ def _build_snapshots_and_dag(
             ignored_snapshot_names,
         )
 
+    def _has_paused_forward_only(self) -> bool:
+        for name, snapshot in self._snapshot_mapping.items():
+            if snapshot.is_paused_forward_only or self._is_forward_only_model(name):
+                return True
+        return False
+
     def _is_forward_only_model(self, model_name: str) -> bool:
         def _is_forward_only_expected(snapshot: Snapshot) -> bool:
             # Returns True if the snapshot is not categorized yet but is expected

tests/core/test_plan.py

@@ -633,8 +633,9 @@ def test_indirectly_modified_forward_only_model(make_snapshot, mocker: MockerFix
     context_diff_mock.previous_plan_id = "previous_plan_id"
     context_diff_mock.directly_modified.side_effect = lambda name: name == "a"
 
-    plan = Plan(context_diff_mock, is_dev=True)
+    plan = Plan(context_diff_mock, is_dev=True, default_start="2023-01-01")
     assert plan.indirectly_modified == {"a": {"b", "c"}}
+    assert plan.start == "2023-01-01"
 
     assert len(plan.directly_modified) == 1
     assert plan.directly_modified[0].snapshot_id == updated_snapshot_a.snapshot_id
@@ -666,7 +667,8 @@ def test_added_model_with_forward_only_parent(make_snapshot, mocker: MockerFixtu
     context_diff_mock.environment = "test_dev"
     context_diff_mock.previous_plan_id = "previous_plan_id"
 
-    Plan(context_diff_mock)
+    plan = Plan(context_diff_mock, is_dev=True, default_start="2023-01-01")
+    assert plan.start == "2023-01-01"
     assert snapshot_b.change_category == SnapshotChangeCategory.FORWARD_ONLY