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 <AVakili@Voleon.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/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: