Filter snapshots to promote in environment

Author: treysp

sqlmesh/core/console.py

@@ -246,11 +246,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
@@ -478,11 +474,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:
@@ -958,11 +950,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:
@@ -2833,11 +2821,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
@@ -2968,11 +2952,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

@@ -169,11 +169,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

@@ -375,14 +375,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,
         )
@@ -396,15 +390,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)
@@ -420,7 +412,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,
@@ -437,7 +428,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(
@@ -446,13 +436,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.
@@ -258,7 +257,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,
             )
@@ -268,7 +266,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.
 
@@ -280,9 +277,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,
             )
 
@@ -948,7 +943,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 = (
@@ -979,19 +973,14 @@ 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.get_adapter(snapshot.model_gateway)
@@ -1003,11 +992,7 @@ def _demote_snapshot(
         )
         _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(