Fix: Avoid unnecessary view recreation for metadata changes during migration

sqlmesh/core/snapshot/evaluator.py

@@ -66,6 +66,7 @@
     SnapshotIdBatch,
     SnapshotInfoLike,
     SnapshotTableCleanupTask,
+    SnapshotChangeCategory,
 )
 from sqlmesh.core.snapshot.execution_tracker import QueryExecutionTracker
 from sqlmesh.utils import random_id, CorrelationId, AttributeDict
@@ -2760,20 +2761,28 @@ def migrate(
         **kwargs: t.Any,
     ) -> None:
         logger.info("Migrating view '%s'", target_table_name)
-        model = snapshot.model
-        render_kwargs = dict(
-            execution_time=now(), snapshots=kwargs["snapshots"], engine_adapter=self.adapter
-        )
+        # Optimization: avoid unnecessary recreation when possible
+        if (
+            snapshot.is_forward_only
+            or bool(snapshot.model.physical_version)
+            or not snapshot.virtual_environment_mode.is_full
+            or snapshot.change_category == SnapshotChangeCategory.INDIRECT_NON_BREAKING
+            or not self.adapter.COMMENT_CREATION_VIEW.is_unsupported
+        ):
+            model = snapshot.model
+            render_kwargs = dict(
+                execution_time=now(), snapshots=kwargs["snapshots"], engine_adapter=self.adapter
+            )
 
-        self.adapter.create_view(
-            target_table_name,
-            model.render_query_or_raise(**render_kwargs),
-            model.columns_to_types,
-            materialized=self._is_materialized_view(model),
-            view_properties=model.render_physical_properties(**render_kwargs),
-            table_description=model.description,
-            column_descriptions=model.column_descriptions,
-        )
+            self.adapter.create_view(
+                target_table_name,
+                model.render_query_or_raise(**render_kwargs),
+                model.columns_to_types,
+                materialized=self._is_materialized_view(model),
+                view_properties=model.render_physical_properties(**render_kwargs),
+                table_description=model.description,
+                column_descriptions=model.column_descriptions,
+            )
 
         # Apply grants after view migration
         deployability_index = kwargs.get("deployability_index")

tests/core/test_snapshot_evaluator.py

@@ -24,6 +24,7 @@
     DataObject,
     DataObjectType,
     InsertOverwriteStrategy,
+    CommentCreationView,
 )
 from sqlmesh.core.environment import EnvironmentNamingInfo
 from sqlmesh.core.macros import RuntimeStage, macro, MacroEvaluator, MacroFunc
@@ -1519,6 +1520,50 @@ def test_migrate_view(
     )
 
 
+@pytest.mark.parametrize(
+    "change_category",
+    [
+        SnapshotChangeCategory.BREAKING,
+        SnapshotChangeCategory.NON_BREAKING,
+        SnapshotChangeCategory.METADATA,
+    ],
+)
+def test_migrate_view_recreation_not_needed(
+    mocker: MockerFixture,
+    make_snapshot,
+    make_mocked_engine_adapter,
+    change_category: SnapshotChangeCategory,
+):
+    model = SqlModel(
+        name="test_schema.test_model",
+        kind=ViewKind(),
+        description="my_description",
+        query=parse_one("SELECT c, a FROM tbl"),
+    )
+    snapshot = make_snapshot(model, version="1")
+    snapshot.change_category = change_category
+    snapshot.forward_only = False
+
+    adapter = make_mocked_engine_adapter(EngineAdapter)
+    adapter.COMMENT_CREATION_VIEW = CommentCreationView.UNSUPPORTED
+    adapter.with_settings = lambda **kwargs: adapter
+    mocker.patch(
+        "sqlmesh.core.engine_adapter.base.EngineAdapter.get_data_objects",
+        return_value=[
+            DataObject(
+                schema="sqlmesh__test_schema",
+                name=f"test_schema__test_model__{snapshot.version}",
+                type="view",
+            )
+        ],
+    )
+
+    evaluator = SnapshotEvaluator(adapter)
+    evaluator.migrate([snapshot], {})
+
+    adapter.cursor.execute.assert_not_called()
+
+
 def test_migrate_snapshot_data_object_type_mismatch(
     mocker: MockerFixture,
     make_snapshot,