diff --git a/sqlmesh/core/plan/explainer.py b/sqlmesh/core/plan/explainer.py index f0a1e44aff..6e3fb8f9c3 100644 --- a/sqlmesh/core/plan/explainer.py +++ b/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 diff --git a/tests/core/integration/test_plan_options.py b/tests/core/integration/test_plan_options.py index a50dc145cd..8988e5d409 100644 --- a/tests/core/integration/test_plan_options.py +++ b/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()