Fix: always validate versions whenever a state sync is accessed (#769)

Author: tobymao

examples/multi/repo_1/config.yaml

@@ -5,4 +5,7 @@ connections:
         type: duckdb
         database: db.db
 
+    memory:
+        type: duckdb
+
 default_connection: local

examples/multi/repo_2/config.yaml

@@ -5,4 +5,7 @@ connections:
         type: duckdb
         database: db.db
 
+    memory:
+        type: duckdb
+
 default_connection: local

sqlmesh/core/context.py

@@ -319,6 +319,7 @@ def state_sync(self) -> StateSync:
 
             if self._state_sync.get_versions(validate=False).schema_version == 0:
                 self._state_sync.migrate()
+            self._state_sync.get_versions()
         return self._state_sync
 
     @property

sqlmesh/core/state_sync/engine_adapter.py

@@ -289,7 +289,6 @@ def _get_snapshots(
         self,
         snapshot_ids: t.Optional[t.Iterable[SnapshotIdLike]] = None,
         lock_for_update: bool = False,
-        validate_versions: bool = True,
     ) -> t.Dict[SnapshotId, Snapshot]:
         """Fetches specified snapshots or all snapshots.
 
@@ -300,9 +299,6 @@ def _get_snapshots(
         Returns:
             A dictionary of snapshot ids to snapshots for ones that could be found.
         """
-        if validate_versions:
-            self.get_versions()
-
         query = (
             exp.select("snapshot")
             .from_(self.snapshots_table)
@@ -406,7 +402,7 @@ def migrate(self) -> None:
         super().migrate()
 
     def _migrate_rows(self) -> None:
-        all_snapshots = self._get_snapshots(lock_for_update=True, validate_versions=False)
+        all_snapshots = self._get_snapshots(lock_for_update=True)
         environments = self.get_environments()
 
         snapshot_mapping = {}

sqlmesh/migrations/v0004_environmnent_add_finalized_at.py

@@ -4,7 +4,7 @@
 
 def migrate(state_sync):  # type: ignore
     engine_adapter = state_sync.engine_adapter
-    environments_table = state_sync.environments_table
+    environments_table = f"{state_sync.schema}._environments"
 
     alter_table_exp = exp.AlterTable(
         this=exp.to_table(environments_table),

tests/core/test_integration.py

@@ -627,7 +627,7 @@ def test_auto_categorization(sushi_context: Context):
 @pytest.mark.integration
 @pytest.mark.core_integration
 def test_multi(mocker):
-    context = Context(paths=["examples/multi/repo_1", "examples/multi/repo_2"])
+    context = Context(paths=["examples/multi/repo_1", "examples/multi/repo_2"], config="memory")
     context.state_sync.reset()
     plan = context.plan()
     assert len(plan.new_snapshots) == 4