fix(state_sync): skip interval removal when there is nothing to remove (#6027)

Signed-off-by: Amir Vakili <AVakili@Voleon.com>
Co-authored-by: Cortland Goffena <30168413+cmgoffena13@users.noreply.github.com>

Author: avakili-voleon

sqlmesh/core/state_sync/db/interval.py

@@ -108,6 +108,9 @@ def remove_intervals(
                 for snapshot in all_snapshots
             ]
 
+        if not intervals_to_remove:
+            return
+
         if logger.isEnabledFor(logging.INFO):
             snapshot_ids = ", ".join(str(s.snapshot_id) for s, _ in intervals_to_remove)
             logger.info("Removing interval for snapshots: %s", snapshot_ids)

tests/core/integration/test_restatement.py

@@ -83,6 +83,44 @@ def test_restatement_plan_ignores_changes(init_and_plan_context: t.Callable):
     context.apply(plan)
 
 
+@time_machine.travel("2023-01-08 15:00:00 UTC")
+def test_prod_restatement_with_unbackfilled_dev_version(init_and_plan_context: t.Callable):
+    """
+    Scenario:
+        Prod is built. A breaking change is planned to `dev` with `--skip-backfill`,
+        so `dev` holds a different snapshot version with no interval rows. Prod is
+        then restated.
+    Outcome:
+        RestatementStage tries to clear `dev` intervals for that other version, finds
+        none in `_intervals`, and no-ops instead of crashing on an empty insert. Prod
+        restatement still applies; the un-backfilled `dev` snapshot still has no
+        intervals.
+    """
+    context, plan = init_and_plan_context("examples/sushi")
+    context.apply(plan)
+
+    prod_snapshot_id = context.get_snapshot("sushi.waiter_revenue_by_day").snapshot_id
+
+    context.upsert_model(
+        add_projection_to_model(t.cast(SqlModel, context.get_model("sushi.waiter_revenue_by_day")))
+    )
+    context.plan("dev", skip_backfill=True, auto_apply=True, no_prompts=True)
+
+    dev_snapshot_id = context.get_snapshot("sushi.waiter_revenue_by_day").snapshot_id
+    assert dev_snapshot_id != prod_snapshot_id
+    assert not context.state_sync.get_snapshots([dev_snapshot_id])[dev_snapshot_id].intervals
+
+    context.plan(
+        restate_models=["sushi.waiter_revenue_by_day"],
+        start="2023-01-07",
+        end="2023-01-08",
+        auto_apply=True,
+        no_prompts=True,
+    )
+
+    assert not context.state_sync.get_snapshots([dev_snapshot_id])[dev_snapshot_id].intervals
+
+
 @time_machine.travel("2023-01-08 15:00:00 UTC")
 def test_restatement_plan_across_environments_snapshot_with_shared_version(
     init_and_plan_context: t.Callable,

tests/core/state_sync/test_state_sync.py

@@ -400,6 +400,40 @@ def test_remove_interval_missing_snapshot(
     ]
 
 
+def test_remove_interval_no_matching_intervals(
+    state_sync: EngineAdapterStateSync, make_snapshot: t.Callable
+) -> None:
+    snapshot = make_snapshot(
+        SqlModel(
+            name="a",
+            cron="@daily",
+            query=parse_one("select 1, ds"),
+        ),
+        version="a",
+    )
+    state_sync.push_snapshots([snapshot])
+
+    # The snapshot has never been backfilled, so there are no rows to expand the shared versions from
+    state_sync.remove_intervals(
+        [(snapshot, snapshot.inclusive_exclusive("2020-01-15", "2020-01-17"))],
+        remove_shared_versions=True,
+    )
+
+    remove_records_count = state_sync.engine_adapter.fetchone(
+        "SELECT COUNT(*) FROM sqlmesh._intervals WHERE name = '\"a\"' AND version = 'a' AND is_removed"
+    )[0]  # type: ignore
+    assert remove_records_count == 0
+
+    assert not state_sync.get_snapshots([snapshot])[snapshot.snapshot_id].intervals
+
+
+def test_remove_interval_empty_input(state_sync: EngineAdapterStateSync) -> None:
+    state_sync.remove_intervals([])
+    state_sync.remove_intervals([], remove_shared_versions=True)
+
+    assert state_sync.engine_adapter.fetchone("SELECT COUNT(*) FROM sqlmesh._intervals")[0] == 0  # type: ignore
+
+
 def test_refresh_snapshot_intervals(
     state_sync: EngineAdapterStateSync, make_snapshot: t.Callable
 ) -> None: