Repository: TobikoData/sqlmesh
Pull request #2551 (2024-05-01 23:01:29)

Fix: Model kind change should not be categorized as forward-only
Before this fix when transitioning from a non-materialized kind like `VIEW` to a materialized incremental kind for which the `forward_only` flag is set, SQLMesh would categorized change as `FORWARD_ONLY` and try to reuse the existing physical table which is actually a view which can't be reused.


Patch:
diff --git a/sqlmesh/core/plan/builder.py b/sqlmesh/core/plan/builder.py
index 09419c19..e1444888 100644
--- a/sqlmesh/core/plan/builder.py
+++ b/sqlmesh/core/plan/builder.py
@@ -432,7 +432,7 @@ class PlanBuilder:
                             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,
@@ -487,7 +487,7 @@ class PlanBuilder:
                         # an indirectly modified snapshot to be created because of a new parent
                         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
                         )
                     else:
@@ -497,7 +497,7 @@ class PlanBuilder:
             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
                 )
 
@@ -529,7 +529,7 @@ class PlanBuilder:
             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(
@@ -577,8 +577,13 @@ class PlanBuilder:
             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


Test patch:
diff --git a/tests/core/test_plan.py b/tests/core/test_plan.py
index f8c28f58..62a15c03 100644
--- a/tests/core/test_plan.py
+++ b/tests/core/test_plan.py
@@ -9,7 +9,13 @@ from sqlglot import parse_one
 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)