Fix: Always recreate virtual views for dev environments (#4413)

Author: izeigerman

sqlmesh/core/state_sync/db/facade.py

@@ -24,6 +24,7 @@
 
 from sqlglot import exp
 
+from sqlmesh.core import constants as c
 from sqlmesh.core.console import Console, get_console
 from sqlmesh.core.engine_adapter import EngineAdapter
 from sqlmesh.core.environment import Environment, EnvironmentStatements, EnvironmentSummary
@@ -229,6 +230,7 @@ def promote(
             and existing_environment.finalized_ts
             and not existing_environment.expired
             and existing_environment.gateway_managed == environment.gateway_managed
+            and existing_environment.name == c.PROD
         ):
             # Only promote new snapshots.
             added_table_infos -= set(existing_environment.promoted_snapshots)

tests/core/state_sync/test_state_sync.py

@@ -865,7 +865,10 @@ def test_promote_snapshots_parent_plan_id_mismatch(
         state_sync.promote(stale_new_environment)
 
 
-def test_promote_environment_expired(state_sync: EngineAdapterStateSync, make_snapshot: t.Callable):
+@pytest.mark.parametrize("environment_name", ["dev", "prod"])
+def test_promote_environment_expired(
+    state_sync: EngineAdapterStateSync, make_snapshot: t.Callable, environment_name: str
+):
     snapshot = make_snapshot(
         SqlModel(
             name="a",
@@ -880,7 +883,7 @@ def test_promote_environment_expired(state_sync: EngineAdapterStateSync, make_sn
     state_sync.invalidate_environment("dev")
 
     new_environment = Environment(
-        name="dev",
+        name=environment_name,
         snapshots=[snapshot.table_info],
         start_at="2022-01-01",
         end_at="2022-01-01",
@@ -901,10 +904,15 @@ def test_promote_environment_expired(state_sync: EngineAdapterStateSync, make_sn
     new_environment.previous_plan_id = new_environment.plan_id
     new_environment.plan_id = "another_plan_id"
     promotion_result = state_sync.promote(new_environment)
+
     #  Should be empty since the environment is no longer expired and nothing has changed
-    assert promotion_result.added == []
     assert promotion_result.removed == []
     assert promotion_result.removed_environment_naming_info is None
+    if environment_name == "prod":
+        assert promotion_result.added == []
+    else:
+        # We should always recreate views in dev environments
+        assert promotion_result.added == [snapshot.table_info]
 
 
 def test_promote_snapshots_no_gaps(state_sync: EngineAdapterStateSync, make_snapshot: t.Callable):

tests/core/test_integration.py

@@ -4257,6 +4257,28 @@ def test_table_name(init_and_plan_context: t.Callable):
     )
 
 
+@time_machine.travel("2023-01-08 15:00:00 UTC")
+def test_full_model_change_with_plan_start_not_matching_model_start(
+    init_and_plan_context: t.Callable,
+):
+    context, plan = init_and_plan_context("examples/sushi")
+    context.apply(plan)
+
+    model = context.get_model("sushi.top_waiters")
+    context.upsert_model(model, kind=model_kind_type_from_name("FULL")())  # type: ignore
+
+    # Apply the change with --skip-backfill first and no plan start
+    context.plan("dev", skip_tests=True, skip_backfill=True, no_prompts=True, auto_apply=True)
+
+    # Apply the plan again but this time don't skip backfill and set start
+    # to be later than the model start
+    context.plan("dev", skip_tests=True, no_prompts=True, auto_apply=True, start="1 day ago")
+
+    # Check that the number of rows is not 0
+    row_num = context.engine_adapter.fetchone(f"SELECT COUNT(*) FROM sushi__dev.top_waiters")[0]
+    assert row_num > 0
+
+
 @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"}