diff --git a/Makefile b/Makefile index 1845611f69..496dcf84e4 100644 --- a/Makefile +++ b/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 \ diff --git a/crates/sqlbuild-analysis/src/planner_scenario/_helpers/boundaries.rs b/crates/sqlbuild-analysis/src/planner_scenario/_helpers/boundaries.rs new file mode 100644 index 0000000000..cfe3befc9b --- /dev/null +++ b/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, +) -> (bool, HashMap) { + 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, +) -> Vec { + let mut memo: HashMap = HashMap::new(); + let mut messages: Vec = 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 = unmocked.iter().map(|dep| format!("'{dep}'")).collect(); + let ctes: Vec = 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 +} diff --git a/crates/sqlbuild-analysis/src/planner_scenario/_helpers/mod.rs b/crates/sqlbuild-analysis/src/planner_scenario/_helpers/mod.rs index dd0049eb8b..3c921d56b8 100644 --- a/crates/sqlbuild-analysis/src/planner_scenario/_helpers/mod.rs +++ b/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; diff --git a/crates/sqlbuild-analysis/src/planner_scenario/constants.rs b/crates/sqlbuild-analysis/src/planner_scenario/constants.rs index 6809122df1..8d058150a1 100644 --- a/crates/sqlbuild-analysis/src/planner_scenario/constants.rs +++ b/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; diff --git a/crates/sqlbuild-analysis/src/planner_scenario/main/graph.rs b/crates/sqlbuild-analysis/src/planner_scenario/main/graph.rs index 6413b4ddeb..1db0a77f76 100644 --- a/crates/sqlbuild-analysis/src/planner_scenario/main/graph.rs +++ b/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 = 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) { diff --git a/crates/sqlbuild-analysis/src/planner_scenario/tests/test_graph.rs b/crates/sqlbuild-analysis/src/planner_scenario/tests/test_graph.rs index 5d51593ed2..2888104c4c 100644 --- a/crates/sqlbuild-analysis/src/planner_scenario/tests/test_graph.rs +++ b/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); diff --git a/crates/sqlbuild-cli/src/diff_report/_helpers/payload.rs b/crates/sqlbuild-cli/src/diff_report/_helpers/payload.rs index c46ae03d65..ccf51df42a 100644 --- a/crates/sqlbuild-cli/src/diff_report/_helpers/payload.rs +++ b/crates/sqlbuild-cli/src/diff_report/_helpers/payload.rs @@ -100,10 +100,31 @@ fn examples(model: &DiffModelFacts, options: DiffExampleOptions) -> JsonValue { let keys = |samples: &[Vec]| { JsonValue::Array(samples.iter().map(|sample| named_values(sample)).collect()) }; - object(vec![ + let no_samples: &[Vec] = &[]; + 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 { diff --git a/crates/sqlbuild-core/src/build_progress/main/batch_progress_message.rs b/crates/sqlbuild-core/src/build_progress/main/batch_progress_message.rs new file mode 100644 index 0000000000..be2d87fb29 --- /dev/null +++ b/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 + ) +} diff --git a/crates/sqlbuild-core/src/build_progress/main/mod.rs b/crates/sqlbuild-core/src/build_progress/main/mod.rs index 09e6110e07..c104e9b92c 100644 --- a/crates/sqlbuild-core/src/build_progress/main/mod.rs +++ b/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; diff --git a/crates/sqlbuild-core/src/build_progress/main/runtime_diagnostic_summary.rs b/crates/sqlbuild-core/src/build_progress/main/runtime_diagnostic_summary.rs index bf42ad0e75..bfc8c88d1e 100644 --- a/crates/sqlbuild-core/src/build_progress/main/runtime_diagnostic_summary.rs +++ b/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) )); } diff --git a/crates/sqlbuild-core/src/build_progress/models.rs b/crates/sqlbuild-core/src/build_progress/models.rs index 7ba6a5c391..80a8ba1452 100644 --- a/crates/sqlbuild-core/src/build_progress/models.rs +++ b/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, + pub elapsed_seconds: f64, +} + /// The node shown on the live spinner row. #[derive(Clone, Debug, Default, PartialEq, Eq)] pub struct SpinnerRow { diff --git a/crates/sqlbuild-core/src/build_progress/tests/test_build_progress.rs b/crates/sqlbuild-core/src/build_progress/tests/test_build_progress.rs index 1fac3f177c..74c9447f7c 100644 --- a/crates/sqlbuild-core/src/build_progress/tests/test_build_progress.rs +++ b/crates/sqlbuild-core/src/build_progress/tests/test_build_progress.rs @@ -1,3 +1,4 @@ +use crate::build_progress::main::batch_progress_message::batch_progress_message; use crate::build_progress::main::format_build_footer::format_build_footer; use crate::build_progress::main::progress_name_width::progress_name_width; use crate::build_progress::main::render_build_spinner_line::render_build_spinner_line; @@ -5,13 +6,13 @@ use crate::build_progress::main::render_failed_test_result::render_failed_test_r use crate::build_progress::main::render_model_result::render_model_result; use crate::build_progress::main::scheduler_diagnostic_message::scheduler_diagnostic_message; use crate::build_progress::models::{ - BuildResultText, ErrorText, LifecycleEventText, ModelResultText, ModelRowFacts, - NameWidthInputs, ProgressLayout, PythonNodeText, SchedulerStateText, SpinnerRow, + BatchProgressText, BuildResultText, ErrorText, LifecycleEventText, ModelResultText, + ModelRowFacts, NameWidthInputs, ProgressLayout, PythonNodeText, SchedulerStateText, SpinnerRow, }; use crate::build_progress::tests::helpers::{audit, failed_model, layout, model, test}; use crate::build_progress::tests::test_types::{ - FailedTestTestCase, FooterTestCase, InlineErrorTestCase, ModelRowTestCase, NameWidthTestCase, - SchedulerTestCase, SpinnerTestCase, + BatchProgressTestCase, FailedTestTestCase, FooterTestCase, InlineErrorTestCase, + ModelRowTestCase, NameWidthTestCase, SchedulerTestCase, SpinnerTestCase, }; use crate::cli_text::main::format_inline_error_lines::format_inline_error_lines; use crate::cli_text::models::CliStyle; @@ -293,3 +294,36 @@ fn given_long_error_when_wrapping_inline_then_truncates_to_two_lines() { ); } } + +#[test] +fn given_completed_batch_when_describing_then_names_model_window_rows_and_time() { + let batch = |rows_affected: Option| BatchProgressText { + model_name: "daily_orders".to_owned(), + batch_number: 2, + batch_count: 12, + window: "2024-02-01..2024-03-01".to_owned(), + rows_affected, + elapsed_seconds: 1.26, + }; + let test_cases = [ + BatchProgressTestCase { + description: "known row count", + batch: batch(Some(40)), + expected_text: "Batch daily_orders 2/12 2024-02-01..2024-03-01 40 rows 1.3s", + }, + BatchProgressTestCase { + description: "unknown row count", + batch: batch(None), + expected_text: "Batch daily_orders 2/12 2024-02-01..2024-03-01 1.3s", + }, + ]; + + for test_case in test_cases { + assert_eq!( + batch_progress_message(&test_case.batch), + test_case.expected_text, + "{}", + test_case.description + ); + } +} diff --git a/crates/sqlbuild-core/src/build_progress/tests/test_types.rs b/crates/sqlbuild-core/src/build_progress/tests/test_types.rs index a258d7dd49..91b62d8060 100644 --- a/crates/sqlbuild-core/src/build_progress/tests/test_types.rs +++ b/crates/sqlbuild-core/src/build_progress/tests/test_types.rs @@ -1,6 +1,6 @@ use crate::build_progress::models::{ - BuildResultText, ModelResultText, NameWidthInputs, PythonNodeText, SchedulerStateText, - SpinnerRow, TestResultText, + BatchProgressText, BuildResultText, ModelResultText, NameWidthInputs, PythonNodeText, + SchedulerStateText, SpinnerRow, TestResultText, }; pub(super) struct FooterTestCase { @@ -50,3 +50,9 @@ pub(super) struct InlineErrorTestCase { pub(super) width: usize, pub(super) expected_lines: &'static [&'static str], } + +pub(super) struct BatchProgressTestCase { + pub(super) description: &'static str, + pub(super) batch: BatchProgressText, + pub(super) expected_text: &'static str, +} diff --git a/crates/sqlbuild-core/src/dbt_events/main/consume_dbt_event.rs b/crates/sqlbuild-core/src/dbt_events/main/consume_dbt_event.rs index f348c24fe7..3d3974b8e6 100644 --- a/crates/sqlbuild-core/src/dbt_events/main/consume_dbt_event.rs +++ b/crates/sqlbuild-core/src/dbt_events/main/consume_dbt_event.rs @@ -3,14 +3,19 @@ use crate::dbt_events::_helpers::parse::{ node_message, node_result, node_start, start_message, unique_id, }; -use crate::dbt_events::models::{DbtEvent, DbtStreamRow, DbtStreamState, DbtStreamUpdate}; +use crate::dbt_events::models::{ + DbtEvent, DbtNodeResult, DbtStreamRow, DbtStreamState, DbtStreamUpdate, +}; /// Buffer node messages, number started nodes, and report each node's first final result once. pub fn consume_dbt_event( mut state: DbtStreamState, event: &DbtEvent, ) -> (DbtStreamState, Option) { - if let Some(message) = node_message(event) { + let result: Option = node_result(event); + if result.is_none() + && let Some(message) = node_message(event) + { if let Some(unique_id) = unique_id(event) { state .pending_messages @@ -43,13 +48,14 @@ pub fn consume_dbt_event( }; return (state, Some(update)); } - let Some(mut result) = node_result(event) else { + let Some(mut result) = result else { return (state, None); }; result.messages = state .pending_messages .remove(&result.unique_id) .unwrap_or_default(); + result.messages.extend(node_message(event)); if !state.recorded_unique_ids.insert(result.unique_id.clone()) { return (state, None); } diff --git a/crates/sqlbuild-core/src/dbt_events/tests/test_dbt_events.rs b/crates/sqlbuild-core/src/dbt_events/tests/test_dbt_events.rs index 6d052de54b..7ad3644bc6 100644 --- a/crates/sqlbuild-core/src/dbt_events/tests/test_dbt_events.rs +++ b/crates/sqlbuild-core/src/dbt_events/tests/test_dbt_events.rs @@ -54,6 +54,50 @@ fn given_dbt_events_when_consuming_then_reports_starts_and_first_results() { ), ], }, + StreamTestCase { + description: "an error-level result with a message still streams its row", + events: vec![ + message_event("model.a.payments", "relation missing"), + DbtEvent { + level: Some("error".to_owned()), + info_msg: Some("1 of 2 ERROR creating table model payments".to_owned()), + ..result_event("LogModelResult", "model.a.payments", "ERROR") + }, + ], + expected_updates: vec![ + None, + finished( + "model.a.payments", + ("ERROR", 1), + vec![ + DbtNodeMessage { + level: "error".to_owned(), + message: "relation missing".to_owned(), + }, + DbtNodeMessage { + level: "error".to_owned(), + message: "1 of 2 ERROR creating table model payments".to_owned(), + }, + ], + ), + ], + }, + StreamTestCase { + description: "a warn-level test result keeps its own message", + events: vec![DbtEvent { + level: Some("warn".to_owned()), + info_msg: Some("FAIL 3 not_null_orders_id".to_owned()), + ..result_event("LogTestResult", "test.a.not_null_orders_id", "fail") + }], + expected_updates: vec![finished( + "test.a.not_null_orders_id", + ("fail", 1), + vec![DbtNodeMessage { + level: "warn".to_owned(), + message: "FAIL 3 not_null_orders_id".to_owned(), + }], + )], + }, ]; for test_case in test_cases { diff --git a/crates/sqlbuild-core/src/scenario_command/main/mod.rs b/crates/sqlbuild-core/src/scenario_command/main/mod.rs index 905a2d5b48..62f178831d 100644 --- a/crates/sqlbuild-core/src/scenario_command/main/mod.rs +++ b/crates/sqlbuild-core/src/scenario_command/main/mod.rs @@ -4,6 +4,7 @@ pub mod capture_dialect_required_error; pub mod format_captured_at; pub mod render_capture_result; pub mod render_namespace_header; +pub mod render_scenario_failures; pub mod render_scenario_result; pub mod render_snapshot_sync_result; pub mod resolve_scenario_namespace; diff --git a/crates/sqlbuild-core/src/scenario_command/main/render_scenario_failures.rs b/crates/sqlbuild-core/src/scenario_command/main/render_scenario_failures.rs new file mode 100644 index 0000000000..8a8fdd35a7 --- /dev/null +++ b/crates/sqlbuild-core/src/scenario_command/main/render_scenario_failures.rs @@ -0,0 +1,61 @@ +//! The closing list of failed scenarios, so failures stay visible after long output. + +use crate::cli_text::models::CliStyle; +use crate::scenario_command::_helpers::rows::result_error; +use crate::scenario_command::constants::{ERROR_STATUS, FAIL_STATUS}; +use crate::scenario_command::models::ScenarioResultRow; +use crate::text::main::python_splitlines::python_splitlines; + +/// The first line of a failed scenario's error, or of its first failed expectation. +fn first_failure_line(row: &ScenarioResultRow, style: CliStyle) -> String { + let first_line = |text: String| -> String { + python_splitlines(&text) + .into_iter() + .next() + .unwrap_or_default() + .to_owned() + }; + if let Some(error) = row.error.as_ref().filter(|error| !error.message.is_empty()) { + return first_line(result_error(error, style)); + } + if let Some(expected) = row.expected.iter().find(|expected| expected.failed) { + return match (&expected.error, expected.mismatched_row_count) { + (Some(error), _) => first_line(result_error(error, style)), + (None, Some(mismatched)) => format!( + "expected {}: {mismatched} mismatched, {} missing", + expected.model_name, + expected.missing_row_count.unwrap_or_default() + ), + (None, None) => format!("expected {} did not run", expected.model_name), + }; + } + if let Some(assertion) = row.assertions.iter().find(|assertion| assertion.failed) { + return match (&assertion.error, assertion.failing_row_count) { + (Some(error), _) => first_line(result_error(error, style)), + (None, Some(count)) => format!("assertion {}: {count} failing rows", assertion.name), + (None, None) => format!("assertion {} did not run", assertion.name), + }; + } + String::new() +} + +/// `Failures:` with each failed scenario and its first error line, or nothing when all passed. +#[must_use] +pub fn render_scenario_failures(rows: &[ScenarioResultRow], style: CliStyle) -> String { + let failed: Vec<&ScenarioResultRow> = rows + .iter() + .filter(|row| row.status == FAIL_STATUS || row.status == ERROR_STATUS) + .collect(); + if failed.is_empty() { + return String::new(); + } + let mut text: String = String::from("\nFailures:\n\n"); + for row in failed { + text.push_str(&format!(" {}\n", row.name)); + let line: String = first_failure_line(row, style); + if !line.is_empty() { + text.push_str(&format!(" {line}\n")); + } + } + text +} diff --git a/crates/sqlbuild-core/src/scenario_command/tests/test_scenario_rows.rs b/crates/sqlbuild-core/src/scenario_command/tests/test_scenario_rows.rs index 6391b897d8..76219f235a 100644 --- a/crates/sqlbuild-core/src/scenario_command/tests/test_scenario_rows.rs +++ b/crates/sqlbuild-core/src/scenario_command/tests/test_scenario_rows.rs @@ -4,6 +4,7 @@ use crate::cli_text::models::CliStyle; use crate::scenario_command::main::build_local_runtime_artifacts::build_local_runtime_artifacts; use crate::scenario_command::main::build_runtime_artifacts::build_runtime_artifacts; use crate::scenario_command::main::render_capture_result::render_capture_result; +use crate::scenario_command::main::render_scenario_failures::render_scenario_failures; use crate::scenario_command::main::render_scenario_result::render_scenario_result; use crate::scenario_command::main::render_snapshot_sync_result::render_snapshot_sync_result; use crate::scenario_command::models::{ @@ -16,7 +17,8 @@ use crate::scenario_command::tests::helpers::{ statements, strings, }; use crate::scenario_command::tests::test_types::{ - CaptureRowTestCase, LocalArtifactsTestCase, ResultRowTestCase, WarehouseArtifactsTestCase, + CaptureRowTestCase, FailuresTestCase, LocalArtifactsTestCase, ResultRowTestCase, + WarehouseArtifactsTestCase, }; const RA0: &str = "orders_paid FAIL 1.23s\n error[X506]: expected comparison failed\n second line\n = help: Compare the CTE.\n expect expected fact_orders FAIL 2 mismatched, 1 missing\n error[X506]: mismatch\n = help: look\n expect expected dim_customers PASS\n expect assertion no_negative FAIL 1 row\n expect assertion few FAIL 3 rows\n plain error\n expect assertion ok PASS\n Retained relations:\n source raw_orders -> main.__sqb_abc__source__raw_orders\n model fact_orders -> main.__sqb_abc__model__fact_orders\n expected fact_orders: actual=3 expected=2 mismatched=2 missing=1\n assertion no_negative: failing_rows=1\n sample: (1, 'a', None)\n assertion few: failing_rows=3\n"; @@ -360,3 +362,59 @@ fn given_local_replay_when_building_artifacts_then_writes_loads_functions_and_mo ); } } + +#[test] +fn given_finished_scenarios_when_summarizing_then_lists_each_failure_with_its_first_line() { + let rows = [ + ScenarioResultRow { + name: "orders_ok".to_owned(), + status: "PASS".to_owned(), + ..ScenarioResultRow::default() + }, + ScenarioResultRow { + name: "orders_unmocked".to_owned(), + status: "FAIL".to_owned(), + error: coded(Some("S504"), "first line\nsecond line", None), + ..ScenarioResultRow::default() + }, + ScenarioResultRow { + name: "orders_mismatch".to_owned(), + status: "FAIL".to_owned(), + expected: vec![ExpectedRow { + model_name: "fact_orders".to_owned(), + failed: true, + mismatched_row_count: Some(2), + missing_row_count: Some(1), + ..ExpectedRow::default() + }], + ..ScenarioResultRow::default() + }, + ScenarioResultRow { + name: "orders_local".to_owned(), + status: "SKIP".to_owned(), + local: true, + ..ScenarioResultRow::default() + }, + ]; + let test_cases = [ + FailuresTestCase { + description: "failed scenarios are listed with their first error line", + rows: rows.to_vec(), + expected_text: "\nFailures:\n\n orders_unmocked\n error[S504]: first line\n orders_mismatch\n expected fact_orders: 2 mismatched, 1 missing\n", + }, + FailuresTestCase { + description: "no failures render nothing", + rows: rows[..1].to_vec(), + expected_text: "", + }, + ]; + + for test_case in test_cases { + assert_eq!( + render_scenario_failures(&test_case.rows, CliStyle { use_color: false }), + test_case.expected_text, + "{}", + test_case.description + ); + } +} diff --git a/crates/sqlbuild-core/src/scenario_command/tests/test_types.rs b/crates/sqlbuild-core/src/scenario_command/tests/test_types.rs index 299696eba4..e613413dc9 100644 --- a/crates/sqlbuild-core/src/scenario_command/tests/test_types.rs +++ b/crates/sqlbuild-core/src/scenario_command/tests/test_types.rs @@ -47,6 +47,12 @@ pub(super) struct ResultRowTestCase { pub(super) expected_text: &'static str, } +pub(super) struct FailuresTestCase { + pub(super) description: &'static str, + pub(super) rows: Vec, + pub(super) expected_text: &'static str, +} + pub(super) struct CaptureRowTestCase { pub(super) description: &'static str, pub(super) row: CaptureRow, diff --git a/crates/sqlbuild-python/src/bindings/_helpers/build_command/execution.rs b/crates/sqlbuild-python/src/bindings/_helpers/build_command/execution.rs index 8813268d4d..389eff2226 100644 --- a/crates/sqlbuild-python/src/bindings/_helpers/build_command/execution.rs +++ b/crates/sqlbuild-python/src/bindings/_helpers/build_command/execution.rs @@ -18,7 +18,9 @@ use crate::bindings::_helpers::build_command::lifecycle::{ use crate::bindings::_helpers::build_command::support::{ boolean, display_text, emit, frozenset, joined, keyword_relay, line_writer, none, text, tuple, }; -use crate::bindings::_helpers::build_progress::build_callbacks::BuildProgressCallbacks; +use crate::bindings::_helpers::build_progress::build_callbacks::{ + BuildProgressCallbacks, batch_complete_callback, +}; use crate::bindings::_helpers::build_progress::progress_reporters::ConnectionProgressReporter; use crate::bindings::_helpers::build_progress::progress_text::write_execution_header; use crate::bindings::_helpers::check::check_core::{ @@ -376,6 +378,7 @@ pub(crate) fn execute_build_plan<'py>( "on_sub_progress", native_field(callbacks, "on_sub_progress")?, ), + ("on_batch_complete", batch_complete_callback(callbacks)?), ( "on_connection_start", native_field(connection, "on_connection_start")?, diff --git a/crates/sqlbuild-python/src/bindings/_helpers/build_progress/build_callbacks.rs b/crates/sqlbuild-python/src/bindings/_helpers/build_progress/build_callbacks.rs index 17dcb39cca..ec6bbad153 100644 --- a/crates/sqlbuild-python/src/bindings/_helpers/build_progress/build_callbacks.rs +++ b/crates/sqlbuild-python/src/bindings/_helpers/build_progress/build_callbacks.rs @@ -5,8 +5,9 @@ use std::sync::{Mutex, MutexGuard, PoisonError}; use std::time::Instant; use pyo3::prelude::{Bound, Py, PyAny, PyAnyMethods, PyModule, PyModuleMethods, PyResult, Python}; -use pyo3::types::{PyDict, PyDictMethods}; +use pyo3::types::{PyCFunction, PyDict, PyDictMethods, PyTuple, PyTupleMethods}; use pyo3::{pyclass, pymethods}; +use sqlbuild_core::build_progress::main::batch_progress_message::batch_progress_message; use sqlbuild_core::build_progress::main::query_diagnostic_message::query_diagnostic_message; use sqlbuild_core::build_progress::main::render_build_spinner_line::render_build_spinner_line; use sqlbuild_core::build_progress::main::render_failed_test_result::render_failed_test_result; @@ -15,8 +16,8 @@ use sqlbuild_core::build_progress::main::render_resource_result::render_resource use sqlbuild_core::build_progress::main::runtime_diagnostic_summary::runtime_diagnostic_summary; use sqlbuild_core::build_progress::main::scheduler_diagnostic_message::scheduler_diagnostic_message; use sqlbuild_core::build_progress::models::{ - ModelRowFacts, ProgressLayout, ResourceResult, SchedulerStateText, SpinnerRow, StatementText, - TestResultText, + BatchProgressText, ModelRowFacts, ProgressLayout, ResourceResult, SchedulerStateText, + SpinnerRow, StatementText, TestResultText, }; use sqlbuild_core::cli_text::models::CliStyle; @@ -132,7 +133,7 @@ struct CallbackState { scheduler_count: usize, scheduler_omitted: usize, last_scheduler: Option, - query_count: usize, + query_counts: HashMap, query_omitted: usize, last_query: Option, } @@ -253,6 +254,11 @@ impl BuildProgressCallbacks { if self.debug { return Ok(()); } + self.write_persistent(py, message) + } + + /// Write one permanent line through the shared line coordinator. + fn write_persistent(&self, py: Python<'_>, message: &str) -> PyResult<()> { let options: Bound<'_, PyDict> = PyDict::new(py); options.set_item("stream", self.stream.bind(py))?; options.set_item("text", format!("{message}\n"))?; @@ -262,6 +268,22 @@ impl BuildProgressCallbacks { Ok(()) } + /// Keep a line per completed microbatch when no live spinner row shows batch progress. + fn write_batch_complete(&self, py: Python<'_>, progress: &Bound<'_, PyAny>) -> PyResult<()> { + if self.is_tty { + return Ok(()); + } + let batch: BatchProgressText = BatchProgressText { + model_name: text(progress, "model_name")?, + batch_number: native_field(progress, "batch_number")?.extract()?, + batch_count: native_field(progress, "batch_count")?.extract()?, + window: text(progress, "window")?, + rows_affected: native_field(progress, "rows_affected")?.extract()?, + elapsed_seconds: native_field(progress, "elapsed_seconds")?.extract()?, + }; + self.write_persistent(py, &batch_progress_message(&batch)) + } + fn write_node_result( slf: &Bound<'_, Self>, result: &Bound<'_, PyAny>, @@ -505,6 +527,11 @@ impl BuildProgressCallbacks { Ok(()) } + /// Keep a line per completed microbatch when no live spinner row shows batch progress. + fn on_batch_complete(&self, py: Python<'_>, progress: &Bound<'_, PyAny>) -> PyResult<()> { + self.write_batch_complete(py, progress) + } + /// Report one deduplicated concurrent scheduler state transition. fn on_scheduler_state(&self, py: Python<'_>, state: &Bound<'_, PyAny>) -> PyResult<()> { let state: SchedulerStateText = scheduler_state(state)?; @@ -525,8 +552,12 @@ impl BuildProgressCallbacks { let telemetry: StatementText = statement(telemetry)?; { let mut state = self.state(); - state.query_count += 1; - if state.query_count > QUERY_DIAGNOSTIC_LIMIT { + let count: &mut usize = state + .query_counts + .entry(telemetry.resource_name.clone()) + .or_default(); + *count += 1; + if *count > QUERY_DIAGNOSTIC_LIMIT { state.query_omitted += 1; state.last_query = Some(telemetry); return Ok(()); @@ -611,6 +642,24 @@ impl BuildProgressCallbacks { } } +/// A callable that forwards each completed microbatch to `callbacks`, built without a Python lookup. +pub(crate) fn batch_complete_callback<'py>( + callbacks: &Bound<'py, PyAny>, +) -> PyResult> { + let owned: Py = callbacks + .downcast::()? + .clone() + .unbind(); + let forward = move |arguments: &Bound<'_, PyTuple>, + _keywords: Option<&Bound<'_, PyDict>>| + -> PyResult<()> { + owned + .get() + .write_batch_complete(arguments.py(), &arguments.get_item(0)?) + }; + Ok(PyCFunction::new_closure(callbacks.py(), None, None, forward)?.into_any()) +} + pub(crate) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> { module.add_class::() } diff --git a/crates/sqlbuild-python/src/bindings/_helpers/clone/clone_command.rs b/crates/sqlbuild-python/src/bindings/_helpers/clone/clone_command.rs index 075ec0e0dc..674c72b338 100644 --- a/crates/sqlbuild-python/src/bindings/_helpers/clone/clone_command.rs +++ b/crates/sqlbuild-python/src/bindings/_helpers/clone/clone_command.rs @@ -20,12 +20,12 @@ use sqlbuild_core::clone_text::models::CloneItemText; use sqlbuild_host::entries::models::ExecutorHost; use crate::bindings::_helpers::boundary::python_calls::keywords; -use crate::bindings::_helpers::boundary::python_calls::{call_with, within}; +use crate::bindings::_helpers::boundary::python_calls::{call_with, imported, within}; use crate::bindings::_helpers::build_progress::progress_reporters::PlanningProgressReporter; use crate::bindings::_helpers::clone::clone_execution::{ destination_entry_count, execute_clone_plan, resource_types_by_name, }; -use crate::bindings::_helpers::clone::clone_pipeline::{ClonePipeline, prepared_clone_pipeline}; +use crate::bindings::_helpers::clone::clone_pipeline::{ClonePipeline, timed_clone_pipeline}; use crate::bindings::_helpers::clone::clone_progress::{clone_item_text, write_line}; use crate::bindings::_helpers::command_runtime::adapters::resolve_cli_adapter; use crate::bindings::_helpers::command_runtime::connection::resolve_target_connection_config; @@ -36,6 +36,7 @@ use crate::bindings::_helpers::contracts::spec_resolution::resolve_target_name; use crate::bindings::_helpers::dbt_interop::reference_resolver::resolve_external_sql_reference_resolver; use crate::bindings::_helpers::native_host::connections::connection_failure; use crate::bindings::_helpers::native_host::host; +use crate::bindings::_helpers::planner_entries::clone_plans::PhaseTimings; use crate::bindings::_helpers::planner_selection::unit_tests::reject_unit_test_selectors; use crate::bindings::_helpers::presentation_text::terminal::stdout_supports_color; use crate::bindings::_helpers::records::construct::error_class; @@ -45,6 +46,9 @@ use crate::bindings::_helpers::records::ids::{ErrorId, RecordId}; use crate::bindings::_helpers::result_output::documents::format_clone_execution_json; use crate::bindings::_helpers::result_output::documents::write_execution_json_output; +const CLONE_LOGGER: &str = "sqlbuild.clone"; +/// The clone phase that copies relations into the destination. +const EXECUTION_PHASE: &str = "execution"; const NO_DESTINATION: (&str, &str) = ( "C406", "clone requires --to when the project has no active target", @@ -233,12 +237,12 @@ fn prepare<'py>( request: &Bound<'py, PyAny>, invocation: &Invocation<'py>, (config, connection): (&Bound<'py, PyAny>, &Bound<'py, PyAny>), -) -> PyResult> { +) -> PyResult<(Bound<'py, PyAny>, PhaseTimings)> { let py: Python<'py> = request.py(); let progress: Bound<'py, PlanningProgressReporter> = progress_reporter(&invocation.stream, invocation.style)?; let started: Instant = Instant::now(); - let prepared: PyResult> = + let prepared: PyResult<(Bound<'py, PyAny>, PhaseTimings)> = within(&lifecycle(py, ("project", "project_compile"))?, |_| { progress.get().on_progress(py, "Preparing clone plan...")?; let resolver: Bound<'py, PyAny> = resolve_external_sql_reference_resolver( @@ -263,7 +267,7 @@ fn prepare<'py>( ("cli_vars", native_field(request, "cli_vars")?), ], )?; - let pipeline: Bound<'py, PyAny> = prepared_clone_pipeline( + let prepared: (Bound<'py, PyAny>, PhaseTimings) = timed_clone_pipeline( &ClonePipeline { discovered: invocation.discovered.clone(), adapter: invocation.adapter.clone(), @@ -283,14 +287,14 @@ fn prepare<'py>( started.elapsed().as_secs_f64() ), )?; - Ok(pipeline) + Ok(prepared) }); - let pipeline: Bound<'py, PyAny> = + let (pipeline, timings) = prepared.or_else(|error| progress.get().finish(py, false).and(Err(error)))?; if destination_entry_count(&pipeline)? == 0 { return Err(user_error(py, NOTHING_TO_CLONE)); } - Ok(pipeline) + Ok((pipeline, timings)) } /// Prepare, announce and execute the clone with an open destination connection. @@ -298,8 +302,8 @@ fn clone<'py>( request: &Bound<'py, PyAny>, invocation: &Invocation<'py>, (config, connection): (&Bound<'py, PyAny>, &Bound<'py, PyAny>), -) -> PyResult<(Bound<'py, PyAny>, Bound<'py, PyAny>, f64)> { - let pipeline: Bound<'py, PyAny> = prepare(request, invocation, (config, connection))?; +) -> PyResult<(Bound<'py, PyAny>, Bound<'py, PyAny>, PhaseTimings)> { + let (pipeline, mut timings) = prepare(request, invocation, (config, connection))?; let origin: String = native_field(request, "origin_target_name")?.extract()?; write_line( &invocation.stream, @@ -310,7 +314,17 @@ fn clone<'py>( ), )?; let (result, elapsed) = execute_clone_plan(request, invocation, (&pipeline, connection))?; - Ok((pipeline, result, elapsed)) + timings.push((EXECUTION_PHASE.to_owned(), elapsed)); + Ok((pipeline, result, timings)) +} + +/// Log each clone phase's duration for `--debug`, the way compile phases are reported. +fn log_phase_timings(py: Python<'_>, timings: &[(String, f64)]) -> PyResult<()> { + let logger: Bound<'_, PyAny> = imported(py, "logging", "getLogger")?.call1((CLONE_LOGGER,))?; + for (phase, seconds) in timings { + logger.call_method1("debug", (format!("Clone timing {phase} {seconds:.3}s"),))?; + } + Ok(()) } /// Run one `sqb clone` request. @@ -322,26 +336,35 @@ pub(crate) fn run_clone_command(py: Python<'_>, request: &Bound<'_, PyAny>) -> P let outcome = clone(request, &invocation, (&config, &connection)); let closed: PyResult> = invocation.adapter.call_method1("close", (&connection,)); - let (pipeline, result, elapsed) = outcome?; + let (pipeline, result, timings) = outcome?; closed?; + log_phase_timings(py, &timings)?; + let elapsed: f64 = timings + .iter() + .find(|(phase, _)| phase == EXECUTION_PHASE) + .map_or(0.0, |(_, seconds)| *seconds); let items: Vec = native_field(&result, "item_results")? .try_iter()? .map(|item| clone_item_text(&item?)) .collect::>()?; - let stdout: Bound<'_, PyAny> = py.import("sys")?.getattr("stdout")?; for line in render_clone_completion(&items, Some(elapsed), invocation.style).lines() { - if !line.is_empty() { - stdout.call_method1("write", (line,))?; - } - stdout.call_method1("write", ("\n",))?; + write_line(&invocation.stream, line)?; + } + let phase_timings: Bound<'_, PyDict> = PyDict::new(py); + for (phase, seconds) in &timings { + phase_timings.set_item(phase, seconds)?; } let resource_types: Bound<'_, PyDict> = PyDict::new(py); for (name, resource_type) in resource_types_by_name(&pipeline)? { resource_types.set_item(name, resource_type)?; } - let payload: Bound<'_, PyAny> = - format_clone_execution_json(py, result.clone(), resource_types.into_any()) - .and_then(|value| pyo3::IntoPyObjectExt::into_bound_py_any(value, py))?; + let payload: Bound<'_, PyAny> = format_clone_execution_json( + py, + result.clone(), + resource_types.into_any(), + Some(phase_timings), + ) + .and_then(|value| pyo3::IntoPyObjectExt::into_bound_py_any(value, py))?; write_execution_json_output( py, payload, diff --git a/crates/sqlbuild-python/src/bindings/_helpers/clone/clone_pipeline.rs b/crates/sqlbuild-python/src/bindings/_helpers/clone/clone_pipeline.rs index 354a5593f5..50d18b2b66 100644 --- a/crates/sqlbuild-python/src/bindings/_helpers/clone/clone_pipeline.rs +++ b/crates/sqlbuild-python/src/bindings/_helpers/clone/clone_pipeline.rs @@ -8,11 +8,14 @@ use crate::bindings::_helpers::boundary::keyword_request::Request; use crate::bindings::_helpers::boundary::python_calls::keywords; use crate::bindings::_helpers::compile_driver::project::build_compiled_project; use crate::bindings::_helpers::contracts::spec_resolution::resolve_target_config; -use crate::bindings::_helpers::planner_entries::clone_plans::run_clone_planning; +use crate::bindings::_helpers::planner_entries::clone_plans::{PhaseTimings, timed_clone_planning}; use crate::bindings::_helpers::records::construct::new_record_with; use crate::bindings::_helpers::records::fields::field as native_field; use crate::bindings::_helpers::records::ids::RecordId; +/// The phase name of one target compile, before the target name. +const COMPILE_PHASE_PREFIX: &str = "compile_"; + /// The `ClonePipelineResult` fields after the two projects, in `run_clone_planning` order. const PLANNED_FIELDS: [&str; 8] = [ "clone_plan", @@ -111,8 +114,17 @@ fn read_target_name<'py>( /// Compile both environments, plan the clone and return the `ClonePipelineResult`. pub(crate) fn prepared_clone_pipeline<'py>( pipeline: &ClonePipeline<'py>, - (origin_target, destination_target): (&Bound<'py, PyAny>, &Bound<'py, PyAny>), + targets: (&Bound<'py, PyAny>, &Bound<'py, PyAny>), ) -> PyResult> { + Ok(timed_clone_pipeline(pipeline, targets)?.0) +} + +/// `prepared_clone_pipeline`, with each phase's name and seconds in order. +pub(crate) fn timed_clone_pipeline<'py>( + pipeline: &ClonePipeline<'py>, + (origin_target, destination_target): (&Bound<'py, PyAny>, &Bound<'py, PyAny>), +) -> PyResult<(Bound<'py, PyAny>, PhaseTimings)> { + let mut timings: PhaseTimings = Vec::new(); let py: Python<'py> = origin_target.py(); let origin_reads: Bound<'py, PyAny> = read_target_name(&pipeline.discovered, origin_target)?; let destination_reads: Bound<'py, PyAny> = @@ -125,14 +137,19 @@ pub(crate) fn prepared_clone_pipeline<'py>( &destination_reads, ] { if environment(&compiled, target).is_err() { + let started = std::time::Instant::now(); compiled.push((target.clone(), compiled_environment(pipeline, target)?)); + timings.push(( + format!("{COMPILE_PHASE_PREFIX}{}", target.str()?), + started.elapsed().as_secs_f64(), + )); } } let origin_project: Bound<'py, PyAny> = environment(&compiled, origin_target)?; let destination_project: Bound<'py, PyAny> = environment(&compiled, destination_target)?; let origin_source_project: Bound<'py, PyAny> = environment(&compiled, &origin_reads)?; let destination_source_project: Bound<'py, PyAny> = environment(&compiled, &destination_reads)?; - let planned: Bound<'py, PyTuple> = run_clone_planning( + let (planned, planning_timings): (Bound<'py, PyTuple>, PhaseTimings) = timed_clone_planning( py, Some(keywords( py, @@ -147,9 +164,8 @@ pub(crate) fn prepared_clone_pipeline<'py>( ("destination_source_project", destination_source_project), ], )?), - ) - .and_then(|value| pyo3::IntoPyObjectExt::into_bound_py_any(value, py))? - .downcast_into()?; + )?; + timings.extend(planning_timings); let mut fields: Vec<(&str, Bound<'py, PyAny>)> = vec![ ("origin_project", origin_project), ("destination_project", destination_project), @@ -157,7 +173,10 @@ pub(crate) fn prepared_clone_pipeline<'py>( for (index, name) in PLANNED_FIELDS.iter().enumerate() { fields.push((*name, planned.get_item(index)?)); } - new_record_with(py, RecordId::ClonePipelineResult, &fields) + Ok(( + new_record_with(py, RecordId::ClonePipelineResult, &fields)?, + timings, + )) } /// Run a clone pipeline with explicit compilation controls. diff --git a/crates/sqlbuild-python/src/bindings/_helpers/dbt_interop/sqlbuild_work.rs b/crates/sqlbuild-python/src/bindings/_helpers/dbt_interop/sqlbuild_work.rs index e6fe96555f..bc10e0eff5 100644 --- a/crates/sqlbuild-python/src/bindings/_helpers/dbt_interop/sqlbuild_work.rs +++ b/crates/sqlbuild-python/src/bindings/_helpers/dbt_interop/sqlbuild_work.rs @@ -14,7 +14,9 @@ use crate::bindings::_helpers::build_command::policies::enforce_plan_safety_poli use crate::bindings::_helpers::build_command::support::{ boolean, emit, joined, keyword_relay, none, text, tuple, }; -use crate::bindings::_helpers::build_progress::build_callbacks::BuildProgressCallbacks; +use crate::bindings::_helpers::build_progress::build_callbacks::{ + BuildProgressCallbacks, batch_complete_callback, +}; use crate::bindings::_helpers::build_progress::footer::format_build_footer; use crate::bindings::_helpers::build_progress::progress_reporters::ConnectionProgressReporter; use crate::bindings::_helpers::build_progress::progress_text::format_execution_header; @@ -202,6 +204,7 @@ pub(crate) fn execute_dbt_sqlbuild_work<'py>( "on_sub_progress", native_field(&callbacks, "on_sub_progress")?, ), + ("on_batch_complete", batch_complete_callback(&callbacks)?), ( "on_connection_start", native_field(&connection, "on_connection_start")?, diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_entries/clone_plans.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_entries/clone_plans.rs index 14c8f0fc90..6896a71a80 100644 --- a/crates/sqlbuild-python/src/bindings/_helpers/planner_entries/clone_plans.rs +++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_entries/clone_plans.rs @@ -37,6 +37,13 @@ const ENABLED_KEY: &str = "enabled"; const VIEW_MATERIALIZATION: &str = "view"; const GRAPH_ARGUMENTS: [&str; 4] = ["selected", "upstream", "is_clonable", "is_view"]; +/// The clone phase that reads source columns from the warehouse. +const SOURCE_INSPECTION_PHASE: &str = "source_inspection"; +/// The clone phase that plans entries from the compiled projects. +const PLANNING_PHASE: &str = "planning"; + +/// Each clone phase name with its seconds, in the order the phases ran. +pub(crate) type PhaseTimings = Vec<(String, f64)>; /// Whether a model is explicitly disabled with a boolean `enabled = false`. fn disabled_model(model: &Bound<'_, PyAny>) -> PyResult { @@ -511,6 +518,15 @@ pub(crate) fn run_clone_planning<'py>( py: Python<'py>, kwargs: Option>, ) -> PyResult> { + Ok(timed_clone_planning(py, kwargs)?.0) +} + +/// `run_clone_planning`, with the seconds spent inspecting sources and planning. +pub(crate) fn timed_clone_planning<'py>( + py: Python<'py>, + kwargs: Option>, +) -> PyResult<(Bound<'py, PyTuple>, PhaseTimings)> { + let started = std::time::Instant::now(); let args = KeywordArguments::read( py, "run_clone_planning", @@ -534,11 +550,13 @@ pub(crate) fn run_clone_planning<'py>( let connection = args.value("connection")?; let destination_sources = args.value("destination_source_project")?; let plan = clone_plan_output(&project, &args.value("select")?, &args.value("exclude")?)?; + let inspection_started = std::time::Instant::now(); let resolution = clone_resolution( (&project, &plan), &destination_sources, (&adapter, &connection), )?; + let inspection_seconds = inspection_started.elapsed().as_secs_f64(); let models = clone_model_entries((&project, &plan), &adapter, &resolution)?; let seeds = seed_entries(&project, "key", &native_field(&plan, "selected_keys")?)?; let functions = clone_function_entries((&project, &plan), &adapter, &resolution)?; @@ -580,7 +598,14 @@ pub(crate) fn run_clone_planning<'py>( kept_sources.push(entry); } } - PyTuple::new( + let timings: PhaseTimings = vec![ + (SOURCE_INSPECTION_PHASE.to_owned(), inspection_seconds), + ( + PLANNING_PHASE.to_owned(), + started.elapsed().as_secs_f64() - inspection_seconds, + ), + ]; + let planned = PyTuple::new( py, [ plan, @@ -592,7 +617,8 @@ pub(crate) fn run_clone_planning<'py>( origin_seeds, tuple_of(py, kept_origins)?, ], - ) + )?; + Ok((planned, timings)) } /// The traversal callbacks and graph of one deferred-clone query. diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_entries/execution_plan.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_entries/execution_plan.rs index aeb70b2f53..49dfa8bfb9 100644 --- a/crates/sqlbuild-python/src/bindings/_helpers/planner_entries/execution_plan.rs +++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_entries/execution_plan.rs @@ -13,6 +13,7 @@ use crate::bindings::_helpers::planner_entries::test_planning::{ use crate::bindings::_helpers::planner_entries::values::{native, replaced}; use crate::bindings::_helpers::planner_phases::buildability::check_selected_scope_buildability; use crate::bindings::_helpers::planner_phases::contexts::class_field; +use crate::bindings::_helpers::planner_phases::deferred_origins::defer_to_migration_origins; use crate::bindings::_helpers::planner_phases::execution_changes::{ build_planner_entry_results, reconcile_execution_changes, }; @@ -25,8 +26,10 @@ use crate::bindings::_helpers::planner_phases::models::PlannerScopeResolution; use crate::bindings::_helpers::planner_phases::models::{ PlanOutputContext, PlanWarningContext, PlannedModelEntries, StoragePolicyContext, }; +use crate::bindings::_helpers::planner_phases::output_assembly::unselected_upstream_models; use crate::bindings::_helpers::planner_phases::output_assembly::{ - add_plan_metadata, add_plan_warnings, add_storage_policies, base_plan_output, + add_deferred_origin_notes, add_plan_metadata, add_plan_warnings, add_storage_policies, + base_plan_output, }; use crate::bindings::_helpers::planner_phases::runtime::warehouse_field; use crate::bindings::_helpers::planner_phases::scopes::resolve_planner_scopes; @@ -112,6 +115,31 @@ fn change_results<'py>( Ok((changes, stale)) } +/// Redirect unselected upstreams to their `migrate_from` relation in the deferred target. +fn deferred_migration_origins<'py>( + args: &KeywordArguments<'py>, + scopes: &Bound<'py, PyAny>, + deferral: &Bound<'py, PyAny>, +) -> PyResult> { + let locations = native_field(deferral, "deferred_locations")?; + let relations = native_field(deferral, "deferred_relations")?; + if locations.is_none() || relations.is_none() { + return Ok(Vec::new()); + } + defer_to_migration_origins( + ( + &args.value("project")?, + &args.value("adapter")?, + &args.value("connection")?, + ), + ( + locations.downcast::()?, + relations.downcast::()?, + ), + &unselected_upstream_models(scopes)?, + ) +} + /// The plan phases after warehouse inspection, run inside the inspection catalog. fn planned_output<'py>( args: &KeywordArguments<'py>, @@ -123,6 +151,7 @@ fn planned_output<'py>( let py = project.py(); let deferral = args.value("deferral")?; let policies = args.value("policies")?; + let origin_failures = deferred_migration_origins(args, scopes, &deferral)?; let warehouse = gather_planner_warehouse_state(runtime, scopes, overrides, &deferral)?; let started = Instant::now(); let identities = build_planner_identity_context( @@ -193,6 +222,8 @@ fn planned_output<'py>( }, (&source_freshness, &policies), )?; + let output = + add_deferred_origin_notes(output, (&project, scopes), (&deferral, &origin_failures))?; let plan_output = add_plan_metadata(output, &source_freshness, &policies)?.build(py)?; let on_progress = args.value("on_progress")?; if !on_progress.is_none() { diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_phases/deferred_origins.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_phases/deferred_origins.rs new file mode 100644 index 0000000000..57590cc3c0 --- /dev/null +++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_phases/deferred_origins.rs @@ -0,0 +1,212 @@ +//! Deferred locations of renamed models the deferred target holds only under their old name. + +use pyo3::exceptions::PyException; +use pyo3::prelude::{Bound, IntoPyObject, PyAny, PyAnyMethods, PyResult, Python}; +use pyo3::types::{PyDict, PyDictMethods, PyString, PyTuple}; + +use crate::bindings::_helpers::boundary::python_calls::call_with; +use crate::bindings::_helpers::model_config::typed_config::{config_value, model_values}; +use crate::bindings::_helpers::records::construct::new_record_with; +use crate::bindings::_helpers::records::fields::field as native_field; +use crate::bindings::_helpers::records::ids::RecordId; + +const MIGRATE_FROM_KEY: &str = "migrate_from"; +/// `database.schema.name`. +const QUALIFIED_RELATION_MAX_PARTS: usize = 3; + +/// The relation in `relations` named `name`, ignoring case. +fn relation_named<'py>( + relations: &Bound<'py, PyDict>, + name: &str, +) -> PyResult>> { + if let Some(relation) = relations.get_item(name)? { + return Ok(Some(relation)); + } + let wanted: String = name.to_lowercase(); + for (key, relation) in relations.iter() { + if key.extract::()?.to_lowercase() == wanted { + return Ok(Some(relation)); + } + } + Ok(None) +} + +/// A model's declared `migrate_from`, split into its dotted parts. +fn migration_origin_parts(model: &Bound<'_, PyAny>) -> PyResult>> { + let py: Python<'_> = model.py(); + let origin: Bound<'_, PyAny> = config_value(py, &model_values(model)?, MIGRATE_FROM_KEY)?; + if !origin.is_instance_of::() { + return Ok(None); + } + let raw: String = origin.extract()?; + let parts: Vec = raw + .split('.') + .map(|part| part.trim().trim_matches(['"', '`']).to_owned()) + .collect(); + if parts.is_empty() + || parts.len() > QUALIFIED_RELATION_MAX_PARTS + || parts.iter().any(String::is_empty) + { + return Ok(None); + } + Ok(Some(parts)) +} + +/// Relations already listed in the deferred target, by `(database, schema)`. +type ListedSchemas<'py> = Vec<(Option, String, Bound<'py, PyDict>)>; + +/// Relations of one deferred `database.schema`, listing it on first use, with the cache. +fn schema_relations<'py>( + (adapter, connection): (&Bound<'py, PyAny>, &Bound<'py, PyAny>), + (database, schema): (&Option, &str), + mut listed: ListedSchemas<'py>, +) -> (PyResult>, ListedSchemas<'py>) { + if let Some((_, _, found)) = listed.iter().find(|(listed_database, listed_schema, _)| { + listed_database == database && listed_schema.eq_ignore_ascii_case(schema) + }) { + return (Ok(found.clone()), listed); + } + let py: Python<'py> = adapter.py(); + let found = (|| -> PyResult> { + let found: Bound<'py, PyDict> = PyDict::new(py); + for relation in call_with( + &native_field(adapter, "list_relations")?, + &[ + ("connection", connection.clone()), + ("database", database.clone().into_pyobject(py)?.into_any()), + ("schemas", PyTuple::new(py, [schema])?.into_any()), + ], + )? + .try_iter()? + { + let relation: Bound<'py, PyAny> = relation?; + found.set_item(native_field(&relation, "name")?, relation)?; + } + Ok(found) + })(); + if let Ok(found) = &found { + listed.push((database.clone(), schema.to_owned(), found.clone())); + } + (found, listed) +} + +/// The `(database, schema, name)` a `migrate_from` names, resolving omitted parts in `location`. +fn migration_origin_relation( + location: &Bound<'_, PyAny>, + parts: &[String], +) -> PyResult, String, String)>> { + let location_database: Option = native_field(location, "database")?.extract()?; + let location_schema: Option = native_field(location, "schema")?.extract()?; + let name: String = parts[parts.len() - 1].clone(); + Ok(match parts.len() { + 1 => location_schema.map(|schema| (location_database, schema, name)), + QUALIFIED_RELATION_MAX_PARTS => Some((Some(parts[0].clone()), parts[1].clone(), name)), + _ => Some((location_database, parts[0].clone(), name)), + }) +} + +/// Redirect unselected `models` unbuilt in the deferred target to their `migrate_from` relation. +pub(crate) fn defer_to_migration_origins<'py>( + (project, adapter, connection): (&Bound<'py, PyAny>, &Bound<'py, PyAny>, &Bound<'py, PyAny>), + (locations, relations): (&Bound<'py, PyDict>, &Bound<'py, PyDict>), + models: &[String], +) -> PyResult> { + let py: Python<'py> = project.py(); + let mut failures: Vec<(String, String)> = Vec::new(); + let mut listed: ListedSchemas<'py> = Vec::new(); + for model in native_field(project, "models")?.try_iter()? { + let model: Bound<'py, PyAny> = model?; + let name: String = native_field(&model, "name")?.extract()?; + if !models.contains(&name) { + continue; + } + let Some(location) = locations.get_item(&name)? else { + continue; + }; + let Some(parts) = migration_origin_parts(&model)? else { + continue; + }; + let Some((origin_database, origin_schema, origin_name)) = + migration_origin_relation(&location, &parts)? + else { + continue; + }; + let current: Option<(Option, String, String)> = + migration_origin_relation(&location, &[native_field(&location, "name")?.extract()?])?; + let mut lookups: Vec<(Option, String, String)> = current.into_iter().collect(); + lookups.push((origin_database, origin_schema, origin_name)); + let mut origin: Option> = None; + for (index, (database, schema, relation_name)) in lookups.iter().enumerate() { + let (schema_found, updated) = + schema_relations((adapter, connection), (database, schema), listed); + listed = updated; + let hit: Option> = match schema_found { + Ok(schema_found) => relation_named(&schema_found, relation_name)?, + Err(error) if error.is_instance_of::(py) => { + let qualified: String = [ + database.as_deref(), + Some(schema.as_str()), + Some(relation_name.as_str()), + ] + .into_iter() + .flatten() + .collect::>() + .join("."); + failures.push(( + name.clone(), + format!( + "could not list {qualified} in the deferred target to look for a pre-migration relation, so model '{name}' is deferred under its current name: {}", + error.value(py) + ), + )); + break; + } + Err(error) => return Err(error), + }; + if index + 1 < lookups.len() { + if hit.is_some() { + break; + } + } else { + origin = hit; + } + } + let Some(origin) = origin else { + continue; + }; + let mut database: Bound<'py, PyAny> = native_field(&origin, "database")?; + if database.is_none() { + database = native_field(&location, "database")?; + } + let schema: Bound<'py, PyAny> = native_field(&origin, "schema")?; + let relation_name: Bound<'py, PyAny> = native_field(&origin, "name")?; + let qualified_name: Bound<'py, PyAny> = call_with( + &native_field(adapter, "render_qualified_name")?, + &[ + ("database", database.clone()), + ("schema", schema.clone()), + ("name", relation_name.clone()), + ], + )?; + locations.set_item( + &name, + new_record_with( + py, + RecordId::CompiledRelationLocation, + &[ + ("database", database), + ("schema", schema), + ("name", relation_name), + ("qualified_name", qualified_name), + ("logical_schema", native_field(&location, "logical_schema")?), + ( + "logical_database", + native_field(&location, "logical_database")?, + ), + ], + )?, + )?; + relations.set_item(&name, origin)?; + } + Ok(failures) +} diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_phases/mod.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_phases/mod.rs index ec478e3ab3..0fd4cce629 100644 --- a/crates/sqlbuild-python/src/bindings/_helpers/planner_phases/mod.rs +++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_phases/mod.rs @@ -3,6 +3,7 @@ pub(crate) mod buildability; pub(crate) mod constants; pub(crate) mod contexts; +pub(crate) mod deferred_origins; pub(crate) mod execution_changes; pub(crate) mod full_refresh; pub(crate) mod getters; diff --git a/crates/sqlbuild-python/src/bindings/_helpers/planner_phases/output_assembly.rs b/crates/sqlbuild-python/src/bindings/_helpers/planner_phases/output_assembly.rs index 7259f08623..5e7c1c134f 100644 --- a/crates/sqlbuild-python/src/bindings/_helpers/planner_phases/output_assembly.rs +++ b/crates/sqlbuild-python/src/bindings/_helpers/planner_phases/output_assembly.rs @@ -21,7 +21,7 @@ use crate::bindings::_helpers::planner_phases::retention::{plan_retention, plan_ use crate::bindings::_helpers::planner_phases::runtime::runtime_field; use crate::bindings::_helpers::planner_phases::runtime::warehouse_field; use crate::bindings::_helpers::planner_warehouse::source_freshness::source_freshness_unknown_warnings; -use crate::bindings::_helpers::records::construct::new_record_with; +use crate::bindings::_helpers::records::construct::{enum_member, new_record_with}; use crate::bindings::_helpers::records::fields::field as native_field; use crate::bindings::_helpers::records::fields::mapping_items; use crate::bindings::_helpers::records::ids::RecordId; @@ -306,6 +306,91 @@ pub(crate) fn add_plan_warnings<'py>( Ok(output) } +/// The unselected models the selected scope reads, sorted. +pub(crate) fn unselected_upstream_models(scopes: &Bound<'_, PyAny>) -> PyResult> { + let py = scopes.py(); + let selected = class_field(scopes, |s: &PlannerScopeResolution| &s.selected_scope)?; + let selected_keys = native_field(&selected, "selected_keys")?; + let upstream_deps = native_field(&selected, "upstream_deps")?; + let model_type = enum_member(py, RecordId::CompiledResourceType, "MODEL")?; + let mut names: Vec = Vec::new(); + for key in selected_keys.try_iter()? { + let Some(deps) = upstream_deps.downcast::()?.get_item(key?)? else { + continue; + }; + for dep in deps.try_iter()? { + let dep = dep?; + if native_field(&dep, "resource_type")?.eq(&model_type)? + && !selected_keys.contains(&dep)? + { + names.push(native_field(&dep, "name")?.extract()?); + } + } + } + names.sort(); + names.dedup(); + Ok(names) +} + +/// Note unselected upstreams deferred to pre-migration names, and each lookup in `failures`. +pub(crate) fn add_deferred_origin_notes<'py>( + mut output: PlanOutputFields<'py>, + (project, scopes): (&Bound<'py, PyAny>, &Bound<'py, PyAny>), + (deferral, failures): (&Bound<'py, PyAny>, &[(String, String)]), +) -> PyResult> { + let py = project.py(); + let locations = native_field(deferral, "deferred_locations")?; + if locations.is_none() { + return Ok(output); + } + let names: Vec = unselected_upstream_models(scopes)?; + let warning = |name: &str, message: &str| -> PyResult> { + native( + py, + RecordId::PlanWarning, + &[ + ("model_name", text(py, name)), + ( + "severity", + enum_member(py, RecordId::WarningSeverity, "WARNING")?, + ), + ("message", text(py, message)), + ], + ) + }; + let mut notes: Vec> = Vec::new(); + for model in native_field(project, "models")?.try_iter()? { + let model = model?; + let name: String = native_field(&model, "name")?.extract()?; + if names.binary_search(&name).is_err() { + continue; + } + let Some(location) = locations.downcast::()?.get_item(&name)? else { + continue; + }; + if native_field(&location, "name")? + .eq(native_field(&native_field(&model, "destination")?, "name")?)? + { + continue; + } + notes.push(warning( + &name, + &format!( + "deferred to pre-migration name {}: the deferred target has not built model '{name}' under its current name yet", + native_field(&location, "qualified_name")? + ), + )?); + } + for (name, message) in failures { + notes.push(warning(name, message)?); + } + if !notes.is_empty() { + let warnings = concat(py, &[output.field("warnings")?, tuple(py, notes)?])?; + output.set("warnings", warnings); + } + Ok(output) +} + fn source_names<'py>(items: &Bound<'py, PyAny>) -> PyResult> { let mut names: Vec<_> = Vec::new(); for item in items.try_iter()? { diff --git a/crates/sqlbuild-python/src/bindings/_helpers/result_output/documents.rs b/crates/sqlbuild-python/src/bindings/_helpers/result_output/documents.rs index a3ccfbe313..7227f41fe8 100644 --- a/crates/sqlbuild-python/src/bindings/_helpers/result_output/documents.rs +++ b/crates/sqlbuild-python/src/bindings/_helpers/result_output/documents.rs @@ -444,11 +444,12 @@ pub(crate) fn format_load_execution_json<'py>( /// Format direct clone command execution results as JSON. #[pyfunction] -#[pyo3(signature = (*, result, resource_types_by_name))] +#[pyo3(signature = (*, result, resource_types_by_name, phase_timings = None))] pub(crate) fn format_clone_execution_json<'py>( py: Python<'py>, result: Bound<'py, PyAny>, resource_types_by_name: Bound<'py, PyAny>, + phase_timings: Option>, ) -> PyResult { let projection: Projection<'py> = Projection::installed(py)?; let mut assets: Vec> = Vec::new(); @@ -482,7 +483,13 @@ pub(crate) fn format_clone_execution_json<'py>( assets, checks: Vec::new(), }, - Extras::default(), + Extras { + trailing: phase_timings + .into_iter() + .map(|timings| ("phase_timings", timings.into_any())) + .collect(), + ..Extras::default() + }, ) } diff --git a/crates/sqlbuild-python/src/bindings/_helpers/scenario_command/test_command.rs b/crates/sqlbuild-python/src/bindings/_helpers/scenario_command/test_command.rs index dc9607e854..700f55b4fe 100644 --- a/crates/sqlbuild-python/src/bindings/_helpers/scenario_command/test_command.rs +++ b/crates/sqlbuild-python/src/bindings/_helpers/scenario_command/test_command.rs @@ -8,9 +8,10 @@ use pyo3::types::{PyDict, PyDictMethods, PyString, PyTuple, PyTupleMethods}; use pyo3::{pyfunction, wrap_pyfunction}; use sqlbuild_core::cli_text::main::format_summary_footer::format_summary_footer; use sqlbuild_core::cli_text::models::{CliStyle, SummaryCount}; +use sqlbuild_core::scenario_command::main::render_scenario_failures::render_scenario_failures; use sqlbuild_core::scenario_command::main::validate_scenario_test_flags::validate_scenario_test_flags; use sqlbuild_core::scenario_command::models::{ - ScenarioNamespace, ScenarioTestFlags, SqlAnalysisCommand, + ScenarioNamespace, ScenarioResultRow, ScenarioTestFlags, SqlAnalysisCommand, }; use sqlbuild_host::entries::models::ExecutorHost; @@ -21,7 +22,7 @@ use crate::bindings::_helpers::native_host::host; use crate::bindings::_helpers::records::fields::field as native_field; use crate::bindings::_helpers::result_output::documents::format_scenario_execution_json; use crate::bindings::_helpers::result_output::documents::write_execution_json_output; -use crate::bindings::_helpers::scenario_command::conversions::succeeded; +use crate::bindings::_helpers::scenario_command::conversions::{scenario_result_row, succeeded}; use crate::bindings::_helpers::scenario_command::presentation::{ CompletionKind, RelayContext, SectionStatus, activity, begin_section, connection_hooks, progress_relay, write_completion, @@ -142,6 +143,14 @@ fn finish( format_summary_footer(run.style, &counts, Some(&format!("{elapsed_seconds:.2}s"))) ), )?; + let rows: Vec = results + .iter() + .map(|result| scenario_result_row(&result)) + .collect::>()?; + let failures: String = render_scenario_failures(&rows, run.style); + if !failures.is_empty() { + write(&run.stream, &format!("{failures}\n"))?; + } flush(&run.stream)?; write_completion(&run.stream, (&run.project.namespace, succeeded))?; let namespace: &ScenarioNamespace = &run.project.namespace; diff --git a/crates/sqlbuild-refactor/src/refactoring/_helpers/planning/model_planning.rs b/crates/sqlbuild-refactor/src/refactoring/_helpers/planning/model_planning.rs index 58177858f1..3808539d64 100644 --- a/crates/sqlbuild-refactor/src/refactoring/_helpers/planning/model_planning.rs +++ b/crates/sqlbuild-refactor/src/refactoring/_helpers/planning/model_planning.rs @@ -301,7 +301,7 @@ fn pending_migration( files: &[ProjectSqlFile], target: &ModelTarget<'_>, ) -> Result, RefactorError> { - if !target.model.declares_migrate_from { + if !target.model.declares_migrate_from || target.request.new_name == target.model.name { return Ok(Vec::new()); } let (contents, text) = files diff --git a/crates/sqlbuild-refactor/src/refactoring/main/with_model_migration.rs b/crates/sqlbuild-refactor/src/refactoring/main/with_model_migration.rs index 095a70d7ae..b9af4104a0 100644 --- a/crates/sqlbuild-refactor/src/refactoring/main/with_model_migration.rs +++ b/crates/sqlbuild-refactor/src/refactoring/main/with_model_migration.rs @@ -53,6 +53,21 @@ fn migrate( if !decision.needed { return migrated; } + if old == new + && before + .iter() + .any(|model| model.name == old && model.declares_migrate_from) + { + migrated.blocking.push(ManualLocation { + path: migrated.changes[index].path.clone(), + line: None, + column: None, + reason: format!( + "model:{old} still declares migrate_from and this move changes its relation; build it on every target and remove migrate_from before moving it again" + ), + }); + return migrated; + } let declaration = format!("{MIGRATE_FROM_KEY} {old}"); let change = &migrated.changes[index]; let contents = originals diff --git a/crates/sqlbuild-refactor/src/refactoring/tests/test_planning.rs b/crates/sqlbuild-refactor/src/refactoring/tests/test_planning.rs index 242a9ea7d7..8b5da21c18 100644 --- a/crates/sqlbuild-refactor/src/refactoring/tests/test_planning.rs +++ b/crates/sqlbuild-refactor/src/refactoring/tests/test_planning.rs @@ -1,6 +1,7 @@ use crate::refactoring::_helpers::planning::model_planning::decide_model_migration; use crate::refactoring::errors::{RefactorError, RefactorErrorKind}; -use crate::refactoring::models::{DeclarationPlacement, RefactorOperation}; +use crate::refactoring::main::with_model_migration::with_model_migration; +use crate::refactoring::models::{DeclarationPlacement, RefactorOperation, RefactorPlan}; use crate::refactoring::tests::helpers::{ blocking_reasons, change, commit_failure, facts, files_on_disk, followed_test_changes, followed_unit_tests, model, model_in_database, move_request, owned_pairs, placed, @@ -9,7 +10,8 @@ use crate::refactoring::tests::helpers::{ }; use crate::refactoring::tests::test_types::{ BatchPlanTestCase, CollisionTestCase, CommitTestCase, DeclarationMovesTestCase, - FollowedUnitTestTestCase, MigrationDecisionTestCase, RefusedTargetTestCase, + FollowedUnitTestTestCase, MigrationDecisionTestCase, PendingMigrationTestCase, + RefusedTargetTestCase, }; #[test] @@ -586,3 +588,109 @@ fn given_several_moves_when_planning_together_then_chains_pass_and_conflicts_blo } Ok(()) } + +#[test] +fn given_pending_migration_when_moving_under_same_name_then_only_relation_moves_block() +-> Result<(), RefactorError> { + let project = tempfile::tempdir().map_err(|error| RefactorError::value(error.to_string()))?; + write_files( + project.path(), + &[( + "models/orders.sql", + "MODEL (materialized table, migrate_from old_orders);\nSELECT 1", + )], + )?; + let mut pending = model("orders", Some("table"), "analytics"); + pending.declares_migrate_from = true; + pending.migrate_from_set = true; + let project_facts = facts(&project.path().to_string_lossy(), vec![pending.clone()]); + let plan = RefactorPlan { + request: request( + RefactorOperation::MoveModel, + "orders", + Some("models/finance/orders.sql"), + ), + batch: Vec::new(), + changes: vec![change( + "models/finance/orders.sql", + "models/orders.sql", + Vec::new(), + )], + manual: Vec::new(), + blocking: Vec::new(), + migrations: Vec::new(), + renamed_columns: Vec::new(), + help: None, + notes: Vec::new(), + }; + let test_cases = [ + PendingMigrationTestCase { + description: "a same-name move passes the pending-migration refusal", + request: request(RefactorOperation::MoveModel, "", Some("models/finance/")), + after: vec![pending.clone()], + expected_refusals: vec![], + expected_blocking: vec![], + expected_migration_count: 0, + }, + PendingMigrationTestCase { + description: "a rename is refused while migrate_from is pending", + request: request(RefactorOperation::RenameModel, "order_lines", None), + after: vec![pending.clone()], + expected_refusals: vec![ + "models/orders.sql: model:orders still declares migrate_from; build it on every target and remove migrate_from before renaming it again", + ], + expected_blocking: vec![], + expected_migration_count: 0, + }, + PendingMigrationTestCase { + description: "a same-name move that changes the relation is blocked", + request: request(RefactorOperation::MoveModel, "", Some("models/finance/")), + after: vec![ + model("orders", Some("table"), "finance"), + model("customers", Some("table"), "analytics"), + ], + expected_refusals: vec![], + expected_blocking: vec![ + "models/finance/orders.sql: model:orders still declares migrate_from and this move changes its relation; build it on every target and remove migrate_from before moving it again", + ], + expected_migration_count: 0, + }, + ]; + for test_case in test_cases { + let refusals = blocking_reasons(&project_facts, &test_case.request)?; + let planned = with_model_migration( + &plan, + std::slice::from_ref(&pending), + Some(&test_case.after), + &Vec::new(), + ); + + assert_eq!( + ( + refusals, + planned + .blocking + .iter() + .map(|item| format!("{}: {}", item.path, item.reason)) + .collect::>(), + planned.migrations.len(), + ), + ( + test_case + .expected_refusals + .iter() + .map(|item| (*item).to_owned()) + .collect::>(), + test_case + .expected_blocking + .iter() + .map(|item| (*item).to_owned()) + .collect::>(), + test_case.expected_migration_count, + ), + "{}", + test_case.description + ); + } + Ok(()) +} diff --git a/crates/sqlbuild-refactor/src/refactoring/tests/test_types.rs b/crates/sqlbuild-refactor/src/refactoring/tests/test_types.rs index fd757d21e4..84ffc994fb 100644 --- a/crates/sqlbuild-refactor/src/refactoring/tests/test_types.rs +++ b/crates/sqlbuild-refactor/src/refactoring/tests/test_types.rs @@ -143,6 +143,18 @@ pub(super) struct MigrationDecisionTestCase { pub(super) expected_decision: (bool, bool, &'static str), } +pub(super) struct PendingMigrationTestCase { + pub(super) description: &'static str, + pub(super) request: RefactorRequest, + /// Models after the move, for the relation check. + pub(super) after: Vec, + /// `path: reason` refusals before planning. + pub(super) expected_refusals: Vec<&'static str>, + /// `path: reason` blockers after planning the move. + pub(super) expected_blocking: Vec<&'static str>, + pub(super) expected_migration_count: usize, +} + pub(super) struct CommitTestCase { pub(super) description: &'static str, /// Files on disk before the commit. diff --git a/fensu.toml b/fensu.toml index ff2c76ed43..5680636ba3 100644 --- a/fensu.toml +++ b/fensu.toml @@ -1677,7 +1677,7 @@ pending_budgets = [ "crates/sqlbuild-python/src/bindings/_helpers/cli/namespace.rs=3", "crates/sqlbuild-python/src/bindings/_helpers/cli/observability.rs=21", "crates/sqlbuild-python/src/bindings/_helpers/cli/surrogates.rs=2", - "crates/sqlbuild-python/src/bindings/_helpers/clone/clone_command.rs=12", + "crates/sqlbuild-python/src/bindings/_helpers/clone/clone_command.rs=11", "crates/sqlbuild-python/src/bindings/_helpers/clone/clone_execution.rs=4", "crates/sqlbuild-python/src/bindings/_helpers/clone/clone_pipeline.rs=1", "crates/sqlbuild-python/src/bindings/_helpers/clone/clone_progress.rs=7", diff --git a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/clone.md b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/clone.md index ac85f9beec..dd95ba1219 100644 --- a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/clone.md +++ b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/clone.md @@ -25,6 +25,10 @@ function SQL from each target's project, so SQL analysis errors never stop it; s such as an unknown reference, still do. The hidden `--no-sql-analysis` flag (alias `--no-sql-validation`) is still accepted but has no effect. +With `--debug`, clone logs how long each phase took: compiling each target it reads, inspecting +sources, planning, and execution. The `--json-output` document reports the same durations in +seconds under `phase_timings`. + ## Usage ```bash diff --git a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/adapters.md b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/adapters.md index 35f105f766..759e8deb0a 100644 --- a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/adapters.md +++ b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/adapters.md @@ -131,4 +131,4 @@ StrictAdapter (fully abstract - all methods must be implemented) - **ConnectionMixin** - `connect`, `close`, `begin`, `commit`, `rollback` - **SchemaMixin** - `relation_exists`, `list_relations`, `get_columns`, `ensure_schema` - **MaterializationMixin** - `create_table_as`, `create_view_as`, `drop`, `rename`, `swap`, `load_seed` -- **DiffMixin** - `diff_schema`, `diff_rows`, `sample_unequal_rows`, `sample_side_only_rows` +- **DiffMixin** - `diff_schema`, `diff_rows`, `sample_unequal_rows`, `sample_side_only_rows`, `diff_unkeyed_rows` diff --git a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/diff.md b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/diff.md index 6a4b8531a1..3f2f8c2245 100644 --- a/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/diff.md +++ b/src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/diff.md @@ -154,6 +154,11 @@ For a model without a stable row key, opt in to exact full-row multiset comparis sqb diff prod:dev --full --select daily_order_totals --unkeyed ``` +Unkeyed diffs show sample complete rows that one side holds more often than the other, up to +`--max-row-only-examples` per side; a row duplicated three more times on one side appears three +times. In `--json`, they are `examples.from_only_rows` and `examples.to_only_rows`, and +`from_only_keys` and `to_only_keys` stay empty because there is no key. + The diff output categorises rows as: - **Equal** - same key, same values on both sides - **Unequal** - same key, different values (with per-column breakdown) diff --git a/src/sqlbuild/_native.pyi b/src/sqlbuild/_native.pyi index 5060fad237..fb65575531 100644 --- a/src/sqlbuild/_native.pyi +++ b/src/sqlbuild/_native.pyi @@ -97,6 +97,7 @@ from sqlbuild.executor.python_nodes.models import ( PythonNodeExecutionResult, ) from sqlbuild.executor.run.models import ( + MicrobatchBatchProgress, ModelExecutionResult, ) from sqlbuild.executor.scenario.models import ( @@ -3768,7 +3769,10 @@ def dlt_destination_arguments( dataset_name: str | None, ) -> tuple[str, dict[str, Any], dict[str, Any], str | None]: ... def format_clone_execution_json( - *, result: CloneExecutionResult, resource_types_by_name: Mapping[str, str] + *, + result: CloneExecutionResult, + resource_types_by_name: Mapping[str, str], + phase_timings: dict[str, float] | None = None, ) -> str: ... def format_test_execution_json(*, results: tuple[SqlTestExecutionResult, ...]) -> str: ... def format_audit_execution_json( @@ -5881,6 +5885,7 @@ class BuildProgressCallbacks: def elapsed(self) -> float: ... def on_node_start(self, *, name: str, resource_kind: ExecutionResourceKind) -> None: ... def on_sub_progress(self, message: str) -> None: ... + def on_batch_complete(self, progress: MicrobatchBatchProgress) -> None: ... def on_scheduler_state(self, state: SchedulerState) -> None: ... def on_statement_complete(self, telemetry: StatementExecutionTelemetry) -> None: ... def clear_transient_line(self) -> None: ... diff --git a/src/sqlbuild/adapter/contract/classes/base_adapter.py b/src/sqlbuild/adapter/contract/classes/base_adapter.py index 23c722f41b..8bf89cfc53 100644 --- a/src/sqlbuild/adapter/contract/classes/base_adapter.py +++ b/src/sqlbuild/adapter/contract/classes/base_adapter.py @@ -182,6 +182,61 @@ def diff_unkeyed_rows( ) -> RowDiffResult: """Compare complete-row multiplicities without a unique key.""" + _, joined_sql = self._unkeyed_multiset_sql( + connection=connection, + left=left, + right=right, + excluded_columns=excluded_columns, + cursor_column=cursor_column, + start_cursor=start_cursor, + end_cursor=end_cursor, + ) + sql: str = ( + f"{joined_sql} " + "SELECT " + "COALESCE(SUM(__left_count), 0), " + "COALESCE(SUM(__right_count), 0), " + "COALESCE(SUM(CASE WHEN __left_count < __right_count " + "THEN __left_count ELSE __right_count END), 0), " + "COALESCE(SUM(CASE WHEN __left_count > __right_count " + "THEN __left_count - __right_count ELSE 0 END), 0), " + "COALESCE(SUM(CASE WHEN __right_count > __left_count " + "THEN __right_count - __left_count ELSE 0 END), 0) " + "FROM __joined" + ) + row: tuple[Any, ...] = self.execute(connection=connection, sql=sql).fetchone() + left_count: int = int(row[0]) + right_count: int = int(row[1]) + equal_count: int = int(row[2]) + left_only_count: int = int(row[3]) + right_only_count: int = int(row[4]) + population_count: int = equal_count + left_only_count + right_only_count + return RowDiffResult( + left_count=left_count, + right_count=right_count, + joined_count=population_count, + equal_count=equal_count, + unequal_count=0, + left_only_count=left_only_count, + right_only_count=right_only_count, + population_count=population_count, + compared_count=population_count, + ) + + def _unkeyed_multiset_sql( + self, + *, + connection: Any, + left: str, + right: str, + excluded_columns: tuple[str, ...], + cursor_column: str | None, + start_cursor: CursorValue | None, + end_cursor: CursorValue | None, + carried_side: str | None = None, + ) -> tuple[tuple[ColumnInfo, ...], str]: + """Return the compared columns and the per-row multiplicity CTEs ending in ``__joined``.""" + left_columns: tuple[ColumnInfo, ...] = self.describe_relation( connection=connection, relation=left ) @@ -214,43 +269,22 @@ def diff_unkeyed_rows( end_cursor=end_cursor, ) where_clause: str = f" WHERE {cursor_filter}" if cursor_filter else "" - sql: str = ( + carried: str = "" + if carried_side is not None: + carried = "".join( + f"__{carried_side}.{self.render_identifier(column.name)} AS __c{index}, " + for index, column in enumerate(compared_columns) + ) + return compared_columns, ( f"WITH __left AS (SELECT {column_list}, COUNT(*) AS {multiplicity_column} " f"FROM {left}{where_clause} GROUP BY {column_list}), " f"__right AS (SELECT {column_list}, COUNT(*) AS {multiplicity_column} " f"FROM {right}{where_clause} GROUP BY {column_list}), " "__joined AS (SELECT " + f"{carried}" f"COALESCE(__left.{multiplicity_column}, 0) AS __left_count, " f"COALESCE(__right.{multiplicity_column}, 0) AS __right_count " - f"FROM __left FULL OUTER JOIN __right ON {equal_conditions}) " - "SELECT " - "COALESCE(SUM(__left_count), 0), " - "COALESCE(SUM(__right_count), 0), " - "COALESCE(SUM(CASE WHEN __left_count < __right_count " - "THEN __left_count ELSE __right_count END), 0), " - "COALESCE(SUM(CASE WHEN __left_count > __right_count " - "THEN __left_count - __right_count ELSE 0 END), 0), " - "COALESCE(SUM(CASE WHEN __right_count > __left_count " - "THEN __right_count - __left_count ELSE 0 END), 0) " - "FROM __joined" - ) - row: tuple[Any, ...] = self.execute(connection=connection, sql=sql).fetchone() - left_count: int = int(row[0]) - right_count: int = int(row[1]) - equal_count: int = int(row[2]) - left_only_count: int = int(row[3]) - right_only_count: int = int(row[4]) - population_count: int = equal_count + left_only_count + right_only_count - return RowDiffResult( - left_count=left_count, - right_count=right_count, - joined_count=population_count, - equal_count=equal_count, - unequal_count=0, - left_only_count=left_only_count, - right_only_count=right_only_count, - population_count=population_count, - compared_count=population_count, + f"FROM __left FULL OUTER JOIN __right ON {equal_conditions})" ) def physical_relation_generation( diff --git a/src/sqlbuild/adapter/contract/classes/unkeyed_diff.py b/src/sqlbuild/adapter/contract/classes/unkeyed_diff.py index b79f47a922..c2e8bb3151 100644 --- a/src/sqlbuild/adapter/contract/classes/unkeyed_diff.py +++ b/src/sqlbuild/adapter/contract/classes/unkeyed_diff.py @@ -5,6 +5,7 @@ from typing import Any, cast from sqlbuild.adapter.contract.classes.base_adapter import BaseAdapter +from sqlbuild.adapter.contract.constants import DIFF_LEFT_SIDE from sqlbuild.adapter.contract.models import CursorValue, RowDiffResult @@ -34,3 +35,67 @@ def diff_unkeyed_rows( start_cursor=start_cursor, end_cursor=end_cursor, ) + + def sample_unkeyed_side_only_rows( + self, + *, + connection: Any, + left: str, + right: str, + side: str, + excluded_columns: tuple[str, ...] = (), + cursor_column: str | None = None, + start_cursor: CursorValue | None = None, + end_cursor: CursorValue | None = None, + limit: int = 20, + ) -> tuple[tuple[tuple[str, object], ...], ...]: + """Return up to ``limit`` complete rows one side holds more often than the other.""" + + if limit <= 0: + return () + columns, joined_sql = BaseAdapter._unkeyed_multiset_sql( + cast(BaseAdapter, self), + connection=connection, + left=left, + right=right, + excluded_columns=excluded_columns, + cursor_column=cursor_column, + start_cursor=start_cursor, + end_cursor=end_cursor, + carried_side=side, + ) + surplus: str = ( + "__left_count - __right_count" + if side == DIFF_LEFT_SIDE + else "__right_count - __left_count" + ) + names: str = ", ".join(f"__c{index}" for index in range(len(columns))) + sql: str = self._render_unkeyed_sample_sql( + joined_sql=joined_sql, + select_list=f"{names}, {surplus}", + predicate=f"{surplus} > 0", + limit=limit, + ) + cursor: Any = cast(BaseAdapter, self).execute(connection=connection, sql=sql) + try: + fetched_rows: list[Any] = cursor.fetchall() + finally: + if cursor is not connection: + cursor.close() + rows: list[tuple[tuple[str, object], ...]] = [] + for fetched in fetched_rows: + values: tuple[Any, ...] = tuple(fetched) + row: tuple[tuple[str, object], ...] = tuple( + (column.name, value) for column, value in zip(columns, values[:-1], strict=True) + ) + rows.extend([row] * min(int(values[-1]), limit - len(rows))) + if len(rows) >= limit: + break + return tuple(rows) + + def _render_unkeyed_sample_sql( + self, *, joined_sql: str, select_list: str, predicate: str, limit: int + ) -> str: + """Render the bounded side-only sample query for LIMIT-capable engines.""" + + return f"{joined_sql} SELECT {select_list} FROM __joined WHERE {predicate} LIMIT {limit}" diff --git a/src/sqlbuild/adapters/sqlserver/classes/sqlserver_adapter.py b/src/sqlbuild/adapters/sqlserver/classes/sqlserver_adapter.py index a89f7d79df..0acea97875 100644 --- a/src/sqlbuild/adapters/sqlserver/classes/sqlserver_adapter.py +++ b/src/sqlbuild/adapters/sqlserver/classes/sqlserver_adapter.py @@ -2885,6 +2885,13 @@ def _render_row_diff_sample_keys_sql( f"ORDER BY {hash_expression}, {tie_break}" ) + def _render_unkeyed_sample_sql( + self, *, joined_sql: str, select_list: str, predicate: str, limit: int + ) -> str: + """Render the bounded SQL Server side-only sample query.""" + + return f"{joined_sql} SELECT TOP ({limit}) {select_list} FROM __joined WHERE {predicate}" + def _render_row_diff_key_hash( self, *, diff --git a/src/sqlbuild/executor/build/classes/build_scheduler.py b/src/sqlbuild/executor/build/classes/build_scheduler.py index 5ebd3dae3b..7170b1956e 100644 --- a/src/sqlbuild/executor/build/classes/build_scheduler.py +++ b/src/sqlbuild/executor/build/classes/build_scheduler.py @@ -102,6 +102,7 @@ from sqlbuild.executor.run.models import ( BatchWindow, BoundViewGuard, + MicrobatchBatchProgress, MicrobatchPhaseOutcome, ModelExecutionResult, ModelMaterializationContext, @@ -219,6 +220,9 @@ def __init__( self._effective_vars: dict[str, object] = runtime.effective_vars or {} self._warehouse_relations: dict[str, RelationInfo] = initial_state.warehouse_relations or {} self._on_sub_progress: Callable[[str], None] | None = callbacks.on_sub_progress + self._on_batch_complete: Callable[[MicrobatchBatchProgress], None] | None = ( + callbacks.on_batch_complete + ) self._on_statement_complete: Callable[[StatementExecutionTelemetry], None] | None = ( callbacks.on_statement_complete ) @@ -1120,6 +1124,7 @@ def _execute_model_node( microbatch_lease_check=self._runtime.microbatch_lease_check, microbatch_global_concurrency=self._max_concurrency, microbatch_batch_runner=microbatch_batch_runner, + on_batch_complete=self._on_batch_complete, watermark_resolver=self._watermark_resolver, bound_views=bound_views, ), diff --git a/src/sqlbuild/executor/build/models.py b/src/sqlbuild/executor/build/models.py index 818b7d9eaf..1a0cedf690 100644 --- a/src/sqlbuild/executor/build/models.py +++ b/src/sqlbuild/executor/build/models.py @@ -27,7 +27,7 @@ from sqlbuild.executor.custom.models import MaterializationResult from sqlbuild.executor.load.models import LoadExecutionResult from sqlbuild.executor.python_nodes.types import PythonIdentityRecorder -from sqlbuild.executor.run.models import ModelExecutionResult +from sqlbuild.executor.run.models import MicrobatchBatchProgress, ModelExecutionResult from sqlbuild.executor.scheduling.types import ExecutionStatus from sqlbuild.executor.testing.models import SqlTestExecutionResult from sqlbuild.microbatches.models import MicrobatchScope @@ -104,6 +104,7 @@ class BuildCallbacks: on_progress: Callable[[str], None] | None = None on_sub_progress: Callable[[str], None] | None = None + on_batch_complete: Callable[[MicrobatchBatchProgress], None] | None = None on_node_start: NodeStartCallback | None = None on_node_complete: Callable[[object], None] | None = None before_model_materialize: BeforeModelMaterializeCallback | None = None diff --git a/src/sqlbuild/executor/diff/_helpers/execution.py b/src/sqlbuild/executor/diff/_helpers/execution.py index fea5d32897..96de9b6fd2 100644 --- a/src/sqlbuild/executor/diff/_helpers/execution.py +++ b/src/sqlbuild/executor/diff/_helpers/execution.py @@ -11,6 +11,8 @@ resolve_row_diff_sampling, ) from sqlbuild.adapter.contract.classes.base_adapter import BaseAdapter +from sqlbuild.adapter.contract.classes.unkeyed_diff import UnkeyedDiffMixin +from sqlbuild.adapter.contract.constants import DIFF_LEFT_SIDE, DIFF_RIGHT_SIDE from sqlbuild.adapter.contract.models import ( ColumnInfo, RowDiffCoverage, @@ -207,6 +209,14 @@ def execute_query_diff( right=right_relation, excluded_columns=excluded_columns, ) + left_only_rows, right_only_rows = _unkeyed_side_only_samples( + adapter=adapter, + connection=connection, + relations=(left_relation, right_relation), + row_result=row_result, + options=options, + excluded_columns=excluded_columns, + ) return replace( result, row_result=replace( @@ -214,6 +224,8 @@ def execute_query_diff( left_coverage=left_coverage, right_coverage=right_coverage, ), + left_only_key_samples=left_only_rows, + right_only_key_samples=right_only_rows, ) sampling: RowDiffSampling | None = resolve_row_diff_sampling( raw_row_limit=None, @@ -424,6 +436,15 @@ def _execute_model_row_diff( start_cursor=start_cursor, end_cursor=end_cursor, ) + unkeyed_left_rows, unkeyed_right_rows = _unkeyed_side_only_samples( + adapter=adapter, + connection=connection, + relations=(left_relation, right_relation), + row_result=row_result, + options=options, + excluded_columns=excluded_columns, + cursor=(cursor_column, start_cursor, end_cursor), + ) return ( replace( row_result, @@ -434,8 +455,8 @@ def _execute_model_row_diff( cursor_end=end_cursor.value if end_cursor is not None else None, ), (), - (), - (), + unkeyed_left_rows, + unkeyed_right_rows, bounded_fallback, excluded_columns, ) @@ -525,6 +546,44 @@ def _execute_model_row_diff( ) +def _unkeyed_side_only_samples( + *, + adapter: BaseAdapter, + connection: Any, + relations: tuple[str, str], + row_result: RowDiffResult, + options: DiffExecutionOptions, + excluded_columns: tuple[str, ...], + cursor: tuple[str | None, Any | None, Any | None] = (None, None, None), +) -> tuple[tuple[tuple[tuple[str, object], ...], ...], tuple[tuple[tuple[str, object], ...], ...]]: + """Return bounded samples of the complete rows each side holds more often than the other.""" + + samples: list[tuple[tuple[tuple[str, object], ...], ...]] = [] + side: str + count: int + for side, count in ( + (DIFF_LEFT_SIDE, row_result.left_only_count), + (DIFF_RIGHT_SIDE, row_result.right_only_count), + ): + if not options.collect_samples or count <= 0 or not isinstance(adapter, UnkeyedDiffMixin): + samples.append(()) + continue + samples.append( + adapter.sample_unkeyed_side_only_rows( + connection=connection, + left=relations[0], + right=relations[1], + side=side, + excluded_columns=excluded_columns, + cursor_column=cursor[0], + start_cursor=cursor[1], + end_cursor=cursor[2], + limit=options.max_row_only_examples, + ) + ) + return samples[0], samples[1] + + def _model_has_unique_key(model: Any) -> bool: raw: object | None = model.config.values.get("unique_key") if isinstance(raw, str): diff --git a/src/sqlbuild/executor/run/_helpers/materializations/microbatch.py b/src/sqlbuild/executor/run/_helpers/materializations/microbatch.py index 422e775eda..acc24fc55f 100644 --- a/src/sqlbuild/executor/run/_helpers/materializations/microbatch.py +++ b/src/sqlbuild/executor/run/_helpers/materializations/microbatch.py @@ -112,6 +112,7 @@ FinalAuditRun, FullRefreshRelations, MicrobatchAccountingInterval, + MicrobatchBatchProgress, MicrobatchLifecycleState, MicrobatchPhaseOutcome, MicrobatchSchemaPhaseOutcome, @@ -1188,11 +1189,22 @@ def _execute_microbatch_batches( raise lifecycle.completed(affected_rows=dml_result if isinstance(dml_result, int) else None) completed_batches += 1 + batch_elapsed: float = time.monotonic() - batch_start_time if on_progress is not None: - batch_elapsed: float = time.monotonic() - batch_start_time on_progress( f"batch {batch.index + 1}/{plan.batch_count} {display_window} {batch_elapsed:.1f}s" ) + if context.on_batch_complete is not None: + context.on_batch_complete( + MicrobatchBatchProgress( + model_name=context.entry.name, + batch_number=batch.index + 1, + batch_count=plan.batch_count, + window=display_window, + rows_affected=dml_result if isinstance(dml_result, int) else None, + elapsed_seconds=batch_elapsed, + ) + ) return MicrobatchPhaseOutcome( state=state, completed_batches=completed_batches, diff --git a/src/sqlbuild/executor/run/models.py b/src/sqlbuild/executor/run/models.py index 598dc135be..9b82f68de4 100644 --- a/src/sqlbuild/executor/run/models.py +++ b/src/sqlbuild/executor/run/models.py @@ -284,6 +284,18 @@ class BoundViewGuard: rebind: Callable[[], int] = int +@dataclass(frozen=True) +class MicrobatchBatchProgress: + """One completed microbatch, for progress output that keeps a line per batch.""" + + model_name: str + batch_number: int + batch_count: int + window: str + rows_affected: int | None + elapsed_seconds: float + + @dataclass(frozen=True) class ModelMaterializationContext: """Shared runtime inputs for one model's materialization lifecycle.""" @@ -312,6 +324,7 @@ class ModelMaterializationContext: microbatch_lease_check: Callable[[], None] | None = None microbatch_global_concurrency: int = 1 microbatch_batch_runner: MicrobatchBatchRunner | None = None + on_batch_complete: Callable[[MicrobatchBatchProgress], None] | None = None watermark_resolver: WatermarkResolver | None = None enforce_explicit_references: bool = False bound_views: BoundViewGuard = field(default_factory=BoundViewGuard) diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/build/_test_types.py b/tests/e2e/src/sqlbuild/cli/commands/main/build/_test_types.py index 19826f9db5..262e84be58 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/build/_test_types.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/build/_test_types.py @@ -2,6 +2,8 @@ from __future__ import annotations +import subprocess +from collections.abc import Callable from dataclasses import dataclass, field @@ -1388,3 +1390,22 @@ class TargetMicrobatchWarningE2ETestCase: description: str command: tuple[str, ...] expected_warning_count: int + + +@dataclass(frozen=True) +class MicrobatchBatchProgressE2ETestCase: + """Per-batch microbatch progress lines for captured and terminal output.""" + + description: str + run: Callable[..., subprocess.CompletedProcess[str]] + expected_batches: tuple[tuple[str, str, str], ...] + + +@dataclass(frozen=True) +class DeferredMigrationOriginE2ETestCase: + """A deferred reference to a renamed model under different deferred-target states.""" + + description: str + prod_sql: str + expected_order_ids: tuple[int, ...] + expect_note: bool diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/build/test_defer_migration_origin.py b/tests/e2e/src/sqlbuild/cli/commands/main/build/test_defer_migration_origin.py new file mode 100644 index 0000000000..7c567d49b9 --- /dev/null +++ b/tests/e2e/src/sqlbuild/cli/commands/main/build/test_defer_migration_origin.py @@ -0,0 +1,113 @@ +"""Real-CLI coverage of deferred references to a model renamed with migrate_from.""" + +from __future__ import annotations + +import subprocess +from pathlib import Path + +import pytest + +from tests.e2e.src.sqlbuild.cli.commands.main.build._test_types import ( + DeferredMigrationOriginE2ETestCase, +) +from tests.e2e.src.sqlbuild.cli.commands.shared.helpers import ( + execute_duckdb, + prepare_inline_project, + query_duckdb, + run_sqb, +) + +_DATABASE: str = "warehouse.duckdb" +_PROJECT_FILES: dict[str, str] = { + "sqlbuild_project.toml": ( + 'name = "deferred_renames"\nadapter = "duckdb"\ndefault_target = "dev"\n\n' + f'[connection]\ndatabase = "{_DATABASE}"\n\n' + '[targets.dev]\nschema = "dev"\n\n' + '[targets.prod]\nschema = "preserve"\n' + ), + "models/order_ledger.sql": ( + 'MODEL (description "Test model order_ledger.", materialized table, ' + "schema prod, migrate_from order_history);\n\nSELECT 1 AS order_id\n" + ), + "models/order_report.sql": ( + 'MODEL (description "Test model order_report.", materialized table, schema prod);\n\n' + 'SELECT order_id FROM __ref("order_ledger")\n' + ), + "models/order_archive.sql": ( + 'MODEL (description "Test model order_archive.", materialized table, schema sales);\n\n' + "SELECT 1 AS order_id\n" + ), +} +_PRE_MIGRATION_NOTE: str = "deferred to pre-migration name prod.order_history" + + +@pytest.mark.parametrize( + "test_case", + [ + DeferredMigrationOriginE2ETestCase( + description="rename not yet deployed reads the pre-migration relation", + prod_sql="CREATE TABLE prod.order_history AS SELECT 7 AS order_id;", + expected_order_ids=(7,), + expect_note=True, + ), + DeferredMigrationOriginE2ETestCase( + description="deployed rename reads the current relation", + prod_sql=( + "CREATE TABLE prod.order_history AS SELECT 7 AS order_id;" + "CREATE TABLE prod.order_ledger AS SELECT 8 AS order_id;" + ), + expected_order_ids=(8,), + expect_note=False, + ), + DeferredMigrationOriginE2ETestCase( + description="bare pre-migration name resolves in the model's own deferred schema", + prod_sql=( + "CREATE SCHEMA sales;" + "CREATE TABLE prod.order_history AS SELECT 7 AS order_id;" + "CREATE TABLE sales.order_history AS SELECT 5 AS order_id;" + ), + expected_order_ids=(7,), + expect_note=True, + ), + DeferredMigrationOriginE2ETestCase( + description="current name built only in another deferred schema is not the model", + prod_sql=( + "CREATE SCHEMA sales;" + "CREATE TABLE prod.order_history AS SELECT 7 AS order_id;" + "CREATE TABLE sales.order_ledger AS SELECT 9 AS order_id;" + ), + expected_order_ids=(7,), + expect_note=True, + ), + ], + ids=lambda case: case.description, +) +def test_given_renamed_upstream_when_building_with_defer_to_then_reads_the_relation_prod_has( + test_case: DeferredMigrationOriginE2ETestCase, + tmp_path: Path, +) -> None: + project_dir: Path = prepare_inline_project( + tmp_path=tmp_path, project_name="deferred_renames", repo_files=_PROJECT_FILES + ) + database: Path = project_dir / _DATABASE + execute_duckdb(db_path=database, sql="CREATE SCHEMA prod;" + test_case.prod_sql) + + result: subprocess.CompletedProcess[str] = run_sqb( + command=("--no-color", "build", "--select", "order_report", "--defer-to", "prod"), + project_dir=project_dir, + ) + + output: str = result.stdout + result.stderr + assert result.returncode == 0, output + assert (_PRE_MIGRATION_NOTE in output) is test_case.expect_note, output + assert ( + tuple( + row[0] + for row in query_duckdb(db_path=database, sql="SELECT order_id FROM dev.order_report") + ) + == test_case.expected_order_ids + ) + + +if __name__ == "__main__": + pytest.main([__file__, "-vv"]) diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/build/test_microbatch_batch_progress.py b/tests/e2e/src/sqlbuild/cli/commands/main/build/test_microbatch_batch_progress.py new file mode 100644 index 0000000000..fbc81f53d2 --- /dev/null +++ b/tests/e2e/src/sqlbuild/cli/commands/main/build/test_microbatch_batch_progress.py @@ -0,0 +1,96 @@ +"""Real-CLI coverage of per-batch microbatch progress when stdout is not a terminal.""" + +from __future__ import annotations + +import re +import subprocess +from pathlib import Path +from textwrap import dedent + +import pytest + +from tests.e2e.src.sqlbuild.cli.commands.main.build._test_types import ( + MicrobatchBatchProgressE2ETestCase, +) +from tests.e2e.src.sqlbuild.cli.commands.shared.helpers import ( + execute_duckdb, + prepare_inline_project, + run_sqb, + run_sqb_with_pty, +) + +_DATABASE: str = "warehouse.duckdb" +_PROJECT_FILES: dict[str, str] = { + "sqlbuild_project.toml": f'name = "batch_progress"\nadapter = "duckdb"\n\n[connection]\ndatabase = "{_DATABASE}"\n', + "sources/raw.yml": ( + "sources:\n - name: raw_orders\n description: Test source raw_orders.\n" + " schema: main\n table: raw_orders\n" + ), + "models/daily_orders.sql": dedent( + """ + MODEL (description "Test model daily_orders.", + materialized incremental, + incremental_strategy delete_insert, + incremental_mode microbatch, + microbatch_strategy watermark, + cursor_watermark_mode all, + cursor order_date, + cursor_type timestamp, + cursor_grain day, + cursor_start '2026-07-01', + cursor_inputs ( + raw_orders (column order_date, roles [filter, watermark]), + ), + batch_size 1d, + ); + SELECT order_id, order_date FROM __source("raw_orders") + """ + ).strip() + + "\n", +} +_RAW_ORDERS: str = ( + "CREATE TABLE raw_orders (order_id INTEGER, order_date TIMESTAMP);" + "INSERT INTO raw_orders VALUES (1, '2026-07-01 08:00:00'), (2, '2026-07-01 09:00:00')," + " (3, '2026-07-02 10:00:00'), (4, '2026-07-03 11:00:00');" +) +_BATCH_LINE: re.Pattern[str] = re.compile( + r"^Batch daily_orders (\d+)/(\d+) \S+ (\d+) rows \d+\.\ds$", re.MULTILINE +) + + +@pytest.mark.parametrize( + "test_case", + [ + MicrobatchBatchProgressE2ETestCase( + description="captured stdout keeps one line per batch with its rows", + run=run_sqb, + expected_batches=(("1", "3", "2"), ("2", "3", "1"), ("3", "3", "1")), + ), + MicrobatchBatchProgressE2ETestCase( + description="a terminal keeps batch progress on the live row only", + run=run_sqb_with_pty, + expected_batches=(), + ), + ], + ids=lambda case: case.description, +) +def test_given_multi_batch_microbatch_when_building_then_batch_lines_follow_the_output_kind( + test_case: MicrobatchBatchProgressE2ETestCase, + tmp_path: Path, +) -> None: + project_dir: Path = prepare_inline_project( + tmp_path=tmp_path, project_name="batch_progress", repo_files=_PROJECT_FILES + ) + execute_duckdb(db_path=project_dir / _DATABASE, sql=_RAW_ORDERS) + + result: subprocess.CompletedProcess[str] = test_case.run( + command=("--no-color", "build"), project_dir=project_dir + ) + + output: str = result.stdout.replace("\r\n", "\n") + assert result.returncode == 0, output + result.stderr + assert tuple(_BATCH_LINE.findall(output)) == test_case.expected_batches, output + + +if __name__ == "__main__": + pytest.main([__file__, "-vv"]) diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/clone/_test_types.py b/tests/e2e/src/sqlbuild/cli/commands/main/clone/_test_types.py index ed85edd284..b1ca648fe1 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/clone/_test_types.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/clone/_test_types.py @@ -66,3 +66,11 @@ class ClonePreparationFailureE2ETestCase: broken_model_path: str broken_model_sql: str expected_screen_fragments: tuple[str, ...] + + +@dataclass(frozen=True) +class CloneTimingsE2ETestCase: + """The clone phases a debug run reports, in order.""" + + description: str + expected_phases: tuple[str, ...] diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/clone/test_clone_timings.py b/tests/e2e/src/sqlbuild/cli/commands/main/clone/test_clone_timings.py new file mode 100644 index 0000000000..d25df752cc --- /dev/null +++ b/tests/e2e/src/sqlbuild/cli/commands/main/clone/test_clone_timings.py @@ -0,0 +1,88 @@ +"""Real-CLI coverage of clone phase timings in --debug output and the JSON document.""" + +from __future__ import annotations + +import json +import subprocess +from pathlib import Path +from typing import Any + +import pytest + +from tests.e2e.src.sqlbuild.cli.commands.main.clone._test_types import CloneTimingsE2ETestCase +from tests.e2e.src.sqlbuild.cli.commands.shared.helpers import ( + execute_duckdb, + prepare_inline_project, + run_sqb, +) + +_PROJECT_FILES: dict[str, str] = { + "sqlbuild_project.toml": ( + 'name = "clone_timings"\nadapter = "duckdb"\ndefault_target = "dev"\n\n' + '[connection]\ndatabase = "clone.duckdb"\n\n' + '[targets.prod]\nschema = "prod"\n\n' + "[targets.prod.clone]\nallow_as_clone_origin = true\n\n" + '[targets.dev]\nschema = "dev"\n\n' + "[targets.dev.clone]\nallow_as_clone_destination = true\n" + ), + "models/fact_orders.sql": ( + 'MODEL (description "Test model fact_orders.", materialized table);\n\n' + "SELECT 1 AS order_id\n" + ), +} + + +@pytest.mark.parametrize( + "test_case", + [ + CloneTimingsE2ETestCase( + description="debug and JSON name every clone phase", + expected_phases=( + "compile_prod", + "compile_dev", + "source_inspection", + "planning", + "execution", + ), + ) + ], + ids=lambda case: case.description, +) +def test_given_clone_when_debugging_then_reports_each_phase_duration( + test_case: CloneTimingsE2ETestCase, + tmp_path: Path, +) -> None: + project_dir: Path = prepare_inline_project( + tmp_path=tmp_path, project_name="clone_timings", repo_files=_PROJECT_FILES + ) + execute_duckdb( + db_path=project_dir / "clone.duckdb", + sql="CREATE SCHEMA prod; CREATE SCHEMA dev; CREATE TABLE prod.fact_orders AS SELECT 1 AS order_id;", + ) + document_path: Path = tmp_path / "clone.json" + + result: subprocess.CompletedProcess[str] = run_sqb( + command=( + "--no-color", + "--debug", + "clone", + "--from", + "prod", + "--json-output", + str(document_path), + ), + project_dir=project_dir, + ) + + output: str = result.stdout + result.stderr + assert result.returncode == 0, output + document: dict[str, Any] = json.loads(document_path.read_text(encoding="utf-8")) + timings: dict[str, float] = document["phase_timings"] + assert tuple(timings) == test_case.expected_phases + assert all(seconds >= 0 for seconds in timings.values()) + for phase in test_case.expected_phases: + assert f"Clone timing {phase} " in output, output + + +if __name__ == "__main__": + pytest.main([__file__, "-vv"]) diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/diff/_test_types.py b/tests/e2e/src/sqlbuild/cli/commands/main/diff/_test_types.py index 2a7c7a25ca..d8f02ab3f0 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/diff/_test_types.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/diff/_test_types.py @@ -96,3 +96,15 @@ class DefaultFullDiffE2ETestCase: unexpected_stdout_fragments: tuple[str, ...] = field(default_factory=tuple) expected_json_status: str | None = None expected_json_blocked_models: tuple[str, ...] = field(default_factory=tuple) + + +@dataclass(frozen=True) +class UnkeyedExamplesE2ETestCase: + """An unkeyed diff and the order IDs of the side-only example rows it reports.""" + + description: str + mutation_sql: tuple[str, ...] + command: tuple[str, ...] + expected_from_order_ids: tuple[int, ...] + expected_to_order_ids: tuple[int, ...] + expected_truncated: bool diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/diff/test_unkeyed_examples.py b/tests/e2e/src/sqlbuild/cli/commands/main/diff/test_unkeyed_examples.py new file mode 100644 index 0000000000..c5c056580c --- /dev/null +++ b/tests/e2e/src/sqlbuild/cli/commands/main/diff/test_unkeyed_examples.py @@ -0,0 +1,94 @@ +"""Real-CLI coverage of side-only example rows in unkeyed diffs.""" + +from __future__ import annotations + +import json +import subprocess +from pathlib import Path +from typing import Any + +import pytest + +from tests.e2e.src.sqlbuild.cli.commands.main.diff._test_types import ( + UnkeyedExamplesE2ETestCase, +) +from tests.e2e.src.sqlbuild.cli.commands.main.diff.helpers import ( + build_both_environments, + execute_duckdb, + prepare_diff_project, +) +from tests.e2e.src.sqlbuild.cli.commands.shared.helpers import run_sqb + + +@pytest.mark.parametrize( + "test_case", + [ + UnkeyedExamplesE2ETestCase( + description="model diff shows the dropped row and the duplicated row", + mutation_sql=( + "DELETE FROM dev.orders_snapshot WHERE order_id = 1", + "INSERT INTO dev.orders_snapshot SELECT * FROM dev.orders_snapshot WHERE order_id = 2", + ), + command=( + "diff", + "prod:dev", + "--full", + "--unkeyed", + "--select", + "orders_snapshot", + "--exclude-column", + "status", + ), + expected_from_order_ids=(1,), + expected_to_order_ids=(2,), + expected_truncated=False, + ), + UnkeyedExamplesE2ETestCase( + description="query diff bounds examples by max-row-only-examples", + mutation_sql=(), + command=( + "diff", + "--left-query", + "SELECT * FROM (VALUES (1), (1), (1), (2)) AS orders(order_id)", + "--right-query", + "SELECT * FROM (VALUES (2)) AS orders(order_id)", + "--unkeyed", + "--max-row-only-examples", + "2", + ), + expected_from_order_ids=(1, 1), + expected_to_order_ids=(), + expected_truncated=True, + ), + ], + ids=lambda case: case.description, +) +def test_given_side_only_rows_when_diffing_unkeyed_then_json_shows_bounded_full_rows( + test_case: UnkeyedExamplesE2ETestCase, + tmp_path: Path, +) -> None: + project_dir: Path = prepare_diff_project(tmp_path) + build_both_environments(project_dir=project_dir) + for mutation_sql in test_case.mutation_sql: + execute_duckdb(db_path=project_dir / "diff.duckdb", sql=mutation_sql) + + result: subprocess.CompletedProcess[str] = run_sqb( + command=(*test_case.command, "--json"), project_dir=project_dir + ) + + assert result.returncode == 1, result.stdout + result.stderr + (model,) = json.loads(result.stdout)["models"] + examples: dict[str, Any] = model["examples"] + + def order_ids(rows: list[dict[str, Any]]) -> tuple[int, ...]: + return tuple(row["order_id"] for row in rows) + + assert examples["from_only_keys"] == [] + assert examples["to_only_keys"] == [] + assert order_ids(examples["from_only_rows"]) == test_case.expected_from_order_ids + assert order_ids(examples["to_only_rows"]) == test_case.expected_to_order_ids + assert examples["truncated_by_count"] is test_case.expected_truncated + + +if __name__ == "__main__": + pytest.main([__file__, "-vv"]) diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/refactor/_test_types.py b/tests/e2e/src/sqlbuild/cli/commands/main/refactor/_test_types.py index b1b0c4960f..ee1dd6d081 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/refactor/_test_types.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/refactor/_test_types.py @@ -29,6 +29,15 @@ class ModelMigrationE2ETestCase: stripped_fingerprints: tuple[str, ...] = () +@dataclass(frozen=True) +class PendingMigrationRefactorE2ETestCase: + description: str + command: tuple[str, ...] + expected_returncode: int + expected_file: str + expected_output: str + + @dataclass(frozen=True) class ColumnRenameE2ETestCase: description: str diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/refactor/test_rename_models.py b/tests/e2e/src/sqlbuild/cli/commands/main/refactor/test_rename_models.py index ee5db6bb35..e7e790d8fc 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/refactor/test_rename_models.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/refactor/test_rename_models.py @@ -12,6 +12,7 @@ DryRunE2ETestCase, ModelMigrationE2ETestCase, ModelRefactorE2ETestCase, + PendingMigrationRefactorE2ETestCase, ) from tests.e2e.src.sqlbuild.cli.commands.main.refactor.helpers import ( CUSTOMER_DECLARATIONS, @@ -266,3 +267,51 @@ def test_given_built_model_when_renamed_then_build_migrates_its_relation( test_case.expected_old_name_type ) assert order_ids(project_dir=project_dir, name="order_ledger") == test_case.expected_order_ids + + +@pytest.mark.parametrize( + "test_case", + [ + PendingMigrationRefactorE2ETestCase( + description="move to another folder under the same name carries migrate_from", + command=("mv", "model:order_ledger", "models/reporting/"), + expected_returncode=0, + expected_file="models/reporting/order_ledger.sql", + expected_output="Compiled: ok", + ), + PendingMigrationRefactorE2ETestCase( + description="rename on top of an undeployed migration is refused", + command=("rename", "model:order_ledger", "order_journal"), + expected_returncode=1, + expected_file="models/marts/order_ledger.sql", + expected_output="still declares migrate_from", + ), + ], + ids=lambda case: case.description, +) +def test_given_model_with_pending_migration_when_refactored_again_then_only_renames_are_refused( + tmp_path: Path, test_case: PendingMigrationRefactorE2ETestCase +) -> None: + """A same-name folder move keeps the relation, so the undeployed migrate_from moves with it.""" + + project_dir: Path = write_orders_project( + tmp_path=tmp_path, files=order_history_files(materialized="table", udf=False) + ) + first: subprocess.CompletedProcess[str] = sqb(project_dir, "build") + renamed: subprocess.CompletedProcess[str] = sqb( + project_dir, "rename", "model:order_history", "order_ledger" + ) + + result: subprocess.CompletedProcess[str] = sqb(project_dir, *test_case.command) + header: str = (project_dir / test_case.expected_file).read_text(encoding="utf-8") + load_raw_orders(project_dir=project_dir, order_ids=(3, 4)) + second: subprocess.CompletedProcess[str] = sqb(project_dir, "build") + + assert first.returncode == 0, first.stdout + first.stderr + assert renamed.returncode == 0, renamed.stdout + renamed.stderr + assert result.returncode == test_case.expected_returncode, result.stdout + result.stderr + assert test_case.expected_output in result.stdout + result.stderr + assert header.count("migrate_from order_history") == 1, header + assert second.returncode == 0, second.stdout + second.stderr + assert relation_type(project_dir=project_dir, name="order_history") == "VIEW" + assert order_ids(project_dir=project_dir, name="order_ledger") == (3, 4) diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/scenario/_test_types.py b/tests/e2e/src/sqlbuild/cli/commands/main/scenario/_test_types.py index 8906a34d34..2841c03b13 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/scenario/_test_types.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/scenario/_test_types.py @@ -296,3 +296,12 @@ class ScenarioHelperShadowingE2ETestCase: check_ctes: str expected_output_fragment: str unexpected_output_fragment: str + + +@dataclass(frozen=True) +class UnmockedInputScenarioE2ETestCase: + """A scenario that mocks one model input and leaves another, source-backed, real.""" + + description: str + expected_fragments: tuple[str, ...] + unexpected_fragment: str diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/scenario/helpers.py b/tests/e2e/src/sqlbuild/cli/commands/main/scenario/helpers.py index a491cd5d19..94f157084d 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/scenario/helpers.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/scenario/helpers.py @@ -937,3 +937,56 @@ def build_helper_scenario_project_files(*, helper_ctes: str, check_ctes: str) -> f"{_EXPECTED_COLUMNS_FIXTURE}{helper_ctes},\n\n{check_ctes}\n\nSELECT 1\n" ), } + + +_UNMOCKED_REP_SOURCES: tuple[str, ...] = tuple(f"raw_reps_{index}" for index in range(8)) + + +def build_unmocked_input_scenario_project_files() -> dict[str, str]: + """Build scenarios that mock one input of order_reps and leave a source-backed one real.""" + + files: dict[str, str] = { + "sqlbuild_project.toml": ( + 'name = "unmocked_inputs"\nadapter = "duckdb"\n\n' + '[connection]\ndatabase = "unmocked_inputs.duckdb"\n' + ), + "sources/raw.yml": "sources:\n" + + "".join( + f" - name: {name}\n description: Test source {name}.\n" + f" schema: main\n table: {name}\n" + for name in ("raw_orders", *_UNMOCKED_REP_SOURCES) + ), + "models/stg_orders.sql": ( + 'MODEL (description "Test model stg_orders.");\n\n' + 'SELECT order_id, rep_id FROM __source("raw_orders")\n' + ), + "models/stg_sales_reps.sql": ( + 'MODEL (description "Test model stg_sales_reps.");\n\n' + + "\nUNION ALL\n".join( + f'SELECT rep_id, rep_name FROM __source("{name}")' for name in _UNMOCKED_REP_SOURCES + ) + + "\n" + ), + "models/order_reps.sql": ( + 'MODEL (description "Test model order_reps.");\n\n' + "SELECT o.order_id, r.rep_name\n" + 'FROM __ref("stg_orders") AS o\n' + 'JOIN __ref("stg_sales_reps") AS r ON o.rep_id = r.rep_id\n' + ), + "tests/scenarios/order_reps_ok.sql": ( + "SCENARIO (description 'Test scenario order_reps_ok.');\n\n" + "WITH\n" + "__ref__stg_orders AS (SELECT 1 AS order_id, 7 AS rep_id),\n" + "__ref__stg_sales_reps AS (SELECT 7 AS rep_id, 'Ana' AS rep_name),\n" + "__expected__order_reps AS (SELECT 1 AS order_id, 'Ana' AS rep_name)\n" + "SELECT 1\n" + ), + "tests/scenarios/order_reps_unmocked.sql": ( + "SCENARIO (description 'Test scenario order_reps_unmocked.');\n\n" + "WITH\n" + "__ref__stg_orders AS (SELECT 1 AS order_id, 7 AS rep_id),\n" + "__expected__order_reps AS (SELECT 1 AS order_id, 'Ana' AS rep_name)\n" + "SELECT 1\n" + ), + } + return files diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/scenario/test_unmocked_inputs.py b/tests/e2e/src/sqlbuild/cli/commands/main/scenario/test_unmocked_inputs.py new file mode 100644 index 0000000000..e2ca9d6955 --- /dev/null +++ b/tests/e2e/src/sqlbuild/cli/commands/main/scenario/test_unmocked_inputs.py @@ -0,0 +1,59 @@ +"""Real-CLI coverage of scenarios that leave some of a model's inputs unmocked.""" + +from __future__ import annotations + +import subprocess +from pathlib import Path + +import pytest + +from tests.e2e.src.sqlbuild.cli.commands.main.scenario._test_types import ( + UnmockedInputScenarioE2ETestCase, +) +from tests.e2e.src.sqlbuild.cli.commands.main.scenario.helpers import ( + build_unmocked_input_scenario_project_files, +) +from tests.e2e.src.sqlbuild.cli.commands.shared.helpers import prepare_inline_project, run_sqb + + +@pytest.mark.parametrize( + "test_case", + [ + UnmockedInputScenarioE2ETestCase( + description="error names the unmocked input and the run ends with the failure", + expected_fragments=( + "model 'order_reps' reads 'stg_sales_reps', which is not mocked", + "Add a CTE named __ref__stg_sales_reps AS (...)", + "requires 8 sources without fixtures: raw_reps_0, raw_reps_1, raw_reps_2," + " raw_reps_3, raw_reps_4 and 3 more.", + "Failures:\n\n order_reps_unmocked\n error[S504]:", + ), + unexpected_fragment="requires source 'raw_reps_0'", + ), + ], + ids=lambda case: case.description, +) +def test_given_scenario_with_unmocked_model_input_when_testing_then_reports_the_boundary( + test_case: UnmockedInputScenarioE2ETestCase, + tmp_path: Path, +) -> None: + project_dir: Path = prepare_inline_project( + tmp_path=tmp_path, + project_name="unmocked_inputs", + repo_files=build_unmocked_input_scenario_project_files(), + ) + + result: subprocess.CompletedProcess[str] = run_sqb( + command=("--no-color", "scenario", "test"), project_dir=project_dir + ) + + output: str = result.stdout + result.stderr + assert result.returncode == 1, output + for fragment in test_case.expected_fragments: + assert fragment in output, output + assert test_case.unexpected_fragment not in output, output + assert output.rstrip().splitlines()[-1].startswith("Scenario execution failed"), output + + +if __name__ == "__main__": + pytest.main([__file__, "-vv"]) diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/sqlserver/helpers.py b/tests/e2e/src/sqlbuild/cli/commands/main/sqlserver/helpers.py index d15a8cbef7..895f163f31 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/sqlserver/helpers.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/sqlserver/helpers.py @@ -283,9 +283,10 @@ def prepare_sqlserver_waffle_shop(*, tmp_path: Path, config: dict[str, object]) (project_dir / "models" / "marts" / "daily_order_partitioned.sql").unlink(missing_ok=True) (project_dir / "tests" / "unit" / "test_is_completed_order_udf.sql").unlink(missing_ok=True) - macro_path: Path = project_dir / "macros" / "time.py" - macro_path.write_text( - macro_path.read_text(encoding="utf-8").replace( + macro_path: Path = project_dir / "models" / "marts" / "_macros" / "time.py" + _rewrite( + macro_path, + ( ' if ctx.adapter_name == "bigquery":\n' ' return f"TIMESTAMP_TRUNC({expression}, {grain.upper()})"\n' " return f\"DATE_TRUNC('{grain}', {expression})\"", @@ -298,26 +299,24 @@ def prepare_sqlserver_waffle_shop(*, tmp_path: Path, config: dict[str, object]) ' return f"CAST(CAST({expression} AS DATE) AS DATETIME2)"\n' " return f\"DATE_TRUNC('{grain}', {expression})\"", ), - encoding="utf-8", ) is_completed_order_path: Path = project_dir / "functions" / "sql" / "is_completed_order.sql" - is_completed_order_path.write_text( - is_completed_order_path.read_text(encoding="utf-8") - .replace("STRING", "NVARCHAR") - .replace("BOOLEAN", "BIT") - .replace( + _rewrite(is_completed_order_path, ("STRING", "NVARCHAR")) + _rewrite(is_completed_order_path, ("BOOLEAN", "BIT")) + _rewrite( + is_completed_order_path, + ( "order_status = 'completed'", "CAST(CASE WHEN order_status = 'completed' THEN 1 ELSE 0 END AS BIT)", ), - encoding="utf-8", ) fact_orders_path: Path = project_dir / "models" / "marts" / "fact_orders.sql" - fact_orders_path.write_text( - fact_orders_path.read_text(encoding="utf-8").replace( + _rewrite( + fact_orders_path, + ( '__udf("is_completed_order_py")(o.status) AS is_completed_order_py,', '__udf("is_completed_order")(o.status) AS is_completed_order_py,', ), - encoding="utf-8", ) project_file_path: Path = project_dir / "sqlbuild_project.toml" project_file_path.write_text( @@ -333,22 +332,28 @@ def prepare_sqlserver_waffle_shop(*, tmp_path: Path, config: dict[str, object]) encoding="utf-8", ) lookups_path: Path = project_dir / "seeds" / "lookups.yml" - lookups_path.write_text( - lookups_path.read_text(encoding="utf-8").replace("type: VARCHAR", "type: NVARCHAR(100)"), - encoding="utf-8", - ) + _rewrite(lookups_path, ("type: VARCHAR", "type: NVARCHAR(100)")) raw_sources_path: Path = project_dir / "sources" / "raw.yml" adapt_sqlserver_text_file(raw_sources_path) - raw_sources_path.write_text( - raw_sources_path.read_text(encoding="utf-8").replace( + _rewrite( + raw_sources_path, + ( " - name: ordered_at\n type: DATETIME2", " - name: ordered_at\n type: DATETIME", ), - encoding="utf-8", ) return project_dir, schema_name +def _rewrite(path: Path, replacement: tuple[str, str]) -> None: + """Replace text in a copied fixture file, failing when the fixture no longer contains it.""" + + old, new = replacement + text: str = path.read_text(encoding="utf-8") + assert old in text, f"{path} no longer contains {old!r}; update the SQL Server fixture" + path.write_text(text.replace(old, new), encoding="utf-8") + + def prepare_sqlserver_source_loader_strategies( *, tmp_path: Path, config: dict[str, object] ) -> tuple[Path, str]: diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/test/_test_types.py b/tests/e2e/src/sqlbuild/cli/commands/main/test/_test_types.py index 6a913f11c7..78c5d80a83 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/test/_test_types.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/test/_test_types.py @@ -483,3 +483,13 @@ class OutputEvaluationErrorE2ETestCase: expected_fragments: tuple[str, ...] expected_json_fragments: tuple[str, ...] unexpected_fragments: tuple[str, ...] + + +@dataclass(frozen=True) +class SharedPartialFixtureE2ETestCase: + """A partial mock read by two real models, with each reader's column spelling.""" + + description: str + enriched_column: str + report_column: str + expected_padding_count: int diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/test/helpers.py b/tests/e2e/src/sqlbuild/cli/commands/main/test/helpers.py index c86779e631..1819396c24 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/test/helpers.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/test/helpers.py @@ -2113,3 +2113,40 @@ def order_totals_test_files(total_sql: str) -> dict[str, str]: "__expected__order_totals AS (\n SELECT 1 AS order_id\n)\nSELECT 1\n" ), } + + +def build_shared_partial_fixture_project_files( + *, enriched_column: str, report_column: str +) -> dict[str, str]: + """Build a test chain whose two real models read one partial mock.""" + + return { + "sqlbuild_project.toml": ( + 'name = "shared_partial_fixture"\nadapter = "duckdb"\n\n' + '[connection]\ndatabase = "shared_partial_fixture.duckdb"\n' + ), + "models/order_lines.sql": ( + 'MODEL (description "Test model order_lines.");\n\n' + "SELECT 1 AS order_id, CAST(1.5 AS DOUBLE) AS amount, CAST(2.5 AS DOUBLE) AS discount\n" + ), + "models/enriched_lines.sql": ( + 'MODEL (description "Test model enriched_lines.");\n\n' + f'SELECT order_id, amount, {enriched_column} FROM __ref("order_lines")\n' + ), + "models/line_report.sql": ( + 'MODEL (description "Test model line_report.");\n\n' + f"SELECT e.order_id, e.amount, l.{report_column} AS line_discount\n" + 'FROM __ref("enriched_lines") AS e\n' + 'JOIN __ref("order_lines") AS l ON e.order_id = l.order_id\n' + ), + "tests/unit/test_line_report.sql": ( + "TEST();\n\n" + "WITH\n" + "__ref__order_lines AS (SELECT 1 AS order_id, CAST(1.5 AS DOUBLE) AS amount),\n" + "__expected__line_report AS (\n" + " SELECT 1 AS order_id, CAST(1.5 AS DOUBLE) AS amount," + " CAST(NULL AS DOUBLE) AS line_discount\n" + ")\n" + "SELECT 1\n" + ), + } diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/test/test_partial_fixture_shared_readers.py b/tests/e2e/src/sqlbuild/cli/commands/main/test/test_partial_fixture_shared_readers.py new file mode 100644 index 0000000000..d4694eb820 --- /dev/null +++ b/tests/e2e/src/sqlbuild/cli/commands/main/test/test_partial_fixture_shared_readers.py @@ -0,0 +1,65 @@ +"""Real-CLI coverage of one partial mock read by two real models of a SQL test chain.""" + +from __future__ import annotations + +import subprocess +from pathlib import Path + +import pytest + +from tests.e2e.src.sqlbuild.cli.commands.main.test._test_types import ( + SharedPartialFixtureE2ETestCase, +) +from tests.e2e.src.sqlbuild.cli.commands.main.test.helpers import ( + build_shared_partial_fixture_project_files, +) +from tests.e2e.src.sqlbuild.cli.commands.shared.helpers import prepare_inline_project, run_sqb + + +@pytest.mark.parametrize( + "test_case", + [ + SharedPartialFixtureE2ETestCase( + description="both readers spell the column the same way", + enriched_column="discount", + report_column="discount", + expected_padding_count=1, + ), + SharedPartialFixtureE2ETestCase( + description="readers spell the column in different case", + enriched_column="DISCOUNT", + report_column="discount", + expected_padding_count=1, + ), + ], + ids=lambda case: case.description, +) +def test_given_partial_mock_read_by_two_real_models_when_testing_then_pads_each_column_once( + test_case: SharedPartialFixtureE2ETestCase, + tmp_path: Path, +) -> None: + project_dir: Path = prepare_inline_project( + tmp_path=tmp_path, + project_name="shared_partial_fixture", + repo_files=build_shared_partial_fixture_project_files( + enriched_column=test_case.enriched_column, report_column=test_case.report_column + ), + ) + + result: subprocess.CompletedProcess[str] = run_sqb( + command=("--no-color", "test"), project_dir=project_dir + ) + + assert result.returncode == 0, result.stdout + result.stderr + statements: str = "".join( + path.read_text(encoding="utf-8") + for path in (project_dir / "target" / "run" / "tests").rglob("*.sql") + ) + assert ( + statements.lower().count('cast(null as double) as "discount"') + == test_case.expected_padding_count + ), statements + + +if __name__ == "__main__": + pytest.main([__file__, "-vv"]) diff --git a/tests/integration/src/sqlbuild/adapters/sqlite/_test_types.py b/tests/integration/src/sqlbuild/adapters/sqlite/_test_types.py index 803668b253..09ec18f774 100644 --- a/tests/integration/src/sqlbuild/adapters/sqlite/_test_types.py +++ b/tests/integration/src/sqlbuild/adapters/sqlite/_test_types.py @@ -210,3 +210,12 @@ class SqliteRetypeIndexTestCase: description: str expected_indexes: tuple[tuple[object, ...], ...] expected_columns: tuple[ColumnInfo, ...] + + +@dataclass(frozen=True) +class SqliteUnkeyedSampleTestCase: + description: str + left_sql: str + limit: int + expected_row_count: int + expected_order_ids: frozenset[object] diff --git a/tests/integration/src/sqlbuild/adapters/sqlite/helpers.py b/tests/integration/src/sqlbuild/adapters/sqlite/helpers.py index 48b708b73f..fcf5cad5f0 100644 --- a/tests/integration/src/sqlbuild/adapters/sqlite/helpers.py +++ b/tests/integration/src/sqlbuild/adapters/sqlite/helpers.py @@ -255,3 +255,16 @@ def _retype_column(adapter: SqliteAdapter, connection: Any) -> None: "retype": _retype_column, } ) + + +class RecordingCursorSqliteAdapter(SqliteAdapter): + """Record every statement and the cursor it returned.""" + + def __init__(self) -> None: + super().__init__() + self.executed: list[tuple[str, Any]] = [] + + def _execute(self, *, connection: Any, sql: str) -> Any: + cursor: Any = super()._execute(connection=connection, sql=sql) + self.executed.append((sql, cursor)) + return cursor diff --git a/tests/integration/src/sqlbuild/adapters/sqlite/test_unkeyed_samples.py b/tests/integration/src/sqlbuild/adapters/sqlite/test_unkeyed_samples.py new file mode 100644 index 0000000000..16a7cf1feb --- /dev/null +++ b/tests/integration/src/sqlbuild/adapters/sqlite/test_unkeyed_samples.py @@ -0,0 +1,85 @@ +"""Unkeyed side-only samples read a bounded result and release their cursor.""" + +from __future__ import annotations + +import sqlite3 +from pathlib import Path +from typing import Any + +import pytest + +from sqlbuild.adapter.contract.constants import DIFF_LEFT_SIDE +from tests.integration.src.sqlbuild.adapters.sqlite._test_types import ( + SqliteUnkeyedSampleTestCase, +) +from tests.integration.src.sqlbuild.adapters.sqlite.helpers import ( + RecordingCursorSqliteAdapter, + recorder, +) + + +@pytest.mark.parametrize( + "test_case", + [ + SqliteUnkeyedSampleTestCase( + description="many distinct side-only rows stop at the limit", + left_sql=( + "SELECT 1 AS order_id UNION ALL SELECT 2 UNION ALL SELECT 3 " + "UNION ALL SELECT 4 UNION ALL SELECT 5" + ), + limit=2, + expected_row_count=2, + expected_order_ids=frozenset({1, 2, 3, 4, 5}), + ), + SqliteUnkeyedSampleTestCase( + description="a repeated side-only row fills the limit by multiplicity", + left_sql="SELECT 7 AS order_id UNION ALL SELECT 7 UNION ALL SELECT 7", + limit=2, + expected_row_count=2, + expected_order_ids=frozenset({7}), + ), + ], + ids=lambda case: case.description, +) +def test_given_side_only_rows_when_sampling_unkeyed_then_query_is_bounded_and_cursor_closed( + test_case: SqliteUnkeyedSampleTestCase, tmp_path: Path +) -> None: + adapter: RecordingCursorSqliteAdapter = RecordingCursorSqliteAdapter() + connection: Any = adapter.connect({"path": str(tmp_path / "shop.sqlite")}) + try: + adapter.create_table_as( + connection=connection, + destination="orders", + sql=f"SELECT order_id FROM ({test_case.left_sql}) ORDER BY order_id", + statement_recorder=recorder(), + ) + adapter.create_table_as( + connection=connection, + destination="orders_next", + sql="SELECT order_id FROM orders WHERE 0 = 1", + statement_recorder=recorder(), + ) + adapter.executed.clear() + + rows: tuple[tuple[tuple[str, object], ...], ...] = adapter.sample_unkeyed_side_only_rows( + connection=connection, + left="orders", + right="orders_next", + side=DIFF_LEFT_SIDE, + limit=test_case.limit, + ) + + sample_sql, sample_cursor = adapter.executed[-1] + assert len(rows) == test_case.expected_row_count, test_case.description + assert {dict(row)["order_id"] for row in rows} <= test_case.expected_order_ids, ( + test_case.description + ) + assert sample_sql.endswith(f"LIMIT {test_case.limit}"), test_case.description + with pytest.raises(sqlite3.ProgrammingError): + sample_cursor.fetchall() + finally: + adapter.close(connection) + + +if __name__ == "__main__": + pytest.main([__file__, "-vv"]) diff --git a/tests/integration/src/sqlbuild/compiler/planner/main/_test_types.py b/tests/integration/src/sqlbuild/compiler/planner/main/_test_types.py index acc3e29866..efc338a23f 100644 --- a/tests/integration/src/sqlbuild/compiler/planner/main/_test_types.py +++ b/tests/integration/src/sqlbuild/compiler/planner/main/_test_types.py @@ -165,3 +165,11 @@ class SourceTableProbeBudgetTestCase: source_table: str expected_source_listings: int expected_probes: tuple[str, ...] + + +@dataclass(frozen=True) +class UnlistedMigrationOriginTestCase: + description: str + select: tuple[str, ...] + expected_warnings: tuple[str, ...] + expected_unlisted_reads: int diff --git a/tests/integration/src/sqlbuild/compiler/planner/main/helpers.py b/tests/integration/src/sqlbuild/compiler/planner/main/helpers.py index 3067f5ba66..2cda67eeaf 100644 --- a/tests/integration/src/sqlbuild/compiler/planner/main/helpers.py +++ b/tests/integration/src/sqlbuild/compiler/planner/main/helpers.py @@ -590,6 +590,63 @@ def probe_relation_read(self, *, connection: Any, relation: str) -> RelationRead return super().probe_relation_read(connection=connection, relation=relation) +class UnlistedDatabaseDuckDbAdapter(CountingSourceDuckDbAdapter): + """Plan against one shared DuckDB connection whose listings of one database fail.""" + + def __init__(self, *, connection: Any, unlisted_database: str) -> None: + super().__init__(connection=connection) + self._unlisted_database: str = unlisted_database + self.listed_databases: list[str | None] = [] + + def list_relations( + self, + *, + connection: Any, + database: str | None, + schemas: tuple[str, ...] | None, + names: tuple[str, ...] | None = None, + ) -> tuple[RelationInfo, ...]: + self.listed_databases.append(database) + _LISTING_GUARDS[database == self._unlisted_database](database) + return super().list_relations( + connection=connection, database=database, schemas=schemas, names=names + ) + + +def _allow_listing(database: str | None) -> None: + del database + + +def _refuse_listing(database: str | None) -> None: + raise RuntimeError(f"cannot list database {database}") + + +_LISTING_GUARDS: MappingProxyType[bool, Callable[[str | None], None]] = MappingProxyType( + {False: _allow_listing, True: _refuse_listing} +) + + +def plan_deferred_project( + *, + project_dir: Path, + adapter: DuckDbAdapter, + project_files: dict[str, str], + select: tuple[str, ...], +) -> CompilePipelineResult: + """Write a project with `dev` and `prod` targets and plan `select` deferring to prod.""" + + relative_path: str + contents: str + for relative_path, contents in project_files.items(): + (project_dir / relative_path).parent.mkdir(parents=True, exist_ok=True) + (project_dir / relative_path).write_text(contents, encoding="utf-8") + return run_compile_pipeline( + discovered_inputs=discover_project_inputs(project_dir=project_dir), + adapter=adapter, + options=CompilePipelineOptions(no_sql_validation=True, defer_to="prod", select=select), + ) + + def plan_source_reading_project( *, project_dir: Path, adapter: DuckDbAdapter, source_schema: str, source_table: str ) -> CompilePipelineResult: diff --git a/tests/integration/src/sqlbuild/compiler/planner/main/test_deferred_migration_origins.py b/tests/integration/src/sqlbuild/compiler/planner/main/test_deferred_migration_origins.py new file mode 100644 index 0000000000..c00902ab45 --- /dev/null +++ b/tests/integration/src/sqlbuild/compiler/planner/main/test_deferred_migration_origins.py @@ -0,0 +1,93 @@ +"""Deferred migrate_from lookups skip selected models and survive unlistable origins.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +import pytest + +from sqlbuild._native import CompilePipelineResult +from tests.integration.src.sqlbuild.compiler.planner.main._test_types import ( + UnlistedMigrationOriginTestCase, +) +from tests.integration.src.sqlbuild.compiler.planner.main.helpers import ( + UnlistedDatabaseDuckDbAdapter, + plan_deferred_project, +) + +_UNLISTED_DATABASE: str = "archive_db" +_PROJECT_FILES: dict[str, str] = { + "sqlbuild_project.toml": ( + 'name = "orders"\nadapter = "duckdb"\ndefault_target = "dev"\n\n' + '[targets.dev]\nschema = "dev"\n\n' + '[targets.prod]\nschema = "prod"\n' + ), + "models/order_ledger.sql": ( + 'MODEL (description "Test model order_ledger.", materialized table, ' + f"migrate_from {_UNLISTED_DATABASE}.prod.order_history);\n\nSELECT 1 AS order_id\n" + ), + "models/order_report.sql": ( + 'MODEL (description "Test model order_report.", materialized table);\n\n' + 'SELECT order_id FROM __ref("order_ledger")\n' + ), +} + + +@pytest.mark.parametrize( + "test_case", + [ + UnlistedMigrationOriginTestCase( + description="unlistable origin of an unselected upstream falls back with a warning", + select=("order_report",), + expected_warnings=( + "could not list archive_db.prod.order_history in the deferred target to look " + "for a pre-migration relation, so model 'order_ledger' is deferred under its " + "current name: cannot list database archive_db", + ), + expected_unlisted_reads=1, + ), + UnlistedMigrationOriginTestCase( + description="selected models are never looked up in the deferred target", + select=("order_ledger", "order_report"), + expected_warnings=(), + expected_unlisted_reads=0, + ), + ], + ids=lambda case: case.description, +) +def test_given_unlistable_migration_origin_when_planning_with_defer_then_planning_continues( + test_case: UnlistedMigrationOriginTestCase, + tmp_path: Path, + connection: Any, +) -> None: + connection.execute("CREATE SCHEMA dev") + connection.execute("CREATE SCHEMA prod") + connection.execute("CREATE TABLE dev.order_ledger AS SELECT 1 AS order_id") + adapter: UnlistedDatabaseDuckDbAdapter = UnlistedDatabaseDuckDbAdapter( + connection=connection, unlisted_database=_UNLISTED_DATABASE + ) + + result: CompilePipelineResult = plan_deferred_project( + project_dir=tmp_path, + adapter=adapter, + project_files=_PROJECT_FILES, + select=test_case.select, + ) + + assert ( + tuple( + filter( + lambda message: "deferred target" in message, + (warning.message for warning in result.plan_output.warnings), + ) + ) + == test_case.expected_warnings + ), test_case.description + assert ( + adapter.listed_databases.count(_UNLISTED_DATABASE) == test_case.expected_unlisted_reads + ), test_case.description + + +if __name__ == "__main__": + pytest.main([__file__, "-vv"]) diff --git a/tests/unit/scripts/native_module_paths/_helpers/test_module_paths.py b/tests/unit/scripts/native_module_paths/_helpers/test_module_paths.py index 5fd81a5f4d..d13e911769 100644 --- a/tests/unit/scripts/native_module_paths/_helpers/test_module_paths.py +++ b/tests/unit/scripts/native_module_paths/_helpers/test_module_paths.py @@ -127,6 +127,7 @@ def test_given_native_source_when_checking_module_paths_then_reports_errors( "sqlbuild.cli.commands.classes.prepared_compile_artifacts", "sqlbuild.cli.compute_logs", "sqlbuild.cli.observability", + "sqlbuild.clone", "sqlbuild.compile", "sqlbuild.compile.reuse", "sqlbuild.cost", diff --git a/tests/unit/src/sqlbuild/cli/commands/shared/_helpers/test_progress.py b/tests/unit/src/sqlbuild/cli/commands/shared/_helpers/test_progress.py index 88223e1e7f..3926dfa2ff 100644 --- a/tests/unit/src/sqlbuild/cli/commands/shared/_helpers/test_progress.py +++ b/tests/unit/src/sqlbuild/cli/commands/shared/_helpers/test_progress.py @@ -113,7 +113,7 @@ def test_given_verbose_runtime_diagnostics_when_reporting_then_writes_concise_ra expected_fragments=( "Scheduler diagnostics 2 transitions omitted after 50", "51 waiting on dependencies", - "Query diagnostics 2 statements omitted after 25", + "Query diagnostics 2 statements omitted after 25 per resource", "query-26", ), expected_absent_fragments=("query-25 success",), @@ -142,6 +142,16 @@ def test_given_high_volume_runtime_diagnostics_when_closing_then_output_is_bound phase="microbatch", ) ) + callbacks.on_statement_complete( + StatementExecutionTelemetry( + query_id="customers-query", + status="success", + elapsed_seconds=0.1, + resource_type="model", + resource_name="customers", + phase="materialize", + ) + ) callbacks.close() output: str = capsys.readouterr().out @@ -152,7 +162,8 @@ def test_given_high_volume_runtime_diagnostics_when_closing_then_output_is_bound for absent_fragment in test_case.expected_absent_fragments: assert absent_fragment not in output assert output.count("Scheduler ") == 51 - assert output.count("Query model") == 26 + assert output.count("Query model orders") == 26 + assert "Query model customers phase=materialize customers-query" in output @pytest.mark.parametrize( diff --git a/website/src/content/docs/docs/cli/clone.mdx b/website/src/content/docs/docs/cli/clone.mdx index af79602e17..dabc1b3103 100644 --- a/website/src/content/docs/docs/cli/clone.mdx +++ b/website/src/content/docs/docs/cli/clone.mdx @@ -24,6 +24,10 @@ function SQL from each target's project, so SQL analysis errors never stop it; s such as an unknown reference, still do. The hidden `--no-sql-analysis` flag (alias `--no-sql-validation`) is still accepted but has no effect. +With `--debug`, clone logs how long each phase took: compiling each target it reads, inspecting +sources, planning, and execution. The `--json-output` document reports the same durations in +seconds under `phase_timings`. + ## Usage ```bash diff --git a/website/src/content/docs/docs/concepts/adapters.mdx b/website/src/content/docs/docs/concepts/adapters.mdx index 0ab7a25970..23ed8255a4 100644 --- a/website/src/content/docs/docs/concepts/adapters.mdx +++ b/website/src/content/docs/docs/concepts/adapters.mdx @@ -124,4 +124,4 @@ StrictAdapter (fully abstract - all methods must be implemented) - **ConnectionMixin** - `connect`, `close`, `begin`, `commit`, `rollback` - **SchemaMixin** - `relation_exists`, `list_relations`, `get_columns`, `ensure_schema` - **MaterializationMixin** - `create_table_as`, `create_view_as`, `drop`, `rename`, `swap`, `load_seed` -- **DiffMixin** - `diff_schema`, `diff_rows`, `sample_unequal_rows`, `sample_side_only_rows` +- **DiffMixin** - `diff_schema`, `diff_rows`, `sample_unequal_rows`, `sample_side_only_rows`, `diff_unkeyed_rows` diff --git a/website/src/content/docs/docs/concepts/diff.mdx b/website/src/content/docs/docs/concepts/diff.mdx index 71231de94b..68b4b49225 100644 --- a/website/src/content/docs/docs/concepts/diff.mdx +++ b/website/src/content/docs/docs/concepts/diff.mdx @@ -142,6 +142,11 @@ For a model without a stable row key, opt in to exact full-row multiset comparis sqb diff prod:dev --full --select daily_order_totals --unkeyed ``` +Unkeyed diffs show sample complete rows that one side holds more often than the other, up to +`--max-row-only-examples` per side; a row duplicated three more times on one side appears three +times. In `--json`, they are `examples.from_only_rows` and `examples.to_only_rows`, and +`from_only_keys` and `to_only_keys` stay empty because there is no key. + The diff output categorises rows as: - **Equal** - same key, same values on both sides - **Unequal** - same key, different values (with per-column breakdown)