Different approach: decide whether we should recreate based on forward only and virtual env mode flags

Author: xardasos

sqlmesh/core/plan/evaluator.py

@@ -15,7 +15,6 @@
 """
 
 import abc
-import itertools
 import logging
 import typing as t
 from sqlmesh.core import analytics
@@ -380,12 +379,6 @@ def visit_migrate_schemas_stage(
                 allow_destructive_snapshots=plan.allow_destructive_models,
                 allow_additive_snapshots=plan.allow_additive_models,
                 deployability_index=stage.deployability_index,
-                directly_or_indirectly_modified_snapshots_ids=set(
-                    itertools.chain(
-                        *plan.indirectly_modified_snapshots.values(),
-                        plan.directly_modified_snapshots,
-                    )
-                ),
             )
         except NodeExecutionFailedError as ex:
             raise PlanError(str(ex.__cause__) if ex.__cause__ else str(ex))

sqlmesh/core/snapshot/evaluator.py

@@ -477,7 +477,6 @@ def migrate(
         allow_destructive_snapshots: t.Optional[t.Set[str]] = None,
         allow_additive_snapshots: t.Optional[t.Set[str]] = None,
         deployability_index: t.Optional[DeployabilityIndex] = None,
-        directly_or_indirectly_modified_snapshots_ids: t.Optional[t.Set[SnapshotId]] = None,
     ) -> None:
         """Alters a physical snapshot table to match its snapshot's schema for the given collection of snapshots.
 
@@ -487,7 +486,6 @@ def migrate(
             allow_destructive_snapshots: Set of snapshots that are allowed to have destructive schema changes.
             allow_additive_snapshots: Set of snapshots that are allowed to have additive schema changes.
             deployability_index: Determines snapshots that are deployable in the context of this evaluation.
-            directly_or_indirectly_modified_snapshots_ids: Set of SnapshotIds with direct or indirect changes.
         """
         deployability_index = deployability_index or DeployabilityIndex.all_deployable()
         target_data_objects = self._get_physical_data_objects(target_snapshots, deployability_index)
@@ -512,10 +510,6 @@ def migrate(
                     allow_additive_snapshots,
                     self.get_adapter(s.model_gateway),
                     deployability_index,
-                    only_metadata_changes=s.snapshot_id
-                    not in directly_or_indirectly_modified_snapshots_ids
-                    if directly_or_indirectly_modified_snapshots_ids is not None
-                    else False,
                 ),
                 self.ddl_concurrent_tasks,
             )
@@ -1117,7 +1111,6 @@ def _migrate_snapshot(
         allow_additive_snapshots: t.Set[str],
         adapter: EngineAdapter,
         deployability_index: DeployabilityIndex,
-        only_metadata_changes: bool,
     ) -> None:
         if not snapshot.is_model or snapshot.is_symbolic:
             return
@@ -1161,7 +1154,6 @@ def _migrate_snapshot(
                     allow_destructive_snapshots=allow_destructive_snapshots,
                     allow_additive_snapshots=allow_additive_snapshots,
                     run_pre_post_statements=True,
-                    only_metadata_changes=only_metadata_changes,
                 )
             else:
                 self._execute_create(
@@ -1198,7 +1190,6 @@ def _migrate_target_table(
         allow_destructive_snapshots: t.Set[str],
         allow_additive_snapshots: t.Set[str],
         run_pre_post_statements: bool = False,
-        only_metadata_changes: bool = False,
     ) -> None:
         adapter = self.get_adapter(snapshot.model.gateway)
 
@@ -1235,7 +1226,6 @@ def _migrate_target_table(
                 ignore_destructive=snapshot.model.on_destructive_change.is_ignore,
                 ignore_additive=snapshot.model.on_additive_change.is_ignore,
                 deployability_index=deployability_index,
-                only_metadata_changes=only_metadata_changes,
             )
         finally:
             if snapshot.is_materialized:
@@ -2770,8 +2760,12 @@ def migrate(
         **kwargs: t.Any,
     ) -> None:
         logger.info("Migrating view '%s'", target_table_name)
-        if not (
-            kwargs["only_metadata_changes"] and self.adapter.COMMENT_CREATION_VIEW.is_unsupported
+        # Optimization: avoid unnecessary recreation when possible
+        if (
+            snapshot.model.forward_only
+            or bool(snapshot.model.physical_version)
+            or not snapshot.virtual_environment_mode.is_full
+            or not self.adapter.COMMENT_CREATION_VIEW.is_unsupported
         ):
             model = snapshot.model
             render_kwargs = dict(

tests/core/test_snapshot_evaluator.py

@@ -1520,10 +1520,20 @@ def test_migrate_view(
     )
 
 
+@pytest.mark.parametrize(
+    "change_category",
+    [
+        SnapshotChangeCategory.BREAKING,
+        SnapshotChangeCategory.NON_BREAKING,
+        SnapshotChangeCategory.METADATA,
+    ],
+)
+@pytest.mark.tz
 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",
@@ -1532,7 +1542,7 @@ def test_migrate_view_recreation_not_needed(
         query=parse_one("SELECT c, a FROM tbl"),
     )
     snapshot = make_snapshot(model, version="1")
-    snapshot.change_category = SnapshotChangeCategory.METADATA
+    snapshot.change_category = change_category
     snapshot.forward_only = False
 
     adapter = make_mocked_engine_adapter(EngineAdapter)
@@ -1550,7 +1560,7 @@ def test_migrate_view_recreation_not_needed(
     )
 
     evaluator = SnapshotEvaluator(adapter)
-    evaluator.migrate([snapshot], {}, directly_or_indirectly_modified_snapshots_ids=set())
+    evaluator.migrate([snapshot], {})
 
     adapter.cursor.execute.assert_not_called()