Fix: Skip preview for metadata-only changes (#2741)

Author: izeigerman

sqlmesh/core/plan/builder.py

@@ -321,7 +321,13 @@ def is_restateable_snapshot(snapshot: Snapshot) -> bool:
             restate_models = {
                 s.name
                 for s in self._context_diff.new_snapshots.values()
-                if s.is_materialized and (self._forward_only or s.model.forward_only)
+                if s.is_materialized
+                and (self._forward_only or s.model.forward_only)
+                and (
+                    # Metadata changes should not be previewed.
+                    self._context_diff.directly_modified(s.name)
+                    or self._context_diff.indirectly_modified(s.name)
+                )
             }
             is_preview = True
 

tests/core/test_integration.py

@@ -395,6 +395,34 @@ def test_full_history_restatement_model_regular_plan_preview_enabled(
     context.apply(plan)
 
 
+@freeze_time("2023-01-08 15:00:00")
+def test_metadata_changed_regular_plan_preview_enabled(init_and_plan_context: t.Callable):
+    context, plan = init_and_plan_context("examples/sushi")
+    context.apply(plan)
+
+    model_name = "sushi.waiter_revenue_by_day"
+
+    model = context.get_model(model_name)
+    model = model.copy(update={"owner": "new_owner"})
+
+    context.upsert_model(model)
+    snapshot = context.get_snapshot(model, raise_if_missing=True)
+    top_waiters_snapshot = context.get_snapshot("sushi.top_waiters", raise_if_missing=True)
+
+    plan = context.plan("dev", no_prompts=True, skip_tests=True, enable_preview=True)
+    assert len(plan.new_snapshots) == 2
+    assert (
+        plan.context_diff.snapshots[snapshot.snapshot_id].change_category
+        == SnapshotChangeCategory.FORWARD_ONLY
+    )
+    assert (
+        plan.context_diff.snapshots[top_waiters_snapshot.snapshot_id].change_category
+        == SnapshotChangeCategory.FORWARD_ONLY
+    )
+    assert not plan.missing_intervals
+    assert not plan.restatements
+
+
 @freeze_time("2023-01-08 15:00:00")
 def test_hourly_model_with_lookback_no_backfill_in_dev(init_and_plan_context: t.Callable):
     context, plan = init_and_plan_context("examples/sushi")

tests/core/test_plan.py

@@ -18,6 +18,7 @@
     SeedKind,
     SeedModel,
     SqlModel,
+    ModelKindName,
 )
 from sqlmesh.core.model.kind import OnDestructiveChange
 from sqlmesh.core.model.seed import Seed
@@ -91,12 +92,22 @@ def test_forward_only_plan_sets_version(make_snapshot, mocker: MockerFixture):
 
 
 def test_forward_only_dev(make_snapshot, mocker: MockerFixture):
-    snapshot_a = make_snapshot(
+    snapshot = make_snapshot(
         SqlModel(
             name="a",
             dialect="duckdb",
             query=parse_one("select 1, ds"),
-            kind=IncrementalByTimeRangeKind(time_column="ds"),
+            kind=dict(name=ModelKindName.INCREMENTAL_BY_TIME_RANGE, time_column="ds"),
+        )
+    )
+
+    # Simulate a direct change.
+    updated_snapshot = make_snapshot(
+        SqlModel(
+            **{
+                **snapshot.model.dict(),
+                "query": parse_one("select 2, ds"),
+            }
         )
     )
 
@@ -111,9 +122,9 @@ def test_forward_only_dev(make_snapshot, mocker: MockerFixture):
         create_from="prod",
         added=set(),
         removed_snapshots={},
-        modified_snapshots={snapshot_a.name: (snapshot_a, snapshot_a)},
-        snapshots={snapshot_a.snapshot_id: snapshot_a},
-        new_snapshots={snapshot_a.snapshot_id: snapshot_a},
+        modified_snapshots={snapshot.name: (updated_snapshot, snapshot)},
+        snapshots={updated_snapshot.snapshot_id: updated_snapshot},
+        new_snapshots={updated_snapshot.snapshot_id: updated_snapshot},
         previous_plan_id=None,
         previously_promoted_snapshot_ids=set(),
         previous_finalized_snapshots=None,
@@ -133,12 +144,66 @@ def test_forward_only_dev(make_snapshot, mocker: MockerFixture):
     ).build()
 
     assert plan.restatements == {
-        snapshot_a.snapshot_id: (to_timestamp(expected_start), expected_interval_end)
+        updated_snapshot.snapshot_id: (to_timestamp(expected_start), expected_interval_end)
     }
     assert plan.start == to_date(expected_start)
     assert plan.end == expected_end
 
 
+def test_forward_only_metadata_change_dev(make_snapshot, mocker: MockerFixture):
+    snapshot = make_snapshot(
+        SqlModel(
+            name="a",
+            dialect="duckdb",
+            query=parse_one("select 1, ds"),
+            kind=dict(name=ModelKindName.INCREMENTAL_BY_TIME_RANGE, time_column="ds"),
+        )
+    )
+
+    # Simulate a metadata change.
+    updated_snapshot = make_snapshot(
+        SqlModel(
+            **{
+                **snapshot.model.dict(),
+                "owner": "new_owner",
+            }
+        )
+    )
+
+    expected_start = to_date("2022-01-02")
+    expected_end = to_date("2022-01-03")
+
+    context_diff = ContextDiff(
+        environment="test_environment",
+        is_new_environment=True,
+        is_unfinalized_environment=False,
+        create_from="prod",
+        added=set(),
+        removed_snapshots={},
+        modified_snapshots={updated_snapshot.name: (updated_snapshot, snapshot)},
+        snapshots={updated_snapshot.snapshot_id: snapshot},
+        new_snapshots={updated_snapshot.snapshot_id: snapshot},
+        previous_plan_id=None,
+        previously_promoted_snapshot_ids=set(),
+        previous_finalized_snapshots=None,
+    )
+
+    yesterday_ds_mock = mocker.patch("sqlmesh.core.plan.builder.yesterday_ds")
+    yesterday_ds_mock.return_value = expected_start
+
+    now_mock = mocker.patch("sqlmesh.core.snapshot.definition.now")
+    now_mock.return_value = expected_end
+
+    mocker.patch("sqlmesh.core.plan.builder.now").return_value = expected_end
+    mocker.patch("sqlmesh.core.plan.definition.now").return_value = expected_end
+
+    plan = PlanBuilder(
+        context_diff, DuckDBEngineAdapter.SCHEMA_DIFFER, forward_only=True, is_dev=True
+    ).build()
+
+    assert not plan.restatements
+
+
 def test_forward_only_plan_added_models(make_snapshot, mocker: MockerFixture):
     snapshot_a = make_snapshot(
         SqlModel(name="a", query=parse_one("select 1 as a, ds"), dialect="duckdb")