refactor: move compile-path logic out of executor and user-code shims

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

@@ -13,3 +13,7 @@ pub(crate) const FUNCTION_TYPE_DIALECTS: [(&str, &str, DialectType); 4] = [
 pub(crate) const CTES_FIELD: &str = "ctes";
 pub(crate) const ALIAS_FIELD: &str = "alias";
 pub(crate) const NAME_FIELD: &str = "name";
+
+/// Opening and closing quotes of identifiers SQL tests list as expected columns.
+pub(crate) const SQL_TEST_IDENTIFIER_QUOTES: [(char, char); 3] =
+    [('"', '"'), ('`', '`'), ('[', ']')];

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

@@ -0,0 +1,43 @@
+//! Explain expected-output columns that a model does not produce.
+
+use sqlbuild_core::text::main::python_casefold::python_casefold;
+
+use crate::compiler::constants::SQL_TEST_IDENTIFIER_QUOTES;
+
+/// The message for listed expected columns absent from casefolded model output names, if any.
+#[must_use]
+pub fn missing_expected_columns(
+    model_name: &str,
+    expected_columns: &[String],
+    available_columns: &[String],
+) -> Option<String> {
+    let missing: Vec<&str> = expected_columns
+        .iter()
+        .filter(|column| !available_columns.contains(&normalized_identifier(column)))
+        .map(String::as_str)
+        .collect();
+    if missing.is_empty() {
+        return None;
+    }
+    Some(format!(
+        "expected output '{model_name}' lists columns that model '{model_name}' does not output: {}. Expected CTEs are compared only on the columns they list, matched by name",
+        missing.join(", ")
+    ))
+}
+
+fn normalized_identifier(identifier: &str) -> String {
+    let closing: Option<char> = identifier.chars().next().and_then(|first| {
+        SQL_TEST_IDENTIFIER_QUOTES
+            .iter()
+            .find(|(open, _)| *open == first)
+            .map(|(_, close)| *close)
+    });
+    match closing {
+        Some(close) if identifier.ends_with(close) => {
+            let inner: &str = identifier.get(1..identifier.len() - 1).unwrap_or_default();
+            let doubled: String = [close, close].iter().collect();
+            python_casefold(&inner.replace(&doubled, &close.to_string()))
+        }
+        _ => python_casefold(identifier),
+    }
+}

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

@@ -1,4 +1,6 @@
 pub mod function_type_validation;
+pub mod missing_expected_columns;
+pub mod scenario_expected_comparison_sql;
 pub mod sql_scenario_classification;
 pub mod sql_test_assembly;
 pub mod sql_test_chains;

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

@@ -0,0 +1,17 @@
+//! The query comparing a scenario's actual output with its expected SQL.
+
+/// Actual, expected and mismatched row counts of one scenario expectation.
+#[must_use]
+pub fn scenario_expected_comparison_sql(
+    actual_sql: &str,
+    expected_sql: &str,
+    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"
+    )
+}

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

@@ -1,4 +1,5 @@
 pub(crate) mod checks;
 pub(crate) mod edges;
 pub(crate) mod nodes;
+pub(crate) mod selection;
 pub(crate) mod values;

crates/sqlbuild-analysis/src/dag_artifact/_helpers/selection.rs

@@ -0,0 +1,219 @@
+//! Breadth-first closures and directed path nodes over string-keyed DAG edges.
+
+use std::collections::{BTreeMap, BTreeSet, VecDeque};
+
+use sqlbuild_core::text::main::python_isspace::python_isspace;
+use sqlbuild_core::text::main::python_str_repr::python_str_repr;
+
+use crate::dag_artifact::constants::{
+    SELECTOR_KIND_CHECK, SELECTOR_KIND_NAME, SELECTOR_KIND_PATH, SELECTOR_KIND_SEPARATOR,
+    SELECTOR_KIND_TAG, SELECTOR_KIND_TEST, SELECTOR_MODELS_PREFIX,
+};
+use crate::dag_artifact::errors::DagSelectionError;
+use crate::dag_artifact::models::{DagSelectionIndex, DagSelectionNode};
+use crate::graph::main::parse_selector::parse_selector;
+use crate::graph::models::ParsedSelector;
+
+/// Upstream and downstream neighbours of each node id.
+pub(crate) type Adjacency = BTreeMap<String, BTreeSet<String>>;
+
+/// Every id reachable from `starts`, plus the starts when `include_starts`.
+pub(crate) fn closure(
+    starts: &BTreeSet<String>,
+    edges: &Adjacency,
+    include_starts: bool,
+) -> BTreeSet<String> {
+    let mut visited: BTreeSet<String> = if include_starts {
+        starts.clone()
+    } else {
+        BTreeSet::new()
+    };
+    let mut frontier: VecDeque<&String> = starts.iter().collect();
+    while let Some(current) = frontier.pop_front() {
+        for neighbor in edges.get(current).into_iter().flatten() {
+            if visited.insert(neighbor.clone()) {
+                frontier.push_back(neighbor);
+            }
+        }
+    }
+    visited
+}
+
+/// Ids on directed paths from `start` to `end`, or `None` when `end` is not downstream.
+pub(crate) fn path_nodes(
+    (start, end): (&str, &str),
+    (upstream, downstream): (&Adjacency, &Adjacency),
+) -> Option<BTreeSet<String>> {
+    let reachable: BTreeSet<String> =
+        closure(&BTreeSet::from([start.to_owned()]), downstream, false);
+    if !reachable.contains(end) {
+        return None;
+    }
+    let mut upstream_of_end: BTreeSet<String> = BTreeSet::new();
+    let mut stack: Vec<String> = vec![end.to_owned()];
+    while let Some(current) = stack.pop() {
+        if !upstream_of_end.insert(current.clone()) {
+            continue;
+        }
+        for parent in upstream.get(&current).into_iter().flatten() {
+            if reachable.contains(parent) || parent == start {
+                stack.push(parent.clone());
+            }
+        }
+    }
+    let mut nodes: BTreeSet<String> = reachable.intersection(&upstream_of_end).cloned().collect();
+    nodes.insert(start.to_owned());
+    nodes.insert(end.to_owned());
+    Some(nodes)
+}
+
+pub(crate) fn resolve_expression(
+    index: &DagSelectionIndex<'_>,
+    expression: &str,
+) -> Result<BTreeSet<String>, DagSelectionError> {
+    let mut selected: BTreeSet<String> = BTreeSet::new();
+    for token in expression
+        .split(python_isspace)
+        .filter(|token| !token.is_empty())
+    {
+        let mut token_selection: Option<BTreeSet<String>> = None;
+        for part in token.split(',') {
+            let matched: BTreeSet<String> = resolve_atomic(index, part)?;
+            token_selection = Some(match token_selection {
+                None => matched,
+                Some(previous) => previous.intersection(&matched).cloned().collect(),
+            });
+        }
+        selected.extend(token_selection.unwrap_or_default());
+    }
+    Ok(selected)
+}
+
+fn resolve_atomic(
+    index: &DagSelectionIndex<'_>,
+    raw: &str,
+) -> Result<BTreeSet<String>, DagSelectionError> {
+    match parse_selector(raw).map_err(DagSelectionError::Selector)? {
+        ParsedSelector::Path {
+            start,
+            end,
+            upstream,
+            downstream,
+        } => {
+            let start_id: String = node_id_for_name(index, &start)?;
+            let end_id: String = node_id_for_name(index, &end)?;
+            let mut matched: BTreeSet<String> =
+                path_nodes((&start_id, &end_id), (&index.upstream, &index.downstream)).ok_or_else(
+                    || {
+                        DagSelectionError::Input(format!(
+                            "no SQLBuild DAG path exists between {} and {}",
+                            python_str_repr(&start_id),
+                            python_str_repr(&end_id)
+                        ))
+                    },
+                )?;
+            if upstream {
+                matched.extend(closure(&BTreeSet::from([start_id]), &index.upstream, true));
+            }
+            if downstream {
+                matched.extend(closure(&BTreeSet::from([end_id]), &index.downstream, true));
+            }
+            Ok(matched)
+        }
+        ParsedSelector::Kind {
+            kind,
+            value,
+            upstream,
+            downstream,
+        } => {
+            let mut matched: BTreeSet<String> = match_kind(index, &kind, &value)?;
+            let anchors: BTreeSet<String> = matched.clone();
+            if upstream {
+                matched.extend(closure(&anchors, &index.upstream, true));
+            }
+            if downstream {
+                matched.extend(closure(&anchors, &index.downstream, true));
+            }
+            Ok(matched)
+        }
+    }
+}
+
+fn match_kind(
+    index: &DagSelectionIndex<'_>,
+    kind: &str,
+    value: &str,
+) -> Result<BTreeSet<String>, DagSelectionError> {
+    let ids = |matches: &dyn Fn(&DagSelectionNode) -> bool| -> BTreeSet<String> {
+        index
+            .nodes
+            .iter()
+            .filter(|(_, node)| matches(node))
+            .map(|(id, _)| (*id).to_owned())
+            .collect()
+    };
+    Ok(match kind {
+        SELECTOR_KIND_TEST => {
+            return Err(DagSelectionError::UnitTest(format!(
+                "{SELECTOR_KIND_TEST}{SELECTOR_KIND_SEPARATOR}{value}"
+            )));
+        }
+        SELECTOR_KIND_TAG => ids(&|node| node.tags.iter().any(|tag| tag == value)),
+        SELECTOR_KIND_PATH => {
+            let folder: String = value.replace('\\', "/");
+            let folder: &str = folder.trim_matches('/');
+            ids(&|node| path_is_under(node.path.as_deref(), folder))
+        }
+        SELECTOR_KIND_CHECK => index
+            .graph
+            .checks
+            .iter()
+            .filter(|check| check.name == value || check.id == value)
+            .flat_map(|check| check.checked_asset_ids.iter().cloned())
+            .collect(),
+        SELECTOR_KIND_NAME => ids(&|node| node.name == value),
+        _ => ids(&|node| node.name == value && node.kind == kind),
+    })
+}
+
+fn path_is_under(path: Option<&str>, folder: &str) -> bool {
+    let Some(path) = path else {
+        return false;
+    };
+    let normalized: String = path.replace('\\', "/");
+    let normalized: &str = normalized.trim_matches('/');
+    let prefixed: String = if !folder.is_empty() && !folder.starts_with(SELECTOR_MODELS_PREFIX) {
+        format!("{SELECTOR_MODELS_PREFIX}{folder}")
+    } else {
+        folder.to_owned()
+    };
+    [folder, prefixed.as_str()]
+        .into_iter()
+        .filter(|candidate| !candidate.is_empty())
+        .any(|candidate| {
+            normalized == candidate
+                || normalized
+                    .strip_prefix(candidate)
+                    .is_some_and(|rest| rest.starts_with('/'))
+        })
+}
+
+fn node_id_for_name(
+    index: &DagSelectionIndex<'_>,
+    name: &str,
+) -> Result<String, DagSelectionError> {
+    let matching: Vec<&str> = index
+        .nodes
+        .iter()
+        .filter(|(_, node)| node.name == name)
+        .map(|(id, _)| *id)
+        .collect();
+    match matching.as_slice() {
+        [only] => Ok((*only).to_owned()),
+        _ => Err(DagSelectionError::Input(format!(
+            "SQLBuild selector name {} matched {} DAG nodes",
+            python_str_repr(name),
+            matching.len()
+        ))),
+    }
+}

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

@@ -27,3 +27,10 @@ pub(crate) const LOADER_NODE_META_KEY: &str = "sqlbuild_loader_node";
 pub(crate) const LOADER_DESTINATION_PREFIX: &str = "__loader__";
 /// The most parts a loader destination may have.
 pub(crate) const LOADER_MAX_PARTS: usize = 3;
+pub(crate) const SELECTOR_KIND_NAME: &str = "name";
+pub(crate) const SELECTOR_KIND_TAG: &str = "tag";
+pub(crate) const SELECTOR_KIND_PATH: &str = "path";
+pub(crate) const SELECTOR_KIND_CHECK: &str = "check";
+pub(crate) const SELECTOR_KIND_TEST: &str = "test";
+pub(crate) const SELECTOR_KIND_SEPARATOR: &str = ":";
+pub(crate) const SELECTOR_MODELS_PREFIX: &str = "models/";

crates/sqlbuild-analysis/src/dag_artifact/errors.rs

@@ -14,3 +14,14 @@ pub enum DagArtifactError {
     /// The emitter refused the encoded artifact.
     Emit(JsonEmitError),
 }
+
+/// Why a Dagster selection over the DAG artifact failed, in the order Python raised them.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum DagSelectionError {
+    /// The selector token did not parse.
+    Selector(crate::graph::errors::SelectorError),
+    /// A `test:` selector, which only `sqb test` and `sqb build` run.
+    UnitTest(String),
+    /// `DagsterDagInputError` with this message.
+    Input(String),
+}

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

@@ -1 +1,2 @@
 pub mod build_dag_json;
+pub mod resolve_dag_selection;

crates/sqlbuild-analysis/src/dag_artifact/main/resolve_dag_selection.rs

@@ -0,0 +1,50 @@
+//! Resolve SQLBuild selectors over a DAG artifact into the selected node ids.
+
+use std::collections::{BTreeMap, BTreeSet};
+
+use crate::dag_artifact::_helpers::selection::{Adjacency, resolve_expression};
+use crate::dag_artifact::errors::DagSelectionError;
+use crate::dag_artifact::models::{DagSelectionGraph, DagSelectionIndex, DagSelectionNode};
+
+/// Node ids `select` matches minus those `exclude` matches.
+pub fn resolve_dag_selection(
+    graph: &DagSelectionGraph,
+    (select, exclude): (&str, Option<&str>),
+) -> Result<BTreeSet<String>, DagSelectionError> {
+    let mut nodes: BTreeMap<&str, &DagSelectionNode> = BTreeMap::new();
+    for node in &graph.nodes {
+        nodes.insert(node.id.as_str(), node);
+    }
+    let mut upstream: Adjacency = Adjacency::new();
+    let mut downstream: Adjacency = Adjacency::new();
+    for id in nodes.keys() {
+        upstream.insert((*id).to_owned(), BTreeSet::new());
+        downstream.insert((*id).to_owned(), BTreeSet::new());
+    }
+    for (from_id, to_id) in &graph.edges {
+        if !nodes.contains_key(from_id.as_str()) || !nodes.contains_key(to_id.as_str()) {
+            continue;
+        }
+        upstream
+            .entry(to_id.clone())
+            .or_default()
+            .insert(from_id.clone());
+        downstream
+            .entry(from_id.clone())
+            .or_default()
+            .insert(to_id.clone());
+    }
+    let index: DagSelectionIndex<'_> = DagSelectionIndex {
+        graph,
+        nodes,
+        upstream,
+        downstream,
+    };
+    let mut selected: BTreeSet<String> = resolve_expression(&index, select)?;
+    if let Some(exclude) = exclude.filter(|exclude| !exclude.is_empty()) {
+        for excluded in resolve_expression(&index, exclude)? {
+            selected.remove(&excluded);
+        }
+    }
+    Ok(selected)
+}