Fix: Make sure backfill is not triggered for models downstream of met…adata changes (#4237)

Author: izeigerman

sqlmesh/core/plan/builder.py

@@ -590,7 +590,9 @@ def _categorize_snapshot(
                     if p_id in snapshot.parents:
                         direct_parent_categories.add(parent.change_category)
 
-            if not direct_parent_categories or direct_parent_categories.intersection(
+            if snapshot.is_model and snapshot.model.forward_only:
+                snapshot.categorize_as(SnapshotChangeCategory.FORWARD_ONLY)
+            elif not direct_parent_categories or direct_parent_categories.intersection(
                 {SnapshotChangeCategory.BREAKING, SnapshotChangeCategory.INDIRECT_BREAKING}
             ):
                 snapshot.categorize_as(SnapshotChangeCategory.INDIRECT_BREAKING)
@@ -619,6 +621,10 @@ def _apply_effective_from(self) -> None:
                 snapshot.effective_from = self._effective_from
 
     def _is_forward_only_change(self, s_id: SnapshotId) -> bool:
+        if not self._context_diff.directly_modified(
+            s_id.name
+        ) and not self._context_diff.indirectly_modified(s_id.name):
+            return False
         snapshot = self._context_diff.snapshots[s_id]
         if snapshot.name in self._context_diff.modified_snapshots:
             _, old = self._context_diff.modified_snapshots[snapshot.name]

sqlmesh/core/snapshot/definition.py

@@ -1473,7 +1473,9 @@ def create(
             )
             if this_deployable:
                 snapshot = snapshots[node]
-                is_forward_only_model = snapshot.is_model and snapshot.model.forward_only
+                is_forward_only_model = (
+                    snapshot.is_model and snapshot.model.forward_only and not snapshot.is_metadata
+                )
                 has_auto_restatement = (
                     snapshot.is_model and snapshot.model.auto_restatement_cron is not None
                 )

tests/core/test_context.py

@@ -197,7 +197,7 @@ def test_render_sql_model(sushi_context, assert_exp_eq, copy_to_temp_path: t.Cal
 def test_render_non_deployable_parent(sushi_context, assert_exp_eq, copy_to_temp_path: t.Callable):
     model = sushi_context.get_model("sushi.waiter_revenue_by_day")
     forward_only_kind = model.kind.copy(update={"forward_only": True})
-    model = model.copy(update={"kind": forward_only_kind})
+    model = model.copy(update={"kind": forward_only_kind, "stamp": "trigger forward-only change"})
     sushi_context.upsert_model(model)
     sushi_context.plan("dev", no_prompts=True, auto_apply=True)
 

tests/core/test_integration.py

@@ -4049,6 +4049,32 @@ def test_plan_repairs_unrenderable_snapshot_state(
     assert target_snapshot_in_state.model.render_query_or_raise()
 
 
+@time_machine.travel("2023-01-08 15:00:00 UTC")
+def test_no_backfill_for_model_downstream_of_metadata_change(init_and_plan_context: t.Callable):
+    context, _ = init_and_plan_context("examples/sushi")
+
+    # Make sushi.waiter_revenue_by_day a forward-only model.
+    forward_only_model = context.get_model("sushi.waiter_revenue_by_day")
+    updated_model_kind = forward_only_model.kind.copy(update={"forward_only": True})
+    forward_only_model = forward_only_model.copy(update={"kind": updated_model_kind})
+    context.upsert_model(forward_only_model)
+
+    context.plan("prod", auto_apply=True, no_prompts=True, skip_tests=True)
+
+    # Make a metadata change upstream of the forward-only model.
+    context.upsert_model("sushi.orders", owner="new_owner")
+
+    plan = context.plan_builder("test_dev").build()
+    assert plan.has_changes
+    assert not plan.directly_modified
+    assert not plan.indirectly_modified
+    assert not plan.missing_intervals
+    assert all(
+        snapshot.change_category == SnapshotChangeCategory.METADATA
+        for snapshot in plan.new_snapshots
+    )
+
+
 @time_machine.travel("2023-01-08 15:00:00 UTC")
 def test_dbt_requirements(sushi_dbt_context: Context):
     assert set(sushi_dbt_context.requirements) == {"dbt-core", "dbt-duckdb"}

tests/core/test_snapshot.py

@@ -2034,11 +2034,11 @@ def test_deployability_index_categorized_forward_only_model(make_snapshot):
 
     snapshot_a = make_snapshot(model_a)
     snapshot_a.previous_versions = snapshot_a_old.all_versions
-    snapshot_a.categorize_as(SnapshotChangeCategory.METADATA)
+    snapshot_a.categorize_as(SnapshotChangeCategory.FORWARD_ONLY)
 
     snapshot_b = make_snapshot(SqlModel(name="b", query=parse_one("SELECT 1")))
     snapshot_b.parents = (snapshot_a.snapshot_id,)
-    snapshot_b.categorize_as(SnapshotChangeCategory.METADATA)
+    snapshot_b.categorize_as(SnapshotChangeCategory.FORWARD_ONLY)
 
     deployability_index = DeployabilityIndex.create(
         {s.snapshot_id: s for s in [snapshot_a, snapshot_b]}