fix: scenario, SQL test and dialect fixes for fixtures, checks and truth tests

Makefile

@@ -51,7 +51,7 @@ test:
 	@mkdir -p /tmp/opencode
 	@log="/tmp/opencode/test-$$(date +%Y%m%d-%H%M%S).log"; \
 	echo "Logging to $$log"; \
-	env PYTHONUNBUFFERED=1 SQLBUILD_CONCURRENCY=$(SQLBUILD_CONCURRENCY) uv run pytest tests/unit tests/integration -m "not real_warehouse and not dbt" -vv --color=yes -n auto --dist loadfile 2>&1 | tee "$$log"; \
+	env PYTHONUNBUFFERED=1 SQLBUILD_CONCURRENCY=$(SQLBUILD_CONCURRENCY) uv run pytest tests/unit tests/integration -m "not real_warehouse and not dbt" -vv --color=yes -n auto --dist loadfile --max-worker-restart=0 2>&1 | tee "$$log"; \
 	status=$${PIPESTATUS[0]}; \
 	echo "TEST_EXIT=$$status (log: $$log)" | tee -a "$$log"; \
 	exit $$status
@@ -139,7 +139,9 @@ E2E_DUCKDB_CLI_DATA_PATHS := \
 	tests/e2e/src/sqlbuild/cli/commands/main/scenario/test_promotion.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/scenario/test_concurrency.py \
 	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_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
 
 E2E_DUCKDB_CLI_PATHS := \
 	tests/e2e/scripts/cli_preview \

crates/sqlbuild-analysis/src/compiler/_helpers/sql_tests/helper_scope.rs

@@ -33,6 +33,7 @@ pub(crate) struct HelperScope {
 }
 
 /// One mock CTE other test CTEs can read by its generated name.
+#[derive(Debug)]
 pub(crate) struct ScopeMock {
     pub(crate) generated_name: String,
     pub(crate) mock_name: String,
@@ -41,6 +42,7 @@ pub(crate) struct ScopeMock {
 }
 
 /// One test-defined helper: its authored name and body, and its top-level CTE name and body.
+#[derive(Debug)]
 struct ScopeHelper {
     name: String,
     original: String,
@@ -51,6 +53,7 @@ struct ScopeHelper {
 }
 
 /// A helper whose relation references were resolved like an assertion's, after the model chain.
+#[derive(Debug)]
 pub(crate) struct ResolvedHelper {
     /// The helper body with every reference replaced by the CTE that stands in for it.
     pub(crate) sql: String,
@@ -67,7 +70,7 @@ pub(crate) struct ScopeDependency<'a> {
 }
 
 /// Where helper CTEs live in the rendered test query.
-#[derive(Clone, Copy, Default, PartialEq, Eq)]
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
 enum HelperPlacement {
     /// Each top-level test CTE carries the helpers it reads in its own WITH.
     #[default]
@@ -95,7 +98,7 @@ enum ScopeNode {
 }
 
 /// Name references between one test's helper and mock CTEs, tokenized once per test.
-#[derive(Default)]
+#[derive(Debug, Default)]
 pub(crate) struct ScopeGraph {
     placement: HelperPlacement,
     helpers: Vec<ScopeHelper>,
@@ -386,6 +389,17 @@ impl ScopeGraph {
         self.collect_scope(self.closure(&identifier_tokens(sql, patterns), true))
     }
 
+    /// Top-level CTEs `sql` reads, transitively and dependencies first; mocks only when asked.
+    pub(crate) fn reader_ctes(
+        &self,
+        sql: &str,
+        patterns: &SqlTestPatterns,
+        through_mocks: bool,
+    ) -> Vec<(String, String)> {
+        self.collect_scope(self.closure(&identifier_tokens(sql, patterns), through_mocks))
+            .ctes
+    }
+
     fn collect_scope(&self, nodes: Vec<ScopeNode>) -> HelperScope {
         let mut scope = HelperScope {
             ctes: Vec::new(),

crates/sqlbuild-analysis/src/compiler/_helpers/sql_tests/mod.rs

@@ -21,4 +21,5 @@ pub mod reference_calls;
 pub mod relation_markers;
 pub mod rendering;
 pub mod scenario_extraction;
+pub(crate) mod scenario_scope;
 pub(crate) mod simple_template;

crates/sqlbuild-analysis/src/compiler/_helpers/sql_tests/planning.rs

@@ -63,8 +63,8 @@ pub(crate) const MOCK_CTE_PREFIXES: [&str; 5] = [
     DBT_REF_PREFIX,
     TABLE_FUNCTION_PREFIX,
 ];
-const EXPECTED_PREFIX: &str = "__expected__";
-const ASSERT_PREFIX: &str = "__assert__";
+pub(crate) const EXPECTED_PREFIX: &str = "__expected__";
+pub(crate) const ASSERT_PREFIX: &str = "__assert__";
 const REF_FUNCTION: &str = "__ref";
 const SOURCE_FUNCTION: &str = "__source";
 const SEED_FUNCTION: &str = "__seed";
@@ -102,7 +102,7 @@ struct ProjectContext {
     reserved_ctes: Arc<Mutex<HashMap<String, Option<String>>>>,
 }
 
-#[derive(Clone)]
+#[derive(Clone, Debug)]
 pub(crate) struct SqlTestPatterns {
     pub(crate) reference: Regex,
     pub(crate) source: Regex,
@@ -116,7 +116,7 @@ pub(crate) struct SqlTestPatterns {
 }
 
 impl SqlTestPatterns {
-    fn new(lexical: LexicalSyntax) -> Result<Self, String> {
+    pub(crate) fn new(lexical: LexicalSyntax) -> Result<Self, String> {
         Ok(Self {
             reference: compile_pattern(r#"(?i)__ref\(\"([^\"]+)\"\)"#)?,
             source: compile_pattern(r#"(?i)__source\(\"([^\"]+)\"\)"#)?,

crates/sqlbuild-analysis/src/compiler/_helpers/sql_tests/rendering.rs

@@ -696,9 +696,26 @@ fn pruned_statement(statement: &str, dialect: SliceDialect) -> String {
 
 /// Columns compared for an expected-output step: the listed expected columns, else every column.
 fn compared_projection(step: &SqlTestChainStep) -> String {
-    step.expected_columns
-        .as_ref()
-        .map_or_else(|| "*".to_string(), |columns| columns.join(", "))
+    compared_columns_projection(step.expected_columns.as_deref())
+}
+
+/// The projection comparing `columns` by name, or every column positionally without a list.
+pub(crate) fn compared_columns_projection(columns: Option<&[String]>) -> String {
+    columns.map_or_else(|| "*".to_string(), |columns| columns.join(", "))
+}
+
+/// Row counts of `actual` and `expected` and of the rows each has that the other lacks.
+pub(crate) fn comparison_counts_sql(
+    (actual, expected): (&str, &str),
+    projection: &str,
+    set_difference_operator: &str,
+) -> String {
+    format!(
+        "(SELECT COUNT(*) FROM {actual}) AS actual_count, \
+         (SELECT COUNT(*) FROM {expected}) AS expected_count, \
+         (SELECT COUNT(*) FROM (SELECT {projection} FROM {actual} {set_difference_operator} SELECT {projection} FROM {expected}) AS __sqlbuild_mismatch) AS unexpected_count, \
+         (SELECT COUNT(*) FROM (SELECT {projection} FROM {expected} {set_difference_operator} SELECT {projection} FROM {actual}) AS __sqlbuild_missing) AS missing_count"
+    )
 }
 
 /// Render one planned test's comparison query, lifting authored CTE text verbatim.
@@ -753,16 +770,14 @@ pub(crate) fn render_comparison_sql(request: &RenderRequest) -> Result<String, S
         };
         let expected_sql = cte_state.expected_step_sql(step, expected_input, enabled)?;
         comparison_ctes.push((expected_cte.clone(), expected_sql));
-        let projection = compared_projection(step);
         select_parts.push(format!(
-            "SELECT {step_index} AS step_index, '{}' AS model_name, \
-             (SELECT COUNT(*) FROM {actual_cte}) AS actual_count, \
-             (SELECT COUNT(*) FROM {expected_cte}) AS expected_count, \
-             (SELECT COUNT(*) FROM (SELECT {projection} FROM {actual_cte} {} SELECT {projection} FROM {expected_cte}) AS __sqlbuild_mismatch) AS unexpected_count, \
-             (SELECT COUNT(*) FROM (SELECT {projection} FROM {expected_cte} {} SELECT {projection} FROM {actual_cte}) AS __sqlbuild_missing) AS missing_count",
+            "SELECT {step_index} AS step_index, '{}' AS model_name, {}",
             escape_sql_string(&step.model_name),
-            request.set_difference_operator,
-            request.set_difference_operator,
+            comparison_counts_sql(
+                (&actual_cte, &expected_cte),
+                &compared_projection(step),
+                &request.set_difference_operator,
+            ),
         ));
     }
 

crates/sqlbuild-analysis/src/compiler/_helpers/sql_tests/scenario_scope.rs

@@ -0,0 +1,95 @@
+//! Scenario helper and fixture CTEs indexed through the SQL-test helper scope graph.
+
+use std::collections::BTreeMap;
+
+use crate::compiler::_helpers::sql_tests::cte_slices::SliceDialect;
+use crate::compiler::_helpers::sql_tests::helper_scope::{ScopeGraph, ScopeRequest};
+use crate::compiler::_helpers::sql_tests::planning::{
+    ASSERT_PREFIX, DBT_REF_PREFIX, EXPECTED_PREFIX, REF_PREFIX, SEED_PREFIX, SOURCE_PREFIX,
+    SqlTestPatterns, TestFixtures,
+};
+use crate::compiler::models::{ScenarioHelperScope, ScenarioScopeInput, SqlTestCte};
+
+const ERROR_PREFIX: &str = "compile_input:";
+const FIXTURE_PREFIXES: [&str; 4] = [REF_PREFIX, SOURCE_PREFIX, SEED_PREFIX, DBT_REF_PREFIX];
+
+/// Index every helper, including ones that call references, and every fixture as a mock.
+pub(crate) fn helper_scope(input: &ScenarioScopeInput<'_>) -> Result<ScenarioHelperScope, String> {
+    let patterns = SqlTestPatterns::new(input.lexical.clone()).map_err(plain)?;
+    let fixtures = scenario_fixtures(input.ctes, input.checks);
+    let read_helper_names: Vec<String> = fixtures
+        .helpers
+        .iter()
+        .map(|cte| cte.name.clone())
+        .collect();
+    let graph = ScopeGraph::new(&ScopeRequest {
+        fixtures: &fixtures,
+        patterns: &patterns,
+        file_label: input.file_label,
+        rename_dialect: None,
+        flat_dialect: true,
+        slice_dialect: SliceDialect::new(input.dialect),
+        read_helper_names: &read_helper_names,
+    })
+    .map_err(plain)?;
+    Ok(ScenarioHelperScope { graph, patterns })
+}
+
+fn scenario_fixtures(ctes: &[(String, String)], checks: &[(String, String)]) -> TestFixtures {
+    let mut fixtures = TestFixtures {
+        mock_refs: BTreeMap::new(),
+        mock_sources: BTreeMap::new(),
+        mock_seeds: BTreeMap::new(),
+        mock_dbt_refs: BTreeMap::new(),
+        mock_table_functions: BTreeMap::new(),
+        helpers: Vec::new(),
+        scope: ScopeGraph::default(),
+        expected: BTreeMap::new(),
+        assertions: Vec::new(),
+    };
+    for (name, sql) in ctes {
+        let fixture: Option<(&str, &str)> = fixture_key(name);
+        let group = match fixture {
+            Some((REF_PREFIX, _)) => &mut fixtures.mock_refs,
+            Some((SOURCE_PREFIX, _)) => &mut fixtures.mock_sources,
+            Some((SEED_PREFIX, _)) => &mut fixtures.mock_seeds,
+            Some(_) => &mut fixtures.mock_dbt_refs,
+            None => {
+                fixtures.helpers.push(SqlTestCte {
+                    name: name.clone(),
+                    sql_body: sql.clone(),
+                });
+                continue;
+            }
+        };
+        if let Some((_, key)) = fixture {
+            group.insert(key.to_owned(), sql.clone());
+        }
+    }
+    for (name, sql) in checks {
+        if let Some(model) = name.strip_prefix(EXPECTED_PREFIX) {
+            fixtures.expected.insert(model.to_owned(), sql.clone());
+        } else if let Some(assertion) = name.strip_prefix(ASSERT_PREFIX) {
+            fixtures
+                .assertions
+                .push((assertion.to_owned(), sql.clone()));
+        }
+    }
+    fixtures
+}
+
+/// The fixture prefix and name of a fixture CTE, or `None` for a helper.
+fn fixture_key(name: &str) -> Option<(&'static str, &str)> {
+    for prefix in FIXTURE_PREFIXES {
+        if let Some(key) = name.strip_prefix(prefix) {
+            return Some((prefix, key));
+        }
+    }
+    None
+}
+
+fn plain(message: String) -> String {
+    message
+        .strip_prefix(ERROR_PREFIX)
+        .map_or_else(|| message.clone(), str::to_owned)
+}

crates/sqlbuild-analysis/src/compiler/main/mod.rs

@@ -1,6 +1,11 @@
 pub mod function_type_validation;
 pub mod missing_expected_columns;
+pub mod scenario_expected_column_message;
+pub mod scenario_expected_columns;
 pub mod scenario_expected_comparison_sql;
+pub mod scenario_helper_scope;
+pub mod scenario_reader_ctes;
+pub mod scenario_reader_sql;
 pub mod sql_scenario_classification;
 pub mod sql_test_assembly;
 pub mod sql_test_chains;

crates/sqlbuild-analysis/src/compiler/main/scenario_expected_column_message.rs

@@ -0,0 +1,33 @@
+//! The pre-execution error for a scenario expected CTE whose columns do not fit its model.
+
+use crate::compiler::models::ScenarioExpectedColumns;
+
+/// Why the expected CTE cannot be compared with its model, or `None` when it can.
+#[must_use]
+pub fn scenario_expected_column_message(
+    scenario_name: &str,
+    model_name: &str,
+    columns: &ScenarioExpectedColumns,
+    all_columns: bool,
+) -> Option<String> {
+    let mut problems: Vec<String> = Vec::new();
+    if all_columns && !columns.unlisted.is_empty() {
+        problems.push(format!(
+            "does not list columns that model '{model_name}' outputs: {}",
+            columns.unlisted.join(", ")
+        ));
+    }
+    if !columns.unknown.is_empty() {
+        problems.push(format!(
+            "lists columns that model '{model_name}' does not output: {}",
+            columns.unknown.join(", ")
+        ));
+    }
+    if problems.is_empty() {
+        return None;
+    }
+    Some(format!(
+        "scenario '{scenario_name}' expected output __expected__{model_name} {}",
+        problems.join("; and ")
+    ))
+}

crates/sqlbuild-analysis/src/compiler/main/scenario_expected_columns.rs

@@ -0,0 +1,49 @@
+//! Compare a scenario `__expected__` CTE's listed columns with its model's output columns.
+
+use crate::compiler::_helpers::sql_tests::expected_columns::{
+    expected_columns, normalized_identifier,
+};
+use crate::compiler::_helpers::sql_tests::rendering::render_dialect;
+use crate::compiler::models::ScenarioExpectedColumns;
+
+/// The columns an expected CTE lists and how they differ from the known model output.
+#[must_use]
+pub fn scenario_expected_columns(
+    expected_sql: &str,
+    dialect: Option<&str>,
+    model_columns: Option<&[String]>,
+) -> ScenarioExpectedColumns {
+    let listed: Option<Vec<String>> = expected_columns(expected_sql, &render_dialect(dialect));
+    let (Some(listed_columns), Some(model_columns)) = (listed.as_deref(), model_columns) else {
+        return ScenarioExpectedColumns {
+            listed,
+            unlisted: Vec::new(),
+            unknown: Vec::new(),
+        };
+    };
+    let listed_names: Vec<String> = listed_columns
+        .iter()
+        .map(|column| normalized_identifier(column))
+        .collect();
+    let model_names: Vec<String> = model_columns
+        .iter()
+        .map(|column| normalized_identifier(column))
+        .collect();
+    let unlisted: Vec<String> = model_columns
+        .iter()
+        .zip(&model_names)
+        .filter(|(_, name)| !listed_names.contains(name))
+        .map(|(column, _)| column.clone())
+        .collect();
+    let unknown: Vec<String> = listed_columns
+        .iter()
+        .zip(&listed_names)
+        .filter(|(_, name)| !model_names.contains(name))
+        .map(|(column, _)| column.clone())
+        .collect();
+    ScenarioExpectedColumns {
+        listed,
+        unlisted,
+        unknown,
+    }
+}

crates/sqlbuild-analysis/src/compiler/main/scenario_expected_comparison_sql.rs

@@ -1,17 +1,23 @@
 //! The query comparing a scenario's actual output with its expected SQL.
 
-/// Actual, expected and mismatched row counts of one scenario expectation.
+use crate::compiler::_helpers::sql_tests::rendering::{
+    compared_columns_projection, comparison_counts_sql,
+};
+
+/// Row counts of one scenario expectation, comparing `compared_columns` by name, else all by position.
 #[must_use]
 pub fn scenario_expected_comparison_sql(
     actual_sql: &str,
     expected_sql: &str,
+    compared_columns: Option<&[String]>,
     set_difference_operator: &str,
 ) -> String {
     format!(
-        "WITH __actual AS ({actual_sql}), __expected AS ({expected_sql}) SELECT \
-         (SELECT COUNT(*) FROM __actual) AS actual_count, \
-         (SELECT COUNT(*) FROM __expected) AS expected_count, \
-         (SELECT COUNT(*) FROM (SELECT * FROM __actual {set_difference_operator} SELECT * FROM __expected\
-         ) AS __sqlbuild_mismatch) AS mismatched_count"
+        "WITH __actual AS ({actual_sql}), __expected AS ({expected_sql}) SELECT {}",
+        comparison_counts_sql(
+            ("__actual", "__expected"),
+            &compared_columns_projection(compared_columns),
+            set_difference_operator,
+        )
     )
 }