FO model partition changes are breaking not FO

Author: treysp

sqlmesh/core/model/definition.py

@@ -1447,6 +1447,9 @@ def is_breaking_change(self, previous: Model) -> t.Optional[bool]:
         if self.lookback != previous.lookback:
             return None
 
+        if self.partitioned_by != previous.partitioned_by:
+            return None
+
         try:
             # the previous model which comes from disk could be unrenderable
             previous_query = previous.render_query()

sqlmesh/core/plan/builder.py

@@ -547,10 +547,13 @@ def _categorize_snapshots(
                     if snapshot.is_seed
                     else SnapshotChangeCategory.FORWARD_ONLY
                 )
-                # If the model kind changes mark as breaking
+                # If the model kind or partitioning 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:
+                    if (
+                        old.model.kind.name != snapshot.model.kind.name
+                        or old.model.partitioned_by != snapshot.model.partitioned_by
+                    ):
                         category = SnapshotChangeCategory.BREAKING
 
                 snapshot.categorize_as(category)
@@ -714,8 +717,11 @@ 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:
+            # If the model kind or partitioning 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
+                or old.model.partitioned_by != snapshot.model.partitioned_by
+            ):
                 return False
         return (
             snapshot.is_model

tests/core/test_plan.py

@@ -1724,6 +1724,54 @@ def test_forward_only_models_model_kind_changed(make_snapshot, mocker: MockerFix
     assert updated_snapshot.change_category == SnapshotChangeCategory.BREAKING
 
 
+def test_forward_only_models_partition_changed(make_snapshot, mocker: MockerFixture):
+    snapshot = make_snapshot(
+        SqlModel(
+            name="a",
+            query=parse_one("select 3, ds, ds2"),
+            kind=IncrementalByTimeRangeKind(time_column="ds", forward_only=True),
+            partitioned_by=[parse_one("ds")],
+        )
+    )
+    snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+    updated_snapshot = make_snapshot(
+        SqlModel(
+            name="a",
+            query=parse_one("select 3, ds, ds2"),
+            kind=IncrementalByTimeRangeKind(time_column="ds", forward_only=True),
+            partitioned_by=[parse_one("ds2")],
+        )
+    )
+    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,
+    )
+
+    PlanBuilder(context_diff, DuckDBEngineAdapter.SCHEMA_DIFFER, is_dev=True).build()
+    assert updated_snapshot.change_category == SnapshotChangeCategory.BREAKING
+
+    PlanBuilder(
+        context_diff, DuckDBEngineAdapter.SCHEMA_DIFFER, is_dev=True, forward_only=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)