Fix: Model kind change should not be categorized as forward-only (#2551)

Author: izeigerman

sqlmesh/core/plan/builder.py

@@ -432,7 +432,7 @@ def _categorize_snapshots(
                             snapshot.categorize_as(SnapshotChangeCategory.FORWARD_ONLY)
                         else:
                             snapshot.categorize_as(SnapshotChangeCategory.NON_BREAKING)
-                    elif self._is_forward_only_model(s_id) and is_directly_modified:
+                    elif self._is_forward_only_change(s_id) and is_directly_modified:
                         self._set_choice(
                             snapshot,
                             SnapshotChangeCategory.FORWARD_ONLY,
@@ -486,14 +486,14 @@ def _categorize_snapshots(
                 ):
                     snapshot.categorize_as(
                         SnapshotChangeCategory.FORWARD_ONLY
-                        if self._is_forward_only_model(s_id)
+                        if self._is_forward_only_change(s_id)
                         else SnapshotChangeCategory.INDIRECT_BREAKING
                     )
 
             elif s_id in self._context_diff.added and self._is_new_snapshot(snapshot):
                 snapshot.categorize_as(
                     SnapshotChangeCategory.FORWARD_ONLY
-                    if self._is_forward_only_model(s_id)
+                    if self._is_forward_only_change(s_id)
                     else SnapshotChangeCategory.BREAKING
                 )
 
@@ -525,7 +525,7 @@ def _set_choice(
             if not self._is_new_snapshot(child_snapshot):
                 continue
 
-            is_forward_only_child = self._is_forward_only_model(child_s_id)
+            is_forward_only_child = self._is_forward_only_change(child_s_id)
 
             if is_breaking_choice:
                 child_snapshot.categorize_as(
@@ -573,8 +573,13 @@ def _apply_effective_from(self) -> None:
             if not snapshot.disable_restatement:
                 snapshot.effective_from = self._effective_from
 
-    def _is_forward_only_model(self, s_id: SnapshotId) -> bool:
+    def _is_forward_only_change(self, s_id: SnapshotId) -> bool:
         snapshot = self._context_diff.snapshots[s_id]
+        if snapshot.name in self._context_diff.modified_snapshots:
+            _, old = self._context_diff.modified_snapshots[snapshot.name]
+            # If the model kind has changed, then we should not consider this to be a forward-only change.
+            if snapshot.is_model and old.model.kind.name != snapshot.model.kind.name:
+                return False
         return (
             snapshot.is_model
             and snapshot.model.forward_only

tests/core/test_plan.py

@@ -9,7 +9,13 @@
 from sqlmesh.core.context import Context
 from sqlmesh.core.context_diff import ContextDiff
 from sqlmesh.core.environment import EnvironmentNamingInfo
-from sqlmesh.core.model import IncrementalByTimeRangeKind, SeedKind, SeedModel, SqlModel
+from sqlmesh.core.model import (
+    FullKind,
+    IncrementalByTimeRangeKind,
+    SeedKind,
+    SeedModel,
+    SqlModel,
+)
 from sqlmesh.core.model.seed import Seed
 from sqlmesh.core.plan import Plan, PlanBuilder, SnapshotIntervals
 from sqlmesh.core.snapshot import (
@@ -858,7 +864,13 @@ def test_new_environment_with_changes(make_snapshot, mocker: MockerFixture):
 
 
 def test_forward_only_models(make_snapshot, mocker: MockerFixture):
-    snapshot = make_snapshot(SqlModel(name="a", query=parse_one("select 1, ds")))
+    snapshot = make_snapshot(
+        SqlModel(
+            name="a",
+            query=parse_one("select 1, ds"),
+            kind=IncrementalByTimeRangeKind(time_column="ds", forward_only=True),
+        )
+    )
     snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
     updated_snapshot = make_snapshot(
         SqlModel(
@@ -898,6 +910,37 @@ def test_forward_only_models(make_snapshot, mocker: MockerFixture):
     assert updated_snapshot.change_category == SnapshotChangeCategory.FORWARD_ONLY
 
 
+def test_forward_only_models_model_kind_changed(make_snapshot, mocker: MockerFixture):
+    snapshot = make_snapshot(SqlModel(name="a", query=parse_one("select 1, ds"), kind=FullKind()))
+    snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+    updated_snapshot = make_snapshot(
+        SqlModel(
+            name="a",
+            query=parse_one("select 3, ds"),
+            kind=IncrementalByTimeRangeKind(time_column="ds", forward_only=True),
+        )
+    )
+    updated_snapshot.previous_versions = snapshot.all_versions
+
+    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: updated_snapshot},
+        new_snapshots={updated_snapshot.snapshot_id: updated_snapshot},
+        previous_plan_id=None,
+        previously_promoted_snapshot_ids=set(),
+        previous_finalized_snapshots=None,
+    )
+
+    PlanBuilder(context_diff, is_dev=True).build()
+    assert updated_snapshot.change_category == SnapshotChangeCategory.BREAKING
+
+
 def test_indirectly_modified_forward_only_model(make_snapshot, mocker: MockerFixture):
     snapshot_a = make_snapshot(SqlModel(name="a", query=parse_one("select 1 as a, ds")))
     snapshot_a.categorize_as(SnapshotChangeCategory.BREAKING)