fix(migrations): keep still-live dev intervals in v0071

The v0071 intervals migration compared a bare dev version string against
`used_dev_versions`, which only ever holds `(name, dev_version)` tuples. A
string is never an element of a set of 2-tuples, so
`dev_version not in used_dev_versions` was always true and the guard collapsed
to `if is_dev: continue`. Every dev interval whose version was still live (it
had already passed the `used_versions` check) was dropped on upgrade, forcing a
re-backfill of dev environments.

Compare the correctly-typed `(name, dev_version)` key so only genuinely unused
dev intervals are removed. Add a regression test that runs the migration against
an in-memory DuckDB state with a still-live dev interval and asserts it survives.

Signed-off-by: Anas Khan <83116240+anxkhn@users.noreply.github.com>

Author: anxkhn

sqlmesh/migrations/v0071_add_dev_version_to_intervals.py

@@ -106,7 +106,7 @@ def _migrate_intervals(
             continue
 
         dev_version = snapshot_ids_to_dev_versions.get((name, identifier))
-        if dev_version not in used_dev_versions and is_dev:
+        if (name, dev_version) not in used_dev_versions and is_dev:
             # If the interval's dev version is no longer used and this is a dev interval, we can safely delete it
             continue
 

tests/core/state_sync/test_state_sync.py

@@ -2937,6 +2937,141 @@ def test_migrate_rows(state_sync: EngineAdapterStateSync, mocker: MockerFixture)
     )
 
 
+def test_migrate_v0071_keeps_still_live_dev_intervals() -> None:
+    from sqlmesh.migrations import v0071_add_dev_version_to_intervals as migration
+
+    engine_adapter = create_engine_adapter(duckdb.connect, "duckdb")
+    engine_adapter.create_schema(c.SQLMESH)
+
+    engine_adapter.create_table(
+        "sqlmesh._snapshots",
+        {
+            "name": exp.DataType.build("text"),
+            "identifier": exp.DataType.build("text"),
+            "version": exp.DataType.build("text"),
+            "snapshot": exp.DataType.build("text"),
+            "kind_name": exp.DataType.build("text"),
+            "updated_ts": exp.DataType.build("bigint"),
+            "unpaused_ts": exp.DataType.build("bigint"),
+            "ttl_ms": exp.DataType.build("bigint"),
+            "unrestorable": exp.DataType.build("boolean"),
+        },
+    )
+    # The intervals table at schema version 70, before v0071 adds the dev_version column.
+    engine_adapter.create_table(
+        "sqlmesh._intervals",
+        {
+            "id": exp.DataType.build("text"),
+            "created_ts": exp.DataType.build("bigint"),
+            "name": exp.DataType.build("text"),
+            "identifier": exp.DataType.build("text"),
+            "version": exp.DataType.build("text"),
+            "start_ts": exp.DataType.build("bigint"),
+            "end_ts": exp.DataType.build("bigint"),
+            "is_dev": exp.DataType.build("boolean"),
+            "is_removed": exp.DataType.build("boolean"),
+            "is_compacted": exp.DataType.build("boolean"),
+            "is_pending_restatement": exp.DataType.build("boolean"),
+        },
+    )
+
+    def snapshot_blob(name: str, identifier: str, version: str, dev_version: str) -> str:
+        return json.dumps(
+            {
+                "name": name,
+                "identifier": identifier,
+                "version": version,
+                "dev_version": dev_version,
+                "fingerprint": {
+                    "data_hash": "data",
+                    "metadata_hash": "metadata",
+                    "parent_data_hash": "parent_data",
+                    "parent_metadata_hash": "parent_metadata",
+                },
+                "previous_versions": [],
+            }
+        )
+
+    engine_adapter.insert_append(
+        "sqlmesh._snapshots",
+        pd.DataFrame(
+            [
+                {
+                    "name": '"db"."prod_model"',
+                    "identifier": "prod_ident",
+                    "version": "prod_ver",
+                    "snapshot": snapshot_blob(
+                        '"db"."prod_model"', "prod_ident", "prod_ver", "prod_dev"
+                    ),
+                    "kind_name": "FULL",
+                    "updated_ts": 1,
+                    "unpaused_ts": None,
+                    "ttl_ms": None,
+                    "unrestorable": False,
+                },
+                {
+                    "name": '"db"."dev_model"',
+                    "identifier": "dev_ident",
+                    "version": "dev_ver",
+                    "snapshot": snapshot_blob(
+                        '"db"."dev_model"', "dev_ident", "dev_ver", "dev_dev"
+                    ),
+                    "kind_name": "FULL",
+                    "updated_ts": 1,
+                    "unpaused_ts": None,
+                    "ttl_ms": None,
+                    "unrestorable": False,
+                },
+            ]
+        ),
+    )
+
+    engine_adapter.insert_append(
+        "sqlmesh._intervals",
+        pd.DataFrame(
+            [
+                # A non-dev interval that is always kept; it forces the migration to
+                # rewrite the intervals table so a wrongly-dropped dev interval is
+                # actually deleted rather than left behind by the empty-result guard.
+                {
+                    "id": "prod_int",
+                    "created_ts": 1,
+                    "name": '"db"."prod_model"',
+                    "identifier": "prod_ident",
+                    "version": "prod_ver",
+                    "start_ts": 100,
+                    "end_ts": 200,
+                    "is_dev": False,
+                    "is_removed": False,
+                    "is_compacted": False,
+                    "is_pending_restatement": False,
+                },
+                # A dev interval whose version and dev version are both still live.
+                # It must be preserved by the migration.
+                {
+                    "id": "dev_int",
+                    "created_ts": 1,
+                    "name": '"db"."dev_model"',
+                    "identifier": "dev_ident",
+                    "version": "dev_ver",
+                    "start_ts": 100,
+                    "end_ts": 200,
+                    "is_dev": True,
+                    "is_removed": False,
+                    "is_compacted": False,
+                    "is_pending_restatement": False,
+                },
+            ]
+        ),
+    )
+
+    migration.migrate_schemas(engine_adapter, c.SQLMESH)
+    migration.migrate_rows(engine_adapter, c.SQLMESH)
+
+    surviving_ids = {row[0] for row in engine_adapter.fetchall("SELECT id FROM sqlmesh._intervals")}
+    assert surviving_ids == {"prod_int", "dev_int"}
+
+
 def test_backup_state(state_sync: EngineAdapterStateSync, mocker: MockerFixture) -> None:
     state_sync.engine_adapter.replace_query(
         "sqlmesh._snapshots",