fix(compiler)!: reject unknown config keys and keep scenario SQL as authored

Makefile

@@ -116,6 +116,7 @@ E2E_DUCKDB_BUILD_INCREMENTAL_PATHS := \
 E2E_DUCKDB_CLI_DATA_PATHS := \
 	tests/e2e/src/sqlbuild/cli/commands/main/load \
 	tests/e2e/src/sqlbuild/cli/commands/main/providers \
+	tests/e2e/src/sqlbuild/cli/commands/main/scenario/test_authored_check_sql.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/scenario/test_scenario.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/scenario/test_capture.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/scenario/test_namespace.py

crates/sqlbuild-rules-native/src/compiler/_helpers/model_headers/tokenization.rs

@@ -438,6 +438,10 @@ fn tokenize(header: &str) -> Result<Vec<HeaderToken>, String> {
             index += 1;
             continue;
         }
+        if starts_comment(&characters, index) {
+            index = comment_end(&characters, index)?;
+            continue;
+        }
         if starts_template(&characters, index) {
             let template_end = find_character(&characters, '}', index + 2)
                 .ok_or_else(|| format!("unterminated template value at position {index}"))?;
@@ -470,7 +474,11 @@ fn tokenize(header: &str) -> Result<Vec<HeaderToken>, String> {
                 continue;
             }
             let next = characters[index];
-            if is_python_whitespace(next) || is_symbol(next) || next == ':' {
+            if is_python_whitespace(next)
+                || is_symbol(next)
+                || next == ':'
+                || starts_comment(&characters, index)
+            {
                 break;
             }
             if matches!(next, '\'' | '"') {
@@ -518,6 +526,27 @@ fn read_quoted_string(characters: &[char], start: usize) -> Result<(String, usiz
     ))
 }
 
+fn starts_comment(characters: &[char], index: usize) -> bool {
+    matches!(
+        (characters.get(index), characters.get(index + 1)),
+        (Some('-'), Some('-')) | (Some('/'), Some('*'))
+    )
+}
+
+fn comment_end(characters: &[char], start: usize) -> Result<usize, String> {
+    if characters[start] == '-' {
+        return Ok(find_character(characters, '\n', start).map_or(characters.len(), |end| end + 1));
+    }
+    let mut index = start + 2;
+    while index + 1 < characters.len() {
+        if characters[index] == '*' && characters[index + 1] == '/' {
+            return Ok(index + 2);
+        }
+        index += 1;
+    }
+    Err(format!("unterminated block comment at position {start}"))
+}
+
 fn starts_template(characters: &[char], index: usize) -> bool {
     characters.get(index) == Some(&'$') && characters.get(index + 1) == Some(&'{')
 }

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

@@ -144,6 +144,47 @@ pub(crate) fn invalid_headers_return_each_exact_error_in_order() -> bool {
     true
 }
 
+pub(crate) fn header_comments_are_skipped_without_shifting_offsets() -> bool {
+    let headers = vec![
+        "materialized table -- keep model metadata\n  columns (id ()) /* it's, (fine) */"
+            .to_owned(),
+        "materialized table--trailing note\n".to_owned(),
+        "materialized table /* unclosed".to_owned(),
+    ];
+
+    let results = parse_batch(&headers).expect("worker pool builds");
+
+    let AuthoredValue::Map(values) = results[0].0.as_ref().expect("header values") else {
+        panic!("header values must be a map");
+    };
+    assert_eq!(
+        values
+            .iter()
+            .map(|(key, _)| key.as_str())
+            .collect::<Vec<_>>(),
+        vec!["materialized", "columns"]
+    );
+    assert_eq!(
+        results[0].1.as_ref().expect("column offsets"),
+        &vec![("id".to_owned(), 53, 2)]
+    );
+    let AuthoredValue::Map(values) = results[1].0.as_ref().expect("header values") else {
+        panic!("header values must be a map");
+    };
+    assert_eq!(
+        values
+            .iter()
+            .map(|(key, _)| key.as_str())
+            .collect::<Vec<_>>(),
+        vec!["materialized"]
+    );
+    assert_eq!(
+        results[2].2.as_deref(),
+        Some("unterminated block comment at position 19")
+    );
+    true
+}
+
 pub(crate) fn nested_and_root_columns_return_only_root_offsets() -> bool {
     let headers = vec![
         "config (columns (nested (type INTEGER))), columns (top (type INTEGER))".to_owned(),

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

@@ -1,5 +1,6 @@
 use crate::compiler::tests::helpers::{
-    batch_sizes_bound_workers_by_contract, invalid_headers_return_each_exact_error_in_order,
+    batch_sizes_bound_workers_by_contract, header_comments_are_skipped_without_shifting_offsets,
+    invalid_headers_return_each_exact_error_in_order,
     nested_and_root_columns_return_only_root_offsets,
     nested_authored_headers_preserve_values_and_offsets,
 };
@@ -18,6 +19,11 @@ fn given_model_header_cases_when_exercising_native_parser_then_expected_behavior
             run: invalid_headers_return_each_exact_error_in_order,
             expected_success: true,
         },
+        ModelHeaderTokenizationTestCase {
+            description: "line and block comments are skipped without shifting offsets",
+            run: header_comments_are_skipped_without_shifting_offsets,
+            expected_success: true,
+        },
         ModelHeaderTokenizationTestCase {
             description: "nested columns do not leak into root offsets",
             run: nested_and_root_columns_return_only_root_offsets,

crates/sqlbuild-rules-native/src/semantic_validation/_helpers/positions.rs

@@ -8,7 +8,7 @@ use pyo3::exceptions::PyValueError;
 use pyo3::{PyResult, pymethods};
 use regex::Regex;
 use std::collections::HashMap;
-use std::sync::LazyLock;
+use std::sync::{LazyLock, OnceLock};
 
 const POPULAR_TOKEN_THRESHOLD: usize = 200;
 static TOKENS: LazyLock<Result<Regex, String>> =
@@ -42,6 +42,7 @@ impl BindingPositions {
                 authored,
                 offsets,
                 passes,
+                words: OnceLock::new(),
             })
         })
     }
@@ -59,21 +60,8 @@ impl BindingPositions {
                 .captures(message)
             {
                 let identifier = captures[1].rsplit('.').next().unwrap_or(&captures[1]);
-                let mut occurrences =
-                    self.authored
-                        .match_indices(identifier)
-                        .filter(|(offset, _)| {
-                            let before = self.authored[..*offset].chars().next_back();
-                            let after = self.authored[*offset + identifier.len()..].chars().next();
-                            !before.is_some_and(word) && !after.is_some_and(word)
-                        });
-                if let Some((offset, _)) = occurrences.next()
-                    && occurrences.next().is_none()
-                {
-                    return Ok(position(
-                        &self.lines,
-                        self.authored[..offset].chars().count(),
-                    ));
+                if let Some(offset) = self.unique_word_offset(identifier) {
+                    return Ok(position(&self.lines, offset));
                 }
             }
             let Some(query_start) = self.query_start else {
@@ -116,6 +104,53 @@ impl BindingPositions {
     }
 }
 
+impl BindingPositions {
+    /// Char offset of the only whole-word occurrence of `identifier` in the authored SQL.
+    pub(crate) fn unique_word_offset(&self, identifier: &str) -> Option<usize> {
+        if identifier.is_empty() || !identifier.chars().all(word) {
+            let mut occurrences = self
+                .authored
+                .match_indices(identifier)
+                .filter(|(offset, _)| {
+                    let before = self.authored[..*offset].chars().next_back();
+                    let after = self.authored[*offset + identifier.len()..].chars().next();
+                    !before.is_some_and(word) && !after.is_some_and(word)
+                });
+            let (offset, _) = occurrences.next()?;
+            return occurrences
+                .next()
+                .is_none()
+                .then(|| self.authored[..offset].chars().count());
+        }
+        let (offset, unique) = *self
+            .words
+            .get_or_init(|| word_occurrences(&self.authored))
+            .get(identifier)?;
+        unique.then_some(offset)
+    }
+}
+
+/// Index every maximal word run once so identifier lookups do not rescan the SQL.
+fn word_occurrences(sql: &str) -> HashMap<String, (usize, bool)> {
+    let mut occurrences: HashMap<String, (usize, bool)> = HashMap::new();
+    let mut current = String::new();
+    let mut start = 0;
+    for (index, character) in sql.chars().chain(std::iter::once(' ')).enumerate() {
+        if word(character) {
+            if current.is_empty() {
+                start = index;
+            }
+            current.push(character);
+        } else if !current.is_empty() {
+            occurrences
+                .entry(std::mem::take(&mut current))
+                .and_modify(|entry| entry.1 = false)
+                .or_insert((start, true));
+        }
+    }
+    occurrences
+}
+
 fn word(character: char) -> bool {
     character == '_' || character.is_alphanumeric()
 }

crates/sqlbuild-rules-native/src/semantic_validation/models.rs

@@ -65,6 +65,7 @@ pub(crate) struct BindingPositions {
     pub(super) cleaned_lines: Vec<usize>,
     pub(super) offsets: Vec<usize>,
     pub(super) passes: Vec<Vec<Expansion>>,
+    pub(super) words: std::sync::OnceLock<HashMap<String, (usize, bool)>>,
 }
 
 impl std::fmt::Debug for ProjectCatalog {

crates/sqlbuild-rules-native/src/semantic_validation/tests.rs

@@ -1,5 +1,7 @@
 #[path = "tests/test_function_names.rs"]
 mod function_names;
+#[path = "tests/test_positions.rs"]
+mod positions;
 #[path = "tests/test_semantic_validation.rs"]
 mod semantic_validation;
 #[path = "tests/test_types.rs"]

crates/sqlbuild-rules-native/src/semantic_validation/tests/test_positions.rs

@@ -0,0 +1,111 @@
+use crate::semantic_validation::models::BindingPositions;
+use crate::semantic_validation::tests::test_types::{
+    UniqueWordOffsetTestCase, UniqueWordScalingTestCase,
+};
+use std::sync::OnceLock;
+use std::time::{Duration, Instant};
+
+#[test]
+fn given_identifier_when_locating_unique_word_then_returns_its_char_offset() {
+    let test_cases = [
+        UniqueWordOffsetTestCase {
+            description: "single whole-word occurrence",
+            authored: "SELECT o.order_id, o.amount FROM orders AS o",
+            identifier: "amount",
+            expected_offset: Some(21),
+        },
+        UniqueWordOffsetTestCase {
+            description: "occurrence inside a longer word is not a match",
+            authored: "SELECT order_id_v2, order_id FROM orders",
+            identifier: "order_id",
+            expected_offset: Some(20),
+        },
+        UniqueWordOffsetTestCase {
+            description: "repeated word is ambiguous",
+            authored: "SELECT amount FROM orders WHERE amount > 0",
+            identifier: "amount",
+            expected_offset: None,
+        },
+        UniqueWordOffsetTestCase {
+            description: "absent word",
+            authored: "SELECT amount FROM orders",
+            identifier: "quantity",
+            expected_offset: None,
+        },
+        UniqueWordOffsetTestCase {
+            description: "offsets count characters after non-ASCII text",
+            authored: "SELECT 'café' AS label, total FROM orders",
+            identifier: "total",
+            expected_offset: Some(24),
+        },
+        UniqueWordOffsetTestCase {
+            description: "identifier with non-word characters falls back to a substring scan",
+            authored: "SELECT \"order total\" FROM orders",
+            identifier: "order total",
+            expected_offset: Some(8),
+        },
+    ];
+
+    for test_case in test_cases {
+        let located = BindingPositions {
+            authored: test_case.authored.to_owned(),
+            query_start: Some(0),
+            lines: vec![0],
+            cleaned_lines: vec![0],
+            offsets: Vec::new(),
+            passes: Vec::new(),
+            words: OnceLock::new(),
+        };
+        for lookup in ["first", "cached"] {
+            assert_eq!(
+                located.unique_word_offset(test_case.identifier),
+                test_case.expected_offset,
+                "{} ({lookup} lookup)",
+                test_case.description
+            );
+        }
+    }
+}
+
+#[test]
+fn given_many_unique_identifiers_when_locating_each_then_work_stays_linear() {
+    let test_cases = [UniqueWordScalingTestCase {
+        description: "one wide query with an unknown column per projection",
+        identifiers: 20_000,
+        expected_max_duration: Duration::from_secs(2),
+    }];
+
+    for test_case in test_cases {
+        let located = BindingPositions {
+            authored: (0..test_case.identifiers)
+                .map(|index| format!("missing_{index}"))
+                .collect::<Vec<_>>()
+                .join(",\n"),
+            query_start: Some(0),
+            lines: vec![0],
+            cleaned_lines: vec![0],
+            offsets: Vec::new(),
+            passes: Vec::new(),
+            words: OnceLock::new(),
+        };
+        let started = Instant::now();
+        let located_count = (0..test_case.identifiers)
+            .filter(|index| {
+                located
+                    .unique_word_offset(&format!("missing_{index}"))
+                    .is_some()
+            })
+            .count();
+        assert_eq!(
+            located_count, test_case.identifiers,
+            "{}",
+            test_case.description
+        );
+        assert!(
+            started.elapsed() < test_case.expected_max_duration,
+            "{}: took {:?}",
+            test_case.description,
+            started.elapsed()
+        );
+    }
+}

crates/sqlbuild-rules-native/src/semantic_validation/tests/test_types.rs

@@ -24,3 +24,16 @@ pub(crate) struct LocatedFunctionDiagnosticTestCase {
     pub(crate) expected_location: (Option<usize>, Option<usize>),
     pub(crate) expected_span: (Option<usize>, Option<usize>),
 }
+
+pub(crate) struct UniqueWordOffsetTestCase {
+    pub(crate) description: &'static str,
+    pub(crate) authored: &'static str,
+    pub(crate) identifier: &'static str,
+    pub(crate) expected_offset: Option<usize>,
+}
+
+pub(crate) struct UniqueWordScalingTestCase {
+    pub(crate) description: &'static str,
+    pub(crate) identifiers: usize,
+    pub(crate) expected_max_duration: std::time::Duration,
+}

crates/sqlbuild-rules-native/src/sql_lint/_helpers/token_layout.rs

@@ -554,7 +554,10 @@ fn separators(
             " ".to_string()
         };
         separators[next] = if substitution {
-            let oracle_gap = gap(next_at - 1, next_at);
+            let oracle_gap = (previous_at + 1..next_at)
+                .rev()
+                .find(|&at| generated[at].key == authored[last].key)
+                .map_or_else(|| gap(next_at - 1, next_at), |at| gap(at, at + 1));
             let unspaced = default_space(&authored[last], &authored[next]).is_empty();
             if unspaced && !oracle_gap.contains('\n') {
                 String::new()

crates/sqlbuild-rules-native/src/sql_lint/tests/test_authored_tokens.rs

@@ -58,6 +58,18 @@ fn given_generator_restructuring_when_formatting_then_authored_tokens_are_laid_o
             sql: "select substring(code, 1, 2) as region from orders",
             expected_sql: "SELECT SUBSTRING(code, 1, 2) AS region\nFROM orders",
         },
+        test_types::DialectFormatTestCase {
+            description: "DuckDB adjacent string literals after a quoted-in-oracle alias keep comma spacing",
+            dialect: "duckdb",
+            sql: "select sku as replace, 'north' 'east' as region, sku from products",
+            expected_sql: "SELECT\n  sku AS replace,\n  'north' 'east' AS region,\n  sku\nFROM products",
+        },
+        test_types::DialectFormatTestCase {
+            description: "DuckDB adjacent string literals after a plain column keep comma spacing",
+            dialect: "duckdb",
+            sql: "select sku, 'north' 'east' as region from products",
+            expected_sql: "SELECT\n  sku,\n  'north' 'east' AS region\nFROM products",
+        },
         test_types::DialectFormatTestCase {
             description: "DuckDB SELECT ALL keeps ALL",
             dialect: "duckdb",