fix: preserve full-replay history and improve SQL test planning (#801)

* fix(sql-tests): render each upstream model once in the textual SQL-test chain

* refactor(sql-tests): make native planner the single SQL-test chain and comparison implementation

The Python SQL-test planner and comparison renderer duplicated the native
implementation used by compile, and the Python textual chain inlined every
upstream model once per dependency path.

- Native planning now returns each test's chain and assertion steps, model
  names and warnings, with optional rendered SQL, in one batch call.
- plan/build/test build SqlTestPlanEntry from the native plan; Python keeps
  fixture validation and plan metadata only.
- Comparison SQL, difference samples, compile target files and runtime
  artifacts all render natively, using the SQL-analysis dialect for CTE
  lifting.
- Each distinct missing mock is reported once per test instead of once per
  dependency path or chain step.
- Textual assertion fallbacks lift their chain CTEs to the top level.
- Remove the Python assembly, analysis assembly, comparison helpers and
  dead wrappers.

* fix(sql-tests): omit SQL from chain steps the comparison never renders

Native plan output carried every chain step's full upstream closure, so
long unmocked chains produced step payloads quadratic in chain length.

- Populate resolvedSql, liftedCtes and comparisonBodySql only for steps
  with an expected CTE or an assertion that reads the step's actual CTE,
  using one shared rendered-step helper for planner and renderer.
- Skip empty steps when building manifest test compiled_code.
- Add a long-chain regression test bounding step SQL by the rendered SQL
  and proving unrendered step SQL cannot change the rendered comparison.

* feat(format): accept file paths and explain path selector errors

* fix(format): keep leading CTE comments on their own lines above the CTE

* feat(lineage): accept several and kind-prefixed targets and expand selections by direction

* feat(sql-tests): compare only the columns listed in the expected CTE

SQL unit tests compared whole rows by position, so an __expected__ CTE had
to list every model column in model order.

- The native planner reads each expected CTE's column list syntactically
  (projection names, VALUES alias lists, first set-operation branch) with
  the SQL-analysis dialect, keeping authored identifier spelling. It works
  without SQL analysis and falls back to full-row comparison when the list
  is not explicit.
- Comparison counts and difference samples project both sides to the listed
  columns; the actual row count still counts every actual row.
- With SQL analysis and authoritative model columns, unknown expected
  columns fail statically before connecting. Otherwise a failed comparison
  probes the actual step's columns with a zero-row query and reports the
  missing columns as T005 instead of the warehouse error.
- Document the rule in the agent testing skill references.

* feat(sql-tests): report missing mocks in sqb test and sqb compile

Planner ERROR warnings for a SQL test, such as a reference with no mock,
were only visible through `sqb test --inspect`.

- `plan_sql_tests` now turns a test's distinct ERROR-severity planning
  messages into fixture diagnostics, so `sqb test`, `sqb build` and
  `sqb test --inspect` stop with the existing "Invalid SQL test fixtures"
  error before execution. WARNING-severity warnings are unchanged.
- Compile artifact planning reports the same messages as S000 test-phase
  compile diagnostics from the single native batch call. They count in the
  compile summary, appear in --json, and make compile exit 1. Tests with
  planning errors are not stored in the SQL-test artifact cache, so warm
  compiles report them again.

* fix(incremental): preserve full history when replay replaces a model

* fix(lineage): honor exclusions and depth for directional selectors

* fix(sql-tests): retain upstream resolution after textual fallback

* fix(plan): hide unresolved cursor placeholders in replay ranges

* perf(compile): omit unused SQL-test execution plans from artifact responses

* test(preview): force a listed-column execution error in failure scenes

Author: kvlonge

Makefile

@@ -95,6 +95,7 @@ E2E_DUCKDB_BUILD_INCREMENTAL_PATHS := \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_microbatch_replay_backfill_lifecycle.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_mixed_timestamp_grain_replay.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_model_backed_cursor_build.py \
+	tests/e2e/src/sqlbuild/cli/commands/main/build/test_replay_on_change_full_build.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_seed_watermark_build.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_snapshot_build.py
 

crates/sqlbuild-rules-native/src/bindings/_helpers/functions.rs

@@ -121,6 +121,24 @@ fn plan_and_render_sql_tests_json(py: Python<'_>, request_json: &str) -> PyResul
         .map_err(value_error)
 }
 
+#[pyfunction]
+fn resolve_sql_test_chains_json(py: Python<'_>, request_json: &str) -> PyResult<String> {
+    py.detach(|| {
+        crate::compiler::main::sql_test_chain_resolution::resolve_chains_json(request_json)
+    })
+    .map_err(value_error)
+}
+
+#[pyfunction]
+fn render_sql_test_difference_sample_json(py: Python<'_>, request_json: &str) -> PyResult<String> {
+    py.detach(|| {
+        crate::compiler::main::sql_test_difference_sampling::render_difference_sample_json(
+            request_json,
+        )
+    })
+    .map_err(value_error)
+}
+
 #[pyfunction]
 fn extract_sql_tests_json(py: Python<'_>, request_json: &str) -> PyResult<String> {
     py.detach(|| crate::compiler::main::sql_test_extraction::extract_batch_json(request_json))
@@ -321,6 +339,11 @@ pub(crate) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> {
     )?)?;
     module.add_function(wrap_pyfunction!(render_sql_test_comparisons_json, module)?)?;
     module.add_function(wrap_pyfunction!(plan_and_render_sql_tests_json, module)?)?;
+    module.add_function(wrap_pyfunction!(resolve_sql_test_chains_json, module)?)?;
+    module.add_function(wrap_pyfunction!(
+        render_sql_test_difference_sample_json,
+        module
+    )?)?;
     module.add_function(wrap_pyfunction!(extract_sql_tests_json, module)?)?;
     module.add_function(wrap_pyfunction!(parse_model_headers, module)?)?;
     module.add_function(wrap_pyfunction!(tokenize_model_header, module)?)?;

crates/sqlbuild-rules-native/src/compiler/_helpers/sql_tests/expected_columns.rs

@@ -0,0 +1,143 @@
+//! Expected-output column names read syntactically from an `__expected__` CTE body.
+
+use std::collections::HashSet;
+
+use polyglot_sql::expressions::{Identifier, Select};
+use polyglot_sql::{Dialect, Expression};
+
+/// Return the authored output column identifiers of `sql`, or `None` when they are not explicit.
+pub(crate) fn expected_columns(sql: &str, dialect: &Dialect) -> Option<Vec<String>> {
+    let mut statements = match dialect.parse(sql) {
+        Ok(statements) => statements,
+        Err(_) => return None,
+    };
+    if statements.len() != 1 {
+        return None;
+    }
+    let statement = statements.pop()?;
+    let source = AuthoredSource::new(sql);
+    let columns: Vec<String> = query_columns(&statement, &source)?;
+    let mut seen: HashSet<String> = HashSet::new();
+    if columns.is_empty()
+        || !columns
+            .iter()
+            .all(|column| seen.insert(normalized_identifier(column)))
+    {
+        return None;
+    }
+    Some(columns)
+}
+
+/// Fold an authored identifier to the case-insensitive name used for column matching.
+pub(crate) fn normalized_identifier(identifier: &str) -> String {
+    let unquoted = match identifier.chars().next() {
+        Some('"') => identifier.trim_matches('"').replace("\"\"", "\""),
+        Some('`') => identifier.trim_matches('`').replace("``", "`"),
+        Some('[') => identifier
+            .trim_start_matches('[')
+            .trim_end_matches(']')
+            .replace("]]", "]"),
+        _ => identifier.to_string(),
+    };
+    unquoted.to_lowercase()
+}
+
+struct AuthoredSource<'a> {
+    sql: &'a str,
+    char_offsets: Vec<usize>,
+}
+
+impl<'a> AuthoredSource<'a> {
+    fn new(sql: &'a str) -> Self {
+        let mut char_offsets: Vec<usize> = sql.char_indices().map(|(offset, _)| offset).collect();
+        char_offsets.push(sql.len());
+        Self { sql, char_offsets }
+    }
+
+    /// Return the identifier exactly as written, including its quoting.
+    fn identifier(&self, identifier: &Identifier) -> Option<String> {
+        let Some(span) = identifier.span.as_ref() else {
+            return (!identifier.quoted).then(|| identifier.name.clone());
+        };
+        let start = *self.char_offsets.get(span.start)?;
+        let end = *self.char_offsets.get(span.end)?;
+        let text = self.sql.get(start..end)?;
+        let consistent = if identifier.quoted {
+            text.chars().count() >= identifier.name.chars().count() + 2
+        } else {
+            text == identifier.name
+        };
+        consistent.then(|| text.to_string())
+    }
+}
+
+fn query_columns(expression: &Expression, source: &AuthoredSource<'_>) -> Option<Vec<String>> {
+    match expression {
+        Expression::Select(select) => select_columns(select, source),
+        Expression::Union(union) => query_columns(&union.left, source),
+        Expression::Intersect(intersect) => query_columns(&intersect.left, source),
+        Expression::Except(except) => query_columns(&except.left, source),
+        Expression::Paren(paren) => query_columns(&paren.this, source),
+        Expression::Subquery(subquery) if subquery.column_aliases.is_empty() => {
+            query_columns(&subquery.this, source)
+        }
+        Expression::Values(values) => identifiers(&values.column_aliases, source),
+        _ => None,
+    }
+}
+
+fn select_columns(select: &Select, source: &AuthoredSource<'_>) -> Option<Vec<String>> {
+    if let [Expression::Star(star)] = select.expressions.as_slice() {
+        if star.table.is_some()
+            || star.except.is_some()
+            || star.replace.is_some()
+            || star.rename.is_some()
+            || !select.joins.is_empty()
+        {
+            return None;
+        }
+        let from = select.from.as_ref()?;
+        let [relation] = from.expressions.as_slice() else {
+            return None;
+        };
+        return aliased_values_columns(relation, source);
+    }
+    select
+        .expressions
+        .iter()
+        .map(|projection| projection_column(projection, source))
+        .collect()
+}
+
+fn aliased_values_columns(
+    relation: &Expression,
+    source: &AuthoredSource<'_>,
+) -> Option<Vec<String>> {
+    match relation {
+        Expression::Subquery(subquery) if matches!(subquery.this, Expression::Values(_)) => {
+            identifiers(&subquery.column_aliases, source)
+        }
+        Expression::Values(values) => identifiers(&values.column_aliases, source),
+        _ => None,
+    }
+}
+
+fn projection_column(projection: &Expression, source: &AuthoredSource<'_>) -> Option<String> {
+    match projection {
+        Expression::Alias(alias) if alias.column_aliases.is_empty() => {
+            source.identifier(&alias.alias)
+        }
+        Expression::Column(column) => source.identifier(&column.name),
+        _ => None,
+    }
+}
+
+fn identifiers(identifiers: &[Identifier], source: &AuthoredSource<'_>) -> Option<Vec<String>> {
+    if identifiers.is_empty() {
+        return None;
+    }
+    identifiers
+        .iter()
+        .map(|identifier| source.identifier(identifier))
+        .collect()
+}

crates/sqlbuild-rules-native/src/compiler/_helpers/sql_tests/mod.rs

@@ -1,3 +1,4 @@
+pub(crate) mod expected_columns;
 pub(crate) mod extraction;
 pub(crate) mod planning;
 pub(crate) mod rendering;