Fix: Remove ensure no forward-only revert check

Author: izeigerman

sqlmesh/core/plan/builder.py

@@ -267,7 +267,6 @@ def build(self) -> Plan:
         self._ensure_no_new_snapshots_with_restatements()
         self._ensure_new_env_with_changes()
         self._ensure_valid_date_range()
-        self._ensure_no_forward_only_revert()
         self._ensure_no_broken_references()
 
         self._apply_effective_from()
@@ -795,27 +794,6 @@ def _ensure_valid_date_range(self) -> None:
                     f"Plan end date: '{time_like_to_str(end)}' cannot be in the future (execution time: '{time_like_to_str(self.execution_time)}')"
                 )
 
-    def _ensure_no_forward_only_revert(self) -> None:
-        """Ensures that a previously superseded breaking / non-breaking snapshot is not being
-        used again to replace an existing forward-only snapshot with the same version.
-
-        In other words there is no going back to the original non-forward-only snapshot with
-        the same version once a forward-only change for that version has been introduced.
-        """
-        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.is_no_rebuild
-                and promoted.version == candidate.version
-            ):
-                raise PlanError(
-                    f"Attempted to revert to an unrevertable version of model '{name}'. Run `sqlmesh plan` again to mitigate the issue."
-                )
-
     def _ensure_no_broken_references(self) -> None:
         for snapshot in self._context_diff.snapshots.values():
             broken_references = {

tests/core/test_plan.py

@@ -37,7 +37,6 @@
 from sqlmesh.utils.dag import DAG
 from sqlmesh.utils.date import (
     now,
-    now_timestamp,
     to_date,
     to_datetime,
     to_timestamp,
@@ -1097,53 +1096,6 @@ def test_end_validation(make_snapshot, mocker: MockerFixture):
     assert restatement_prod_plan_builder.build().end == "2022-01-04"
 
 
-def test_forward_only_revert_not_allowed(make_snapshot, mocker: MockerFixture):
-    snapshot = make_snapshot(SqlModel(name="a", query=parse_one("select 1, ds")))
-    snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
-    assert not snapshot.is_forward_only
-
-    forward_only_snapshot = make_snapshot(SqlModel(name="a", query=parse_one("select 2, ds")))
-    forward_only_snapshot.categorize_as(SnapshotChangeCategory.BREAKING, forward_only=True)
-    forward_only_snapshot.version = snapshot.version
-    forward_only_snapshot.unpaused_ts = now_timestamp()
-    assert forward_only_snapshot.is_forward_only
-
-    context_diff = ContextDiff(
-        environment="test_environment",
-        is_new_environment=True,
-        is_unfinalized_environment=False,
-        normalize_environment_name=True,
-        create_from="prod",
-        create_from_env_exists=True,
-        added=set(),
-        removed_snapshots={},
-        modified_snapshots={snapshot.name: (snapshot, forward_only_snapshot)},
-        snapshots={snapshot.snapshot_id: snapshot},
-        new_snapshots={},
-        previous_plan_id=None,
-        previously_promoted_snapshot_ids=set(),
-        previous_finalized_snapshots=None,
-        previous_gateway_managed_virtual_layer=False,
-        gateway_managed_virtual_layer=False,
-        environment_statements=[],
-    )
-
-    with pytest.raises(
-        PlanError,
-        match=r"Attempted to revert to an unrevertable version of model.*",
-    ):
-        PlanBuilder(context_diff, forward_only=True).build()
-
-    # Make sure the plan can be created if a new snapshot version was enforced.
-    new_version_snapshot = make_snapshot(
-        SqlModel(name="a", query=parse_one("select 1, ds"), stamp="test_stamp")
-    )
-    snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
-    context_diff.modified_snapshots = {snapshot.name: (new_version_snapshot, forward_only_snapshot)}
-    context_diff.new_snapshots = {new_version_snapshot.snapshot_id: new_version_snapshot}
-    PlanBuilder(context_diff, forward_only=True).build()
-
-
 def test_forward_only_plan_seed_models(make_snapshot, mocker: MockerFixture):
     snapshot_a = make_snapshot(
         SeedModel(