Overhaul CLI evaluation progress bar

Author: treysp

sqlmesh/core/config/root.py

@@ -147,7 +147,7 @@ class Config(BaseConfig):
     }
 
     _connection_config_validator = connection_config_validator
-    _scheduler_config_validator = scheduler_config_validator
+    _scheduler_config_validator = scheduler_config_validator  # type: ignore
     _variables_validator = variables_validator
 
     @field_validator("gateways", mode="before")

sqlmesh/core/console.py

@@ -165,6 +165,8 @@ def evaluate(
         execution_time: TimeLike,
         deployability_index: DeployabilityIndex,
         batch_index: int,
+        environment_naming_info: EnvironmentNamingInfo,
+        default_catalog: t.Optional[str],
         **kwargs: t.Any,
     ) -> None:
         """Evaluate a snapshot and add the processed interval to the state sync.
@@ -205,6 +207,7 @@ def evaluate(
             wap_id=wap_id,
             **kwargs,
         )
+        get_console().store_evaluation_audit_results(snapshot, audit_results)
 
         audit_errors_to_raise: t.List[AuditError] = []
         for audit_result in (result for result in audit_results if result.count):
@@ -224,8 +227,13 @@ def evaluate(
             if audit_result.blocking:
                 audit_errors_to_raise.append(error)
             else:
+                display_name = snapshot.display_name(
+                    environment_naming_info,
+                    default_catalog,
+                    self.snapshot_evaluator.adapter.dialect,
+                )
                 get_console().log_warning(
-                    f"\n{error}.",
+                    f"\n{display_name}: {error}.",
                     long_message=f"{error}. Audit query:\n{error.query.sql(error.adapter_dialect)}",
                 )
 
@@ -424,7 +432,7 @@ def run_merged_intervals(
         batched_intervals = self.batch_intervals(merged_intervals)
 
         self.console.start_evaluation_progress(
-            {snapshot: len(intervals) for snapshot, intervals in batched_intervals.items()},
+            batched_intervals,
             environment_naming_info,
             self.default_catalog,
         )
@@ -473,6 +481,8 @@ def evaluate_node(node: SchedulingUnit) -> None:
                     execution_time=execution_time,
                     deployability_index=deployability_index,
                     batch_index=batch_idx,
+                    environment_naming_info=environment_naming_info,
+                    default_catalog=self.default_catalog,
                 )
                 evaluation_duration_ms = now_timestamp() - execution_start_ts
             finally:

sqlmesh/core/scheduler.py

@@ -132,18 +132,18 @@ def assert_model_versions_created(result) -> None:
     assert "Model versions created successfully" in result.output
 
 
-def assert_model_batches_executed(result) -> None:
-    assert "Model batches executed successfully" in result.output
+def assert_model_batches_evaluated(result) -> None:
+    assert "Model batches evaluated successfully" in result.output
 
 
-def assert_target_env_updated(result) -> None:
-    assert "Target environment updated successfully" in result.output
+def assert_env_views_updated(result) -> None:
+    assert "Environment views updated successfully" in result.output
 
 
 def assert_backfill_success(result) -> None:
     assert_model_versions_created(result)
-    assert_model_batches_executed(result)
-    assert_target_env_updated(result)
+    assert_model_batches_evaluated(result)
+    assert_env_views_updated(result)
 
 
 def assert_plan_success(result, new_env="prod", from_env="prod") -> None:
@@ -220,9 +220,9 @@ def test_plan_restate_model(runner, tmp_path):
     assert result.exit_code == 0
     assert_duckdb_test(result)
     assert "No changes to plan: project files match the `prod` environment" in result.output
-    assert "sqlmesh_example.full_model evaluated in" in result.output
-    assert_model_batches_executed(result)
-    assert_target_env_updated(result)
+    assert "sqlmesh_example.full_model [full refresh" in result.output
+    assert_model_batches_evaluated(result)
+    assert_env_views_updated(result)
 
 
 @pytest.mark.parametrize("flag", ["--skip-backfill", "--dry-run"])
@@ -260,7 +260,7 @@ def test_plan_auto_apply(runner, tmp_path):
 
     # confirm verbose output not present
     assert "sqlmesh_example.seed_model created" not in result.output
-    assert "sqlmesh_example.seed_model promoted" not in result.output
+    assert "sqlmesh_example.seed_model updated" not in result.output
 
 
 def test_plan_verbose(runner, tmp_path):
@@ -272,7 +272,7 @@ 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 promoted" in result.output
+    assert "sqlmesh_example.seed_model updated" in result.output
 
 
 def test_plan_very_verbose(runner, tmp_path, copy_to_temp_path):
@@ -374,7 +374,7 @@ def test_plan_dev_create_from_virtual(runner, tmp_path):
     )
     assert result.exit_code == 0
     assert_new_env(result, "dev2", "dev", initialize=False)
-    assert_target_env_updated(result)
+    assert_env_views_updated(result)
     assert_virtual_update(result)
 
 
@@ -511,7 +511,7 @@ def test_plan_dev_no_changes(runner, tmp_path):
     )
     assert result.exit_code == 0
     assert_new_env(result, "dev", initialize=False)
-    assert_target_env_updated(result)
+    assert_env_views_updated(result)
     assert_virtual_update(result)
 
 
@@ -530,8 +530,8 @@ def test_plan_nonbreaking(runner, tmp_path):
     assert "+  'a' AS new_col" in result.output
     assert "Directly Modified: sqlmesh_example.incremental_model (Non-breaking)" in result.output
     assert "sqlmesh_example.full_model (Indirect Non-breaking)" in result.output
-    assert "sqlmesh_example.incremental_model evaluated in" in result.output
-    assert "sqlmesh_example.full_model evaluated in" not in result.output
+    assert "sqlmesh_example.incremental_model [insert" in result.output
+    assert "sqlmesh_example.full_model evaluated [full refresh" not in result.output
     assert_backfill_success(result)
 
 
@@ -588,8 +588,8 @@ def test_plan_breaking(runner, tmp_path):
     assert result.exit_code == 0
     assert "+  item_id + 1 AS item_id," in result.output
     assert "Directly Modified: sqlmesh_example.full_model (Breaking)" in result.output
-    assert "sqlmesh_example.full_model evaluated in" in result.output
-    assert "sqlmesh_example.incremental_model evaluated in" not in result.output
+    assert "sqlmesh_example.full_model [full refresh" in result.output
+    assert "sqlmesh_example.incremental_model [insert" not in result.output
     assert_backfill_success(result)
 
 
@@ -627,8 +627,8 @@ def test_plan_dev_select(runner, tmp_path):
     assert "+  item_id + 1 AS item_id," not in result.output
     assert "Directly Modified: sqlmesh_example__dev.full_model (Breaking)" not in result.output
     # only incremental_model backfilled
-    assert "sqlmesh_example__dev.incremental_model evaluated in" in result.output
-    assert "sqlmesh_example__dev.full_model evaluated in" not in result.output
+    assert "sqlmesh_example__dev.incremental_model [insert" in result.output
+    assert "sqlmesh_example__dev.full_model [full refresh" not in result.output
     assert_backfill_success(result)
 
 
@@ -666,8 +666,8 @@ def test_plan_dev_backfill(runner, tmp_path):
         "Directly Modified: sqlmesh_example__dev.incremental_model (Non-breaking)" in result.output
     )
     # only incremental_model backfilled
-    assert "sqlmesh_example__dev.incremental_model evaluated in" in result.output
-    assert "sqlmesh_example__dev.full_model evaluated in" not in result.output
+    assert "sqlmesh_example__dev.incremental_model [insert" in result.output
+    assert "sqlmesh_example__dev.full_model [full refresh" not in result.output
     assert_backfill_success(result)
 
 
@@ -696,7 +696,7 @@ def test_run_dev(runner, tmp_path, flag):
     # Confirm backfill occurs when we run non-backfilled dev env
     result = runner.invoke(cli, ["--log-file-dir", tmp_path, "--paths", tmp_path, "run", "dev"])
     assert result.exit_code == 0
-    assert_model_batches_executed(result)
+    assert_model_batches_evaluated(result)
 
 
 @time_machine.travel(FREEZE_TIME)
@@ -728,7 +728,7 @@ def test_run_cron_elapsed(runner, tmp_path):
         result = runner.invoke(cli, ["--log-file-dir", tmp_path, "--paths", tmp_path, "run"])
 
     assert result.exit_code == 0
-    assert_model_batches_executed(result)
+    assert_model_batches_evaluated(result)
 
 
 def test_clean(runner, tmp_path):

tests/cli/test_cli.py

@@ -470,7 +470,9 @@ def test_override_builtin_audit_blocking_mode():
         plan = context.plan(auto_apply=True, no_prompts=True)
         new_snapshot = next(iter(plan.context_diff.new_snapshots.values()))
 
-        assert mock_logger.call_args_list[0][0][0] == "\n'not_null' audit error: 1 row failed."
+        assert (
+            mock_logger.call_args_list[0][0][0] == "\ndb.x: 'not_null' audit error: 1 row failed."
+        )
 
     # Even though there are two builtin audits referenced in the above definition, we only
     # store the one that overrides `blocking` in the snapshot; the other one isn't needed
@@ -1401,7 +1403,7 @@ def test_plan_runs_audits_on_dev_previews(sushi_context: Context, capsys, caplog
     log = caplog.text
     assert "'not_null' audit error:" in log
     assert "'at_least_one_non_blocking' audit error:" in log
-    assert "Target environment updated successfully" in stdout
+    assert "Environment views updated successfully" in stdout
 
 
 def test_environment_statements(tmp_path: pathlib.Path):

tests/core/test_context.py

@@ -528,6 +528,8 @@ def _evaluate():
             to_datetime("2022-01-03"),
             DeployabilityIndex.all_deployable(),
             0,
+            EnvironmentNamingInfo(),
+            None,
         )
 
     evaluator_audit_mock.return_value = [

tests/core/test_scheduler.py

@@ -291,7 +291,7 @@ def test_plan(
 
     # TODO: Should this be going to stdout? This is printing the status updates for when each batch finishes for
     # the models and how long it took
-    assert len(output.stdout.strip().split("\n")) == 24
+    assert len(output.stdout.strip().split("\n")) == 46
     assert not output.stderr
     assert len(output.outputs) == 4
     text_output = convert_all_html_output_to_text(output)
@@ -326,7 +326,7 @@ def test_run_dag(
     assert not output.stderr
     assert len(output.outputs) == 2
     assert convert_all_html_output_to_text(output) == [
-        "Model batches executed successfully",
+        "Model batches evaluated successfully",
         "Run finished for environment 'prod'",
     ]
     assert get_all_html_output(output) == [
@@ -337,7 +337,7 @@ def test_run_dag(
                 h(
                     "span",
                     {"style": SUCCESS_STYLE},
-                    "Model batches executed successfully",
+                    "Model batches evaluated successfully",
                     autoescape=False,
                 ),
                 autoescape=False,

tests/integrations/jupyter/test_magics.py

@@ -12,6 +12,7 @@
 from sqlmesh.core.environment import EnvironmentNamingInfo
 from sqlmesh.core.plan.definition import EvaluatablePlan
 from sqlmesh.core.snapshot import Snapshot, SnapshotInfoLike
+from sqlmesh.core.snapshot.definition import Intervals
 from sqlmesh.core.test import ModelTest
 from sqlmesh.utils.date import now_timestamp
 from web.server import models
@@ -91,7 +92,7 @@ def stop_restate_progress(self, success: bool) -> None:
 
     def start_evaluation_progress(
         self,
-        batches: t.Dict[Snapshot, int],
+        batched_intervals: t.Dict[Snapshot, Intervals],
         environment_naming_info: EnvironmentNamingInfo,
         default_catalog: t.Optional[str],
     ) -> None:
@@ -104,7 +105,7 @@ def start_evaluation_progress(
                     name=snapshot.name,
                     view_name=snapshot.display_name(environment_naming_info, default_catalog),
                 )
-                for snapshot, total_tasks in batches.items()
+                for snapshot, total_tasks in batched_intervals.items()
             }
             self.plan_apply_stage_tracker.add_stage(
                 models.PlanStage.backfill,