From 8e8ad74cc6d786527d3a0ec9ede8f2c7bdfea84c Mon Sep 17 00:00:00 2001 From: Amir Vakili Date: Fri, 4 Sep 2026 11:09:33 -0700 Subject: [PATCH 1/2] fix(state_sync): skip interval removal when there is nothing to remove `remove_intervals` unconditionally called `insert_append`, which raises "Cannot construct source query from an empty DataFrame" when there are no intervals to remove. With `remove_shared_versions=True` the rows to write are rebuilt by querying the `_intervals` table for rows sharing the target name/versions. When none of the affected snapshots have any rows there, for example an un-backfilled snapshot version held by another environment, the resulting set is empty and the insert fails. This surfaced as a non-zero exit code at the end of an otherwise successful prod restatement, after the model batches had already been backfilled and committed. Return early instead, matching the guards already present on the neighbouring insert paths `add_snapshots_intervals` and `_push_snapshot_intervals`. Fixes #5909 Signed-off-by: Amir Vakili --- sqlmesh/core/state_sync/db/interval.py | 3 +++ tests/core/state_sync/test_state_sync.py | 34 ++++++++++++++++++++++++ 2 files changed, 37 insertions(+) diff --git a/sqlmesh/core/state_sync/db/interval.py b/sqlmesh/core/state_sync/db/interval.py index 8ccdc58fa0..227ee27d01 100644 --- a/sqlmesh/core/state_sync/db/interval.py +++ b/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) diff --git a/tests/core/state_sync/test_state_sync.py b/tests/core/state_sync/test_state_sync.py index 348a883fd5..562afc6ac7 100644 --- a/tests/core/state_sync/test_state_sync.py +++ b/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: From cccd4f6bfa401e8682f8b5ebe7493f03ed73aa6f Mon Sep 17 00:00:00 2001 From: Amir Vakili Date: Wed, 9 Sep 2026 13:05:30 -0700 Subject: [PATCH 2/2] test: cover prod restatement with an unbackfilled dev snapshot A skip-backfill dev environment can hold a newer snapshot version with no interval rows. Restating prod then tried to clear those intervals and crashed on an empty insert. Add a plan-level regression for that path. Signed-off-by: Amir Vakili --- tests/core/integration/test_restatement.py | 38 ++++++++++++++++++++++ 1 file changed, 38 insertions(+) diff --git a/tests/core/integration/test_restatement.py b/tests/core/integration/test_restatement.py index 3694efce31..af4f8965a6 100644 --- a/tests/core/integration/test_restatement.py +++ b/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,