fix manual categorization for the model kind change

Author: izeigerman

sqlmesh/core/plan/builder.py

@@ -586,7 +586,12 @@ def _categorize_snapshots(
             if not snapshot or not self._is_new_snapshot(snapshot):
                 continue
 
-            forward_only = self._is_forward_only_change(s_id) or self._forward_only
+            forward_only = self._forward_only or self._is_forward_only_change(s_id)
+            if forward_only and s_id.name in self._context_diff.modified_snapshots:
+                new, old = self._context_diff.modified_snapshots[s_id.name]
+                if _should_force_rebuild(old, new) or snapshot.is_seed:
+                    # Breaking kind changes and seed changes can't be forward-only.
+                    forward_only = False
 
             if s_id in self._choices:
                 snapshot.categorize_as(self._choices[s_id], forward_only)
@@ -607,15 +612,10 @@ def _categorize_snapshot(
         s_id = snapshot.snapshot_id
 
         if self._context_diff.directly_modified(s_id.name):
-            new, old = self._context_diff.modified_snapshots[s_id.name]
-            should_force_rebuild = _should_force_rebuild(old, new)
-            if should_force_rebuild or snapshot.is_seed:
-                # Breaking kind changes and seed changes can't be forward-only.
-                forward_only = False
-
             if self._auto_categorization_enabled:
-                if should_force_rebuild:
-                    snapshot.categorize_as(SnapshotChangeCategory.BREAKING, forward_only)
+                new, old = self._context_diff.modified_snapshots[s_id.name]
+                if _should_force_rebuild(old, new):
+                    snapshot.categorize_as(SnapshotChangeCategory.BREAKING, False)
                     return
 
                 s_id_with_missing_columns: t.Optional[SnapshotId] = None

tests/core/test_integration.py

@@ -2557,8 +2557,6 @@ def test_virtual_environment_mode_dev_only_model_kind_change(init_and_plan_conte
     assert len(data_objects) == 1
     assert data_objects[0].type == "table"
 
-    context.state_sync.clear_cache()
-
     # Change back to view
     model = context.get_model("sushi.top_waiters")
     model = model.copy(update={"kind": ViewKind()})
@@ -2605,6 +2603,46 @@ def test_virtual_environment_mode_dev_only_model_kind_change(init_and_plan_conte
     assert data_objects[0].type == "table"
 
 
+@time_machine.travel("2023-01-08 15:00:00 UTC")
+def test_virtual_environment_mode_dev_only_model_kind_change_manual_categorization(
+    init_and_plan_context: t.Callable,
+):
+    context, plan = init_and_plan_context(
+        "examples/sushi", config="test_config_virtual_environment_mode_dev_only"
+    )
+    context.apply(plan)
+
+    model = context.get_model("sushi.top_waiters")
+    model = model.copy(update={"kind": FullKind()})
+    context.upsert_model(model)
+    dev_plan_builder = context.plan_builder("dev", skip_tests=True, no_auto_categorization=True)
+    dev_plan_builder.set_choice(
+        dev_plan_builder._context_diff.snapshots[context.get_snapshot(model.name).snapshot_id],
+        SnapshotChangeCategory.NON_BREAKING,
+    )
+    dev_plan = dev_plan_builder.build()
+    assert dev_plan.requires_backfill
+    assert len(dev_plan.missing_intervals) == 1
+    context.apply(dev_plan)
+
+    prod_plan = context.plan_builder("prod", skip_tests=True).build()
+    assert prod_plan.requires_backfill
+    assert prod_plan.missing_intervals == [
+        SnapshotIntervals(
+            snapshot_id=context.get_snapshot("sushi.top_waiters").snapshot_id,
+            intervals=[
+                (to_timestamp("2023-01-01"), to_timestamp("2023-01-02")),
+                (to_timestamp("2023-01-02"), to_timestamp("2023-01-03")),
+                (to_timestamp("2023-01-03"), to_timestamp("2023-01-04")),
+                (to_timestamp("2023-01-04"), to_timestamp("2023-01-05")),
+                (to_timestamp("2023-01-05"), to_timestamp("2023-01-06")),
+                (to_timestamp("2023-01-06"), to_timestamp("2023-01-07")),
+                (to_timestamp("2023-01-07"), to_timestamp("2023-01-08")),
+            ],
+        ),
+    ]
+
+
 @time_machine.travel("2023-01-08 15:00:00 UTC")
 def test_restatement_plan_ignores_changes(init_and_plan_context: t.Callable):
     context, plan = init_and_plan_context("examples/sushi")