Fix: Migration of indirect non-breaking views (#4020)

Author: izeigerman

sqlmesh/core/plan/evaluator.py

@@ -143,7 +143,9 @@ def evaluate(
                 deployability_index_for_evaluation,
                 circuit_breaker=circuit_breaker,
             )
-            promotion_result = self._promote(plan, snapshots, before_promote_snapshots)
+            promotion_result = self._promote(
+                plan, snapshots, before_promote_snapshots, deployability_index_for_creation
+            )
             self._backfill(
                 plan,
                 snapshots_by_name,
@@ -301,6 +303,7 @@ def _promote(
         plan: EvaluatablePlan,
         snapshots: t.Dict[SnapshotId, Snapshot],
         no_gaps_snapshot_names: t.Optional[t.Set[str]] = None,
+        deployability_index: t.Optional[DeployabilityIndex] = None,
     ) -> PromotionResult:
         """Promote a plan.
 
@@ -320,7 +323,8 @@ def _promote(
                 self.snapshot_evaluator.migrate(
                     [s for s in snapshots.values() if s.is_paused],
                     snapshots,
-                    plan.allow_destructive_models,
+                    allow_destructive_snapshots=plan.allow_destructive_models,
+                    deployability_index=deployability_index,
                 )
             except NodeExecutionFailedError as ex:
                 raise PlanError(str(ex.__cause__) if ex.__cause__ else str(ex))

sqlmesh/core/snapshot/evaluator.py

@@ -57,7 +57,6 @@
     DeployabilityIndex,
     Intervals,
     Snapshot,
-    SnapshotChangeCategory,
     SnapshotId,
     SnapshotInfoLike,
     SnapshotTableCleanupTask,
@@ -390,15 +389,17 @@ def migrate(
         target_snapshots: t.Iterable[Snapshot],
         snapshots: t.Dict[SnapshotId, Snapshot],
         allow_destructive_snapshots: t.Set[str] = set(),
+        deployability_index: t.Optional[DeployabilityIndex] = None,
     ) -> None:
         """Alters a physical snapshot table to match its snapshot's schema for the given collection of snapshots.
 
         Args:
             target_snapshots: Target snapshots.
             snapshots: Mapping of snapshot ID to snapshot.
             allow_destructive_snapshots: Set of snapshots that are allowed to have destructive schema changes.
+            deployability_index: Determines snapshots that are deployable in the context of this evaluation.
         """
-
+        deployability_index = deployability_index or DeployabilityIndex.all_deployable()
         with self.concurrent_context():
             concurrent_apply_to_snapshots(
                 target_snapshots,
@@ -407,6 +408,7 @@ def migrate(
                     snapshots,
                     allow_destructive_snapshots,
                     self._get_adapter(s.model_gateway),
+                    deployability_index,
                 ),
                 self.ddl_concurrent_tasks,
             )
@@ -850,15 +852,13 @@ def _migrate_snapshot(
         snapshots: t.Dict[SnapshotId, Snapshot],
         allow_destructive_snapshots: t.Set[str],
         adapter: EngineAdapter,
+        deployability_index: DeployabilityIndex,
     ) -> None:
-        if not snapshot.is_paused or not snapshot.is_model:
-            return
-
-        needs_migration = snapshot.model.forward_only or snapshot.change_category in (
-            SnapshotChangeCategory.FORWARD_ONLY,
-            SnapshotChangeCategory.INDIRECT_NON_BREAKING,
-        )
-        if not needs_migration:
+        if (
+            not snapshot.is_paused
+            or not snapshot.is_model
+            or deployability_index.is_representative(snapshot)
+        ):
             return
 
         target_table_name = snapshot.table_name()

tests/core/test_integration.py

@@ -2140,6 +2140,71 @@ def test_indirect_non_breaking_view_model_non_representative_snapshot(
     )
 
 
+@time_machine.travel("2023-01-08 15:00:00 UTC")
+def test_indirect_non_breaking_view_model_non_representative_snapshot_migration(
+    init_and_plan_context: t.Callable,
+):
+    context, _ = init_and_plan_context("examples/sushi")
+
+    forward_only_model_expr = d.parse(
+        """
+        MODEL (
+            name memory.sushi.forward_only_model,
+            kind INCREMENTAL_BY_TIME_RANGE (
+                time_column ds,
+                forward_only TRUE,
+                on_destructive_change 'allow',
+            ),
+        );
+
+        SELECT '2023-01-07' AS ds, 1 AS a;
+        """
+    )
+    forward_only_model = load_sql_based_model(forward_only_model_expr)
+    context.upsert_model(forward_only_model)
+
+    downstream_view_a_expr = d.parse(
+        """
+        MODEL (
+            name memory.sushi.downstream_view_a,
+            kind VIEW,
+        );
+
+        SELECT a from memory.sushi.forward_only_model;
+        """
+    )
+    downstream_view_a = load_sql_based_model(downstream_view_a_expr)
+    context.upsert_model(downstream_view_a)
+
+    downstream_view_b_expr = d.parse(
+        """
+        MODEL (
+            name memory.sushi.downstream_view_b,
+            kind VIEW,
+        );
+
+        SELECT a from memory.sushi.downstream_view_a;
+        """
+    )
+    downstream_view_b = load_sql_based_model(downstream_view_b_expr)
+    context.upsert_model(downstream_view_b)
+
+    context.plan(auto_apply=True, no_prompts=True, skip_tests=True)
+
+    # Make a forward-only change
+    context.upsert_model(add_projection_to_model(t.cast(SqlModel, forward_only_model)))
+    # Make a non-breaking change downstream
+    context.upsert_model(add_projection_to_model(t.cast(SqlModel, downstream_view_a)))
+
+    context.plan(auto_apply=True, no_prompts=True, skip_tests=True)
+
+    # Make sure the downstrean indirect non-breaking view is available in prod
+    count = context.engine_adapter.fetchone("SELECT COUNT(*) FROM memory.sushi.downstream_view_b")[
+        0
+    ]
+    assert count > 0
+
+
 @time_machine.travel("2023-01-08 15:00:00 UTC")
 def test_unaligned_start_snapshot_with_non_deployable_downstream(init_and_plan_context: t.Callable):
     context, _ = init_and_plan_context("examples/sushi")

tests/core/test_snapshot_evaluator.py

@@ -1191,7 +1191,7 @@ def columns(table_name):
     snapshot = make_snapshot(model, version="1")
     snapshot.change_category = SnapshotChangeCategory.FORWARD_ONLY
 
-    evaluator.migrate([snapshot], {})
+    evaluator.migrate([snapshot], {}, deployability_index=DeployabilityIndex.none_deployable())
 
     cursor_mock.execute.assert_has_calls(
         [
@@ -1226,7 +1226,7 @@ def test_migrate_missing_table(mocker: MockerFixture, make_snapshot):
     snapshot = make_snapshot(model, version="1")
     snapshot.change_category = SnapshotChangeCategory.FORWARD_ONLY
 
-    evaluator.migrate([snapshot], {})
+    evaluator.migrate([snapshot], {}, deployability_index=DeployabilityIndex.none_deployable())
 
     cursor_mock.execute.assert_has_calls(
         [
@@ -1262,7 +1262,7 @@ def test_migrate_view(
     snapshot = make_snapshot(model, version="1")
     snapshot.change_category = change_category
 
-    evaluator.migrate([snapshot], {})
+    evaluator.migrate([snapshot], {}, deployability_index=DeployabilityIndex.none_deployable())
 
     cursor_mock.execute.assert_has_calls(
         [
@@ -1722,7 +1722,7 @@ def columns(table_name):
     snapshot.change_category = SnapshotChangeCategory.FORWARD_ONLY
 
     with pytest.raises(NodeExecutionFailedError) as ex:
-        evaluator.migrate([snapshot], {})
+        evaluator.migrate([snapshot], {}, deployability_index=DeployabilityIndex.none_deployable())
 
     destructive_change_err = ex.value.__cause__
     assert isinstance(destructive_change_err, DestructiveChangeError)
@@ -1744,15 +1744,20 @@ def columns(table_name):
 
     logger = logging.getLogger("sqlmesh.core.snapshot.evaluator")
     with patch.object(logger, "warning") as mock_logger:
-        evaluator.migrate([snapshot], {})
+        evaluator.migrate([snapshot], {}, deployability_index=DeployabilityIndex.none_deployable())
         assert (
             mock_logger.call_args[0][0]
             == "\nPlan requires a destructive change to forward-only model '\"test_schema\".\"test_model\"'s schema that drops column 'b'.\n\nSchema changes:\n  ALTER TABLE sqlmesh__test_schema.test_schema__test_model__1 DROP COLUMN b\n  ALTER TABLE sqlmesh__test_schema.test_schema__test_model__1 ADD COLUMN a INT"
         )
 
     # allow destructive
     with patch.object(logger, "warning") as mock_logger:
-        evaluator.migrate([snapshot], {}, {'"test_schema"."test_model"'})
+        evaluator.migrate(
+            [snapshot],
+            {},
+            {'"test_schema"."test_model"'},
+            deployability_index=DeployabilityIndex.none_deployable(),
+        )
         assert mock_logger.call_count == 0
 
 
@@ -3638,7 +3643,7 @@ def test_migrate_snapshot(snapshot: Snapshot, mocker: MockerFixture, adapter_moc
     assert new_snapshot.table_name() == snapshot.table_name()
 
     evaluator.create([new_snapshot], {})
-    evaluator.migrate([new_snapshot], {})
+    evaluator.migrate([new_snapshot], {}, deployability_index=DeployabilityIndex.none_deployable())
 
     common_kwargs: t.Dict[str, t.Any] = dict(
         table_format=None,
@@ -3706,7 +3711,11 @@ def test_migrate_managed(adapter_mock, make_snapshot, mocker: MockerFixture):
 
     # no schema changes - no-op
     adapter_mock.get_alter_expressions.return_value = []
-    evaluator.migrate(target_snapshots=[snapshot], snapshots={})
+    evaluator.migrate(
+        target_snapshots=[snapshot],
+        snapshots={},
+        deployability_index=DeployabilityIndex.none_deployable(),
+    )
 
     adapter_mock.create_table.assert_not_called()
     adapter_mock.create_managed_table.assert_not_called()
@@ -3716,7 +3725,11 @@ def test_migrate_managed(adapter_mock, make_snapshot, mocker: MockerFixture):
     adapter_mock.get_alter_expressions.return_value = [exp.Alter()]
 
     with pytest.raises(NodeExecutionFailedError) as ex:
-        evaluator.migrate(target_snapshots=[snapshot], snapshots={})
+        evaluator.migrate(
+            target_snapshots=[snapshot],
+            snapshots={},
+            deployability_index=DeployabilityIndex.none_deployable(),
+        )
 
     sqlmesh_err = ex.value.__cause__
     assert isinstance(sqlmesh_err, SQLMeshError)
@@ -3907,7 +3920,9 @@ def columns(table_name):
     )
     snapshot_2 = make_snapshot(model_2, version="1")
     snapshot_2.change_category = SnapshotChangeCategory.FORWARD_ONLY
-    evaluator.migrate([snapshot_1, snapshot_2], {})
+    evaluator.migrate(
+        [snapshot_1, snapshot_2], {}, deployability_index=DeployabilityIndex.none_deployable()
+    )
 
     cursor_mock.execute.assert_has_calls(
         [