Chore: Refactor migration script application (#2514)

* Chore: Refactor migration script application

* PR feedback

Author: vchan

sqlmesh/core/state_sync/engine_adapter.py

@@ -917,23 +917,13 @@ def migrate(
     ) -> None:
         """Migrate the state sync to the latest SQLMesh / SQLGlot version."""
         versions = self.get_versions(validate=False)
-        migrations = MIGRATIONS[versions.schema_version :]
 
-        if (
-            not migrations
-            and major_minor(SQLGLOT_VERSION) == versions.minor_sqlglot_version
-            and major_minor(SQLMESH_VERSION) == versions.minor_sqlmesh_version
-        ):
-            return
+        try:
+            migrate_rows = self._apply_migrations(default_catalog, skip_backup)
 
-        migrate_rows = migrations or major_minor(SQLGLOT_VERSION) != versions.minor_sqlglot_version
-        if not skip_backup and migrate_rows:
-            self._backup_state()
+            if not migrate_rows and major_minor(SQLMESH_VERSION) == versions.minor_sqlmesh_version:
+                return
 
-        try:
-            for migration in migrations:
-                logger.info(f"Applying migration {migration}")
-                migration.migrate(self, default_catalog=default_catalog)
             if migrate_rows:
                 self._migrate_rows(promoted_snapshots_only)
                 # Cleanup plan DAGs since we currently don't migrate snapshot records that are in there.
@@ -992,6 +982,24 @@ def _backup_state(self) -> None:
                         backup_name, exp.select("*").from_(table), exists=False
                     )
 
+    def _apply_migrations(
+        self,
+        default_catalog: t.Optional[str],
+        skip_backup: bool,
+    ) -> bool:
+        versions = self.get_versions(validate=False)
+        migrations = MIGRATIONS[versions.schema_version :]
+
+        migrate_rows = migrations or major_minor(SQLGLOT_VERSION) != versions.minor_sqlglot_version
+        if not skip_backup and migrate_rows:
+            self._backup_state()
+
+        for migration in migrations:
+            logger.info(f"Applying migration {migration}")
+            migration.migrate(self, default_catalog=default_catalog)
+
+        return bool(migrate_rows)
+
     def _migrate_rows(self, promoted_snapshots_only: bool) -> None:
         logger.info("Fetching environments")
         environments = self.get_environments()