@@ -110,6 +110,10 @@ E2E_DUCKDB_BUILD_CORE_PATHS := \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_target_concurrency.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_template_expressions.py \
 	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
 
 E2E_DUCKDB_BUILD_INCREMENTAL_PATHS := \
@@ -147,6 +151,9 @@ E2E_DUCKDB_CLI_DATA_PATHS := \
 	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_fixture_columns.py \
+	tests/e2e/src/sqlbuild/cli/commands/main/scenario/test_local_variant.py \
+	tests/e2e/src/sqlbuild/cli/commands/main/scenario/test_snapshot_staleness.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/scenario/test_unmocked_inputs.py
 
 E2E_DUCKDB_CLI_PATHS := \
 
@@ -129,7 +129,8 @@ failures are the baseline that better translation should remove:
 The other three scenarios (`campaign_roas__attributes_orders_to_channel_spend`,
 `carrier_performance__weekly_rates_from_shipment_facts` and
 `payments__fully_refunded_orders_collect_nothing`) never reach translation: their snapshots are
-reported stale straight after capture, a SQLBuild issue described under known limitations.
+reported stale straight after capture. The snapshot fingerprint now hashes the authored fixture
+SQL, so these results predate that fix; recapture the snapshots to replay them.
 
 ## Expected divergences
 
@@ -148,53 +149,15 @@ them as expected differences or refusals instead of errors:
 These SQLBuild limitations shape parts of the project; each workaround is a reviewed exception or
 noted here:
 
-- Fixture checking asks for `CONNECT BY` pseudo-columns and lambda parameters as mock columns, so
-  `int_category_paths` and `int_cart_funnel` have no unit test. Scenarios cover both, and the
-  category scenario builds the hierarchy from a source fixture for the same reason.
-- Parameterised tests (`parameters` and `cases`) fail with `L005` while `SQBRSQL` is selected, so
-  repeated cases use multi-row fixtures or `VALUES` case tables instead.
-- Macro tests can only reach project-wide macros, so folder-scoped staging macros such as
-  `cdc_latest` are covered through the models that call them.
-- Scenarios cannot build incremental models that use cursor intrinsics, so scenarios that read
-  them provide the incremental model as a fixture.
-- `sqb format` rewrites a typed `WHERE FALSE` fixture into `__empty_fixture()` even where the
-  relation's schema is not authoritative, so the empty fixtures here use `WHERE 1 = 0`.
-- A macro mock must be an unaliased `SELECT` of one string literal, which `SQBRSQL022` rejects.
-- A file with several `TEST` blocks cannot satisfy `SQBRTEST102`, which names the file after each
-  block.
-- Loaders that return rows write JSON values as `PARSE_JSON(...)` inside a `VALUES` list, which
-  Snowflake rejects because `VALUES` accepts only constants. The DuckShop loaders therefore write
-  their own tables, parsing JSON in the select list of each `INSERT` (`python/loaders/_writer.py`).
-- Once a source table exists, Snowflake planning rebinds models to the table's inspected columns.
-  SQLBuild reads those names in lower case and then quotes them, so they only match quoted
-  lower-case references, and every unquoted column reference fails with `B002`. The connection
-  therefore sets `QUOTED_IDENTIFIERS_IGNORE_CASE = true` in `session_parameters`, which the binder
-  honours. DuckShop quotes no identifier whose case matters, so the setting changes no results.
 - Snowflake does not support concurrent microbatch `delete_insert` batches (`P001`), so
   `int_web_sessions` runs its batches one at a time.
-- `sqb build --no-load` and `auto_load_sources = false` still load managed sources when the build
-  has no `--select`, because the default selection includes the sources.
 - Self-managed loaders always report `rows=0`, since SQLBuild does not see the rows they insert.
 - Compilation accepts SQL that Snowflake rejects: `CONNECT_BY_ISLEAF`, which Snowflake does not
   have, and a CTE whose body is a bare `VALUES` list. `int_category_paths` finds leaf categories
   with a join instead, and test case tables use `SELECT ... FROM VALUES ... AS v(...)`.
 - `SQBRSQL018` can only take proof from single-column keys. Append-only feeds redeliver rows, so
   their id is unique only together with `_loaded_at` (the `unique_per_delivery` audit), and the
   staging models that keep the latest delivery carry reviewed exceptions.
-- Source fixtures are not cast to the source's declared types, so a typed function over an
-  untyped `NULL` fixture column (such as `YEAR(NULL)`) fails on Snowflake; such fixtures cast it.
-- The scenario fixture checker treats lambda parameters (`i -> ...`, `(acc, q) -> ...`) as columns
-  the model reads, so the `cart_funnel` scenario mocks carry placeholder `acc`, `i` and `q` columns.
-- A `relationships` audit with a quoted relation name, `to "fct_orders"`, compiles cleanly but
-  queries an unqualified table and does not wait for it; the relation must be `to __ref(...)`.
-- A local scenario snapshot is reported stale (`X602`) straight after capture when a fixture has a
-  bare `NULL` column, which SQLBuild wraps in a typed `__sqlbuild_partial_fixture` subquery. The
-  snapshot fingerprint hashes that adapter-rendered SQL: capture renders the alias in Snowflake's
-  upper case and local replay in DuckDB's lower case, and capture can also cast columns whose types
-  it only knows from the warehouse. The fingerprint can never match, so the snapshot cannot be
-  replayed. Three of the eight scenarios are affected.
-- A unit test of a table function, and an end-of-build audit, still run when the function or the
-  relation they read was skipped, and then fail with a missing-object error.
 
 ## Compile
 
 
@@ -235,18 +235,3 @@ reason = """
 The KPI grid is every reporting day for every region and metric pair that has ever had a value, so
 days without activity report zero instead of disappearing. Both sides are small and bounded.
 """
-
-[[rules.rule_exceptions]]
-rule = "SQBRSQL022"
-path = "tests/unit/staging/storefront/test_stg_wishlist_items__uses_the_mocked_sku_normaliser.sql"
-reason = """
-A macro mock must be a bare SELECT of one string literal, so its projection cannot carry an alias.
-"""
-
-[[rules.rule_exceptions]]
-rule = "SQBRTEST102"
-path = "tests/unit/marts/catalogue/test_rpt_wishlist_conversion__edge_cases.sql"
-reason = """
-This file keeps two related edge cases for one report together as separate TEST blocks. Each block
-has its own name, so the file cannot be named after both.
-"""
@@ -22,12 +22,7 @@ WITH __ref__int_web_sessions AS (
     NULL AS landing_page_url,
     FALSE AS is_bounce,
     TRUE AS has_add_to_cart,
-    FALSE AS has_order,
-    -- Not real columns: the fixture checker asks for int_cart_funnel's lambda parameters
-    -- (i, q, acc) as mock columns. Remove these once that is fixed.
-    NULL AS acc,
-    NULL AS i,
-    NULL AS q
+    FALSE AS has_order
   UNION ALL
   SELECT
     'S2',
@@ -46,10 +41,7 @@ WITH __ref__int_web_sessions AS (
     NULL,
     FALSE,
     FALSE,
-    TRUE,
-    NULL,
-    NULL,
-    NULL
+    TRUE
   UNION ALL
   SELECT
     'S3',
@@ -68,10 +60,7 @@ WITH __ref__int_web_sessions AS (
     NULL,
     FALSE,
     FALSE,
-    FALSE,
-    NULL,
-    NULL,
-    NULL
+    FALSE
 ),
 
 __ref__stg_web_events AS (
@@ -103,9 +92,7 @@ __ref__stg_web_events AS (
       OBJECT_CONSTRUCT('sku', 'DUCK-2', 'price_minor', 900),
       OBJECT_CONSTRUCT('quantity', 1, 'price_minor', 100),
       OBJECT_CONSTRUCT('sku', 'duck-3', 'quantity', 0)
-    )::ARRAY AS items,
-    -- See the note on int_web_sessions: i is a lambda parameter, not a column.
-    NULL AS i
+    )::ARRAY AS items
   UNION ALL
   SELECT
     'e2',
@@ -125,8 +112,7 @@ __ref__stg_web_events AS (
     NULL,
     ARRAY_CONSTRUCT(
       OBJECT_CONSTRUCT('sku', 'duck-1', 'quantity', 1)
-    )::ARRAY,
-    NULL
+    )::ARRAY
   UNION ALL
   SELECT
     'e3',
@@ -153,8 +139,7 @@ __ref__stg_web_events AS (
         'price_minor',
         500
       )
-    )::ARRAY,
-    NULL
+    )::ARRAY
   UNION ALL
   SELECT
     'e4',
@@ -174,8 +159,7 @@ __ref__stg_web_events AS (
     NULL,
     ARRAY_CONSTRUCT(
       OBJECT_CONSTRUCT('sku', 'DUCK-9', 'price_minor', 9900)
-    )::ARRAY,
-    NULL
+    )::ARRAY
 ),
 
 __ref__stg_web_checkout_steps AS (
 
@@ -1,7 +1,7 @@
 TEST (name "stg_wishlist_items__uses_the_mocked_sku_normaliser");
 
 WITH __macro__normalise_sku AS (
-  SELECT '''MOCKED-SKU'''
+  SELECT '''MOCKED-SKU''' AS normalised_sku
 ),
 
 __ref__stg_wishlists AS (
 
@@ -12,8 +12,9 @@ use crate::assembly::analysis_session::constants::{
     COLUMN_AST_KIND, CTE_ALIAS_PAYLOAD_KEY, CTE_LIST_PAYLOAD_KEY, DEPENDENCY_FUNCTION_NAMES,
     FUNCTION_AST_KIND, FUNCTION_FIXTURE_TYPES, FUNCTION_PAYLOAD_KIND, HELPER_READS_DIALECT,
     LITERAL_FIXTURE_TYPES, LITERAL_PAYLOAD_KIND, MAX_FUNCTION_CALL_DEPTH, NAME_PAYLOAD_KEY,
-    NULL_PAYLOAD_KIND, RENDERED_TYPE_NAMES, SCHEMA_TABLE_ARG, SELECT_PAYLOAD_KEY, STAR_AST_KIND,
-    STRUCTURED_PAYLOAD_KINDS, TABLE_AST_KIND, VARIANT_FIXTURE_TYPE, WILDCARD,
+    NULL_PAYLOAD_KIND, RENDERED_TYPE_NAMES, SCHEMA_TABLE_ARG, SELECT_PAYLOAD_KEY,
+    SET_OPERATION_PAYLOAD_KEYS, STAR_AST_KIND, STRUCTURED_PAYLOAD_KINDS, TABLE_AST_KIND,
+    VARIANT_FIXTURE_TYPE, WILDCARD,
 };
 
 /// Python's `bool(value)`.
@@ -452,34 +453,101 @@ pub(crate) fn fixture_projection_facts(
         return Ok(ProjectionFacts::default());
     };
     let mut facts: ProjectionFacts = collected_facts(select);
+    let top: Option<&Map<String, Value>> = payload.as_object();
     let direct: bool = matches!(
-        payload
-            .as_object()
-            .and_then(|map| map.get(SELECT_PAYLOAD_KEY)),
+        top.and_then(|map| map.get(SELECT_PAYLOAD_KEY)),
         Some(Value::Object(_))
     );
-    if !direct {
+    let set_operation: bool = top.is_some_and(|map| {
+        SET_OPERATION_PAYLOAD_KEYS
+            .iter()
+            .any(|key| matches!(map.get(*key), Some(Value::Object(_))))
+    });
+    if !direct && !set_operation {
         facts.null_literal_names.clear();
     }
+    if set_operation {
+        let branches: Vec<&Map<String, Value>> = branch_selects(&payload);
+        let null_in_every_branch = |index: usize| {
+            branches
+                .iter()
+                .all(|branch| null_positions(branch).get(index) == Some(&true))
+        };
+        let kept: Vec<String> = null_names_by_position(select)
+            .into_iter()
+            .filter(|(index, _)| null_in_every_branch(*index))
+            .map(|(_, name)| name)
+            .collect();
+        facts.null_literal_names.retain(|name| kept.contains(name));
+    }
     Ok(facts)
 }
 
-/// Python's recursive `_first_select`, depth first in payload order, without recursion.
-fn first_select(payload: &Value) -> Option<&Map<String, Value>> {
+/// Every branch `SELECT` of a set operation, without the selects nested inside them.
+fn branch_selects(payload: &Value) -> Vec<&Map<String, Value>> {
+    let mut branches: Vec<&Map<String, Value>> = Vec::new();
     let mut pending: Vec<&Value> = vec![payload];
     while let Some(node) = pending.pop() {
         match node {
             Value::Object(map) => {
                 if let Some(Value::Object(select)) = map.get(SELECT_PAYLOAD_KEY) {
-                    return Some(select);
+                    branches.push(select);
+                    continue;
                 }
                 pending.extend(map.values().rev());
             }
             Value::Array(items) => pending.extend(items.iter().rev()),
             _ => {}
         }
     }
-    None
+    branches
+}
+
+/// The projection payloads of a select, each with any alias removed.
+fn projected_values(select: &Map<String, Value>) -> Vec<&Map<String, Value>> {
+    select
+        .get("expressions")
+        .and_then(Value::as_array)
+        .map_or(&[][..], Vec::as_slice)
+        .iter()
+        .filter_map(|expression| {
+            match expression.get("alias").and_then(|alias| alias.get("this")) {
+                Some(Value::Object(value)) => Some(value),
+                _ => expression.as_object(),
+            }
+        })
+        .collect()
+}
+
+/// Whether each projection of a select is a bare `NULL`.
+fn null_positions(select: &Map<String, Value>) -> Vec<bool> {
+    projected_values(select)
+        .into_iter()
+        .map(|value| value.keys().next().map(String::as_str) == Some(NULL_PAYLOAD_KIND))
+        .collect()
+}
+
+/// The folded alias of each bare `NULL` projection of a select, with its position.
+fn null_names_by_position(select: &Map<String, Value>) -> Vec<(usize, String)> {
+    let expressions: &[Value] = select
+        .get("expressions")
+        .and_then(Value::as_array)
+        .map_or(&[], Vec::as_slice);
+    expressions
+        .iter()
+        .enumerate()
+        .filter_map(|(index, expression)| {
+            let alias_payload = expression.get("alias")?;
+            let name: &str = alias_payload.get("alias")?.get("name")?.as_str()?;
+            let kind: &str = alias_payload.get("this")?.as_object()?.keys().next()?;
+            (kind == NULL_PAYLOAD_KIND).then(|| (index, str::to_lowercase(name)))
+        })
+        .collect()
+}
+
+/// Python's recursive `_first_select`, depth first in payload order, without recursion.
+fn first_select(payload: &Value) -> Option<&Map<String, Value>> {
+    branch_selects(payload).into_iter().next()
 }
 
 fn collected_facts(select: &Map<String, Value>) -> ProjectionFacts {
 
@@ -212,6 +212,8 @@ pub(crate) const ARRAY_PAYLOAD_KIND: &str = "array_func";
 pub(crate) const STRUCTURED_PAYLOAD_KINDS: [&str; 2] = ["parse_json", "struct"];
 pub(crate) const FUNCTION_PAYLOAD_KIND: &str = "function";
 pub(crate) const SELECT_PAYLOAD_KEY: &str = "select";
+/// Set operations whose output columns are named and typed by their first branch.
+pub(crate) const SET_OPERATION_PAYLOAD_KEYS: [&str; 3] = ["union", "intersect", "except"];
 /// `{literal_type: type}` Python's `_simple_expression_type` maps a literal to.
 pub(crate) const LITERAL_FIXTURE_TYPES: [(&str, &str); 3] = [
     ("string", "VARCHAR"),
 
@@ -683,6 +683,14 @@ fn macro_mock_value(
     }
     let (value, end) = read_string(body, index)?;
     index = skip_ignorable(body, end, syntax)?;
+    if let Some(alias_start) = consume_keyword(body, index, "AS") {
+        let alias: usize = skip_ignorable(body, alias_start, syntax)?;
+        let alias_end: usize = identifier_end(body, alias);
+        if alias_end == alias {
+            return Err(macro_mock_shape(file, &cte.0, true));
+        }
+        index = skip_ignorable(body, alias_end, syntax)?;
+    }
     if byte_at(body, index) == Some(b';') {
         index = skip_ignorable(body, index + 1, syntax)?;
     }
@@ -692,6 +700,20 @@ fn macro_mock_value(
     Ok(value)
 }
 
+/// The end of the plain or double-quoted identifier at `start`, or `start` when there is none.
+fn identifier_end(sql: &str, start: usize) -> usize {
+    if byte_at(sql, start) == Some(b'"') {
+        return sql[start + 1..]
+            .find('"')
+            .map_or(start, |close| start + close + 2);
+    }
+    let mut end: usize = start;
+    while char_at(sql, end).is_some_and(is_identifier_continue) {
+        end += char_len(sql, end);
+    }
+    end
+}
+
 fn macro_mock_shape(file: &str, name: &str, trailing: bool) -> String {
     if trailing {
         format!(
 
@@ -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;
         }