Fix: Correctly recreate views for snapshots categorized as indirect non-breaking (#1285)

Author: izeigerman

sqlmesh/core/plan/definition.py

@@ -448,7 +448,7 @@ def _add_restatements(self, restate_models: t.Iterable[str]) -> None:
 
             snapshots = self.context_diff.snapshots
             downstream = [
-                d for d in downstream if snapshots[d].is_materialized and not snapshots[d].is_seed
+                d for d in downstream if not snapshots[d].is_symbolic and not snapshots[d].is_seed
             ]
 
             if not self.is_dev:

sqlmesh/core/scheduler.py

@@ -209,7 +209,7 @@ def run(
         else:
             environment_naming_info = environment
 
-        is_dev = environment != c.PROD
+        is_dev = environment_naming_info.name != c.PROD
         execution_time = execution_time or now()
         batches = self.batches(
             start,

sqlmesh/core/snapshot/evaluator.py

@@ -419,7 +419,10 @@ def _create_snapshot(
     def _migrate_snapshot(
         self, snapshot: Snapshot, snapshots: t.Dict[SnapshotId, Snapshot]
     ) -> None:
-        if snapshot.change_category != SnapshotChangeCategory.FORWARD_ONLY:
+        if not snapshot.is_paused or snapshot.change_category not in (
+            SnapshotChangeCategory.FORWARD_ONLY,
+            SnapshotChangeCategory.INDIRECT_NON_BREAKING,
+        ):
             return
 
         parent_snapshots_by_name = {

tests/core/test_plan.py

@@ -145,7 +145,7 @@ def test_restate_models(sushi_context_pre_scheduling: Context):
     plan = sushi_context_pre_scheduling.plan(
         restate_models=["sushi.waiter_revenue_by_day"], no_prompts=True
     )
-    assert plan.restatements == {"sushi.waiter_revenue_by_day"}
+    assert plan.restatements == {"sushi.waiter_revenue_by_day", "sushi.top_waiters"}
     assert plan.requires_backfill
 
     with pytest.raises(PlanError, match=r"Cannot restate from 'unknown_model'.*"):
@@ -157,7 +157,7 @@ def test_restate_model_with_merge_strategy(make_snapshot, mocker: MockerFixture)
         SqlModel(
             name="a",
             query=parse_one("select 1, key"),
-            kind="VIEW",
+            kind="EMBEDDED",
         )
     )
 

tests/core/test_snapshot_evaluator.py

@@ -427,7 +427,13 @@ def columns(table_name: t.Union[str, exp.Table]) -> t.Dict[str, exp.DataType]:
     )
 
 
-def test_migrate_view(mocker: MockerFixture, make_snapshot):
+@pytest.mark.parametrize(
+    "change_category",
+    [SnapshotChangeCategory.FORWARD_ONLY, SnapshotChangeCategory.INDIRECT_NON_BREAKING],
+)
+def test_migrate_view(
+    mocker: MockerFixture, make_snapshot, change_category: SnapshotChangeCategory
+):
     connection_mock = mocker.NonCallableMock()
     cursor_mock = mocker.Mock()
     connection_mock.cursor.return_value = cursor_mock
@@ -442,7 +448,7 @@ def test_migrate_view(mocker: MockerFixture, make_snapshot):
         query=parse_one("SELECT c, a FROM tbl"),
     )
     snapshot = make_snapshot(model, version="1")
-    snapshot.change_category = SnapshotChangeCategory.FORWARD_ONLY
+    snapshot.change_category = change_category
 
     evaluator.migrate([snapshot], {})