Fix: Don't fail because of an unrestorable change if the target model…

… is forward-only (#3835)

sqlmesh/core/plan/builder.py

@@ -641,6 +641,8 @@ def _ensure_no_forward_only_revert(self) -> None:
         for name, (candidate, promoted) in self._context_diff.modified_snapshots.items():
             if (
                 candidate.snapshot_id not in self._context_diff.new_snapshots
+                and candidate.is_model
+                and not candidate.model.forward_only
                 and promoted.is_forward_only
                 and not promoted.is_paused
                 and not candidate.reuses_previous_version

tests/core/test_integration.py

@@ -2964,7 +2964,7 @@ def test_new_forward_only_model_concurrent_versions(init_and_plan_context: t.Cal
     )
     new_model_alt = load_sql_based_model(new_model_alt_expr)
 
-    # Add the second version of the model and apply it to dev_b.
+    # Add the second version of the model but don't apply it yet
     context.upsert_model(new_model_alt)
     snapshot_b = context.get_snapshot(new_model_alt.name)
     plan_b = context.plan_builder("dev_b").build()
@@ -2977,8 +2977,6 @@ def test_new_forward_only_model_concurrent_versions(init_and_plan_context: t.Cal
     assert snapshot_b.fingerprint != snapshot_a.fingerprint
     assert snapshot_b.version == snapshot_a.version
 
-    context.apply(plan_b)
-
     # Apply the 1st version to prod
     context.upsert_model(new_model)
     plan_prod_a = context.plan_builder("prod").build()
@@ -2992,10 +2990,20 @@ def test_new_forward_only_model_concurrent_versions(init_and_plan_context: t.Cal
     df = context.fetchdf("SELECT * FROM memory.sushi.new_model")
     assert df.to_dict() == {"ds": {0: "2023-01-07"}, "a": {0: 1}}
 
+    # Modify the 1st version in prod to trigger a forward-only change
+    new_model = add_projection_to_model(t.cast(SqlModel, new_model))
+    context.upsert_model(new_model)
+    context.plan("prod", auto_apply=True, no_prompts=True, skip_tests=True)
+
+    # Apply the 2nd version to dev_b.
+    # At this point the snapshot of the 2nd version has already been categorized but not
+    # persisted in the state. This means that when the snapshot of the 1st version was
+    # being unpaused during promotion to prod, the state of the 2nd version snapshot was not updated
+    context.apply(plan_b)
+
     # Apply the 2nd version to prod
     context.upsert_model(new_model_alt)
     plan_prod_b = context.plan_builder("prod").build()
-    assert snapshot_b.snapshot_id in plan_prod_b.snapshots
     assert (
         plan_prod_b.snapshots[snapshot_b.snapshot_id].change_category
         == SnapshotChangeCategory.BREAKING