fix: VIEW model with metadata-only changes in VDE=dev_only should always migrate

Author: newtonapple

sqlmesh/core/plan/stages.py

@@ -270,7 +270,6 @@ def build(self, plan: EvaluatablePlan) -> t.List[PlanStage]:
             }
             after_promote_snapshots = all_selected_for_backfill_snapshots - before_promote_snapshots
             deployability_index = DeployabilityIndex.all_deployable()
-
             snapshots_with_schema_migration = [
                 s for s in snapshots.values() if s.requires_schema_migration_in_prod
             ]
@@ -364,6 +363,19 @@ def build(self, plan: EvaluatablePlan) -> t.List[PlanStage]:
             )
         )
 
+        if snapshots_with_schema_migration:
+            missing_ids_before = {s.snapshot_id for s in missing_intervals_before_promote}
+            missing_ids_after = {s.snapshot_id for s in missing_intervals_after_promote}
+            # Only skip migrate for VIEWs that are already scheduled for recreation in Backfill.
+            snapshots_with_schema_migration = [
+                s
+                for s in snapshots_with_schema_migration
+                if not s.is_view
+                or (
+                    s.snapshot_id not in missing_ids_before
+                    and s.snapshot_id not in missing_ids_after
+                )
+            ]
         if snapshots_with_schema_migration:
             stages.append(
                 MigrateSchemasStage(

sqlmesh/core/snapshot/definition.py

@@ -1483,7 +1483,7 @@ def requires_schema_migration_in_prod(self) -> bool:
         return (
             self.is_paused
             and self.is_model
-            and self.is_materialized
+            and not self.is_symbolic
             and (
                 (self.previous_version and self.previous_version.version == self.version)
                 or self.model.forward_only

tests/core/test_plan_stages.py

@@ -1658,7 +1658,7 @@ def test_build_plan_stages_indirect_non_breaking_view_migration(
     # Build plan stages
     stages = build_plan_stages(plan, state_reader, None)
 
-    # Verify stages
+    # Verify stages (no migrate for VIEW here because the view is already scheduled to be recreated in Backfill)
     assert len(stages) == 8
 
     assert isinstance(stages[0], CreateSnapshotRecordsStage)
@@ -1936,6 +1936,76 @@ def test_build_plan_stages_virtual_environment_mode_no_updates(
     assert len(virtual_stages) == 0
 
 
+def test_build_plan_stages_virtual_environment_mode_metadata_only_changes(
+    mocker: MockerFixture,
+) -> None:
+    snapshot_view: Snapshot = Snapshot.from_node(
+        SqlModel(
+            name="dev_only_view",
+            query=parse_one("select 1 as id"),
+            kind=dict(name=ModelKindName.VIEW),
+            virtual_environment_mode=VirtualEnvironmentMode.DEV_ONLY,
+        ),
+        nodes={},
+        ttl="in 1 week",
+    )
+    snapshot_view.categorize_as(SnapshotChangeCategory.METADATA)
+
+    state_reader = mocker.Mock(spec=StateReader)
+    state_reader.get_snapshots.return_value = {}
+    state_reader.get_environment.return_value = None
+
+    # Production environment with the view promoted
+    environment = Environment(
+        name="prod",
+        snapshots=[snapshot_view.table_info],
+        start_at="2023-01-01",
+        end_at="2023-01-02",
+        plan_id="test_plan",
+        previous_plan_id=None,
+        promoted_snapshot_ids=[snapshot_view.snapshot_id],
+    )
+
+    plan = EvaluatablePlan(
+        start="2023-01-01",
+        end="2023-01-02",
+        new_snapshots=[snapshot_view],
+        environment=environment,
+        no_gaps=False,
+        skip_backfill=True,
+        empty_backfill=False,
+        restatements={},
+        is_dev=False,
+        allow_destructive_models=set(),
+        allow_additive_models=set(),
+        forward_only=False,
+        end_bounded=True,
+        ensure_finalized_snapshots=False,
+        ignore_cron=False,
+        directly_modified_snapshots=[snapshot_view.snapshot_id],
+        indirectly_modified_snapshots={},
+        metadata_updated_snapshots=[snapshot_view.snapshot_id],
+        removed_snapshots=[],
+        requires_backfill=False,
+        models_to_backfill=None,
+        execution_time="2023-01-02",
+        disabled_restatement_models=set(),
+        environment_statements=None,
+        user_provided_flags=None,
+    )
+
+    stages = build_plan_stages(plan, state_reader, None)
+
+    migrate_stages = [s for s in stages if isinstance(s, MigrateSchemasStage)]
+    assert migrate_stages, (
+        "Expected a MigrateSchemasStage for dev_only VIEW with metadata-only change"
+    )
+    migrate_stage = migrate_stages[0]
+    assert snapshot_view in migrate_stage.snapshots
+
+    assert not any(isinstance(s, VirtualLayerUpdateStage) for s in stages)
+
+
 def test_adjust_intervals_new_forward_only_dev_intervals(
     make_snapshot, mocker: MockerFixture
 ) -> None: