Fix: Forward-only models can't be categorized manually

Author: izeigerman

sqlmesh/core/console.py

@@ -1625,6 +1625,8 @@ def _prompt_categorize(
             self._show_categorized_snapshots(plan, default_catalog)
 
         for snapshot in plan.uncategorized:
+            if snapshot.is_model and snapshot.model.forward_only:
+                continue
             if not no_diff:
                 self.show_sql(plan.context_diff.text_diff(snapshot.name))
             tree = Tree(

sqlmesh/core/plan/builder.py

@@ -33,7 +33,7 @@
     yesterday_ds,
     to_timestamp,
 )
-from sqlmesh.utils.errors import NoChangesPlanError, PlanError, SQLMeshError
+from sqlmesh.utils.errors import NoChangesPlanError, PlanError
 
 logger = logging.getLogger(__name__)
 
@@ -197,16 +197,16 @@ def set_choice(self, snapshot: Snapshot, choice: SnapshotChangeCategory) -> Plan
         if self._forward_only:
             raise PlanError("Choice setting is not supported by a forward-only plan.")
         if not self._is_new_snapshot(snapshot):
-            raise SQLMeshError(
-                f"A choice can't be changed for the existing version of '{snapshot.name}'."
+            raise PlanError(
+                f"A choice can't be changed for the existing version of {snapshot.name}."
             )
         if (
             not self._context_diff.directly_modified(snapshot.name)
             and snapshot.snapshot_id not in self._context_diff.added
         ):
-            raise SQLMeshError(
-                f"Only directly modified models can be categorized ({snapshot.name})."
-            )
+            raise PlanError(f"Only directly modified models can be categorized ({snapshot.name}).")
+        if snapshot.is_model and snapshot.model.forward_only:
+            raise PlanError(f"Forward-only model {snapshot.name} cannot be categorized manually.")
 
         self._choices[snapshot.snapshot_id] = choice
         self._latest_plan = None
@@ -215,7 +215,7 @@ def set_choice(self, snapshot: Snapshot, choice: SnapshotChangeCategory) -> Plan
     def apply(self) -> None:
         """Builds and applies the plan."""
         if not self._apply:
-            raise SQLMeshError("Plan was not initialized with an applier.")
+            raise PlanError("Plan was not initialized with an applier.")
         self._apply(self.build())
 
     def build(self) -> Plan:

tests/core/test_plan.py

@@ -3053,3 +3053,55 @@ def test_plan_environment_statements_diff(make_snapshot):
     )
     assert stripped == expected_output
     console_output.close()
+
+
+def test_set_choice_for_forward_only_model(make_snapshot):
+    snapshot = make_snapshot(
+        SqlModel(
+            name="a",
+            query=parse_one("select 1, ds"),
+            dialect="duckdb",
+            kind=IncrementalByTimeRangeKind(time_column="ds", forward_only=True),
+        )
+    )
+    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),
+            dialect="duckdb",
+        )
+    )
+    updated_snapshot.previous_versions = snapshot.all_versions
+
+    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={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,
+        previous_gateway_managed_virtual_layer=False,
+        gateway_managed_virtual_layer=False,
+    )
+
+    schema_differ = DuckDBEngineAdapter.SCHEMA_DIFFER
+    plan_builder = PlanBuilder(context_diff, schema_differ, is_dev=True)
+
+    with pytest.raises(PlanError, match='Forward-only model "a" cannot be categorized manually.'):
+        plan_builder.set_choice(updated_snapshot, SnapshotChangeCategory.BREAKING)
+
+    plan = plan_builder.build()
+    assert (
+        plan.snapshots[updated_snapshot.snapshot_id].change_category
+        == SnapshotChangeCategory.FORWARD_ONLY
+    )