fix: deferral, refactor, microbatch progress, scenario, diff and clone fixes

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)
         ));
     }

crates/sqlbuild-core/src/build_progress/models.rs

@@ -184,6 +184,17 @@ pub struct StatementText {
     pub phase: String,
 }
 
+/// One completed microbatch of a model.
+#[derive(Clone, Debug, Default, PartialEq)]
+pub struct BatchProgressText {
+    pub model_name: String,
+    pub batch_number: usize,
+    pub batch_count: usize,
+    pub window: String,
+    pub rows_affected: Option<i64>,
+    pub elapsed_seconds: f64,
+}
+
 /// The node shown on the live spinner row.
 #[derive(Clone, Debug, Default, PartialEq, Eq)]
 pub struct SpinnerRow {