Fix: Ignore local changes when restating (#2823)

sqlmesh/core/context.py

@@ -1089,7 +1089,7 @@ def plan_builder(
  environment or c.PROD,
  snapshots=snapshots,
  create_from=create_from,
- force_no_diff=(restate_models is not None and not expanded_restate_models)
+ force_no_diff=restate_models is not None
  or (backfill_models is not None and not backfill_models),
  ensure_finalized_snapshots=self.config.plan.use_finalized_state,
  )
@@ -1896,10 +1896,7 @@ def _context_diff(
  ) -> ContextDiff:
  environment = Environment.sanitize_name(environment)
  if force_no_diff:
- return ContextDiff.create_no_diff(
- self.state_reader.get_environment(environment.lower())
- or EnvironmentNamingInfo(name=environment)
- )
+ return ContextDiff.create_no_diff(environment, self.state_reader)
 
  return ContextDiff.create(
  environment,

sqlmesh/core/context_diff.py

@@ -21,7 +21,6 @@
 from sqlmesh.utils.pydantic import PydanticModel
 
 if t.TYPE_CHECKING:
- from sqlmesh.core.environment import EnvironmentNamingInfo
  from sqlmesh.core.state_sync import StateReader
 
 logger = logging.getLogger(__name__)
@@ -178,29 +177,36 @@ def create(
  )
 
  @classmethod
- def create_no_diff(cls, environment: EnvironmentNamingInfo) -> ContextDiff:
+ def create_no_diff(cls, environment: str, state_reader: StateReader) -> ContextDiff:
  """Create a no-op ContextDiff object.
 
  Args:
- environment: The environment to diff.
+ environment: The target environment.
+ state_reader: StateReader to access the remote environment record.
 
  Returns:
  The ContextDiff object.
  """
+ env = state_reader.get_environment(environment.lower())
+ if not env:
+ raise SQLMeshError(f"Environment '{environment}' must exist for this operation.")
+
+ snapshots = state_reader.get_snapshots(env.snapshots)
+
  return ContextDiff(
- environment=environment.name,
+ environment=env.name,
  is_new_environment=False,
  is_unfinalized_environment=False,
- normalize_environment_name=environment.normalize_name,
+ normalize_environment_name=env.normalize_name,
  create_from="",
  added=set(),
  removed_snapshots={},
  modified_snapshots={},
- snapshots={},
+ snapshots=snapshots,
  new_snapshots={},
- previous_plan_id=None,
- previously_promoted_snapshot_ids=set(),
- previous_finalized_snapshots=None,
+ previous_plan_id=env.plan_id,
+ previously_promoted_snapshot_ids={s.snapshot_id for s in env.promoted_snapshots},
+ previous_finalized_snapshots=env.previous_finalized_snapshots,
  )
 
  @property

tests/core/test_integration.py

@@ -1360,6 +1360,37 @@ def test_ignored_snapshot_with_non_deployable_downstream(init_and_plan_context:
  assert not plan.missing_intervals
 
 
+@freeze_time("2023-01-08 15:00:00")
+def test_restatement_plan_ignores_changes(init_and_plan_context: t.Callable):
+ context, plan = init_and_plan_context("examples/sushi")
+ context.apply(plan)
+
+ restated_snapshot = context.get_snapshot("sushi.top_waiters")
+
+ # Simulate a change.
+ model = context.get_model("sushi.waiter_revenue_by_day")
+ context.upsert_model(add_projection_to_model(t.cast(SqlModel, model)))
+
+ plan = context.plan(no_prompts=True, restate_models=["sushi.top_waiters"], start="2023-01-07")
+ assert plan.snapshots != context.snapshots
+
+ assert not plan.directly_modified
+ assert not plan.has_changes
+ assert not plan.new_snapshots
+ assert plan.requires_backfill
+ assert plan.restatements == {
+ restated_snapshot.snapshot_id: (to_timestamp("2023-01-07"), to_timestamp("2023-01-08"))
+ }
+ assert plan.missing_intervals == [
+ SnapshotIntervals(
+ snapshot_id=restated_snapshot.snapshot_id,
+ intervals=[(to_timestamp("2023-01-07"), to_timestamp("2023-01-08"))],
+ )
+ ]
+
+ context.apply(plan)
+
+
 @pytest.mark.parametrize(
  "context_fixture",
  ["sushi_context", "sushi_dbt_context", "sushi_test_dbt_context", "sushi_no_default_catalog"],

tests/core/test_plan.py

@@ -703,16 +703,12 @@ def test_restate_models(sushi_context_pre_scheduling: Context):
  assert plan.requires_backfill
 
  plan = sushi_context_pre_scheduling.plan(restate_models=["unknown_model"], no_prompts=True)
- assert plan.snapshots == {}
- assert plan.missing_intervals == []
  assert not plan.has_changes
- assert not plan.requires_backfill
+ assert not plan.restatements
 
  plan = sushi_context_pre_scheduling.plan(restate_models=["tag:unknown_tag"], no_prompts=True)
- assert plan.snapshots == {}
- assert plan.missing_intervals == []
  assert not plan.has_changes
- assert not plan.requires_backfill
+ assert not plan.restatements
 
 
 @pytest.mark.slow