fix: consolidate shared behavior and correct snapshot histories

AGENTS.md

@@ -120,6 +120,7 @@
 - Treat a genuine duplicate as evidence of possible wider drift, not only a cleanup task: a missing shared owner, parallel implementations of one concept, logic on the wrong side of a boundary, or copies that have already diverged (use `--diff`; a fix present in only one copy may be a bug in the others). Diagnose and record that wider smell before consolidating, because removing the copies erases the only deterministic signal of it.
 - Consolidate genuine duplication the change introduces or touches. Report unrelated findings as follow-ups instead of refactoring them opportunistically.
 - Members marked `[forced]` are adapter contract overrides required by `test_strict_adapter.py`. Record other intentional mirrors in `[dupes]` in `fensu.toml` with a reason.
+- Public adapter contract methods are duplicated per adapter on purpose: `test_strict_adapter.py` requires every first-class adapter to define them, and `[dupes]` exempts them. The private helpers behind them must not be copied. Keep them in shared builders, such as the snapshot SQL builders in `src/sqlbuild/adapter/contract/classes/`, and let each contract method delegate. Express real dialect differences as explicit parameters, such as a `SnapshotSqlDialect` field, not as a forked copy of the helper.
 
 ## Bounded Review Process
 

Makefile

@@ -73,6 +73,7 @@ E2E_DUCKDB_BUILD_CORE_PATHS := \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_no_tests_no_audits_flags.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_plan_command_surface.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_python_hooks.py \
+	tests/e2e/src/sqlbuild/cli/commands/main/build/test_sql_hook_identity.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_query_change_tracking.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_query_propagation.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_remove_column_semantics.py \

crates/sqlbuild-rules-native/src/compiler/_helpers/sql_references/extraction.rs

@@ -1,6 +1,9 @@
 //! Conservative native fast path for logical SQL reference extraction.
 
 use crate::constants::{DBT_REFERENCE_KIND, TABLE_FUNCTION_REFERENCE_KIND};
+use crate::sql_scan::main::non_code_end::non_code_end;
+use crate::sql_scan::main::quote_end::quote_end;
+use crate::sql_scan::models::QuotePolicy;
 
 const PREFIXES: [(&str, &str); 6] = [
     ("__dbt_ref(", "dbt_ref"),
@@ -18,20 +21,13 @@ pub(crate) fn extract(sql: &str) -> Option<Vec<StaticReference>> {
     let mut references: Vec<StaticReference> = Vec::new();
     let mut index = 0;
     while index < bytes.len() {
-        if bytes[index..].starts_with(b"--") {
-            index = sql[index + 2..]
-                .find('\n')
-                .map_or(bytes.len(), |offset| index + offset + 3);
-            continue;
-        }
-        if bytes[index..].starts_with(b"/*") {
-            let offset = sql[index + 2..].find("*/")?;
-            index += offset + 4;
-            continue;
-        }
-        if matches!(bytes[index], b'\'' | b'"' | b'`') {
-            index = quoted_end(bytes, index)?;
-            continue;
+        match non_code_end(bytes, index, QuotePolicy::COMPILER) {
+            Ok(Some(end)) => {
+                index = end;
+                continue;
+            }
+            Ok(None) => {}
+            Err(_) => return None,
         }
         let Some((prefix, kind)) = PREFIXES
             .iter()
@@ -70,7 +66,9 @@ fn reference_name(sql: &str, start: usize) -> Option<(String, usize)> {
     let start = ascii_whitespace_end(bytes, start);
     let first = bytes.get(start).copied()?;
     if matches!(first, b'\'' | b'"') {
-        let end = quoted_end(bytes, start)?;
+        let Ok(end) = quote_end(bytes, start, QuotePolicy::COMPILER) else {
+            return None;
+        };
         return Some((sql[start + 1..end - 1].to_owned(), end));
     }
     if !is_identifier_start(first) {
@@ -86,23 +84,6 @@ fn reference_name(sql: &str, start: usize) -> Option<(String, usize)> {
     Some((sql[start..end].to_owned(), end))
 }
 
-fn quoted_end(bytes: &[u8], start: usize) -> Option<usize> {
-    let quote = bytes[start];
-    let mut index = start + 1;
-    while index < bytes.len() {
-        if bytes[index] != quote {
-            index += 1;
-            continue;
-        }
-        if matches!(quote, b'\'' | b'"') && bytes.get(index + 1).copied() == Some(quote) {
-            index += 2;
-            continue;
-        }
-        return Some(index + 1);
-    }
-    None
-}
-
 fn ascii_whitespace_end(bytes: &[u8], start: usize) -> usize {
     let mut end = start;
     while bytes

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

@@ -1,6 +1,7 @@
 //! CTE definition and leading-WITH assembly shared by SQL-test planning and rendering.
 
-use crate::compiler::_helpers::sql_tests::sql_scan::{comment_end, skip_whitespace};
+use crate::sql_scan::main::comment_end::comment_end;
+use crate::sql_scan::main::skip_whitespace::skip_whitespace;
 
 /// Render one `name AS (body)` definition, closing a trailing line comment first.
 pub(crate) fn cte_definition_sql(name: &str, sql: &str) -> String {
@@ -53,7 +54,7 @@ pub(crate) fn leading_with_prefix_end(sql: &str) -> Option<usize> {
 fn skip_leading_ignorable(sql: &str, mut index: usize) -> usize {
     loop {
         index = skip_whitespace(sql, index);
-        match comment_end(sql, index) {
+        match comment_end(sql.as_bytes(), index) {
             Ok(Some(end)) => index = end,
             Ok(None) | Err(_) => return index,
         }

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

@@ -4,12 +4,16 @@ use std::collections::{HashMap, HashSet, VecDeque};
 
 use serde::{Deserialize, Serialize};
 
-use crate::compiler::_helpers::sql_tests::sql_scan::{
-    self, Unclosed, comment_end, quoted_end, skip_whitespace,
-};
 use crate::constants::{
     DIRECT_DEPENDENCY_PATH_LENGTH, MACRO_TEST_MODE, TABLE_FUNCTION_TEST_MODE, UDF_TEST_MODE,
 };
+use crate::sql_scan::main::comment_end::comment_end;
+use crate::sql_scan::main::matching_paren::matching_paren as scan_matching_paren;
+use crate::sql_scan::main::non_code_end::non_code_end;
+use crate::sql_scan::main::quote_end::quote_end;
+use crate::sql_scan::main::skip_whitespace::skip_whitespace;
+use crate::sql_scan::models::QuotePolicy;
+use crate::sql_scan::models::Unclosed;
 
 const MODEL_PREFIXES: [&str; 8] = [
     "__macro__",
@@ -430,13 +434,13 @@ fn validate_expected(
     let branches = split_unions(&cte.1)?;
     let mut names: Vec<Vec<String>> = Vec::new();
     for branch in branches {
-        names.push(projection_names(branch, file)?);
+        names.push(projection_names(branch, file, label)?);
     }
     if let Some(first) = names.first() {
         for (index, branch) in names.iter().enumerate().skip(1) {
             if branch != first {
                 return Err(format!(
-                    "SQL test '{file}' must use the same __expected__<model> projection names and order in every set-operation branch; branch {} does not match branch 1",
+                    "SQL test '{file}' must use the same {label} projection names and order in every set-operation branch; branch {} does not match branch 1",
                     index + 1
                 ));
             }
@@ -477,9 +481,11 @@ pub(crate) fn empty_fixture_marker_matches(sql: &str) -> Result<bool, String> {
     Ok(index == sql.len())
 }
 
-fn projection_names(branch: &str, file: &str) -> Result<Vec<String>, String> {
+fn projection_names(branch: &str, file: &str, label: &str) -> Result<Vec<String>, String> {
     let start = skip_ignorable(branch, 0)?;
-    let select_end = consume_keyword(branch, start, "SELECT").ok_or_else(|| format!("SQL test '{file}' must define each __expected__<model> set-operation branch as a SELECT query"))?;
+    let select_end = consume_keyword(branch, start, "SELECT").ok_or_else(|| {
+        format!("SQL test '{file}' must define each {label} set-operation branch as a SELECT query")
+    })?;
     let mut end = branch.len();
     for keyword in [
         "FROM", "WHERE", "GROUP", "HAVING", "QUALIFY", "WINDOW", "ORDER", "LIMIT", "OFFSET",
@@ -492,16 +498,16 @@ fn projection_names(branch: &str, file: &str) -> Result<Vec<String>, String> {
     let expressions = split_top_level(&branch[select_end..end], b',')?;
     if expressions.is_empty() {
         return Err(format!(
-            "SQL test '{file}' must project at least one column in __expected__<model>"
+            "SQL test '{file}' must project at least one column in {label}"
         ));
     }
     expressions
         .into_iter()
-        .map(|expression| projection_name(expression, file))
+        .map(|expression| projection_name(expression, file, label))
         .collect()
 }
 
-fn projection_name(expression: &str, file: &str) -> Result<String, String> {
+fn projection_name(expression: &str, file: &str, label: &str) -> Result<String, String> {
     if let Some(position) = find_last_top_level_keyword(expression, "AS")? {
         let alias_start = skip_ignorable(expression, position + 2)?;
         if let Some((alias, end)) = read_identifier(expression, alias_start)
@@ -521,7 +527,7 @@ fn projection_name(expression: &str, file: &str) -> Result<String, String> {
         return Ok(alias);
     }
     Err(format!(
-        "SQL test '{file}' must alias every non-trivial __expected__<model> projection"
+        "SQL test '{file}' must alias every non-trivial {label} projection"
     ))
 }
 
@@ -862,8 +868,10 @@ fn split_unions(sql: &str) -> Result<Vec<&str>, String> {
                 values.push(value);
             }
             index = skip_ignorable(sql, end)?;
-            if let Some(all_end) = consume_keyword(sql, index, "ALL") {
-                index = skip_ignorable(sql, all_end)?;
+            if let Some(quantifier_end) = consume_keyword(sql, index, "ALL")
+                .or_else(|| consume_keyword(sql, index, "DISTINCT"))
+            {
+                index = skip_ignorable(sql, quantifier_end)?;
             }
             start = index;
             continue;
@@ -1060,34 +1068,31 @@ fn char_len(sql: &str, index: usize) -> usize {
 fn skip_ignorable(sql: &str, mut index: usize) -> Result<usize, String> {
     loop {
         index = skip_whitespace(sql, index);
-        match comment_end(sql, index).map_err(|error| scan_error_message(error, "SQL test"))? {
+        match comment_end(sql.as_bytes(), index)
+            .map_err(|error| scan_error_message(error, "SQL test"))?
+        {
             Some(end) => index = end,
             None => return Ok(index),
         }
     }
 }
 
 fn skip_non_code(sql: &str, index: usize) -> Result<usize, String> {
-    if let Some(end) =
-        comment_end(sql, index).map_err(|error| scan_error_message(error, "SQL test"))?
-    {
-        return Ok(end);
-    }
-    if matches!(byte_at(sql, index), Some(b'\'') | Some(b'"') | Some(b'`')) {
-        return skip_quote(sql, index, "SQL test");
-    }
-    Ok(index)
+    Ok(non_code_end(sql.as_bytes(), index, QuotePolicy::COMPILER)
+        .map_err(|error| scan_error_message(error, "SQL test"))?
+        .unwrap_or(index))
 }
 
 fn skip_quote(sql: &str, start: usize, context: &str) -> Result<usize, String> {
     if byte_at(sql, start).is_none() {
         return Err(format!("{context} expected a quote"));
     }
-    quoted_end(sql, start).map_err(|error| scan_error_message(error, context))
+    quote_end(sql.as_bytes(), start, QuotePolicy::COMPILER)
+        .map_err(|error| scan_error_message(error, context))
 }
 
 fn matching_paren(sql: &str, open: usize, context: &str) -> Result<usize, String> {
-    sql_scan::matching_paren(sql, open).map_err(|error| match error {
+    scan_matching_paren(sql.as_bytes(), open, QuotePolicy::COMPILER).map_err(|error| match error {
         Unclosed::Parenthesis => scan_error_message(error, context),
         _ => scan_error_message(error, "SQL test"),
     })

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

@@ -5,7 +5,10 @@ use std::collections::HashSet;
 use regex::{Captures, Regex};
 
 use crate::compiler::_helpers::sql_tests::planning::compile_error;
-use crate::compiler::_helpers::sql_tests::sql_scan::{Unclosed, matching_paren, skip_whitespace};
+use crate::sql_scan::main::matching_paren::matching_paren;
+use crate::sql_scan::main::skip_whitespace::skip_whitespace;
+use crate::sql_scan::models::QuotePolicy;
+use crate::sql_scan::models::Unclosed;
 
 pub(crate) fn replace_callable_markers<F>(
     sql: &str,
@@ -51,7 +54,7 @@ where
 }
 
 fn matching_paren_end(sql: &str, open: usize) -> Result<usize, String> {
-    matching_paren(sql, open)
+    matching_paren(sql.as_bytes(), open, QuotePolicy::COMPILER)
         .map(|close| close + 1)
         .map_err(|error| {
             compile_error(match error {

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

@@ -5,4 +5,3 @@ pub(crate) mod helper_scope;
 pub(crate) mod markers;
 pub(crate) mod planning;
 pub(crate) mod rendering;
-pub(crate) mod sql_scan;

crates/sqlbuild-rules-native/src/compiler/tests/helpers.rs

@@ -200,6 +200,28 @@ pub(crate) fn empty_model_fixture_marker_preserves_direct_mode_validation() -> b
     true
 }
 
+pub(crate) fn expected_projection_errors_name_the_expected_cte() -> bool {
+    let cases = [
+        (
+            r#"{"tests":[{"sql":"WITH __macro_actual__ AS (SELECT @status() AS status), __macro_expected__ AS (SELECT 1 + 1) SELECT 1","fileLabel":"tests/status.sql","mode":"macro"}]}"#,
+            "SQL test 'tests/status.sql' must alias every non-trivial __macro_expected__ projection",
+        ),
+        (
+            r#"{"tests":[{"sql":"WITH __udf_actual__ AS (SELECT 1 AS value), __udf_expected__ AS (SELECT 1 AS value UNION ALL SELECT 2 AS other) SELECT 1","fileLabel":"tests/function.sql","mode":"udf"}]}"#,
+            "SQL test 'tests/function.sql' must use the same __udf_expected__ projection names and order in every set-operation branch; branch 2 does not match branch 1",
+        ),
+        (
+            r#"{"tests":[{"sql":"WITH __source__raw_orders AS (SELECT 1 AS id), __expected__orders AS (SELECT 1 + 1) SELECT 1","fileLabel":"tests/orders.sql","mode":"model"}]}"#,
+            "SQL test 'tests/orders.sql' must alias every non-trivial __expected__<model> projection",
+        ),
+    ];
+    for (request, expected_error) in cases {
+        let error = extract_batch_json(request).expect_err("expected projection is rejected");
+        assert_eq!(error, expected_error);
+    }
+    true
+}
+
 pub(crate) fn concurrent_requests_initialize_shared_template_once() -> bool {
     let templates = Arc::new(Mutex::new(HashMap::new()));
     let starts = Arc::new(Barrier::new(4));