Chore: Treat the target environment as missing during planning if it's expired (#3051)

Author: izeigerman

sqlmesh/core/context.py

@@ -1249,6 +1249,7 @@ def invalidate_environment(self, name: str, sync: bool = False) -> None:
             sync: If True, the call blocks until the environment is deleted. Otherwise, the environment will
                 be deleted asynchronously by the janitor process.
         """
+        name = Environment.sanitize_name(name)
         self.state_sync.invalidate_environment(name)
         if sync:
             self._cleanup_environments()

sqlmesh/core/context_diff.py

@@ -87,7 +87,7 @@ def create(
         environment = environment.lower()
         env = state_reader.get_environment(environment)
 
-        if env is None:
+        if env is None or env.expired:
             env = state_reader.get_environment(create_from.lower())
             is_new_environment = True
             previously_promoted_snapshot_ids = set()

sqlmesh/core/environment.py

@@ -10,7 +10,7 @@
 from sqlmesh.core.config import EnvironmentSuffixTarget
 from sqlmesh.core.snapshot import SnapshotId, SnapshotTableInfo
 from sqlmesh.utils import word_characters_only
-from sqlmesh.utils.date import TimeLike
+from sqlmesh.utils.date import TimeLike, now_timestamp
 from sqlmesh.utils.pydantic import PydanticModel, field_validator
 
 T = t.TypeVar("T", bound="EnvironmentNamingInfo")
@@ -152,3 +152,7 @@ def naming_info(self) -> EnvironmentNamingInfo:
             catalog_name_override=self.catalog_name_override,
             normalize_name=self.normalize_name,
         )
+
+    @property
+    def expired(self) -> bool:
+        return self.expiration_ts is not None and self.expiration_ts <= now_timestamp()

sqlmesh/core/state_sync/engine_adapter.py

@@ -284,19 +284,20 @@ def promote(
                 )
                 != table_infos[name].qualified_view_name.for_environment(environment.naming_info)
             }
-            if environment.previous_plan_id != existing_environment.plan_id:
-                raise SQLMeshError(
-                    f"Plan '{environment.plan_id}' is no longer valid for the target environment '{environment.name}'. "
-                    f"Expected previous plan ID: '{environment.previous_plan_id}', actual previous plan ID: '{existing_environment.plan_id}'. "
-                    "Please recreate the plan and try again"
-                )
-            if no_gaps_snapshot_names != set():
-                snapshots = self._get_snapshots(environment.snapshots).values()
-                self._ensure_no_gaps(
-                    snapshots,
-                    existing_environment,
-                    no_gaps_snapshot_names,
-                )
+            if not existing_environment.expired:
+                if environment.previous_plan_id != existing_environment.plan_id:
+                    raise SQLMeshError(
+                        f"Plan '{environment.plan_id}' is no longer valid for the target environment '{environment.name}'. "
+                        f"Expected previous plan ID: '{environment.previous_plan_id}', actual previous plan ID: '{existing_environment.plan_id}'. "
+                        "Please recreate the plan and try again"
+                    )
+                if no_gaps_snapshot_names != set():
+                    snapshots = self._get_snapshots(environment.snapshots).values()
+                    self._ensure_no_gaps(
+                        snapshots,
+                        existing_environment,
+                        no_gaps_snapshot_names,
+                    )
             demoted_snapshots = set(existing_environment.snapshots) - set(environment.snapshots)
             # Update the updated_at attribute.
             self._update_snapshots(demoted_snapshots)

tests/core/test_integration.py

@@ -1559,6 +1559,37 @@ def test_restatement_plan_ignores_changes(init_and_plan_context: t.Callable):
     context.apply(plan)
 
 
+@freeze_time("2023-01-08 15:00:00")
+def test_plan_against_expired_environment(init_and_plan_context: t.Callable):
+    context, plan = init_and_plan_context("examples/sushi")
+    context.apply(plan)
+
+    model = context.get_model("sushi.waiter_revenue_by_day")
+    context.upsert_model(add_projection_to_model(t.cast(SqlModel, model)))
+
+    modified_models = {model.fqn, context.get_model("sushi.top_waiters").fqn}
+
+    plan = context.plan("dev", no_prompts=True)
+    assert plan.has_changes
+    assert set(plan.context_diff.modified_snapshots) == modified_models
+    assert plan.missing_intervals
+    context.apply(plan)
+
+    # Make sure there are no changes when comparing against the existing environment.
+    plan = context.plan("dev", no_prompts=True)
+    assert not plan.has_changes
+    assert not plan.context_diff.modified_snapshots
+    assert not plan.missing_intervals
+
+    # Invalidate the environment and make sure that the plan detects the changes.
+    context.invalidate_environment("dev")
+    plan = context.plan("dev", no_prompts=True)
+    assert plan.has_changes
+    assert set(plan.context_diff.modified_snapshots) == modified_models
+    assert not plan.missing_intervals
+    context.apply(plan)
+
+
 def test_plan_twice_with_star_macro_yields_no_diff(tmp_path: Path):
     init_example_project(tmp_path, dialect="duckdb")
 

tests/core/test_state_sync.py

@@ -862,6 +862,32 @@ 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):
+    snapshot = make_snapshot(
+        SqlModel(
+            name="a",
+            query=parse_one("select 1, ds"),
+        ),
+    )
+    snapshot.categorize_as(SnapshotChangeCategory.BREAKING)
+
+    state_sync.push_snapshots([snapshot])
+    promote_snapshots(state_sync, [snapshot], "dev")
+    state_sync.invalidate_environment("dev")
+
+    new_environment = Environment(
+        name="dev",
+        snapshots=[snapshot.table_info],
+        start_at="2022-01-01",
+        end_at="2022-01-01",
+        plan_id="new_plan_id",
+        previous_plan_id=None,  # No previous plan ID since it's technically a new environment
+    )
+
+    # This call shouldn't fail.
+    state_sync.promote(new_environment)
+
+
 def test_promote_snapshots_no_gaps(state_sync: EngineAdapterStateSync, make_snapshot: t.Callable):
     model = SqlModel(
         name="a",