Only include snapshots with virtual layer views in promotion progress bar

Author: treysp

sqlmesh/core/console.py

@@ -172,7 +172,12 @@ 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,
+        snapshots_with_virtual_views: t.List[SnapshotId],
+    ) -> None:
         """Update the snapshot promotion progress."""
 
     @abc.abstractmethod
@@ -406,7 +411,12 @@ 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,
+        snapshots_with_virtual_views: t.List[SnapshotId],
+    ) -> None:
         pass
 
     def stop_promotion_progress(self, success: bool = True) -> None:
@@ -799,17 +809,7 @@ def start_promotion_progress(
     ) -> None:
         """Indicates that a new snapshot promotion progress has begun."""
         if self.promotion_progress is None:
-            self.promotion_progress = Progress(
-                TextColumn(
-                    "[bold blue]Updating virtual layer ",  # space to align with other progress bars
-                    justify="right",
-                ),
-                BarColumn(bar_width=PROGRESS_BAR_WIDTH),
-                "[progress.percentage]{task.percentage:>3.1f}%",
-                "•",
-                TimeElapsedColumn(),
-                console=self.console,
-            )
+            self.promotion_progress = make_progress_bar("Updating virtual layer ", self.console)
 
             self.promotion_progress.start()
             self.promotion_task = self.promotion_progress.add_task(
@@ -820,14 +820,29 @@ 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,
+        snapshots_with_virtual_views: t.List[SnapshotId],
+    ) -> None:
         """Update the snapshot promotion progress."""
-        if self.promotion_progress is not None and self.promotion_task is not None:
+        if (
+            self.promotion_progress is not None
+            and self.promotion_task is not None
+            and snapshot.snapshot_id in snapshots_with_virtual_views
+        ):
             if self.verbosity >= Verbosity.VERBOSE:
-                check_mark = f"{GREEN_CHECK_MARK} " if promoted else "  "
-                action_str = "[green]updated[/green]" if promoted else "[yellow]demoted[/yellow]"
+                action_str = ""
+                if promoted:
+                    action_str = (
+                        "[yellow]updated[/yellow]"
+                        if snapshot.previous_version
+                        else "[green]created[/green]"
+                    )
+                action_str = action_str or "[red]dropped[/red]"
                 self.promotion_progress.live.console.print(
-                    f"{check_mark}{snapshot.display_name(self.environment_naming_info, self.default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None, dialect=self.dialect).ljust(self.PROGRESS_BAR_COLUMN_WIDTHS['name'])} {action_str}"
+                    f"{GREEN_CHECK_MARK} {snapshot.display_name(self.environment_naming_info, self.default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None, dialect=self.dialect).ljust(self.PROGRESS_BAR_COLUMN_WIDTHS['name'])} {action_str}"
                 )
             self.promotion_progress.update(self.promotion_task, refresh=True, advance=1)
 
@@ -2410,7 +2425,12 @@ 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,
+        snapshots_with_virtual_views: t.List[SnapshotId],
+    ) -> None:
         """Update the snapshot promotion progress."""
         num_promotions, total_promotions = self.promotion_status
         num_promotions += 1
@@ -2541,7 +2561,12 @@ 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,
+        snapshots_with_virtual_views: t.List[SnapshotId],
+    ) -> None:
         self._write(f"Promoting {snapshot.name}")
 
     def stop_promotion_progress(self, success: bool = True) -> None:
@@ -2628,8 +2653,7 @@ def show_row_diff(
         self._write(row_diff)
 
 
-# _CONSOLE: Console = NoopConsole()
-_CONSOLE: Console = TerminalConsole()
+_CONSOLE: Console = NoopConsole()
 
 
 def set_console(console: Console) -> None:
@@ -2683,6 +2707,8 @@ def _format_missing_intervals(snapshot: Snapshot, missing: SnapshotIntervals) ->
         if snapshot.is_incremental
         else "recreate view"
         if snapshot.is_view
+        else "run audits"
+        if snapshot.is_external
         else "full refresh"
     )
 

sqlmesh/core/plan/evaluator.py

@@ -160,40 +160,26 @@ def evaluate(
             if not plan.requires_backfill:
                 self.console.log_success("Virtual Update executed successfully")
 
-            model_kind_counts: t.Dict[str, int] = {}
-            audit_counts: t.Dict[str, int] = {}
-            for snapshot in snapshots.values():
-                if snapshot.name in all_names:
-                    if snapshot.is_audit:
-                        audit_counts["standalone"] = audit_counts.get("standalone", 0) + 1
-                    if (
-                        snapshot.is_model
-                        and snapshot.model_kind_name
-                        and snapshot.model.kind
-                        and not snapshot.model.kind.is_external
-                        and not snapshot.model.kind.is_embedded
-                    ):
-                        kind_name = snapshot.model_kind_name
-                        model_kind_counts[kind_name] = model_kind_counts.get(kind_name, 0) + 1
-                    if snapshot.is_model and snapshot.model.audits:
-                        if snapshot.model.kind.is_external:
-                            audit_counts["EXTERNAL model"] = audit_counts.get(
-                                "EXTERNAL model", 0
-                            ) + len(snapshot.model.audits)
-                        else:
-                            audit_counts["model"] = audit_counts.get("model", 0) + len(
-                                snapshot.model.audits
-                            )
-
-            summary_str = ", ".join(
-                [f"{v} {k} model{'s' if v > 1 else ''}" for k, v in model_kind_counts.items()]
+            snapshots_to_count: t.List[SnapshotId] = [
+                *plan.directly_modified_snapshots,
+                *plan.removed_snapshots,
+            ] + [
+                snapshot_id
+                for snapshot_ids in plan.indirectly_modified_snapshots.values()
+                for snapshot_id in snapshot_ids
+            ]
+
+            removed_snapshots = [
+                s
+                for s in (plan.environment.previous_finalized_snapshots or [])
+                if s.snapshot_id in snapshots_to_count
+            ]
+            summary_msg = _create_plan_summary_msg(
+                [*snapshots.values(), *removed_snapshots], snapshots_to_count
             )
-            for audit_type in ["EXTERNAL model", "model", "standalone"]:
-                if audit_type in audit_counts:
-                    count = audit_counts[audit_type]
-                    summary_str += f", {count} {audit_type} audit{'s' if count > 1 else ''}"
-            if summary_str:
-                self.console.log_status_update(f"Plan applied for {summary_str}")
+
+            if summary_msg:
+                self.console.log_status_update(f"Plan applied for {summary_msg}")
 
             execute_environment_statements(
                 adapter=self.snapshot_evaluator.adapter,
@@ -396,8 +382,14 @@ 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(promotion_result.added) + len(promotion_result.removed),
+            len(snapshots_with_virtual_views),
             environment.naming_info,
             self.default_catalog,
         )
@@ -409,15 +401,19 @@ def _update_views(
                 [snapshots[s.snapshot_id] for s in promotion_result.added],
                 environment.naming_info,
                 deployability_index=deployability_index,
-                on_complete=lambda s: self.console.update_promotion_progress(s, True),
+                on_complete=lambda s: self.console.update_promotion_progress(
+                    s, True, snapshots_with_virtual_views
+                ),
                 snapshots=snapshots,
             )
             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),
+                    on_complete=lambda s: self.console.update_promotion_progress(
+                        s, False, snapshots_with_virtual_views
+                    ),
                 )
 
             self.state_sync.finalize(environment)
@@ -772,3 +768,46 @@ def update_intervals_for_new_snapshots(
 
     if snapshots_intervals:
         state_sync.add_snapshots_intervals(snapshots_intervals)
+
+
+def _create_plan_summary_msg(
+    snapshots: t.Iterable[SnapshotInfoLike], snapshots_to_count: t.List[SnapshotId]
+) -> str:
+    model_kind_counts: t.Dict[str, int] = {}
+    audit_counts: t.Dict[str, int] = {}
+    for snapshot in snapshots:
+        if snapshot.snapshot_id in snapshots_to_count:
+            if snapshot.is_audit:
+                audit_counts["standalone"] = audit_counts.get("standalone", 0) + 1
+            if (
+                snapshot.is_model
+                and snapshot.model_kind_name
+                and not snapshot.model_kind_name.is_symbolic
+            ):
+                kind_name = snapshot.model_kind_name
+                model_kind_counts[kind_name] = model_kind_counts.get(kind_name, 0) + 1
+            if (
+                snapshot.is_model
+                # removed snapshot SnapshotTableInfos don't have `model`
+                and hasattr(snapshot, "model")
+                and snapshot.model
+                and snapshot.model.audits
+            ):
+                if snapshot.model.kind.is_external:  # type: ignore
+                    audit_counts["EXTERNAL model"] = audit_counts.get("EXTERNAL model", 0) + len(
+                        snapshot.model.audits  # type: ignore
+                    )
+                else:
+                    audit_counts["model"] = audit_counts.get("model", 0) + len(
+                        snapshot.model.audits  # type: ignore
+                    )
+
+    summary_str = ", ".join(
+        [f"{v} {k} model{'s' if v > 1 else ''}" for k, v in model_kind_counts.items()]
+    )
+    for audit_type in ["EXTERNAL model", "model", "standalone"]:
+        if audit_type in audit_counts:
+            count = audit_counts[audit_type]
+            summary_str += f", {count} {audit_type} audit{'s' if count > 1 else ''}"
+
+    return summary_str

sqlmesh/core/snapshot/evaluator.py

@@ -944,8 +944,8 @@ def _promote_snapshot(
             )
             adapter.execute(snapshot.model.render_on_virtual_update(**render_kwargs))
 
-        if on_complete is not None:
-            on_complete(snapshot)
+            if on_complete is not None:
+                on_complete(snapshot)
 
     def _demote_snapshot(
         self,

tests/cli/test_cli.py

@@ -294,7 +294,6 @@ def test_plan_verbose(runner, tmp_path):
     )
     assert_plan_success(result)
     assert "sqlmesh_example.seed_model                         created" in result.output
-    assert "sqlmesh_example.seed_model                         updated" in result.output
 
 
 def test_plan_very_verbose(runner, tmp_path, copy_to_temp_path):

web/server/console.py

@@ -11,7 +11,7 @@
 from sqlmesh.core.console import TerminalConsole
 from sqlmesh.core.environment import EnvironmentNamingInfo
 from sqlmesh.core.plan.definition import EvaluatablePlan
-from sqlmesh.core.snapshot import Snapshot, SnapshotInfoLike
+from sqlmesh.core.snapshot import Snapshot, SnapshotInfoLike, SnapshotId
 from sqlmesh.core.test import ModelTest
 from sqlmesh.utils.date import now_timestamp
 from web.server import models
@@ -170,7 +170,12 @@ def start_promotion_progress(
 
         self.log_event_plan_apply()
 
-    def update_promotion_progress(self, snapshot: SnapshotInfoLike, promoted: bool) -> None:
+    def update_promotion_progress(
+        self,
+        snapshot: SnapshotInfoLike,
+        promoted: bool,
+        snapshots_with_virtual_views: t.List[SnapshotId],
+    ) -> None:
         if self.plan_apply_stage_tracker and self.plan_apply_stage_tracker.promote:
             self.plan_apply_stage_tracker.promote.update(
                 {"num_tasks": self.plan_apply_stage_tracker.promote.num_tasks + 1}