fix: deferral, refactor, microbatch progress, scenario, diff and clone fixes (#1212)

* fix(build): defer renamed models to their pre-migration relation

When --defer-to resolves an unselected model that the deferred target has
not built under its current name, and the model declares migrate_from, read
the migrate_from relation there instead and say so in the plan.

* fix(refactor): let sqb mv keep a pending migrate_from on a same-name move

A move that keeps the model's name and relation carries its undeployed
migrate_from with the file. Renames, and moves that change the relation,
are still refused until the earlier migration is deployed.

* fix(build): keep a line per microbatch when stdout is not a terminal

Captured and debug output now gets one permanent line per completed batch
with its position, cursor window, rows and elapsed time; terminals keep the
live spinner. Verbose query diagnostics are capped per resource instead of
per run, so one model's statements no longer hide another's.

* test: cover a partial mock read by two real models of a test chain

Each missing column is padded once whichever real models read the mock and
however they spell it. The duplicate padding was fixed when required fixture
columns started being keyed by casefolded name; this pins the reported shape.

* fix(scenario): name unmocked model inputs in S504 and list failures at the end

When a scenario mocks some of a model's inputs but leaves others that read
sources real, S504 names that model and its unmocked inputs with the CTEs to
add, and counts the sources instead of listing one line each. sqb scenario
test now ends with the failed scenarios and their first error line.

* fix(diff): show example rows for side-only differences in unkeyed diffs

Unkeyed diffs now sample the complete rows one side holds more often than
the other, bounded by --max-row-only-examples, in the text output and as
examples.from_only_rows and to_only_rows in --json. Key example lists stay
empty, so truncated_by_count reflects the rows actually collected.

* test: point the SQL Server waffle shop helper at the real macro path

The helper rewrote macros/time.py, which the fixture no longer has, so it
failed before reaching the warehouse. Use models/marts/_macros/time.py and
fail loudly when any fixture rewrite no longer matches.

* feat(clone): report per-phase timings in --debug and the JSON document

Clone logs the duration of each target compile, source inspection, planning
and execution under --debug, and the --json-output document reports them in
seconds under phase_timings.

* fix(dbt): stream a result row for error-level dbt model results

An error-level LogModelResult carries a message, so it was buffered as a node
message instead of reporting the node's result. Result events now take
precedence, and the row streams as soon as dbt reports it.

* fix(build): look up deferred pre-migration relations in the model's own schema

A bare migrate_from name now resolves in the model's deferred database and
schema, and the current-name check lists that schema instead of matching a
same-named relation in any deferred schema. Origins are looked up only for
unselected upstreams, after selection, and a schema that cannot be listed
keeps the previous deferral with a plan warning instead of failing planning.

* fix(diff): bound the unkeyed side-only sample query and close its cursor

The sample query now ends in LIMIT (TOP on SQL Server), so client-side
cursors no longer download every differing row, and the cursor is closed
after the fetch.

* fix(dbt): keep the message of warn and error-level dbt results

A warn or error-level dbt result event now carries its own message on the
streamed row, so test, seed and batch rows keep lines such as FAIL 3.

* test: expect the clone logger among native logger names

Author: kvlonge

Makefile

@@ -93,6 +93,7 @@ E2E_DUCKDB_BUILD_CORE_PATHS := \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_python_hooks.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_python_node_selection.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_python_source_deferral.py \
+	tests/e2e/src/sqlbuild/cli/commands/main/build/test_defer_migration_origin.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_sql_hook_reads.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_python_node_identity.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_sql_hook_identity.py \
@@ -123,6 +124,7 @@ E2E_DUCKDB_BUILD_INCREMENTAL_PATHS := \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_microbatch_aware_cursor_date_input.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_microbatch_direct_lifecycle.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_microbatch_failure_windows.py \
+	tests/e2e/src/sqlbuild/cli/commands/main/build/test_microbatch_batch_progress.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_microbatch_replay_backfill_lifecycle.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_mixed_timestamp_grain_replay.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_model_backed_cursor_build.py \
@@ -144,7 +146,8 @@ E2E_DUCKDB_CLI_DATA_PATHS := \
 	tests/e2e/src/sqlbuild/cli/commands/main/scenario/test_interrupts.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/scenario/test_empty_fixture.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/scenario/test_expected_columns.py \
-	tests/e2e/src/sqlbuild/cli/commands/main/scenario/test_helper_ctes.py
+	tests/e2e/src/sqlbuild/cli/commands/main/scenario/test_helper_ctes.py \
+	tests/e2e/src/sqlbuild/cli/commands/main/scenario/test_unmocked_inputs.py
 
 E2E_DUCKDB_CLI_PATHS := \
 	tests/e2e/scripts/cli_preview \

crates/sqlbuild-analysis/src/planner_scenario/_helpers/boundaries.rs

@@ -0,0 +1,86 @@
+//! Real models whose unmocked model inputs pull unmocked sources into a scenario.
+
+use std::collections::{BTreeSet, HashMap};
+
+use crate::planner_scenario::models::ScenarioGraphInput;
+
+/// Whether `name`, run as a real model, reads a source the scenario does not mock, with the memo.
+fn reaches_unmocked_source(
+    name: &str,
+    input: &ScenarioGraphInput,
+    mut memo: HashMap<String, bool>,
+) -> (bool, HashMap<String, bool>) {
+    if let Some(known) = memo.get(name) {
+        return (*known, memo);
+    }
+    memo.insert(name.to_owned(), false);
+    let mut reaches: bool = false;
+    for (kind, dep) in input.models.get(name).into_iter().flatten() {
+        reaches = match kind.as_str() {
+            "source" => !input.sources.contains(dep),
+            "model" if !input.refs.contains(dep) => {
+                let (found, updated) = reaches_unmocked_source(dep, input, memo);
+                memo = updated;
+                found
+            }
+            _ => false,
+        };
+        if reaches {
+            break;
+        }
+    }
+    memo.insert(name.to_owned(), reaches);
+    (reaches, memo)
+}
+
+/// Real models that mock some model inputs but leave others, which read unmocked sources, real.
+pub(crate) fn unmocked_boundaries(
+    input: &ScenarioGraphInput,
+    models: &BTreeSet<String>,
+) -> Vec<String> {
+    let mut memo: HashMap<String, bool> = HashMap::new();
+    let mut messages: Vec<String> = Vec::new();
+    for name in models {
+        let Some(deps) = input.models.get(name) else {
+            continue;
+        };
+        let model_deps = || {
+            deps.iter()
+                .filter(|(kind, _)| matches!(kind.as_str(), "model"))
+        };
+        if !model_deps().any(|(_, dep)| input.refs.contains(dep)) {
+            continue;
+        }
+        let mut unmocked: Vec<&str> = Vec::new();
+        for (_, dep) in model_deps() {
+            if input.refs.contains(dep)
+                || !input.models.contains_key(dep)
+                || unmocked.contains(&dep.as_str())
+            {
+                continue;
+            }
+            let (reaches, updated) = reaches_unmocked_source(dep, input, memo);
+            memo = updated;
+            if reaches {
+                unmocked.push(dep);
+            }
+        }
+        if unmocked.is_empty() {
+            continue;
+        }
+        let quoted: Vec<String> = unmocked.iter().map(|dep| format!("'{dep}'")).collect();
+        let ctes: Vec<String> = unmocked
+            .iter()
+            .map(|dep| format!("__ref__{dep} AS (...)"))
+            .collect();
+        messages.push(format!(
+            "Scenario '{}': model '{name}' reads {}, which {} not mocked, while its other model inputs are. Add {} named {}.",
+            input.name,
+            quoted.join(" and "),
+            if unmocked.len() == 1 { "is" } else { "are" },
+            if unmocked.len() == 1 { "a CTE" } else { "CTEs" },
+            ctes.join(" and ")
+        ));
+    }
+    messages
+}

crates/sqlbuild-analysis/src/planner_scenario/_helpers/mod.rs

@@ -1,3 +1,4 @@
 //! Shared scenario SQL text operations.
 
+pub(super) mod boundaries;
 pub(super) mod markers;

crates/sqlbuild-analysis/src/planner_scenario/constants.rs

@@ -3,3 +3,5 @@
 pub(super) const MARKERS: [&str; 4] = ["__ref(", "__seed(", "__source(", "__dbt_ref("];
 pub(super) const DBT_REF_MARKER: usize = 3;
 pub(super) const HASH_LENGTH: usize = 12;
+/// Missing sources named in full before the rest are counted.
+pub(super) const LISTED_MISSING_SOURCES: usize = 5;

crates/sqlbuild-analysis/src/planner_scenario/main/graph.rs

@@ -1,5 +1,7 @@
 //! Infer scenario closure while preserving dependency and diagnostic order.
 
+use crate::planner_scenario::_helpers::boundaries::unmocked_boundaries;
+use crate::planner_scenario::constants::LISTED_MISSING_SOURCES;
 use crate::planner_scenario::models::{ScenarioGraphInput, ScenarioGraphOutput};
 
 /// Infer required relations, cutting traversal at authored reference fixtures.
@@ -47,13 +49,51 @@ pub fn graph(input: &ScenarioGraphInput) -> ScenarioGraphOutput {
     for target in &input.targets {
         output = walk(target, input, output);
     }
-    for name in &output.sources {
-        if !input.sources.contains(name) {
+    let missing: Vec<&String> = output
+        .sources
+        .iter()
+        .filter(|name| !input.sources.contains(name))
+        .collect();
+    let boundaries: Vec<String> = if missing.is_empty() {
+        Vec::new()
+    } else {
+        unmocked_boundaries(input, &output.models)
+    };
+    if boundaries.is_empty() {
+        for name in &missing {
             output.warnings.push((format!(
                 "Scenario '{}' requires source '{name}', but no fixture was provided. Add a CTE named __source__{} AS (...).",
                 input.name, name.replace('.', "__")
             ), false));
         }
+    } else {
+        output
+            .warnings
+            .extend(boundaries.into_iter().map(|message| (message, false)));
+        let listed: Vec<&str> = missing
+            .iter()
+            .take(LISTED_MISSING_SOURCES)
+            .map(|name| name.as_str())
+            .collect();
+        let more: String = if missing.len() > LISTED_MISSING_SOURCES {
+            format!(" and {} more", missing.len() - LISTED_MISSING_SOURCES)
+        } else {
+            String::new()
+        };
+        output.warnings.push((
+            format!(
+                "Scenario '{}' requires {} {} without fixtures: {}{more}.",
+                input.name,
+                missing.len(),
+                if missing.len() == 1 {
+                    "source"
+                } else {
+                    "sources"
+                },
+                listed.join(", ")
+            ),
+            false,
+        ));
     }
     for name in &output.dbt_refs {
         if !input.dbt_refs.contains(name) {

crates/sqlbuild-analysis/src/planner_scenario/tests/test_graph.rs

@@ -76,6 +76,56 @@ fn given_scenario_dependencies_when_planning_then_preserves_closure_and_warning_
                 ),
             ],
         },
+        GraphTestCase {
+            description: "unmocked inputs beside mocked ones are named before the sources",
+            input: ScenarioGraphInput {
+                name: "orders".into(),
+                targets: vec!["summary".into()],
+                refs: vec!["staging".into()],
+                models: [
+                    (
+                        "summary".into(),
+                        vec![
+                            ("model".into(), "staging".into()),
+                            ("model".into(), "sales_rep".into()),
+                            ("model".into(), "manager".into()),
+                        ],
+                    ),
+                    ("staging".into(), vec![("source".into(), "hidden".into())]),
+                    (
+                        "sales_rep".into(),
+                        vec![
+                            ("source".into(), "raw.reps".into()),
+                            ("source".into(), "raw.a".into()),
+                            ("source".into(), "raw.b".into()),
+                        ],
+                    ),
+                    (
+                        "manager".into(),
+                        vec![
+                            ("source".into(), "raw.c".into()),
+                            ("source".into(), "raw.d".into()),
+                            ("source".into(), "raw.e".into()),
+                        ],
+                    ),
+                ]
+                .into(),
+                ..ScenarioGraphInput::default()
+            },
+            expected_models: vec!["manager", "sales_rep", "summary"],
+            expected_refs: vec!["staging"],
+            expected_functions: vec![],
+            expected_warnings: vec![
+                (
+                    "Scenario 'orders': model 'summary' reads 'sales_rep' and 'manager', which are not mocked, while its other model inputs are. Add CTEs named __ref__sales_rep AS (...) and __ref__manager AS (...).",
+                    false,
+                ),
+                (
+                    "Scenario 'orders' requires 6 sources without fixtures: raw.a, raw.b, raw.c, raw.d, raw.e and 1 more.",
+                    false,
+                ),
+            ],
+        },
     ];
     for test_case in test_cases {
         let result = graph(&test_case.input);

crates/sqlbuild-cli/src/diff_report/_helpers/payload.rs

@@ -100,10 +100,31 @@ fn examples(model: &DiffModelFacts, options: DiffExampleOptions) -> JsonValue {
     let keys = |samples: &[Vec<DiffNamedValue>]| {
         JsonValue::Array(samples.iter().map(|sample| named_values(sample)).collect())
     };
-    object(vec![
+    let no_samples: &[Vec<DiffNamedValue>] = &[];
+    let (side_keys, side_rows) = if model.unkeyed {
+        (
+            (no_samples, no_samples),
+            Some((&model.left_only_keys, &model.right_only_keys)),
+        )
+    } else {
+        (
+            (
+                model.left_only_keys.as_slice(),
+                model.right_only_keys.as_slice(),
+            ),
+            None,
+        )
+    };
+    let mut entries: Vec<(&str, JsonValue)> = vec![
         ("unequal_rows", JsonValue::Array(unequal_rows)),
-        ("from_only_keys", keys(&model.left_only_keys)),
-        ("to_only_keys", keys(&model.right_only_keys)),
+        ("from_only_keys", keys(side_keys.0)),
+        ("to_only_keys", keys(side_keys.1)),
+    ];
+    if let Some((from_rows, to_rows)) = side_rows {
+        entries.push(("from_only_rows", keys(from_rows)));
+        entries.push(("to_only_rows", keys(to_rows)));
+    }
+    entries.extend([
         ("truncated_by_count", JsonValue::Bool(truncated_by_count)),
         ("values_truncated", JsonValue::Bool(values_truncated)),
         (
@@ -116,7 +137,8 @@ fn examples(model: &DiffModelFacts, options: DiffExampleOptions) -> JsonValue {
                 .max_value_length
                 .map_or(JsonValue::Null, |limit| integer(limit as i64)),
         ),
-    ])
+    ]);
+    object(entries)
 }
 
 fn model_payload(model: &DiffModelFacts, options: &DiffDocumentOptions) -> JsonValue {

crates/sqlbuild-core/src/build_progress/main/batch_progress_message.rs

@@ -0,0 +1,20 @@
+//! The permanent line of one completed microbatch, for output without a live spinner.
+
+use crate::build_progress::models::BatchProgressText;
+
+/// The model, batch position, cursor window, rows written and elapsed time of a batch.
+#[must_use]
+pub fn batch_progress_message(batch: &BatchProgressText) -> String {
+    let rows: String = batch
+        .rows_affected
+        .map(|rows| format!("  {rows} rows"))
+        .unwrap_or_default();
+    format!(
+        "Batch  {}  {}/{}  {}{rows}  {:.1}s",
+        batch.model_name,
+        batch.batch_number,
+        batch.batch_count,
+        batch.window,
+        batch.elapsed_seconds
+    )
+}

crates/sqlbuild-core/src/build_progress/main/mod.rs

@@ -1,3 +1,4 @@
+pub mod batch_progress_message;
 pub mod format_build_footer;
 pub mod progress_name_width;
 pub mod query_diagnostic_message;

crates/sqlbuild-core/src/build_progress/main/runtime_diagnostic_summary.rs

@@ -20,7 +20,7 @@ pub fn runtime_diagnostic_summary(
     }
     if let Some((omitted, statement)) = query.filter(|(omitted, _)| *omitted > 0) {
         lines.push(format!(
-            "Query diagnostics  {omitted} statements omitted after {QUERY_DIAGNOSTIC_LIMIT}; final query: {}",
+            "Query diagnostics  {omitted} statements omitted after {QUERY_DIAGNOSTIC_LIMIT} per resource; final query: {}",
             query_diagnostic_message(statement)
         ));
     }