fix(build): audit unread sources only with loading off or when selected

End-of-build source audits ran for every source no model read, on every
build, so a plain build over an unmanaged source with a missing table
failed. They now run only when loading is off (--no-load or
auto_load_sources = false), or for a source selected by name. With
loading on and no such selection, a source is audited only before a
model that reads it, as before.

An unbuilt model is now one that was not built and whose relation the
inspected warehouse does not hold, rather than one planned as a first
run, so a full refresh on a fresh warehouse skips the end audits over a
failed model.

Author: kvlonge

Makefile

@@ -112,6 +112,7 @@ E2E_DUCKDB_BUILD_CORE_PATHS := \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_terminal_failure.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_no_load_sources.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_duckdb_storage_key.py \
+	tests/e2e/src/sqlbuild/cli/commands/main/build/test_source_end_audits.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_unbuilt_dependencies.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_validation_failures.py
 

crates/sqlbuild-analysis/src/planner_scope/main/source_selectors.rs

@@ -7,11 +7,12 @@ use crate::graph::models::ParsedSelector;
 use crate::planner_scope::constants::{NAME, SOURCE};
 use std::collections::HashSet;
 
-/// Find selected sources explicitly named by upstream source or bare-name selectors.
+/// Selected sources named by a `name` or `source` selector, only `+`-prefixed ones if `upstream_only`.
 pub fn source_selectors(
     select: &[String],
     names: &[(String, ObjectKey)],
     selected: &[ObjectKey],
+    upstream_only: bool,
 ) -> Result<HashSet<ObjectKey>, SelectorError> {
     let mut keys: HashSet<ObjectKey> = HashSet::new();
     for part in select
@@ -22,12 +23,15 @@ pub fn source_selectors(
         let ParsedSelector::Kind {
             kind,
             value,
-            upstream: true,
+            upstream,
             ..
         } = parse_selector(part)?
         else {
             continue;
         };
+        if upstream_only && !upstream {
+            continue;
+        }
         if kind != NAME && kind != SOURCE {
             continue;
         }

crates/sqlbuild-python/src/bindings/_helpers/planner_entries/plan_output.rs

@@ -552,6 +552,21 @@ fn planned_sql_tests<'py>(
     )
 }
 
+/// The names of the planned models whose relation the inspected warehouse already holds.
+fn existing_model_names<'py>(
+    model_entries: &Bound<'py, PyAny>,
+    existing_relations: &Bound<'py, PyAny>,
+) -> PyResult<Bound<'py, PyAny>> {
+    let mut names: Vec<Bound<'py, PyAny>> = Vec::new();
+    for entry in model_entries.try_iter()? {
+        let name = native_field(&entry?, "name")?;
+        if existing_relations.contains(&name)? {
+            names.push(name);
+        }
+    }
+    Ok(PyFrozenSet::new(model_entries.py(), names)?.into_any())
+}
+
 /// The plan output's fields from planned entries and resolved relation inputs.
 pub(crate) fn plan_output_fields<'py>(
     inputs: PlanOutputInputs<'py>,
@@ -669,6 +684,17 @@ pub(crate) fn plan_output_fields<'py>(
         ),
         ("selected_keys", selected_keys),
         ("warnings", tuple_of(py, warnings)?),
+        (
+            "existing_model_names",
+            existing_model_names(
+                &model_entries,
+                &native_field(&snapshot, "existing_relations")?,
+            )?,
+        ),
+        (
+            "end_audit_source_keys",
+            native_field(&scope, "end_audit_source_keys")?,
+        ),
         ("upstream_deps", native_field(&scope, "upstream_deps")?),
         ("downstream_deps", native_field(&scope, "downstream_deps")?),
         (

crates/sqlbuild-python/src/bindings/_helpers/planner_records/constants.rs

@@ -52,6 +52,7 @@ const R_PLANNER_SCOPE: RecordSpec = RecordSpec {
             FieldDefault::NativeCall(RecordId::SqlTestSelection),
         ),
         FieldSpec::new("load_sources", FieldDefault::True),
+        FieldSpec::new("end_audit_source_keys", FieldDefault::EmptyFrozenset),
     ],
 };
 const R_SQL_TEST_ASSERTION_STEP: RecordSpec = RecordSpec {
@@ -926,6 +927,8 @@ const R_PLAN_OUTPUT: RecordSpec = RecordSpec {
         FieldSpec::new("python_identity_fingerprints", FieldDefault::EmptyDict),
         FieldSpec::new("metadata", FieldDefault::EmptyDict),
         FieldSpec::new("cursor_overrides", FieldDefault::None),
+        FieldSpec::new("existing_model_names", FieldDefault::EmptyFrozenset),
+        FieldSpec::new("end_audit_source_keys", FieldDefault::EmptyFrozenset),
     ],
 };
 

crates/sqlbuild-python/src/bindings/_helpers/planner_scope/scope.rs

@@ -94,14 +94,15 @@ pub(crate) fn build_planner_scope<'py>(
     let carrier = Carrier::new(py, &texts, || {
         Ok(names.iter().map(|(name, _)| name.clone()).collect())
     })?;
-    let source_keys = source_selectors(&carrier.send_all(&texts)?, &names, &keys(&selected)?)
+    let sent = carrier.send_all(&texts)?;
+    let selected_names = keys(&selected)?;
+    let source_keys = source_selectors(&sent, &names, &selected_names, true)
+        .map_err(|error| carrier.error(error))?;
+    let named_sources = source_selectors(&sent, &names, &selected_names, false)
         .map_err(|error| carrier.error(error))?;
     let cache = KeyCache::new(py)?;
-    let mut executable = PyFrozenSet::new(
-        py,
-        cache.keys(&source_keys.into_iter().collect::<Vec<_>>())?,
-    )?
-    .into_any();
+    let explicit_sources: Vec<(String, String)> = source_keys.into_iter().collect();
+    let mut executable = PyFrozenSet::new(py, cache.keys(&explicit_sources)?)?.into_any();
     if !auto_load {
         executable = PyFrozenSet::empty(py)?.into_any();
     }
@@ -118,6 +119,11 @@ pub(crate) fn build_planner_scope<'py>(
         upstream = deps.into_any();
         downstream = build_downstream_deps(&upstream)?.into_any();
     }
+    let end_audit_sources: Vec<(String, String)> = keys(&selected)?
+        .into_iter()
+        .filter(|key| key.0 == SOURCE && (!auto_load || named_sources.contains(key)))
+        .collect();
+    let end_audit_sources = PyFrozenSet::new(py, cache.keys(&end_audit_sources)?)?.into_any();
     let origins = build_execution_edge_origins(&project)?.into_any();
     let order = topologically_order_keys(&upstream, Some(origins))?.into_any();
     let models = PyDict::new(py);
@@ -150,6 +156,7 @@ pub(crate) fn build_planner_scope<'py>(
                 "load_sources",
                 PyBool::new(py, auto_load).to_owned().into_any(),
             ),
+            ("end_audit_source_keys", end_audit_sources),
         ],
     )
 }

src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/build.md

@@ -91,7 +91,7 @@ This replaces the former `sqb run` command. The full lifecycle (tests + audits)
 
 1. Managed sources are loaded (unless `--no-load`, which audits their existing data instead)
 2. Selected seeds are loaded
-3. Source audits run before their dependent models (unless `--no-audits`)
+3. Source audits run before their dependent models (unless `--no-audits`); with `--no-load`, or for a source selected by name, a source no model reads is audited at the end
 4. SQL unit tests run before their target model (unless `--no-tests`)
 5. Models are materialized in DAG topological order
 6. Error-severity audits run against the staging table before promotion to the target (unless `--no-audits`)

src/sqlbuild/.agents/skills/sqlbuild/references/docs/cli/load.md

@@ -95,7 +95,7 @@ sqb build --reload
 
 With loading off, through `--no-load` or `auto_load_sources = false`, no loader runs, even for a
 source selected by name such as `sqb build --no-load --select raw_orders`. Managed sources in the
-selection stay in the plan, and their audits check the data they already hold. A source outside
-the selection is neither loaded nor audited.
+selection stay in the plan, and their audits check the data they already hold, at the end of the
+build if no model reads them. A source outside the selection is neither loaded nor audited.
 
 See [Loaders](../concepts/python-nodes/loaders.md) for full documentation on write strategies, the loader context API, and source deferral.

src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/audits.md

@@ -379,6 +379,11 @@ sources:
 
 If a source audit with `error` severity fails, all downstream models that depend on that source are blocked. This lets you catch data quality issues at the source before any transformations run.
 
+A source's audits run before the first model in the build that reads it. A source no selected model
+reads is audited at the end of the build only when loading is off (`--no-load` or
+`auto_load_sources = false`) or when the source is selected by name, such as
+`sqb build --select raw_orders`; otherwise its audits do not run.
+
 A source audit that reads other resources waits for them before any model that depends on the
 source is built. It may read a model that is not built from the source; reading one that is fails
 with `P005`. See [Audits that read other resources](#audits-that-read-other-resources).

src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/python-nodes/loaders.md

@@ -340,8 +340,8 @@ sqb build --reload
 
 With loading off, through `--no-load` or `auto_load_sources = false`, no loader runs, even for a
 source selected by name such as `sqb build --no-load --select raw_orders`. Managed sources in the
-selection stay in the plan, and their audits check the data they already hold. A source outside
-the selection is neither loaded nor audited.
+selection stay in the plan, and their audits check the data they already hold, at the end of the
+build if no model reads them. A source outside the selection is neither loaded nor audited.
 
 When `--reload` is passed, `ctx.is_reload` is `True` in the loader function. This lets loaders implement different behavior for full reloads versus normal incremental loads.
 

src/sqlbuild/.agents/skills/sqlbuild/references/docs/concepts/sources.md

@@ -76,6 +76,11 @@ sources:
 
 If a source audit with `error` severity fails, all downstream models that depend on that source are blocked.
 
+A source's audits run before the first model in the build that reads it. A source no selected model
+reads is audited at the end of the build only when loading is off (`--no-load` or
+`auto_load_sources = false`) or when the source is selected by name, such as
+`sqb build --select raw_orders`; otherwise its audits do not run.
+
 A source audit may also read other resources; SQLBuild builds them first. See
 [Audits that read other resources](audits.md#audits-that-read-other-resources).