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

[avakili-voleon]

Description

Fixes #5909.

remove_intervals unconditionally called insert_append, which raises "Cannot construct source query from an empty DataFrame" when there is nothing to remove.

With remove_shared_versions=True (used when restating prod) the rows to write are rebuilt by querying _intervals for rows sharing the target name/versions. If none of the affected snapshots have rows there — for example an un-backfilled snapshot version held by another environment — the set is empty and the insert fails. Since RestatementStage runs after the backfill, this surfaced as a non-zero exit code at the end of an otherwise successful prod restatement, aborting any script wrapping it.

Return early instead, matching the guards already present on the neighbouring insert paths add_snapshots_intervals and _push_snapshot_intervals.

Test Plan

• Added test_remove_interval_no_matching_intervals and test_remove_interval_empty_input; both fail without this change
• Validated against the reproduction project attached to remove_intervals raises "Cannot construct source query from an empty DataFrame" during a prod restatement when another environment holds an un-backfilled snapshot version #5909: with the guard removed the restatement backfills and then exits 1 with the reported error; with the guard in place it exits 0, a follow-up plan reports no changes, and the restated models hold the expected rows
• make style passes
• make fast-test passes

Both make style and make fast-test also report one failure unrelated to this change: tests/web/test_main.py imports httpx2, which is not declared in setup.py. It reproduces on a clean main.

Checklist

• I have run make style and fixed any issues
• I have added tests for my changes (if applicable)
• All existing tests pass (make fast-test)

[cmgoffena13]

Hey @avakili-voleon -- this looks good to me. If you're up for it, I think another test in tests/core/integration/test_restatement.py would be good. Something that replicates the issue exactly, but at the plan level:

sqlmesh plan --auto-apply

sqlmesh plan dev --skip-backfill --auto-apply

sqlmesh plan --restate-model test.incremental_model --auto-apply

and assert it doesn't crash

Something like this (AI Output):

def test_prod_restatement_plan_with_unbackfilled_dev_version(tmp_path: Path):
    """
    Scenario:
        Prod is built. A breaking change is planned to devenv with --skip-backfill,
        so devenv holds a different snapshot version with no interval rows. The
        local files are restored to match prod, then prod is restated.

    Outcome:
        RestatementStage tries to clear devenv intervals for that other version,
        finds none in `_intervals`, and no-ops instead of crashing on an empty
        insert. Prod restatement still applies; devenv still has no intervals.
    """
    original_model = """
    MODEL (
        name test.incremental_model,
        kind INCREMENTAL_BY_TIME_RANGE (
            time_column "date"
        ),
        start '2024-01-01',
        cron '@daily'
    );

    select account_id, date from test.external_table;
    """

    models_dir = tmp_path / "models"
    models_dir.mkdir()
    (models_dir / "model.sql").write_text(original_model)

    config = Config(model_defaults=ModelDefaultsConfig(dialect="duckdb"))
    ctx = Context(paths=[tmp_path], config=config)

    engine_adapter = ctx.engine_adapter
    engine_adapter.create_schema("test")

    df = pd.DataFrame(
        {
            "account_id": [1001, 1002, 1003, 1004, 1005],
            "date": ["2024-01-01", "2024-01-02", "2024-01-03", "2024-01-04", "2024-01-05"],
        }
    )
    columns_to_types = {
        "account_id": exp.DataType.build("int"),
        "date": exp.DataType.build("date"),
    }
    external_table = exp.table_(table="external_table", db="test", quoted=True)
    engine_adapter.create_table(table_name=external_table, target_columns_to_types=columns_to_types)
    engine_adapter.insert_append(
        table_name=external_table, query_or_df=df, target_columns_to_types=columns_to_types
    )

    ctx.plan(auto_apply=True, no_prompts=True)
    assert engine_adapter.fetchone("select count(*) from test.incremental_model") == (5,)

    (models_dir / "model.sql").write_text("""
    MODEL (
        name test.incremental_model,
        kind INCREMENTAL_BY_TIME_RANGE (
            time_column "date"
        ),
        start '2024-01-01',
        cron '@daily'
    );

    select 1 as account_id, date from test.external_table;
    """)
    ctx.load()
    ctx.plan(environment="dev", skip_backfill=True, auto_apply=True, no_prompts=True)

    dev_snapshot_id = ctx.get_snapshot("test.incremental_model").snapshot_id
    assert not ctx.state_sync.get_snapshots([dev_snapshot_id])[dev_snapshot_id].intervals

    (models_dir / "model.sql").write_text(original_model)
    ctx.load()
    prod_snapshot_id = ctx.get_snapshot("test.incremental_model").snapshot_id
    assert prod_snapshot_id != dev_snapshot_id

    engine_adapter.execute("delete from test.external_table where date = '2024-01-01'")

    ctx.plan(
        start="2024-01-01",
        end="2024-01-02",
        restate_models=["test.incremental_model"],
        auto_apply=True,
        no_prompts=True,
    )

    assert engine_adapter.fetchone("select count(*) from test.incremental_model") == (4,)
    assert engine_adapter.fetchone(
        "select count(*) from test.incremental_model where date = '2024-01-01'"
    ) == (0,)
    assert not ctx.state_sync.get_snapshots([dev_snapshot_id])[dev_snapshot_id].intervals

    plan = ctx.plan_builder("prod").build()
    assert not plan.has_changes
    assert not plan.missing_intervals
    ```