Filter snapshots to promote in environment

Author: treysp

sqlmesh/core/console.py

@@ -175,11 +175,7 @@ def start_promotion_progress(
         """Indicates that a new snapshot promotion progress has begun."""
 
     @abc.abstractmethod
-    def update_promotion_progress(
-        self,
-        snapshot: SnapshotInfoLike,
-        promoted: bool,
-    ) -> None:
+    def update_promotion_progress(self, snapshot: SnapshotInfoLike, promoted: bool) -> None:
         """Update the snapshot promotion progress."""
 
     @abc.abstractmethod
@@ -473,11 +469,7 @@ def start_promotion_progress(
     ) -> None:
         pass
 
-    def update_promotion_progress(
-        self,
-        snapshot: SnapshotInfoLike,
-        promoted: bool,
-    ) -> None:
+    def update_promotion_progress(self, snapshot: SnapshotInfoLike, promoted: bool) -> None:
         pass
 
     def stop_promotion_progress(self, success: bool = True) -> None:
@@ -953,11 +945,7 @@ def start_promotion_progress(
             self.environment_naming_info = environment_naming_info
             self.default_catalog = default_catalog
 
-    def update_promotion_progress(
-        self,
-        snapshot: SnapshotInfoLike,
-        promoted: bool,
-    ) -> None:
+    def update_promotion_progress(self, snapshot: SnapshotInfoLike, promoted: bool) -> None:
         """Update the snapshot promotion progress."""
         if self.promotion_progress is not None and self.promotion_task is not None:
             if self.verbosity >= Verbosity.VERBOSE:
@@ -2798,11 +2786,7 @@ def start_promotion_progress(
         self.promotion_status = (0, total_tasks)
         print(f"Virtually Updating '{environment_naming_info.name}'")
 
-    def update_promotion_progress(
-        self,
-        snapshot: SnapshotInfoLike,
-        promoted: bool,
-    ) -> None:
+    def update_promotion_progress(self, snapshot: SnapshotInfoLike, promoted: bool) -> None:
         """Update the snapshot promotion progress."""
         num_promotions, total_promotions = self.promotion_status
         num_promotions += 1
@@ -2933,11 +2917,7 @@ def start_promotion_progress(
     ) -> None:
         self._write(f"Starting promotion for {total_tasks} snapshots")
 
-    def update_promotion_progress(
-        self,
-        snapshot: SnapshotInfoLike,
-        promoted: bool,
-    ) -> None:
+    def update_promotion_progress(self, snapshot: SnapshotInfoLike, promoted: bool) -> None:
         self._write(f"Promoting {snapshot.name}")
 
     def stop_promotion_progress(self, success: bool = True) -> None:

sqlmesh/core/environment.py

@@ -164,11 +164,20 @@ def promoted_snapshot_id_dicts(self) -> t.List[dict]:
 
     @property
     def promoted_snapshots(self) -> t.List[SnapshotTableInfo]:
+        def has_virtual_view(s: SnapshotTableInfo) -> bool:
+            return (
+                s.is_model and s.model_kind_name is not None and not s.model_kind_name.is_symbolic
+            )
+
         if self.promoted_snapshot_ids is None:
-            return self.snapshots
+            return [s for s in self.snapshots if has_virtual_view(s)]
 
         promoted_snapshot_ids = set(self.promoted_snapshot_ids)
-        return [s for s in self.snapshots if s.snapshot_id in promoted_snapshot_ids]
+        return [
+            s
+            for s in self.snapshots
+            if s.snapshot_id in promoted_snapshot_ids and has_virtual_view(s)
+        ]
 
     @property
     def previous_finalized_snapshots(self) -> t.Optional[t.List[SnapshotTableInfo]]:

sqlmesh/core/plan/evaluator.py

@@ -361,14 +361,8 @@ def _update_views(
 
         environment = plan.environment
 
-        # progress bar should only show snapshots that have a virtual layer view
-        snapshots_with_virtual_views = [
-            s.snapshot_id
-            for s in [*promotion_result.added, *promotion_result.removed]
-            if s.is_model and s.model_kind_name and not s.model_kind_name.is_symbolic
-        ]
         self.console.start_promotion_progress(
-            len(snapshots_with_virtual_views),
+            len(promotion_result.added) + len(promotion_result.removed),
             environment.naming_info,
             self.default_catalog,
         )
@@ -382,15 +376,13 @@ def _update_views(
                 deployability_index=deployability_index,
                 on_complete=lambda s: self.console.update_promotion_progress(s, True),
                 snapshots=snapshots,
-                snapshots_with_virtual_views=snapshots_with_virtual_views,
             )
             if promotion_result.removed_environment_naming_info:
                 self._demote_snapshots(
                     plan,
                     promotion_result.removed,
                     promotion_result.removed_environment_naming_info,
                     on_complete=lambda s: self.console.update_promotion_progress(s, False),
-                    snapshots_with_virtual_views=snapshots_with_virtual_views,
                 )
 
             self.state_sync.finalize(environment)
@@ -406,7 +398,6 @@ def _promote_snapshots(
         snapshots: t.Dict[SnapshotId, Snapshot],
         deployability_index: t.Optional[DeployabilityIndex] = None,
         on_complete: t.Optional[t.Callable[[SnapshotInfoLike], None]] = None,
-        snapshots_with_virtual_views: t.Optional[t.List[SnapshotId]] = None,
     ) -> None:
         self.snapshot_evaluator.promote(
             target_snapshots,
@@ -423,7 +414,6 @@ def _promote_snapshots(
             environment_naming_info=environment_naming_info,
             deployability_index=deployability_index,
             on_complete=on_complete,
-            snapshots_with_virtual_views=snapshots_with_virtual_views,
         )
 
     def _demote_snapshots(
@@ -432,13 +422,11 @@ def _demote_snapshots(
         target_snapshots: t.Iterable[SnapshotTableInfo],
         environment_naming_info: EnvironmentNamingInfo,
         on_complete: t.Optional[t.Callable[[SnapshotInfoLike], None]] = None,
-        snapshots_with_virtual_views: t.Optional[t.List[SnapshotId]] = None,
     ) -> None:
         self.snapshot_evaluator.demote(
             target_snapshots,
             environment_naming_info,
             on_complete=on_complete,
-            snapshots_with_virtual_views=snapshots_with_virtual_views,
         )
 
     def _restate(self, plan: EvaluatablePlan, snapshots_by_name: t.Dict[str, Snapshot]) -> None:

sqlmesh/core/snapshot/evaluator.py

@@ -216,7 +216,6 @@ def promote(
         snapshots: t.Optional[t.Dict[SnapshotId, Snapshot]] = None,
         table_mapping: t.Optional[t.Dict[str, str]] = None,
         on_complete: t.Optional[t.Callable[[SnapshotInfoLike], None]] = None,
-        snapshots_with_virtual_views: t.Optional[t.List[SnapshotId]] = None,
     ) -> None:
         """Promotes the given collection of snapshots in the target environment by replacing a corresponding
         view with a physical table associated with the given snapshot.
@@ -250,7 +249,6 @@ def promote(
                     environment_naming_info=environment_naming_info,
                     deployability_index=deployability_index,  # type: ignore
                     on_complete=on_complete,
-                    snapshots_with_virtual_views=snapshots_with_virtual_views,
                 ),
                 self.ddl_concurrent_tasks,
             )
@@ -260,7 +258,6 @@ def demote(
         target_snapshots: t.Iterable[SnapshotInfoLike],
         environment_naming_info: EnvironmentNamingInfo,
         on_complete: t.Optional[t.Callable[[SnapshotInfoLike], None]] = None,
-        snapshots_with_virtual_views: t.Optional[t.List[SnapshotId]] = None,
     ) -> None:
         """Demotes the given collection of snapshots in the target environment by removing its view.
 
@@ -272,9 +269,7 @@ def demote(
         with self.concurrent_context():
             concurrent_apply_to_snapshots(
                 target_snapshots,
-                lambda s: self._demote_snapshot(
-                    s, environment_naming_info, on_complete, snapshots_with_virtual_views
-                ),
+                lambda s: self._demote_snapshot(s, environment_naming_info, on_complete),
                 self.ddl_concurrent_tasks,
             )
 
@@ -923,7 +918,6 @@ def _promote_snapshot(
         execution_time: t.Optional[TimeLike] = None,
         snapshots: t.Optional[t.Dict[SnapshotId, Snapshot]] = None,
         table_mapping: t.Optional[t.Dict[str, str]] = None,
-        snapshots_with_virtual_views: t.Optional[t.List[SnapshotId]] = None,
     ) -> None:
         if snapshot.is_model:
             adapter = self.adapter
@@ -950,31 +944,22 @@ def _promote_snapshot(
             )
             adapter.execute(snapshot.model.render_on_virtual_update(**render_kwargs))
 
-            if (
-                on_complete is not None
-                and snapshots_with_virtual_views
-                and snapshot.snapshot_id in snapshots_with_virtual_views
-            ):
+            if on_complete is not None:
                 on_complete(snapshot)
 
     def _demote_snapshot(
         self,
         snapshot: SnapshotInfoLike,
         environment_naming_info: EnvironmentNamingInfo,
         on_complete: t.Optional[t.Callable[[SnapshotInfoLike], None]],
-        snapshots_with_virtual_views: t.Optional[t.List[SnapshotId]] = None,
     ) -> None:
         adapter = self.adapter
         view_name = snapshot.qualified_view_name.for_environment(
             environment_naming_info, dialect=adapter.dialect
         )
         _evaluation_strategy(snapshot, adapter).demote(view_name)
 
-        if (
-            on_complete is not None
-            and snapshots_with_virtual_views
-            and snapshot.snapshot_id in snapshots_with_virtual_views
-        ):
+        if on_complete is not None:
             on_complete(snapshot)
 
     def _cleanup_snapshot(