Skip to content

Commit 8a588fc

Browse files
authored
Merge branch 'main' into refactor/planner-native-records
2 parents a8ac282 + e1a2850 commit 8a588fc

80 files changed

Lines changed: 2509 additions & 132 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎Makefile‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -93,6 +93,7 @@ E2E_DUCKDB_BUILD_CORE_PATHS := \
9393
tests/e2e/src/sqlbuild/cli/commands/main/build/test_python_hooks.py \
9494
tests/e2e/src/sqlbuild/cli/commands/main/build/test_python_node_selection.py \
9595
tests/e2e/src/sqlbuild/cli/commands/main/build/test_python_source_deferral.py \
96+
tests/e2e/src/sqlbuild/cli/commands/main/build/test_defer_migration_origin.py \
9697
tests/e2e/src/sqlbuild/cli/commands/main/build/test_sql_hook_reads.py \
9798
tests/e2e/src/sqlbuild/cli/commands/main/build/test_python_node_identity.py \
9899
tests/e2e/src/sqlbuild/cli/commands/main/build/test_sql_hook_identity.py \
@@ -123,6 +124,7 @@ E2E_DUCKDB_BUILD_INCREMENTAL_PATHS := \
123124
tests/e2e/src/sqlbuild/cli/commands/main/build/test_microbatch_aware_cursor_date_input.py \
124125
tests/e2e/src/sqlbuild/cli/commands/main/build/test_microbatch_direct_lifecycle.py \
125126
tests/e2e/src/sqlbuild/cli/commands/main/build/test_microbatch_failure_windows.py \
127+
tests/e2e/src/sqlbuild/cli/commands/main/build/test_microbatch_batch_progress.py \
126128
tests/e2e/src/sqlbuild/cli/commands/main/build/test_microbatch_replay_backfill_lifecycle.py \
127129
tests/e2e/src/sqlbuild/cli/commands/main/build/test_mixed_timestamp_grain_replay.py \
128130
tests/e2e/src/sqlbuild/cli/commands/main/build/test_model_backed_cursor_build.py \
@@ -144,7 +146,8 @@ E2E_DUCKDB_CLI_DATA_PATHS := \
144146
tests/e2e/src/sqlbuild/cli/commands/main/scenario/test_interrupts.py \
145147
tests/e2e/src/sqlbuild/cli/commands/main/scenario/test_empty_fixture.py \
146148
tests/e2e/src/sqlbuild/cli/commands/main/scenario/test_expected_columns.py \
147-
tests/e2e/src/sqlbuild/cli/commands/main/scenario/test_helper_ctes.py
149+
tests/e2e/src/sqlbuild/cli/commands/main/scenario/test_helper_ctes.py \
150+
tests/e2e/src/sqlbuild/cli/commands/main/scenario/test_unmocked_inputs.py
148151

149152
E2E_DUCKDB_CLI_PATHS := \
150153
tests/e2e/scripts/cli_preview \
Lines changed: 86 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,86 @@
1+
//! Real models whose unmocked model inputs pull unmocked sources into a scenario.
2+
3+
use std::collections::{BTreeSet, HashMap};
4+
5+
use crate::planner_scenario::models::ScenarioGraphInput;
6+
7+
/// Whether `name`, run as a real model, reads a source the scenario does not mock, with the memo.
8+
fn reaches_unmocked_source(
9+
name: &str,
10+
input: &ScenarioGraphInput,
11+
mut memo: HashMap<String, bool>,
12+
) -> (bool, HashMap<String, bool>) {
13+
if let Some(known) = memo.get(name) {
14+
return (*known, memo);
15+
}
16+
memo.insert(name.to_owned(), false);
17+
let mut reaches: bool = false;
18+
for (kind, dep) in input.models.get(name).into_iter().flatten() {
19+
reaches = match kind.as_str() {
20+
"source" => !input.sources.contains(dep),
21+
"model" if !input.refs.contains(dep) => {
22+
let (found, updated) = reaches_unmocked_source(dep, input, memo);
23+
memo = updated;
24+
found
25+
}
26+
_ => false,
27+
};
28+
if reaches {
29+
break;
30+
}
31+
}
32+
memo.insert(name.to_owned(), reaches);
33+
(reaches, memo)
34+
}
35+
36+
/// Real models that mock some model inputs but leave others, which read unmocked sources, real.
37+
pub(crate) fn unmocked_boundaries(
38+
input: &ScenarioGraphInput,
39+
models: &BTreeSet<String>,
40+
) -> Vec<String> {
41+
let mut memo: HashMap<String, bool> = HashMap::new();
42+
let mut messages: Vec<String> = Vec::new();
43+
for name in models {
44+
let Some(deps) = input.models.get(name) else {
45+
continue;
46+
};
47+
let model_deps = || {
48+
deps.iter()
49+
.filter(|(kind, _)| matches!(kind.as_str(), "model"))
50+
};
51+
if !model_deps().any(|(_, dep)| input.refs.contains(dep)) {
52+
continue;
53+
}
54+
let mut unmocked: Vec<&str> = Vec::new();
55+
for (_, dep) in model_deps() {
56+
if input.refs.contains(dep)
57+
|| !input.models.contains_key(dep)
58+
|| unmocked.contains(&dep.as_str())
59+
{
60+
continue;
61+
}
62+
let (reaches, updated) = reaches_unmocked_source(dep, input, memo);
63+
memo = updated;
64+
if reaches {
65+
unmocked.push(dep);
66+
}
67+
}
68+
if unmocked.is_empty() {
69+
continue;
70+
}
71+
let quoted: Vec<String> = unmocked.iter().map(|dep| format!("'{dep}'")).collect();
72+
let ctes: Vec<String> = unmocked
73+
.iter()
74+
.map(|dep| format!("__ref__{dep} AS (...)"))
75+
.collect();
76+
messages.push(format!(
77+
"Scenario '{}': model '{name}' reads {}, which {} not mocked, while its other model inputs are. Add {} named {}.",
78+
input.name,
79+
quoted.join(" and "),
80+
if unmocked.len() == 1 { "is" } else { "are" },
81+
if unmocked.len() == 1 { "a CTE" } else { "CTEs" },
82+
ctes.join(" and ")
83+
));
84+
}
85+
messages
86+
}
Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
11
//! Shared scenario SQL text operations.
22
3+
pub(super) mod boundaries;
34
pub(super) mod markers;

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,3 +3,5 @@
33
pub(super) const MARKERS: [&str; 4] = ["__ref(", "__seed(", "__source(", "__dbt_ref("];
44
pub(super) const DBT_REF_MARKER: usize = 3;
55
pub(super) const HASH_LENGTH: usize = 12;
6+
/// Missing sources named in full before the rest are counted.
7+
pub(super) const LISTED_MISSING_SOURCES: usize = 5;

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

Lines changed: 42 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11
//! Infer scenario closure while preserving dependency and diagnostic order.
22
3+
use crate::planner_scenario::_helpers::boundaries::unmocked_boundaries;
4+
use crate::planner_scenario::constants::LISTED_MISSING_SOURCES;
35
use crate::planner_scenario::models::{ScenarioGraphInput, ScenarioGraphOutput};
46

57
/// Infer required relations, cutting traversal at authored reference fixtures.
@@ -47,13 +49,51 @@ pub fn graph(input: &ScenarioGraphInput) -> ScenarioGraphOutput {
4749
for target in &input.targets {
4850
output = walk(target, input, output);
4951
}
50-
for name in &output.sources {
51-
if !input.sources.contains(name) {
52+
let missing: Vec<&String> = output
53+
.sources
54+
.iter()
55+
.filter(|name| !input.sources.contains(name))
56+
.collect();
57+
let boundaries: Vec<String> = if missing.is_empty() {
58+
Vec::new()
59+
} else {
60+
unmocked_boundaries(input, &output.models)
61+
};
62+
if boundaries.is_empty() {
63+
for name in &missing {
5264
output.warnings.push((format!(
5365
"Scenario '{}' requires source '{name}', but no fixture was provided. Add a CTE named __source__{} AS (...).",
5466
input.name, name.replace('.', "__")
5567
), false));
5668
}
69+
} else {
70+
output
71+
.warnings
72+
.extend(boundaries.into_iter().map(|message| (message, false)));
73+
let listed: Vec<&str> = missing
74+
.iter()
75+
.take(LISTED_MISSING_SOURCES)
76+
.map(|name| name.as_str())
77+
.collect();
78+
let more: String = if missing.len() > LISTED_MISSING_SOURCES {
79+
format!(" and {} more", missing.len() - LISTED_MISSING_SOURCES)
80+
} else {
81+
String::new()
82+
};
83+
output.warnings.push((
84+
format!(
85+
"Scenario '{}' requires {} {} without fixtures: {}{more}.",
86+
input.name,
87+
missing.len(),
88+
if missing.len() == 1 {
89+
"source"
90+
} else {
91+
"sources"
92+
},
93+
listed.join(", ")
94+
),
95+
false,
96+
));
5797
}
5898
for name in &output.dbt_refs {
5999
if !input.dbt_refs.contains(name) {

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

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -76,6 +76,56 @@ fn given_scenario_dependencies_when_planning_then_preserves_closure_and_warning_
7676
),
7777
],
7878
},
79+
GraphTestCase {
80+
description: "unmocked inputs beside mocked ones are named before the sources",
81+
input: ScenarioGraphInput {
82+
name: "orders".into(),
83+
targets: vec!["summary".into()],
84+
refs: vec!["staging".into()],
85+
models: [
86+
(
87+
"summary".into(),
88+
vec![
89+
("model".into(), "staging".into()),
90+
("model".into(), "sales_rep".into()),
91+
("model".into(), "manager".into()),
92+
],
93+
),
94+
("staging".into(), vec![("source".into(), "hidden".into())]),
95+
(
96+
"sales_rep".into(),
97+
vec![
98+
("source".into(), "raw.reps".into()),
99+
("source".into(), "raw.a".into()),
100+
("source".into(), "raw.b".into()),
101+
],
102+
),
103+
(
104+
"manager".into(),
105+
vec![
106+
("source".into(), "raw.c".into()),
107+
("source".into(), "raw.d".into()),
108+
("source".into(), "raw.e".into()),
109+
],
110+
),
111+
]
112+
.into(),
113+
..ScenarioGraphInput::default()
114+
},
115+
expected_models: vec!["manager", "sales_rep", "summary"],
116+
expected_refs: vec!["staging"],
117+
expected_functions: vec![],
118+
expected_warnings: vec![
119+
(
120+
"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 (...).",
121+
false,
122+
),
123+
(
124+
"Scenario 'orders' requires 6 sources without fixtures: raw.a, raw.b, raw.c, raw.d, raw.e and 1 more.",
125+
false,
126+
),
127+
],
128+
},
79129
];
80130
for test_case in test_cases {
81131
let result = graph(&test_case.input);

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

Lines changed: 26 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -100,10 +100,31 @@ fn examples(model: &DiffModelFacts, options: DiffExampleOptions) -> JsonValue {
100100
let keys = |samples: &[Vec<DiffNamedValue>]| {
101101
JsonValue::Array(samples.iter().map(|sample| named_values(sample)).collect())
102102
};
103-
object(vec![
103+
let no_samples: &[Vec<DiffNamedValue>] = &[];
104+
let (side_keys, side_rows) = if model.unkeyed {
105+
(
106+
(no_samples, no_samples),
107+
Some((&model.left_only_keys, &model.right_only_keys)),
108+
)
109+
} else {
110+
(
111+
(
112+
model.left_only_keys.as_slice(),
113+
model.right_only_keys.as_slice(),
114+
),
115+
None,
116+
)
117+
};
118+
let mut entries: Vec<(&str, JsonValue)> = vec![
104119
("unequal_rows", JsonValue::Array(unequal_rows)),
105-
("from_only_keys", keys(&model.left_only_keys)),
106-
("to_only_keys", keys(&model.right_only_keys)),
120+
("from_only_keys", keys(side_keys.0)),
121+
("to_only_keys", keys(side_keys.1)),
122+
];
123+
if let Some((from_rows, to_rows)) = side_rows {
124+
entries.push(("from_only_rows", keys(from_rows)));
125+
entries.push(("to_only_rows", keys(to_rows)));
126+
}
127+
entries.extend([
107128
("truncated_by_count", JsonValue::Bool(truncated_by_count)),
108129
("values_truncated", JsonValue::Bool(values_truncated)),
109130
(
@@ -116,7 +137,8 @@ fn examples(model: &DiffModelFacts, options: DiffExampleOptions) -> JsonValue {
116137
.max_value_length
117138
.map_or(JsonValue::Null, |limit| integer(limit as i64)),
118139
),
119-
])
140+
]);
141+
object(entries)
120142
}
121143

122144
fn model_payload(model: &DiffModelFacts, options: &DiffDocumentOptions) -> JsonValue {
Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
//! The permanent line of one completed microbatch, for output without a live spinner.
2+
3+
use crate::build_progress::models::BatchProgressText;
4+
5+
/// The model, batch position, cursor window, rows written and elapsed time of a batch.
6+
#[must_use]
7+
pub fn batch_progress_message(batch: &BatchProgressText) -> String {
8+
let rows: String = batch
9+
.rows_affected
10+
.map(|rows| format!(" {rows} rows"))
11+
.unwrap_or_default();
12+
format!(
13+
"Batch {} {}/{} {}{rows} {:.1}s",
14+
batch.model_name,
15+
batch.batch_number,
16+
batch.batch_count,
17+
batch.window,
18+
batch.elapsed_seconds
19+
)
20+
}

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
1+
pub mod batch_progress_message;
12
pub mod format_build_footer;
23
pub mod progress_name_width;
34
pub mod query_diagnostic_message;

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@ pub fn runtime_diagnostic_summary(
2020
}
2121
if let Some((omitted, statement)) = query.filter(|(omitted, _)| *omitted > 0) {
2222
lines.push(format!(
23-
"Query diagnostics {omitted} statements omitted after {QUERY_DIAGNOSTIC_LIMIT}; final query: {}",
23+
"Query diagnostics {omitted} statements omitted after {QUERY_DIAGNOSTIC_LIMIT} per resource; final query: {}",
2424
query_diagnostic_message(statement)
2525
));
2626
}

0 commit comments

Comments
 (0)