Feat: print CLI messages when no physical layer or model evals occurred (#4113)

Author: treysp

sqlmesh/core/console.py

@@ -1303,7 +1303,7 @@ def show_model_difference_summary(
         ):
             self._print(
                 Tree(
-                    f"[bold]Differences from the `{context_diff.create_from if context_diff.is_new_environment else context_diff.environment}` environment:\n"
+                    f"\n[bold]Differences from the `{context_diff.create_from if context_diff.is_new_environment else context_diff.environment}` environment:\n"
                 )
             )
 

sqlmesh/core/plan/evaluator.py

@@ -37,6 +37,7 @@
     SnapshotInfoLike,
     SnapshotTableInfo,
 )
+from sqlmesh.utils import CompletionStatus
 from sqlmesh.core.state_sync import StateSync
 from sqlmesh.core.state_sync.base import PromotionResult
 from sqlmesh.core.user import User
@@ -133,10 +134,14 @@ def evaluate(
                 execution_time=plan.execution_time,
             )
 
-            self._push(plan, snapshots, deployability_index_for_creation)
+            push_completion_status = self._push(plan, snapshots, deployability_index_for_creation)
+            if push_completion_status.is_nothing_to_do:
+                self.console.log_status_update(
+                    "\n[green]SKIP: No physical layer updates to perform[/green]\n"
+                )
             update_intervals_for_new_snapshots(plan.new_snapshots, self.state_sync)
             self._restate(plan, snapshots_by_name)
-            self._backfill(
+            first_bf_completion_status = self._backfill(
                 plan,
                 snapshots_by_name,
                 before_promote_snapshots,
@@ -146,20 +151,22 @@ def evaluate(
             promotion_result = self._promote(
                 plan, snapshots, before_promote_snapshots, deployability_index_for_creation
             )
-            self._backfill(
+            second_bf_completion_status = self._backfill(
                 plan,
                 snapshots_by_name,
                 after_promote_snapshots,
                 deployability_index_for_evaluation,
                 circuit_breaker=circuit_breaker,
             )
+            if (
+                first_bf_completion_status.is_nothing_to_do
+                and second_bf_completion_status.is_nothing_to_do
+            ):
+                self.console.log_status_update("[green]SKIP: No model batches to execute[/green]\n")
             self._update_views(
                 plan, snapshots, promotion_result, deployability_index_for_evaluation
             )
 
-            if not plan.requires_backfill:
-                self.console.log_success("Virtual Update executed successfully")
-
             execute_environment_statements(
                 adapter=self.snapshot_evaluator.adapter,
                 environment_statements=plan.environment_statements or [],
@@ -187,7 +194,7 @@ def _backfill(
         selected_snapshots: t.Set[str],
         deployability_index: DeployabilityIndex,
         circuit_breaker: t.Optional[t.Callable[[], bool]] = None,
-    ) -> None:
+    ) -> CompletionStatus:
         """Backfill missing intervals for snapshots that are part of the given plan.
 
         Args:
@@ -212,10 +219,10 @@ def _backfill(
                     )
                 )
             self.state_sync.add_snapshots_intervals(intervals_to_add)
-            return
+            return CompletionStatus.NOTHING_TO_DO
 
         if not plan.requires_backfill or not selected_snapshots:
-            return
+            return CompletionStatus.NOTHING_TO_DO
 
         scheduler = self.create_scheduler(snapshots_by_name.values())
         completion_status = scheduler.run(
@@ -236,12 +243,14 @@ def _backfill(
         if completion_status.is_failure:
             raise PlanError("Plan application failed.")
 
+        return completion_status
+
     def _push(
         self,
         plan: EvaluatablePlan,
         snapshots: t.Dict[SnapshotId, Snapshot],
         deployability_index: t.Optional[DeployabilityIndex] = None,
-    ) -> None:
+    ) -> CompletionStatus:
         """Push the snapshots to the state sync.
 
         As a part of plan pushing, snapshot tables are created.
@@ -268,10 +277,10 @@ def _should_create(s: Snapshot) -> bool:
 
         snapshots_to_create = [s for s in snapshots.values() if _should_create(s)]
 
-        completed = False
+        completion_status = None
         progress_stopped = False
         try:
-            self.snapshot_evaluator.create(
+            completion_status = self.snapshot_evaluator.create(
                 snapshots_to_create,
                 snapshots,
                 allow_destructive_snapshots=plan.allow_destructive_models,
@@ -281,7 +290,6 @@ def _should_create(s: Snapshot) -> bool:
                 ),
                 on_complete=self.console.update_creation_progress,
             )
-            completed = True
         except NodeExecutionFailedError as ex:
             self.console.stop_creation_progress(success=False)
             progress_stopped = True
@@ -292,13 +300,16 @@ def _should_create(s: Snapshot) -> bool:
             raise PlanError("Plan application failed.")
         finally:
             if not progress_stopped:
-                self.console.stop_creation_progress(success=completed)
+                self.console.stop_creation_progress(
+                    success=completion_status is not None and completion_status.is_success
+                )
 
         self.state_sync.push_snapshots(plan.new_snapshots)
 
         analytics.collector.on_snapshots_created(
             new_snapshots=plan.new_snapshots, plan_id=plan.plan_id
         )
+        return completion_status
 
     def _promote(
         self,

sqlmesh/core/scheduler.py

@@ -1,5 +1,4 @@
 from __future__ import annotations
-from enum import Enum
 import logging
 import typing as t
 from sqlglot import exp
@@ -31,6 +30,7 @@
     parent_snapshots_by_name,
 )
 from sqlmesh.core.state_sync import StateSync
+from sqlmesh.utils import CompletionStatus
 from sqlmesh.utils.concurrency import concurrent_apply_to_dag, NodeExecutionFailedError
 from sqlmesh.utils.dag import DAG
 from sqlmesh.utils.date import (
@@ -48,24 +48,6 @@
 SchedulingUnit = t.Tuple[str, t.Tuple[Interval, int]]
 
 
-class CompletionStatus(Enum):
-    SUCCESS = "success"
-    FAILURE = "failure"
-    NOTHING_TO_DO = "nothing_to_do"
-
-    @property
-    def is_success(self) -> bool:
-        return self == CompletionStatus.SUCCESS
-
-    @property
-    def is_failure(self) -> bool:
-        return self == CompletionStatus.FAILURE
-
-    @property
-    def is_nothing_to_do(self) -> bool:
-        return self == CompletionStatus.NOTHING_TO_DO
-
-
 class Scheduler:
     """Schedules and manages the evaluation of snapshots.
 

sqlmesh/core/snapshot/evaluator.py

@@ -51,7 +51,7 @@
     ViewKind,
     CustomKind,
 )
-
+from sqlmesh.utils import CompletionStatus
 from sqlmesh.core.schema_diff import has_drop_alteration, get_dropped_column_names
 from sqlmesh.core.snapshot import (
     DeployabilityIndex,
@@ -289,7 +289,7 @@ def create(
         on_start: t.Optional[t.Callable] = None,
         on_complete: t.Optional[t.Callable[[SnapshotInfoLike], None]] = None,
         allow_destructive_snapshots: t.Optional[t.Set[str]] = None,
-    ) -> None:
+    ) -> CompletionStatus:
         """Creates a physical snapshot schema and table for the given collection of snapshots.
 
         Args:
@@ -299,6 +299,9 @@ def create(
             on_start: A callback to initialize the snapshot creation progress bar.
             on_complete: A callback to call on each successfully created snapshot.
             allow_destructive_snapshots: Set of snapshots that are allowed to have destructive schema changes.
+
+        Returns:
+            CompletionStatus: The status of the creation operation (success, failure, nothing to do).
         """
         snapshots_with_table_names = defaultdict(set)
         tables_by_gateway_and_schema: t.Dict[t.Union[str, None], t.Dict[exp.Table, set[str]]] = (
@@ -366,7 +369,7 @@ def _get_data_objects(
                 target_deployability_flags[snapshot.name].sort()
 
         if not snapshots_to_create:
-            return
+            return CompletionStatus.NOTHING_TO_DO
         if on_start:
             on_start(len(snapshots_to_create))
 
@@ -381,6 +384,7 @@ def _get_data_objects(
             on_complete=on_complete,
             allow_destructive_snapshots=allow_destructive_snapshots,
         )
+        return CompletionStatus.SUCCESS
 
     def _create_snapshots(
         self,

sqlmesh/utils/__init__.py

@@ -16,7 +16,7 @@
 from collections import defaultdict
 from contextlib import contextmanager
 from copy import deepcopy
-from enum import IntEnum
+from enum import IntEnum, Enum
 from functools import lru_cache, reduce, wraps
 from pathlib import Path
 
@@ -346,3 +346,21 @@ class Verbosity(IntEnum):
     DEFAULT = 0
     VERBOSE = 1
     VERY_VERBOSE = 2
+
+
+class CompletionStatus(Enum):
+    SUCCESS = "success"
+    FAILURE = "failure"
+    NOTHING_TO_DO = "nothing_to_do"
+
+    @property
+    def is_success(self) -> bool:
+        return self == CompletionStatus.SUCCESS
+
+    @property
+    def is_failure(self) -> bool:
+        return self == CompletionStatus.FAILURE
+
+    @property
+    def is_nothing_to_do(self) -> bool:
+        return self == CompletionStatus.NOTHING_TO_DO

tests/cli/test_cli.py

@@ -154,10 +154,6 @@ def assert_plan_success(result, new_env="prod", from_env="prod") -> None:
     assert_backfill_success(result)
 
 
-def assert_virtual_update(result) -> None:
-    assert "Virtual Update executed" in result.output
-
-
 def test_version(runner, tmp_path):
     from sqlmesh import __version__ as SQLMESH_VERSION
 
@@ -275,7 +271,7 @@ def test_plan_skip_backfill(runner, tmp_path, flag):
         input="y\n",
     )
     assert result.exit_code == 0
-    assert_virtual_update(result)
+    assert_virtual_layer_updated(result)
     assert "Model batches executed" not in result.output
 
 
@@ -404,8 +400,9 @@ def test_plan_dev_create_from_virtual(runner, tmp_path):
     )
     assert result.exit_code == 0
     assert_new_env(result, "dev2", "dev", initialize=False)
+    assert "SKIP: No physical layer updates to perform" in result.output
+    assert "SKIP: No model batches to execute" in result.output
     assert_virtual_layer_updated(result)
-    assert_virtual_update(result)
 
 
 def test_plan_dev_create_from(runner, tmp_path):
@@ -542,7 +539,6 @@ def test_plan_dev_no_changes(runner, tmp_path):
     assert result.exit_code == 0
     assert_new_env(result, "dev", initialize=False)
     assert_virtual_layer_updated(result)
-    assert_virtual_update(result)
 
 
 def test_plan_nonbreaking(runner, tmp_path):