fix: preserve full-replay history and improve SQL test planning

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;