Repository: TobikoData/sqlmesh
Pull request #3207 (2024-10-01 11:19:44)

Fix: mark kind changes as breaking in forward only plan
Related to #3196 

This PR addresses the issue by marking all changes in model types as breaking even in the case of a `forward-only` plan. 

The test hopefully ensures the validity of the work, but I was also able to recreate the issue in a more "complete" environment with the following steps:

1.`sqlmesh init duckdb`
2. `sqlmesh plan` and apply
3. Changing the incremental model to 

```
MODEL (
  name sqlmesh_example.incremental_model,
  kind VIEW,
  start '2020-01-01',
  cron '@daily',
  grain (id, event_date)
);

SELECT
  id,
  item_id,
  event_date,
FROM
  sqlmesh_example.seed_model 
```

4. Running `sqlmesh plan --forward-only` and apply again. This gave the error:

```
  File "..../sqlmesh/sqlmesh/core/engine_adapter/base.py", line 2007, in _execute
    self.cursor.execute(sql, **kwargs)
duckdb.duckdb.CatalogException: Catalog Error: Existing object sqlmesh_example__incremental_model__888251848 is of type Table, trying to replace with type View
```

A final observation made along the way that I made that may want to be improved is that if you run `sqlmesh plan --forward-only` on an empty project (e.g. one where plan has never run), you get the exception 

```
  File ".../sqlmesh/sqlmesh/core/plan/builder.py", line 227, in build
    self._check_destructive_changes(directly_modified)
  File ".../sqlmesh/sqlmesh/core/plan/builder.py", line 485, in _check_destructive_changes
    new, old = self._context_diff.modified_snapshots[snapshot.name]
               ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^
KeyError: '"db"."sqlmesh_example"."incremental_model"'
```





Patch:
diff --git a/sqlmesh/core/plan/builder.py b/sqlmesh/core/plan/builder.py
index 8dcf5b96..48eb7ede 100644
--- a/sqlmesh/core/plan/builder.py
+++ b/sqlmesh/core/plan/builder.py
@@ -532,11 +532,18 @@ class PlanBuilder:
                 # In case of the forward only plan any modifications result in reuse of the
                 # previous version for non-seed models.
                 # New snapshots of seed models are considered non-breaking ones.
-                snapshot.categorize_as(
+                category = (
                     SnapshotChangeCategory.NON_BREAKING
                     if snapshot.is_seed
                     else SnapshotChangeCategory.FORWARD_ONLY
                 )
+                # If the model kind changes mark as breaking
+                if snapshot.is_model and snapshot.name in self._context_diff.modified_snapshots:
+                    _, old = self._context_diff.modified_snapshots[snapshot.name]
+                    if old.model.kind.name != snapshot.model.kind.name:
+                        category = SnapshotChangeCategory.BREAKING
+
+                snapshot.categorize_as(category)
             elif s_id.name in self._context_diff.modified_snapshots:
                 self._categorize_snapshot(snapshot, dag, indirectly_modified)
 


Test patch:
diff --git a/tests/core/test_plan.py b/tests/core/test_plan.py
index 664991ec..6675f223 100644
--- a/tests/core/test_plan.py
+++ b/tests/core/test_plan.py
@@ -245,6 +245,49 @@ def test_forward_only_plan_added_models(make_snapshot, mocker: MockerFixture):
     assert snapshot_b.change_category == SnapshotChangeCategory.BREAKING
 
 
+def test_forward_only_plan_categorizes_change_model_kind_as_breaking(
+    make_snapshot, mocker: MockerFixture
+):
+    snapshot_old = make_snapshot(
+        SqlModel(
+            name="a",
+            dialect="duckdb",
+            query=parse_one("select 1, ds"),
+            kind=dict(name=ModelKindName.INCREMENTAL_BY_TIME_RANGE, time_column="ds"),
+        )
+    )
+
+    # Simulate a change in the model kind.
+    updated_snapshot = make_snapshot(
+        SqlModel(
+            **{
+                **snapshot_old.model.dict(),
+                "kind": dict(name=ModelKindName.VIEW),
+            }
+        )
+    )
+
+    context_diff = ContextDiff(
+        environment="prod",
+        is_new_environment=False,
+        is_unfinalized_environment=False,
+        normalize_environment_name=True,
+        create_from="prod",
+        added=set(),
+        removed_snapshots={},
+        modified_snapshots={updated_snapshot.name: (updated_snapshot, snapshot_old)},
+        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, DuckDBEngineAdapter.SCHEMA_DIFFER, forward_only=True).build()
+
+    assert updated_snapshot.change_category == SnapshotChangeCategory.BREAKING
+
+
 def test_paused_forward_only_parent(make_snapshot, mocker: MockerFixture):
     snapshot_a = make_snapshot(SqlModel(name="a", query=parse_one("select 1, ds")))
     snapshot_a.previous_versions = (