fix(plan): add explainer visit method for PhysicalLayerSchemaCreationStage

`RichExplainerConsole` had no `visit_physical_layer_schema_creation_stage`,
so explaining a plan that includes `PhysicalLayerSchemaCreationStage` logged
`ERROR - Unexpected stage: PhysicalLayerSchemaCreationStage` and silently
omitted the stage from the explained plan.

Add the missing visit method. It mirrors the schema derivation in
`SnapshotEvaluator.create_physical_schemas` (non-symbolic models only, table
name resolved through the deployability index) and lists the distinct physical
schemas that will be created.

`test_plan_explain` now asserts that no stage is reported as unexpected, which
covers this stage and guards against the same omission for future ones.

Fixes #5619

Signed-off-by: nkwork9999 <143652584+nkwork9999@users.noreply.github.com>

Author: nkwork9999

sqlmesh/core/plan/explainer.py

@@ -8,9 +8,11 @@
 
 from rich.console import Console as RichConsole
 from rich.tree import Tree
+from sqlglot import exp
 from sqlglot.dialects.dialect import DialectType
 from sqlmesh.core import constants as c
 from sqlmesh.core.console import Console, TerminalConsole, get_console
+from sqlmesh.core.dialect import schema_
 from sqlmesh.core.environment import EnvironmentNamingInfo
 from sqlmesh.core.plan.common import (
     SnapshotIntervalClearRequest,
@@ -145,6 +147,29 @@ def visit_before_all_stage(self, stage: stages.BeforeAllStage) -> Tree:
     def visit_after_all_stage(self, stage: stages.AfterAllStage) -> Tree:
         return Tree("[bold]Execute after all statements[/bold]")
 
+    def visit_physical_layer_schema_creation_stage(
+        self, stage: stages.PhysicalLayerSchemaCreationStage
+    ) -> t.Optional[Tree]:
+        tables = [
+            exp.to_table(
+                snapshot.table_name(is_deployable=stage.deployability_index.is_deployable(snapshot))
+            )
+            for snapshot in stage.snapshots
+            if snapshot.is_model and not snapshot.is_symbolic
+        ]
+        schemas = {
+            schema_(table.args["db"], table.args.get("catalog")).sql(dialect=self.dialect)
+            for table in tables
+            if table.db
+        }
+        if not schemas:
+            return None
+
+        tree = Tree("[bold]Create physical schemas if they do not exist[/bold]")
+        for schema in sorted(schemas):
+            tree.add(schema)
+        return tree
+
     def visit_physical_layer_update_stage(self, stage: stages.PhysicalLayerUpdateStage) -> Tree:
         snapshots = [
             s for s in stage.snapshots if s.snapshot_id in stage.snapshots_with_missing_intervals

tests/core/integration/test_plan_options.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import logging
 import typing as t
 import pytest
 from sqlmesh.core.console import (
@@ -112,7 +113,7 @@ def test_empty_backfill_new_model(init_and_plan_context: t.Callable):
 
 
 @time_machine.travel("2023-01-08 15:00:00 UTC")
-def test_plan_explain(init_and_plan_context: t.Callable):
+def test_plan_explain(init_and_plan_context: t.Callable, caplog):
     old_console = get_console()
     set_console(TerminalConsole())
 
@@ -131,16 +132,20 @@ def test_plan_explain(init_and_plan_context: t.Callable):
     common_kwargs = dict(skip_tests=True, no_prompts=True, explain=True)
 
     # For now just making sure the plan doesn't error
-    context.plan("dev", **common_kwargs)
-    context.plan("dev", **common_kwargs, skip_backfill=True)
-    context.plan("dev", **common_kwargs, empty_backfill=True)
-    context.plan("dev", **common_kwargs, forward_only=True, enable_preview=True)
-    context.plan("prod", **common_kwargs)
-    context.plan("prod", **common_kwargs, forward_only=True)
-    context.plan("prod", **common_kwargs, restate_models=[waiter_revenue_by_day_model.name])
+    with caplog.at_level(logging.ERROR, logger="sqlmesh.core.plan.explainer"):
+        context.plan("dev", **common_kwargs)
+        context.plan("dev", **common_kwargs, skip_backfill=True)
+        context.plan("dev", **common_kwargs, empty_backfill=True)
+        context.plan("dev", **common_kwargs, forward_only=True, enable_preview=True)
+        context.plan("prod", **common_kwargs)
+        context.plan("prod", **common_kwargs, forward_only=True)
+        context.plan("prod", **common_kwargs, restate_models=[waiter_revenue_by_day_model.name])
 
     set_console(old_console)
 
+    # Every stage produced by the plan must have a corresponding visit method on the explainer
+    assert "Unexpected stage" not in caplog.text
+
     # Make sure that the now changes were actually applied
     for target_env in ("dev", "prod"):
         plan = context.plan_builder(target_env, skip_tests=True).build()