Feat: Introduce plan explain mode

Author: izeigerman

sqlmesh/cli/main.py

@@ -457,7 +457,13 @@ def diff(ctx: click.Context, environment: t.Optional[str] = None) -> None:
 @click.option(
     "--diff-rendered",
     is_flag=True,
-    help="Output text differences for the rendered versions of the models and standalone audits",
+    help="Output text differences for the rendered versions of the models and standalone audits.",
+    default=None,
+)
+@click.option(
+    "--explain",
+    is_flag=True,
+    help="Explain the plan instead of applying it.",
     default=None,
 )
 @opt.verbose

sqlmesh/core/context.py

@@ -89,7 +89,7 @@
     NotificationTarget,
     NotificationTargetManager,
 )
-from sqlmesh.core.plan import Plan, PlanBuilder, SnapshotIntervals
+from sqlmesh.core.plan import Plan, PlanBuilder, SnapshotIntervals, PlanExplainer
 from sqlmesh.core.plan.definition import UserProvidedFlags
 from sqlmesh.core.reference import ReferenceGraph
 from sqlmesh.core.scheduler import Scheduler, CompletionStatus
@@ -1215,6 +1215,7 @@ def plan(
         run: t.Optional[bool] = None,
         diff_rendered: t.Optional[bool] = None,
         skip_linter: t.Optional[bool] = None,
+        explain: t.Optional[bool] = None,
     ) -> Plan:
         """Interactively creates a plan.
 
@@ -1260,6 +1261,7 @@ def plan(
             run: Whether to run latest intervals as part of the plan application.
             diff_rendered: Whether the diff should compare raw vs rendered models
             skip_linter: Linter runs by default so this will skip it if enabled
+            explain: Whether to explain the plan instead of applying it.
 
         Returns:
             The populated Plan object.
@@ -1287,6 +1289,7 @@ def plan(
             run=run,
             diff_rendered=diff_rendered,
             skip_linter=skip_linter,
+            explain=explain,
         )
 
         plan = plan_builder.build()
@@ -1296,6 +1299,9 @@ def plan(
             # or if there are any uncategorized snapshots in the plan
             no_prompts = False
 
+        if explain:
+            auto_apply = True
+
         self.console.plan(
             plan_builder,
             auto_apply if auto_apply is not None else self.config.plan.auto_apply,
@@ -1332,6 +1338,7 @@ def plan_builder(
         run: t.Optional[bool] = None,
         diff_rendered: t.Optional[bool] = None,
         skip_linter: t.Optional[bool] = None,
+        explain: t.Optional[bool] = None,
     ) -> PlanBuilder:
         """Creates a plan builder.
 
@@ -1548,6 +1555,7 @@ def plan_builder(
             interval_end_per_model=max_interval_end_per_model,
             console=self.console,
             user_provided_flags=user_provided_flags,
+            explain=explain or False,
         )
 
     def apply(
@@ -1572,6 +1580,16 @@ def apply(
             return
         if plan.uncategorized:
             raise UncategorizedPlanError("Can't apply a plan with uncategorized changes.")
+
+        if plan.explain:
+            explainer = PlanExplainer(
+                state_reader=self.state_reader,
+                default_catalog=self.default_catalog,
+                console=self.console,
+            )
+            explainer.evaluate(plan.to_evaluatable())
+            return
+
         self.notification_target_manager.notify(
             NotificationEvent.APPLY_START,
             environment=plan.environment_naming_info.name,

sqlmesh/core/environment.py

@@ -229,6 +229,19 @@ def summary(self) -> EnvironmentSummary:
             finalized_ts=self.finalized_ts,
         )
 
+    def can_partially_promote(self, existing_environment: Environment) -> bool:
+        """Returns True if the existing environment can be partially promoted to the current environment.
+
+        Partial promotion means that we don't need to re-create views for snapshots that are already promoted in the
+        target environment.
+        """
+        return (
+            bool(existing_environment.finalized_ts)
+            and not existing_environment.expired
+            and existing_environment.gateway_managed == self.gateway_managed
+            and existing_environment.name == c.PROD
+        )
+
     def _convert_list_to_models_and_store(
         self, field: str, type_: t.Type[PydanticType]
     ) -> t.Optional[t.List[PydanticType]]:

sqlmesh/core/plan/__init__.py

@@ -10,3 +10,4 @@
     PlanEvaluator as PlanEvaluator,
     update_intervals_for_new_snapshots as update_intervals_for_new_snapshots,
 )
+from sqlmesh.core.plan.explainer import PlanExplainer as PlanExplainer

sqlmesh/core/plan/builder.py

@@ -83,6 +83,7 @@ class PlanBuilder:
             environment state, or to use whatever snapshots are in the current environment state even if
             the environment is not finalized.
         interval_end_per_model: The mapping from model FQNs to target end dates.
+        explain: Whether to explain the plan instead of applying it.
     """
 
     def __init__(
@@ -112,6 +113,7 @@ def __init__(
         enable_preview: bool = False,
         end_bounded: bool = False,
         ensure_finalized_snapshots: bool = False,
+        explain: bool = False,
         interval_end_per_model: t.Optional[t.Dict[str, int]] = None,
         console: t.Optional[PlanBuilderConsole] = None,
         user_provided_flags: t.Optional[t.Dict[str, UserProvidedFlags]] = None,
@@ -142,6 +144,7 @@ def __init__(
         self._console = console or get_console()
         self._choices: t.Dict[SnapshotId, SnapshotChangeCategory] = {}
         self._user_provided_flags = user_provided_flags
+        self._explain = explain
 
         self._start = start
         if not self._start and (
@@ -273,6 +276,7 @@ def build(self) -> Plan:
             empty_backfill=self._empty_backfill,
             no_gaps=self._no_gaps,
             forward_only=self._forward_only,
+            explain=self._explain,
             allow_destructive_models=t.cast(t.Set, self._allow_destructive_models),
             include_unmodified=self._include_unmodified,
             environment_ttl=self._environment_ttl,

sqlmesh/core/plan/definition.py

@@ -46,6 +46,7 @@ class Plan(PydanticModel, frozen=True):
     include_unmodified: bool
     end_bounded: bool
     ensure_finalized_snapshots: bool
+    explain: bool
 
     environment_ttl: t.Optional[str] = None
     environment_naming_info: EnvironmentNamingInfo

sqlmesh/core/plan/evaluator.py

@@ -36,7 +36,7 @@
     SnapshotCreationFailedError,
 )
 from sqlmesh.utils import CompletionStatus
-from sqlmesh.core.state_sync import StateSync
+from sqlmesh.core.state_sync import StateSync, StateReader
 from sqlmesh.core.state_sync.base import PromotionResult
 from sqlmesh.utils.concurrency import NodeExecutionFailedError
 from sqlmesh.utils.errors import PlanError
@@ -284,23 +284,7 @@ def _push(
             new_snapshots=plan.new_snapshots, plan_id=plan.plan_id
         )
 
-        promoted_snapshot_ids = (
-            set(plan.environment.promoted_snapshot_ids)
-            if plan.environment.promoted_snapshot_ids is not None
-            else None
-        )
-
-        def _should_create(s: Snapshot) -> bool:
-            if not s.is_model or s.is_symbolic:
-                return False
-            # Only create tables for snapshots that we're planning to promote or that were selected for backfill
-            return (
-                plan.is_selected_for_backfill(s.name)
-                or promoted_snapshot_ids is None
-                or s.snapshot_id in promoted_snapshot_ids
-            )
-
-        snapshots_to_create = [s for s in snapshots.values() if _should_create(s)]
+        snapshots_to_create = get_snapshots_to_create(plan, snapshots)
 
         completion_status = None
         progress_stopped = False
@@ -573,32 +557,7 @@ def _run_audits_for_metadata_snapshots(
         plan: EvaluatablePlan,
         new_snapshots: t.Dict[SnapshotId, Snapshot],
     ) -> None:
-        # Filter out snapshots that are not categorized as metadata changes on models
-        metadata_snapshots = []
-        for snapshot in new_snapshots.values():
-            if not snapshot.is_metadata or not snapshot.is_model or not snapshot.evaluatable:
-                continue
-
-            metadata_snapshots.append(snapshot)
-
-        # Bulk load all the previous snapshots
-        previous_snapshots = self.state_sync.get_snapshots(
-            [
-                s.previous_version.snapshot_id(s.name)
-                for s in metadata_snapshots
-                if s.previous_version
-            ]
-        ).values()
-
-        # Check if any of the snapshots have modifications to the audits field by comparing the hashes
-        audit_snapshots = {}
-        for snapshot, previous_snapshot in zip(metadata_snapshots, previous_snapshots):
-            new_audits_hash = snapshot.model.audit_metadata_hash()
-            previous_audit_hash = previous_snapshot.model.audit_metadata_hash()
-
-            if snapshot.model.audits and previous_audit_hash != new_audits_hash:
-                audit_snapshots[snapshot.snapshot_id] = snapshot
-
+        audit_snapshots = get_audit_only_snapshots(new_snapshots, self.state_sync)
         if not audit_snapshots:
             return
 
@@ -636,3 +595,52 @@ def update_intervals_for_new_snapshots(
 
     if snapshots_intervals:
         state_sync.add_snapshots_intervals(snapshots_intervals)
+
+
+def get_audit_only_snapshots(
+    new_snapshots: t.Dict[SnapshotId, Snapshot], state_reader: StateReader
+) -> t.Dict[SnapshotId, Snapshot]:
+    metadata_snapshots = []
+    for snapshot in new_snapshots.values():
+        if not snapshot.is_metadata or not snapshot.is_model or not snapshot.evaluatable:
+            continue
+
+        metadata_snapshots.append(snapshot)
+
+    # Bulk load all the previous snapshots
+    previous_snapshots = state_reader.get_snapshots(
+        [s.previous_version.snapshot_id(s.name) for s in metadata_snapshots if s.previous_version]
+    ).values()
+
+    # Check if any of the snapshots have modifications to the audits field by comparing the hashes
+    audit_snapshots = {}
+    for snapshot, previous_snapshot in zip(metadata_snapshots, previous_snapshots):
+        new_audits_hash = snapshot.model.audit_metadata_hash()
+        previous_audit_hash = previous_snapshot.model.audit_metadata_hash()
+
+        if snapshot.model.audits and previous_audit_hash != new_audits_hash:
+            audit_snapshots[snapshot.snapshot_id] = snapshot
+
+    return audit_snapshots
+
+
+def get_snapshots_to_create(
+    plan: EvaluatablePlan, snapshots: t.Dict[SnapshotId, Snapshot]
+) -> t.List[Snapshot]:
+    promoted_snapshot_ids = (
+        set(plan.environment.promoted_snapshot_ids)
+        if plan.environment.promoted_snapshot_ids is not None
+        else None
+    )
+
+    def _should_create(s: Snapshot) -> bool:
+        if not s.is_model or s.is_symbolic:
+            return False
+        # Only create tables for snapshots that we're planning to promote or that were selected for backfill
+        return (
+            plan.is_selected_for_backfill(s.name)
+            or promoted_snapshot_ids is None
+            or s.snapshot_id in promoted_snapshot_ids
+        )
+
+    return [s for s in snapshots.values() if _should_create(s)]