From e8ef20606b42f8b1a0f71e08e03b6966d0b12fce Mon Sep 17 00:00:00 2001 From: Vamsi-klu Date: Sun, 27 Sep 2026 00:40:15 +0000 Subject: [PATCH] feat(console): show running model batches Signed-off-by: Vamsi-klu --- sqlmesh/core/console.py | 131 ++++++++++++------ sqlmesh/core/scheduler.py | 48 +++++-- tests/cli/test_cli.py | 4 + tests/core/test_console.py | 248 ++++++++++++++++++++++++++++++++++- tests/core/test_scheduler.py | 214 ++++++++++++++++++++++++++++++ 5 files changed, 593 insertions(+), 52 deletions(-) diff --git a/sqlmesh/core/console.py b/sqlmesh/core/console.py index ca11b6e607..0abd06f2dd 100644 --- a/sqlmesh/core/console.py +++ b/sqlmesh/core/console.py @@ -431,6 +431,15 @@ def start_snapshot_evaluation_progress( ) -> None: """Starts the snapshot evaluation progress.""" + def start_snapshot_evaluation_batch( + self, snapshot: Snapshot, interval: Interval, batch_idx: int + ) -> None: + """Starts progress for a snapshot evaluation batch. + + Consoles without per-batch output can inherit this default no-op implementation. + """ + pass + @abc.abstractmethod def update_snapshot_evaluation_progress( self, @@ -1083,6 +1092,63 @@ def start_evaluation_progress( self.environment_naming_info = environment_naming_info self.default_catalog = default_catalog + def _print_evaluation_row( + self, + snapshot: Snapshot, + interval: Interval, + batch_idx: int, + status: str, + audit_only: bool = False, + num_audits_passed: int = 0, + num_audits_failed: int = 0, + execution_stats: t.Optional[QueryExecutionStats] = None, + ) -> None: + """Print one evaluation row (Running or completion) to the live console. + + Both start and completion rows share this formatter so their columns align. + """ + if not ( + self.evaluation_total_progress + and self.evaluation_model_progress + and self.evaluation_progress_live + ): + return + + total_batches = self.evaluation_model_batch_sizes[snapshot] + batch_num = str(batch_idx + 1).rjust(len(str(total_batches))) + batch = f"[{batch_num}/{total_batches}]".ljust(self.evaluation_column_widths["batch"]) + + display_name = snapshot.display_name( + self.environment_naming_info, + self.default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None, + dialect=self.dialect, + ).ljust(self.evaluation_column_widths["name"]) + + annotation = _create_evaluation_model_annotation( + snapshot, + _format_evaluation_model_interval(snapshot, interval), + execution_stats, + ) + audits_str = "" + if num_audits_passed: + audits_str += f" {self.AUDIT_PASS_MARK}{num_audits_passed}" + if num_audits_failed: + audits_str += f" {self.AUDIT_FAIL_MARK}{num_audits_failed}" + audits_str = f", audits{audits_str}" if audits_str else "" + annotation_len = self.evaluation_column_widths["annotation"] + # don't adjust the annotation_len if we're using AUDIT_PADDING + annotation = f"\\[{annotation + audits_str}]".ljust( + annotation_len - 1 if num_audits_failed and self.AUDIT_PADDING == 0 else annotation_len + ) + + status_col = status.ljust(self.evaluation_column_widths["duration"]) + + msg = ( + f"{f'{batch} ' if not audit_only else ''}{display_name} {annotation} {status_col}" + ).replace(self.AUDIT_PASS_MARK, self.GREEN_AUDIT_PASS_MARK) + + self.evaluation_progress_live.console.print(msg, soft_wrap=True) + def start_snapshot_evaluation_progress( self, snapshot: Snapshot, audit_only: bool = False ) -> None: @@ -1098,6 +1164,12 @@ def start_snapshot_evaluation_progress( total=self.evaluation_model_batch_sizes[snapshot], ) + def start_snapshot_evaluation_batch( + self, snapshot: Snapshot, interval: Interval, batch_idx: int + ) -> None: + """Print a durable running row so the active batch appears in captured output.""" + self._print_evaluation_row(snapshot, interval, batch_idx, "Running") + def update_snapshot_evaluation_progress( self, snapshot: Snapshot, @@ -1117,42 +1189,21 @@ def update_snapshot_evaluation_progress( and self.evaluation_progress_live ): total_batches = self.evaluation_model_batch_sizes[snapshot] - batch_num = str(batch_idx + 1).rjust(len(str(total_batches))) - batch = f"[{batch_num}/{total_batches}]".ljust(self.evaluation_column_widths["batch"]) - if duration_ms: - display_name = snapshot.display_name( - self.environment_naming_info, - self.default_catalog if self.verbosity < Verbosity.VERY_VERBOSE else None, - dialect=self.dialect, - ).ljust(self.evaluation_column_widths["name"]) - - annotation = _create_evaluation_model_annotation( - snapshot, _format_evaluation_model_interval(snapshot, interval), execution_stats - ) - audits_str = "" - if num_audits_passed: - audits_str += f" {self.AUDIT_PASS_MARK}{num_audits_passed}" - if num_audits_failed: - audits_str += f" {self.AUDIT_FAIL_MARK}{num_audits_failed}" - audits_str = f", audits{audits_str}" if audits_str else "" - annotation_len = self.evaluation_column_widths["annotation"] - # don't adjust the annotation_len if we're using AUDIT_PADDING - annotation = f"\\[{annotation + audits_str}]".ljust( - annotation_len - 1 - if num_audits_failed and self.AUDIT_PADDING == 0 - else annotation_len - ) - - duration = f"{(duration_ms / 1000.0):.2f}s".ljust( - self.evaluation_column_widths["duration"] - ) - - msg = f"{f'{batch} ' if not audit_only else ''}{display_name} {annotation} {duration}".replace( - self.AUDIT_PASS_MARK, self.GREEN_AUDIT_PASS_MARK + if duration_ms is not None: + duration = f"{(duration_ms / 1000.0):.2f}s" + self._print_evaluation_row( + snapshot, + interval, + batch_idx, + duration, + audit_only=audit_only, + num_audits_passed=num_audits_passed, + num_audits_failed=num_audits_failed, + execution_stats=execution_stats, ) - - self.evaluation_progress_live.console.print(msg) + elif not audit_only: + self._print_evaluation_row(snapshot, interval, batch_idx, "Failed") self.evaluation_total_progress.update( self.evaluation_total_task or TaskID(0), refresh=True, advance=1 @@ -1160,10 +1211,7 @@ def update_snapshot_evaluation_progress( model_task_id = self.evaluation_model_tasks[snapshot.name] self.evaluation_model_progress.update(model_task_id, refresh=True, advance=1) - if ( - self.evaluation_model_progress._tasks[model_task_id].completed >= total_batches - or audit_only - ): + if self.evaluation_model_progress._tasks[model_task_id].completed >= total_batches: self.evaluation_model_progress.remove_task(model_task_id) def stop_evaluation_progress(self, success: bool = True) -> None: @@ -3925,6 +3973,13 @@ def start_snapshot_evaluation_progress( ) -> None: self._write(f"{'Evaluating' if not audit_only else 'Auditing'} {snapshot.name}") + def start_snapshot_evaluation_batch( + self, snapshot: Snapshot, interval: Interval, batch_idx: int + ) -> None: + # The debugger console already logs the model start above and intentionally has no + # TerminalConsole progress state because it doesn't call TerminalConsole.__init__. + pass + def update_snapshot_evaluation_progress( self, snapshot: Snapshot, diff --git a/sqlmesh/core/scheduler.py b/sqlmesh/core/scheduler.py index 5eb0ff40ff..7c6ad0d1ba 100644 --- a/sqlmesh/core/scheduler.py +++ b/sqlmesh/core/scheduler.py @@ -526,7 +526,14 @@ def run_node(node: SchedulingUnit) -> None: snapshot = self.snapshots_by_name[node.snapshot_name] if isinstance(node, EvaluateNode): - self.console.start_snapshot_evaluation_progress(snapshot) + if audit_only: + self.console.start_snapshot_evaluation_progress(snapshot, audit_only=True) + else: + self.console.start_snapshot_evaluation_progress(snapshot) + if not audit_only: + self.console.start_snapshot_evaluation_batch( + snapshot, node.interval, node.batch_index + ) execution_start_ts = now_timestamp() evaluation_duration_ms: t.Optional[int] = None start, end = node.interval @@ -574,18 +581,33 @@ def run_node(node: SchedulingUnit) -> None: SnapshotIdBatch(snapshot_id=snapshot.snapshot_id, batch_id=node.batch_index) ) - self.console.update_snapshot_evaluation_progress( - snapshot, - batched_intervals[snapshot][node.batch_index], - node.batch_index, - evaluation_duration_ms, - num_audits - num_audits_failed, - num_audits_failed, - execution_stats=execution_stats, - auto_restatement_triggers=auto_restatement_triggers.get( - snapshot.snapshot_id - ), - ) + if audit_only: + self.console.update_snapshot_evaluation_progress( + snapshot, + batched_intervals[snapshot][node.batch_index], + node.batch_index, + evaluation_duration_ms, + num_audits - num_audits_failed, + num_audits_failed, + audit_only=True, + execution_stats=execution_stats, + auto_restatement_triggers=auto_restatement_triggers.get( + snapshot.snapshot_id + ), + ) + else: + self.console.update_snapshot_evaluation_progress( + snapshot, + batched_intervals[snapshot][node.batch_index], + node.batch_index, + evaluation_duration_ms, + num_audits - num_audits_failed, + num_audits_failed, + execution_stats=execution_stats, + auto_restatement_triggers=auto_restatement_triggers.get( + snapshot.snapshot_id + ), + ) elif isinstance(node, CreateNode): self.snapshot_evaluator.create_snapshot( snapshot=snapshot, diff --git a/tests/cli/test_cli.py b/tests/cli/test_cli.py index f1540727b1..078214022b 100644 --- a/tests/cli/test_cli.py +++ b/tests/cli/test_cli.py @@ -942,6 +942,10 @@ def test_run_dev(runner, tmp_path, flag): result = runner.invoke(cli, ["--log-file-dir", tmp_path, "--paths", tmp_path, "run", "dev"]) assert result.exit_code == 0 assert_model_batches_executed(result) + running_lines = [ + line for line in result.output.splitlines() if line.rstrip().endswith("Running") + ] + assert any("[1/1]" in line and "[insert " in line for line in running_lines) @time_machine.travel(FREEZE_TIME) diff --git a/tests/core/test_console.py b/tests/core/test_console.py index f899713235..131018b714 100644 --- a/tests/core/test_console.py +++ b/tests/core/test_console.py @@ -1,4 +1,17 @@ -from sqlmesh.core.console import MarkdownConsole +import io +from concurrent.futures import ThreadPoolExecutor + +import pytest +from rich.console import Console as RichConsole +from sqlglot import parse_one + +from sqlmesh.core.console import DebuggerTerminalConsole, MarkdownConsole, TerminalConsole +from sqlmesh.core.environment import EnvironmentNamingInfo +from sqlmesh.core.model import SqlModel +from sqlmesh.core.model.kind import IncrementalByTimeRangeKind, TimeColumn +from sqlmesh.core.snapshot import Snapshot, SnapshotChangeCategory +from sqlmesh.utils.date import to_timestamp +from sqlmesh.utils.rich import strip_ansi_codes def test_markdown_console_warning_block(): @@ -129,3 +142,236 @@ def test_markdown_console_error_block(): ) assert console.consume_captured_errors() == "" + + +def _make_evaluation_console(width: int = 200, force_terminal: bool = True) -> tuple: + """Return (buf, TerminalConsole) backed by a no-color in-memory console.""" + buf = io.StringIO() + rich = RichConsole(file=buf, force_terminal=force_terminal, no_color=True, width=width) + return buf, TerminalConsole(console=rich) + + +def _readable_lines(buf: io.StringIO) -> list: + """Strip ANSI codes and return non-blank lines from buf.""" + return [line for line in strip_ansi_codes(buf.getvalue()).splitlines() if line.strip()] + + +@pytest.mark.parametrize("force_terminal", [True, False]) +def test_snapshot_evaluation_running_row(make_snapshot, force_terminal: bool): + """Running row appears before completion; both rows share the same column alignment.""" + model = SqlModel( + name="silver.model4", + kind=IncrementalByTimeRangeKind(time_column=TimeColumn(column="ts")), + cron="@hourly", + query=parse_one("SELECT ts FROM tbl"), + ) + snapshot = make_snapshot(model) + snapshot.categorize_as(SnapshotChangeCategory.BREAKING) + + # Hourly interval: exclusive end 15:00:00 => inclusive end 14:59:59 + interval = (to_timestamp("2026-07-17 13:45:00"), to_timestamp("2026-07-17 15:00:00")) + env_info = EnvironmentNamingInfo() + batched_intervals = {snapshot: [interval]} + + buf, tc = _make_evaluation_console(force_terminal=force_terminal) + tc.start_evaluation_progress(batched_intervals, env_info, default_catalog=None) + + tc.start_snapshot_evaluation_progress(snapshot) + tc.start_snapshot_evaluation_batch(snapshot, interval, 0) + + # Completion row with fixed 1234 ms => "1.23s" + tc.update_snapshot_evaluation_progress(snapshot, interval, 0, 1234, 0, 0) + tc.stop_evaluation_progress(success=False) + + lines = _readable_lines(buf) + + running_lines = [ln for ln in lines if ln.rstrip().endswith("Running")] + assert len(running_lines) == 1, f"Expected exactly one Running line, got: {running_lines!r}" + + duration_lines = [ln for ln in lines if "1.23s" in ln] + assert len(duration_lines) == 1, f"Expected exactly one duration line, got: {duration_lines!r}" + + # Running must precede completion + running_pos = next(i for i, ln in enumerate(lines) if ln.rstrip().endswith("Running")) + duration_pos = next(i for i, ln in enumerate(lines) if "1.23s" in ln) + assert running_pos < duration_pos, "Running row must appear before the completion row" + + # Both rows carry batch indicator, model name, and inclusive interval end + for row in (running_lines[0], duration_lines[0]): + assert "[1/1]" in row, f"Batch indicator missing: {row!r}" + assert "silver.model4" in row, f"Model name missing: {row!r}" + assert "13:45:00-14:59:59" in row, f"Inclusive end missing: {row!r}" + + # Status/duration starts at the same column in both rows + running_stripped = running_lines[0].rstrip() + duration_stripped = duration_lines[0].rstrip() + running_prefix_len = len(running_stripped) - len("Running") + duration_prefix_len = len(duration_stripped) - len("1.23s") + assert running_prefix_len == duration_prefix_len, ( + f"Status not aligned: Running col={running_prefix_len}, duration col={duration_prefix_len}\n" + f" Running : {running_stripped!r}\n" + f" Duration: {duration_stripped!r}" + ) + + +def test_snapshot_evaluation_zero_duration(make_snapshot): + """A zero-millisecond batch must print '0.00s', not be silently skipped.""" + model = SqlModel( + name="gold.zeromodel", + kind=IncrementalByTimeRangeKind(time_column=TimeColumn(column="ts")), + cron="@hourly", + query=parse_one("SELECT ts FROM tbl"), + ) + snapshot = make_snapshot(model) + snapshot.categorize_as(SnapshotChangeCategory.BREAKING) + + interval = (to_timestamp("2026-07-17 00:00:00"), to_timestamp("2026-07-17 01:00:00")) + env_info = EnvironmentNamingInfo() + batched_intervals = {snapshot: [interval]} + + buf, tc = _make_evaluation_console() + tc.start_evaluation_progress(batched_intervals, env_info, default_catalog=None) + tc.start_snapshot_evaluation_progress(snapshot) + tc.start_snapshot_evaluation_batch(snapshot, interval, 0) + tc.update_snapshot_evaluation_progress(snapshot, interval, 0, 0, 0, 0) + tc.stop_evaluation_progress(success=False) + + lines = _readable_lines(buf) + zero_dur_lines = [ln for ln in lines if "0.00s" in ln] + assert len(zero_dur_lines) == 1, f"Expected one '0.00s' line, got: {zero_dur_lines!r}" + + +def test_snapshot_evaluation_failed_row(make_snapshot): + """A failed batch replaces the durable running state with a terminal status.""" + model = SqlModel( + name="silver.failed_model", + kind=IncrementalByTimeRangeKind(time_column=TimeColumn(column="ts")), + cron="@hourly", + query=parse_one("SELECT ts FROM tbl"), + ) + snapshot = make_snapshot(model) + snapshot.categorize_as(SnapshotChangeCategory.BREAKING) + + interval = (to_timestamp("2026-07-17 13:00:00"), to_timestamp("2026-07-17 14:00:00")) + buf, tc = _make_evaluation_console(width=80, force_terminal=False) + tc.start_evaluation_progress( + {snapshot: [interval]}, EnvironmentNamingInfo(), default_catalog=None + ) + tc.start_snapshot_evaluation_progress(snapshot) + tc.start_snapshot_evaluation_batch(snapshot, interval, 0) + + running_lines = [line for line in _readable_lines(buf) if line.rstrip().endswith("Running")] + assert len(running_lines) == 1 + assert "[1/1]" in running_lines[0] + assert "silver.failed_model" in running_lines[0] + assert "13:00:00-13:59:59" in running_lines[0] + + # A missing duration indicates that evaluation raised before it completed. + tc.update_snapshot_evaluation_progress(snapshot, interval, 0, None, 0, 0) + tc.stop_evaluation_progress(success=False) + + failed_lines = [line for line in _readable_lines(buf) if line.rstrip().endswith("Failed")] + assert len(failed_lines) == 1 + assert "[1/1]" in failed_lines[0] + assert "silver.failed_model" in failed_lines[0] + assert "13:00:00-13:59:59" in failed_lines[0] + + +def test_snapshot_evaluation_audit_only_multiple_batches(make_snapshot): + """Audit-only progress retains its model task until every interval completes.""" + model = SqlModel( + name="silver.audited_model", + kind=IncrementalByTimeRangeKind(time_column=TimeColumn(column="ts")), + cron="@hourly", + query=parse_one("SELECT ts FROM tbl"), + ) + snapshot = make_snapshot(model) + snapshot.categorize_as(SnapshotChangeCategory.BREAKING) + intervals = [ + (to_timestamp("2026-07-17 13:00:00"), to_timestamp("2026-07-17 14:00:00")), + (to_timestamp("2026-07-17 14:00:00"), to_timestamp("2026-07-17 15:00:00")), + ] + _, console = _make_evaluation_console(force_terminal=False) + console.start_evaluation_progress( + {snapshot: intervals}, + EnvironmentNamingInfo(), + default_catalog=None, + audit_only=True, + ) + + for batch_idx, interval in enumerate(intervals): + console.start_snapshot_evaluation_progress(snapshot, audit_only=True) + console.update_snapshot_evaluation_progress( + snapshot, + interval, + batch_idx, + duration_ms=100, + num_audits_passed=1, + num_audits_failed=0, + audit_only=True, + ) + + assert console.evaluation_model_progress is not None + model_task_id = console.evaluation_model_tasks[snapshot.name] + assert model_task_id not in console.evaluation_model_progress._tasks + console.stop_evaluation_progress(success=False) + + +def test_snapshot_evaluation_concurrent_running_rows(make_snapshot): + """Concurrent models each emit one complete running row.""" + snapshots = [] + for model_name in ("silver.concurrent_a", "silver.concurrent_b"): + model = SqlModel( + name=model_name, + kind=IncrementalByTimeRangeKind(time_column=TimeColumn(column="ts")), + cron="@hourly", + query=parse_one("SELECT ts FROM tbl"), + ) + snapshot = make_snapshot(model) + snapshot.categorize_as(SnapshotChangeCategory.BREAKING) + snapshots.append(snapshot) + + interval = (to_timestamp("2026-07-17 13:00:00"), to_timestamp("2026-07-17 14:00:00")) + buf, tc = _make_evaluation_console(force_terminal=False) + tc.start_evaluation_progress( + {snapshot: [interval] for snapshot in snapshots}, + EnvironmentNamingInfo(), + default_catalog=None, + ) + + def start_batch(snapshot: Snapshot) -> None: + tc.start_snapshot_evaluation_progress(snapshot) + tc.start_snapshot_evaluation_batch(snapshot, interval, 0) + + with ThreadPoolExecutor(max_workers=2) as executor: + list(executor.map(start_batch, snapshots)) + + tc.stop_evaluation_progress(success=False) + running_lines = [line for line in _readable_lines(buf) if line.rstrip().endswith("Running")] + + assert len(running_lines) == 2 + for model_name in ("silver.concurrent_a", "silver.concurrent_b"): + assert sum(model_name in line for line in running_lines) == 1 + + +def test_debugger_console_evaluation_batch_start_is_safe(make_snapshot): + """The debugger console does not depend on TerminalConsole progress-bar state.""" + model = SqlModel( + name="silver.debugger_model", + kind=IncrementalByTimeRangeKind(time_column=TimeColumn(column="ts")), + cron="@hourly", + query=parse_one("SELECT ts FROM tbl"), + ) + snapshot = make_snapshot(model) + snapshot.categorize_as(SnapshotChangeCategory.BREAKING) + interval = (to_timestamp("2026-07-17 13:00:00"), to_timestamp("2026-07-17 14:00:00")) + buf = io.StringIO() + console = DebuggerTerminalConsole( + RichConsole(file=buf, force_terminal=False, no_color=True, width=200) + ) + + console.start_snapshot_evaluation_progress(snapshot) + console.start_snapshot_evaluation_batch(snapshot, interval, 0) + + assert "Evaluating" in buf.getvalue() + assert "debugger_model" in buf.getvalue() diff --git a/tests/core/test_scheduler.py b/tests/core/test_scheduler.py index cd32d2451d..5b996fd5e8 100644 --- a/tests/core/test_scheduler.py +++ b/tests/core/test_scheduler.py @@ -6,6 +6,7 @@ from sqlglot.helper import first from sqlmesh.core.context import Context, ExecutionContext +from sqlmesh.core.console import NoopConsole from sqlmesh.core.environment import EnvironmentNamingInfo from sqlmesh.core.macros import RuntimeStage from sqlmesh.core.model import load_sql_based_model @@ -1213,3 +1214,216 @@ def test_dag_upstream_dependency_caching_with_complex_diamond(mocker: MockerFixt expected_g_node: {expected_a_node}, expected_h_node: {expected_a_node}, } + + +def test_snapshot_evaluation_progress_start_hooks(mocker: MockerFixture, make_snapshot) -> None: + """The stable model hook and the batch hook run before evaluate().""" + model = SqlModel( + name="test.hourly_model", + kind=IncrementalByTimeRangeKind(time_column=TimeColumn(column="ds")), + cron="@hourly", + query=parse_one("SELECT ds FROM tbl"), + ) + snapshot = make_snapshot(model) + snapshot.categorize_as(SnapshotChangeCategory.BREAKING) + + mock_console = mocker.MagicMock() + mock_evaluator = mocker.MagicMock() + mock_evaluator.get_snapshots_to_create.return_value = [] + mock_evaluator.adapter = mocker.MagicMock() + + call_order: t.List[str] = [] + + def record_start(snapshot_arg: t.Any, **kwargs: t.Any) -> None: + call_order.append("start") + + mock_console.start_snapshot_evaluation_progress.side_effect = record_start + + def record_batch_start(snapshot_arg: t.Any, interval_arg: t.Any, batch_idx_arg: int) -> None: + call_order.append("batch_start") + + mock_console.start_snapshot_evaluation_batch.side_effect = record_batch_start + + interval = (to_timestamp("2026-07-17 13:00:00"), to_timestamp("2026-07-17 14:00:00")) + + scheduler = Scheduler( + snapshots=[snapshot], + snapshot_evaluator=mock_evaluator, + state_sync=mocker.MagicMock(), + default_catalog=None, + console=mock_console, + ) + + def mock_evaluate(snapshot: t.Any, **kwargs: t.Any) -> t.List: + call_order.append("evaluate") + return [] + + mocker.patch.object(scheduler, "evaluate", side_effect=mock_evaluate) + + merged_intervals = {snapshot: [interval]} + deployability_index = DeployabilityIndex.create([snapshot]) + + errors, skipped_intervals = scheduler.run_merged_intervals( + merged_intervals=merged_intervals, + deployability_index=deployability_index, + environment_naming_info=EnvironmentNamingInfo(), + ) + + assert not errors + assert not skipped_intervals + + # Keep the existing start hook contract stable for custom Console implementations. + mock_console.start_snapshot_evaluation_progress.assert_called_once_with(snapshot) + + # The new optional hook receives the exact interval and zero-based batch index. + mock_console.start_snapshot_evaluation_batch.assert_called_once_with( + snapshot, + interval, + 0, + ) + + # Both start hooks must run before evaluate. + assert call_order == ["start", "batch_start", "evaluate"], ( + f"Unexpected call order: {call_order}" + ) + + # completion update runs after evaluation + mock_console.update_snapshot_evaluation_progress.assert_called_once() + + +def test_snapshot_evaluation_progress_supports_legacy_console( + mocker: MockerFixture, make_snapshot +) -> None: + """A custom Console with the historical start-hook signature remains usable.""" + + class LegacyConsole(NoopConsole): + def __init__(self) -> None: + self.started_snapshots: t.List[Snapshot] = [] + self.updated_snapshots: t.List[Snapshot] = [] + + def start_snapshot_evaluation_progress( # type: ignore[override] + self, snapshot: Snapshot + ) -> None: + self.started_snapshots.append(snapshot) + + def update_snapshot_evaluation_progress( # type: ignore[override] + self, + snapshot: Snapshot, + interval: t.Tuple[int, int], + batch_idx: int, + duration_ms: t.Optional[int], + num_audits_passed: int, + num_audits_failed: int, + execution_stats: t.Any = None, + auto_restatement_triggers: t.Any = None, + ) -> None: + self.updated_snapshots.append(snapshot) + + model = SqlModel( + name="test.legacy_console_model", + kind=IncrementalByTimeRangeKind(time_column=TimeColumn(column="ds")), + cron="@hourly", + query=parse_one("SELECT ds FROM tbl"), + ) + snapshot = make_snapshot(model) + snapshot.categorize_as(SnapshotChangeCategory.BREAKING) + interval = (to_timestamp("2026-07-17 13:00:00"), to_timestamp("2026-07-17 14:00:00")) + console = LegacyConsole() + mock_evaluator = mocker.MagicMock() + mock_evaluator.get_snapshots_to_create.return_value = [] + + scheduler = Scheduler( + snapshots=[snapshot], + snapshot_evaluator=mock_evaluator, + state_sync=mocker.MagicMock(), + default_catalog=None, + console=console, + ) + mocker.patch.object(scheduler, "evaluate", return_value=[]) + + errors, skipped_intervals = scheduler.run_merged_intervals( + merged_intervals={snapshot: [interval]}, + deployability_index=DeployabilityIndex.create([snapshot]), + environment_naming_info=EnvironmentNamingInfo(), + ) + + assert not errors + assert not skipped_intervals + assert console.started_snapshots == [snapshot] + assert console.updated_snapshots == [snapshot] + + +def test_snapshot_evaluation_progress_audit_only(mocker: MockerFixture, make_snapshot) -> None: + """Audit-only runs preserve audit semantics and never emit a running batch row.""" + model = SqlModel( + name="test.audit_model", + kind=IncrementalByTimeRangeKind(time_column=TimeColumn(column="ds")), + cron="@hourly", + query=parse_one("SELECT ds FROM tbl"), + ) + snapshot = make_snapshot(model) + snapshot.categorize_as(SnapshotChangeCategory.BREAKING) + interval = (to_timestamp("2026-07-17 13:00:00"), to_timestamp("2026-07-17 14:00:00")) + + console = mocker.MagicMock() + evaluator = mocker.MagicMock() + evaluator.get_snapshots_to_create.return_value = [] + scheduler = Scheduler( + snapshots=[snapshot], + snapshot_evaluator=evaluator, + state_sync=mocker.MagicMock(), + default_catalog=None, + console=console, + ) + mocker.patch.object(scheduler, "_audit_snapshot", return_value=[]) + + errors, skipped_intervals = scheduler.run_merged_intervals( + merged_intervals={snapshot: [interval]}, + deployability_index=DeployabilityIndex.create([snapshot]), + environment_naming_info=EnvironmentNamingInfo(), + audit_only=True, + ) + + assert not errors + assert not skipped_intervals + console.start_snapshot_evaluation_progress.assert_called_once_with(snapshot, audit_only=True) + console.start_snapshot_evaluation_batch.assert_not_called() + assert console.update_snapshot_evaluation_progress.call_args.kwargs["audit_only"] is True + + +def test_snapshot_evaluation_failure_reports_no_duration( + mocker: MockerFixture, make_snapshot +) -> None: + """Evaluation failures reach the console with no duration so it can print Failed.""" + model = SqlModel( + name="test.failed_model", + kind=IncrementalByTimeRangeKind(time_column=TimeColumn(column="ds")), + cron="@hourly", + query=parse_one("SELECT ds FROM tbl"), + ) + snapshot = make_snapshot(model) + snapshot.categorize_as(SnapshotChangeCategory.BREAKING) + interval = (to_timestamp("2026-07-17 13:00:00"), to_timestamp("2026-07-17 14:00:00")) + + console = mocker.MagicMock() + evaluator = mocker.MagicMock() + evaluator.get_snapshots_to_create.return_value = [] + scheduler = Scheduler( + snapshots=[snapshot], + snapshot_evaluator=evaluator, + state_sync=mocker.MagicMock(), + default_catalog=None, + console=console, + ) + mocker.patch.object(scheduler, "evaluate", side_effect=RuntimeError("evaluation failed")) + + errors, _ = scheduler.run_merged_intervals( + merged_intervals={snapshot: [interval]}, + deployability_index=DeployabilityIndex.create([snapshot]), + environment_naming_info=EnvironmentNamingInfo(), + ) + + assert len(errors) == 1 + update_args = console.update_snapshot_evaluation_progress.call_args.args + assert update_args[:3] == (snapshot, interval, 0) + assert update_args[3] is None