refactor: host migration state, plan entries and SQL-test planning in Rust

crates/sqlbuild-analysis/src/planner_fixtures/_helpers/mod.rs

@@ -1 +1,3 @@
+pub(crate) mod resolved_reads;
 pub(crate) mod schema_order;
+pub(crate) mod stub_markers;

crates/sqlbuild-analysis/src/planner_fixtures/_helpers/resolved_reads.rs

@@ -0,0 +1,48 @@
+//! Bounded reuse of native compiler read resolution.
+
+use crate::planner_fixtures::models::{FixtureReadCache, FixtureReadKey};
+use crate::required_columns::errors::RequiredColumnsError;
+use crate::required_columns::main::resolve_required_columns::resolve_required_columns;
+use crate::required_columns::models::{ColumnRead, ReferenceInput};
+
+/// Cache successful read resolution by the full original Python cache identity.
+pub(crate) fn resolved_reads(
+    mut cache: FixtureReadCache,
+    key: &FixtureReadKey,
+) -> (
+    FixtureReadCache,
+    Result<Vec<ColumnRead>, RequiredColumnsError>,
+) {
+    cache.clock += 1;
+    let clock = cache.clock;
+    if let Some((used, reads)) = cache.entries.get_mut(key) {
+        cache.recency.remove(used);
+        cache.recency.insert(clock, key.clone());
+        *used = clock;
+        let reads = reads.clone();
+        return (cache, Ok(reads));
+    }
+    let references: Vec<ReferenceInput> = key
+        .references
+        .iter()
+        .map(|(kind, name, _, _)| ReferenceInput {
+            kind: kind.clone(),
+            name: name.clone(),
+        })
+        .collect();
+    cache.misses += 1;
+    let reads = match resolve_required_columns(&key.sql, &references, key.dialect.as_deref()) {
+        Ok(reads) => reads,
+        Err(error) => return (cache, Err(error)),
+    };
+    if cache.entries.len() >= crate::planner_fixtures::constants::READ_CACHE_CAPACITY
+        && let Some((_, oldest)) = cache.recency.pop_first()
+    {
+        cache.entries.remove(&oldest);
+    }
+    cache.clock += 1;
+    let clock = cache.clock;
+    cache.recency.insert(clock, key.clone());
+    cache.entries.insert(key.clone(), (clock, reads.clone()));
+    (cache, Ok(reads))
+}

crates/sqlbuild-analysis/src/planner_fixtures/_helpers/stub_markers.rs

@@ -0,0 +1,19 @@
+//! Legacy fixture star-analysis marker substitution.
+
+use crate::planner_fixtures::constants::MARKER_PATTERNS;
+use regex::Regex;
+use std::sync::LazyLock;
+
+static MARKERS: LazyLock<Result<Vec<Regex>, regex::Error>> =
+    LazyLock::new(|| MARKER_PATTERNS.into_iter().map(Regex::new).collect());
+
+/// Substitute exact marker forms, including matches in SQL comments and strings.
+pub(crate) fn stub_markers(sql: &str) -> Result<String, regex::Error> {
+    Ok(MARKERS
+        .as_ref()
+        .map_err(Clone::clone)?
+        .iter()
+        .fold(sql.to_owned(), |sql, pattern| {
+            pattern.replace_all(&sql, "$1").into_owned()
+        }))
+}

crates/sqlbuild-analysis/src/planner_fixtures/constants.rs

@@ -1,3 +1,14 @@
 //! Empty fixture marker pattern.
 
+pub(crate) const MARKER_PATTERNS: [&str; 4] = [
+    r#"__ref\("([^"]+)"\)"#,
+    r#"__seed\("([^"]+)"\)"#,
+    r#"__source\("([^"]+)"\)"#,
+    r#"__dbt_ref\("[^"]+"[\s\x1c-\x1f]*,[\s\x1c-\x1f]*"([^"]+)"\)"#,
+];
+
 pub(crate) const EMPTY_PATTERN: &str = r"(?i)^[\s\x1c-\x1f]*SELECT[\s\x1c-\x1f]+\*[\s\x1c-\x1f]+FROM[\s\x1c-\x1f]+__empty_fixture[\s\x1c-\x1f]*\([\s\x1c-\x1f]*\)[\s\x1c-\x1f]*$";
+pub(crate) const READ_CACHE_CAPACITY: usize = 4096;
+pub(crate) const FIXTURE_WHITESPACE: &str = r"[\s\x1c-\x1f]";
+pub(crate) const COLLECTION_TYPES: [&str; 6] =
+    ["ARRAY", "LIST", "MAP", "OBJECT", "STRUCT", "VARIANT"];

crates/sqlbuild-analysis/src/planner_fixtures/main/collection_mismatch.rs

@@ -0,0 +1,19 @@
+//! Static collection-versus-scalar fixture type conflicts.
+
+use sqlbuild_core::text::main::python_strip::python_strip;
+
+use crate::planner_fixtures::constants::COLLECTION_TYPES;
+
+/// Whether exactly one of two known fixture types is a collection type.
+pub fn collection_mismatch(expected: Option<&str>, inferred: Option<&str>) -> bool {
+    let (Some(expected), Some(inferred)) = (expected, inferred) else {
+        return false;
+    };
+    is_collection(expected) != is_collection(inferred)
+}
+
+fn is_collection(sql_type: &str) -> bool {
+    let upper = sql_type.to_uppercase();
+    let base = python_strip(upper.split('(').next().unwrap_or_default());
+    COLLECTION_TYPES.contains(&base)
+}

crates/sqlbuild-analysis/src/planner_fixtures/main/fixture_line.rs

@@ -0,0 +1,16 @@
+//! Source line of one SQL-test fixture CTE definition.
+
+use regex::Regex;
+
+use crate::planner_fixtures::constants::FIXTURE_WHITESPACE;
+
+/// Return the one-based line that starts a fixture CTE definition, or line one when absent.
+pub fn fixture_line(contents: &str, cte_name: &str) -> Result<usize, regex::Error> {
+    let pattern = Regex::new(&format!(
+        r"(?im)^{FIXTURE_WHITESPACE}*{}{FIXTURE_WHITESPACE}+AS{FIXTURE_WHITESPACE}*\(",
+        regex::escape(cte_name)
+    ))?;
+    Ok(pattern.find(contents).map_or(1, |found| {
+        contents[..found.start()].matches('\n').count() + 1
+    }))
+}

crates/sqlbuild-analysis/src/planner_fixtures/main/mod.rs

@@ -1,10 +1,14 @@
 pub mod align_columns;
+pub mod collection_mismatch;
 pub mod empty_fixture_error;
 pub mod empty_fixture_marker;
+pub mod fixture_line;
 pub mod fixture_schema;
 pub mod merge_metadata;
 pub mod propagate_types;
 pub mod render_completed;
 pub mod render_empty;
+pub mod required_columns;
 pub mod requires_analysis;
+pub mod star_keys;
 pub mod unknown_columns;

crates/sqlbuild-analysis/src/planner_fixtures/main/required_columns.rs

@@ -0,0 +1,97 @@
+//! Required fixture columns derived from compiler read evidence.
+
+use sqlbuild_core::text::main::python_casefold::python_casefold;
+use std::collections::HashMap;
+
+use crate::planner_fixtures::_helpers::resolved_reads::resolved_reads;
+use crate::planner_fixtures::models::{
+    FixtureReadCache, FixtureReadModel, FixtureReadRequirements, FixtureReadSchemas, RequiredColumn,
+};
+use crate::required_columns::errors::RequiredColumnsError;
+
+/// Collect direct reads and authoritative star columns in model discovery order.
+pub fn required_columns(
+    mut cache: FixtureReadCache,
+    models: &[FixtureReadModel],
+    schemas: FixtureReadSchemas<'_>,
+) -> (
+    FixtureReadCache,
+    Result<FixtureReadRequirements, RequiredColumnsError>,
+) {
+    let FixtureReadSchemas {
+        keys: fixture_keys,
+        authoritative,
+        relations,
+    } = schemas;
+    let mut result = FixtureReadRequirements::default();
+    for model in models {
+        let mut reads = model.high_confidence.clone();
+        if model.key.references.iter().any(|(kind, name, _, _)| {
+            let resource_type = match kind.as_str() {
+                "ref" => "model",
+                "source" => "source",
+                "seed" => "seed",
+                _ => return false,
+            };
+            fixture_keys.contains(&(resource_type.to_owned(), name.clone()))
+        }) {
+            let (updated, resolved) = resolved_reads(cache, &model.key);
+            cache = updated;
+            match resolved {
+                Ok(resolved) => reads.extend(resolved),
+                Err(error) => return (cache, Err(error)),
+            }
+        }
+        for read in reads {
+            let key = (read.resource_type, read.resource_name);
+            if !fixture_keys.contains(&key)
+                || authoritative
+                    .get(&key)
+                    .is_some_and(|names| !names.contains(&python_casefold(&read.column_name)))
+            {
+                continue;
+            }
+            result = add_required(result, key, &read.column_name, &model.name);
+        }
+        for key in &model.stars {
+            result.stars.insert(key.clone());
+            if let Some(relation) = relations
+                .iter()
+                .find(|relation| &relation.key == key && relation.authoritative)
+            {
+                for column in &relation.columns {
+                    result = add_required(result, key.clone(), &column.name, &model.name);
+                }
+            }
+        }
+    }
+    (cache, Ok(result))
+}
+
+fn add_required(
+    mut result: FixtureReadRequirements,
+    key: (String, String),
+    name: &str,
+    model: &str,
+) -> FixtureReadRequirements {
+    let position = *result.indices.entry(key.clone()).or_insert_with(|| {
+        result.columns.push((key, Vec::new()));
+        result.column_indices.push(HashMap::new());
+        result.columns.len() - 1
+    });
+    let columns = &mut result.columns[position].1;
+    if let Some(index) = result.column_indices[position].get(name) {
+        let column = &mut columns[*index];
+        if column.readers.last().is_none_or(|reader| reader != model) {
+            column.readers.push(model.to_owned());
+        }
+    } else {
+        result.column_indices[position].insert(name.to_owned(), columns.len());
+        columns.push(RequiredColumn {
+            name: name.to_owned(),
+            folded_name: python_casefold(name),
+            readers: vec![model.to_owned()],
+        });
+    }
+    result
+}

crates/sqlbuild-analysis/src/planner_fixtures/main/star_keys.rs

@@ -0,0 +1,125 @@
+//! Fixture dependencies selected by star projections.
+
+use std::collections::{HashMap, HashSet};
+use std::str::FromStr;
+
+use polyglot_sql::{
+    AnalyzeQueryOptions, DialectType, SchemaColumn, SchemaTable, ValidationSchema, analyze_query,
+};
+use sqlbuild_core::text::main::python_casefold::python_casefold;
+
+use crate::planner_fixtures::_helpers::stub_markers::stub_markers;
+use crate::planner_fixtures::models::FixtureStarRelation;
+
+/// Resolve star inputs using the same semantic analysis as the compiler.
+pub fn star_keys(
+    sql: &str,
+    relations: &[FixtureStarRelation],
+    dialect: Option<&str>,
+    proven_inputs: Option<&[String]>,
+) -> Vec<(String, String)> {
+    if let Some(inputs) = proven_inputs {
+        let names: HashSet<String> = inputs.iter().map(|name| python_casefold(name)).collect();
+        return relations
+            .iter()
+            .filter(|relation| names.contains(&python_casefold(&relation.key.1)))
+            .map(|relation| relation.key.clone())
+            .collect();
+    }
+    if !relations
+        .iter()
+        .any(|relation| !relation.columns.is_empty())
+    {
+        return Vec::new();
+    }
+    let Ok(dialect) =
+        DialectType::from_str(dialect.filter(|name| !name.is_empty()).unwrap_or("generic"))
+    else {
+        return Vec::new();
+    };
+    let tables: Vec<SchemaTable> = relations
+        .iter()
+        .filter(|relation| !relation.columns.is_empty())
+        .map(schema_table)
+        .collect();
+    let options = AnalyzeQueryOptions {
+        dialect,
+        schema: Some(ValidationSchema {
+            tables,
+            strict: None,
+        }),
+        complexity_guard: None,
+    };
+    let Ok(sql) = stub_markers(sql) else {
+        return Vec::new();
+    };
+    let Ok(analysis) = analyze_query(&sql, options) else {
+        return Vec::new();
+    };
+    let mut aliases: HashMap<String, String> = HashMap::new();
+    for relation in analysis.relations {
+        let name = python_casefold(&relation.name);
+        aliases.insert(name.clone(), name.clone());
+        if let Some(alias) = relation.alias {
+            aliases.insert(python_casefold(&alias), name);
+        }
+    }
+    let mut names: HashSet<String> = HashSet::new();
+    for projection in analysis.star_projections {
+        if let Some(table) = projection.table {
+            if let Some(name) = aliases.get(&python_casefold(&table)) {
+                names.insert(name.clone());
+            }
+        } else {
+            let expanded: HashSet<String> = projection
+                .expanded_columns
+                .iter()
+                .map(|name| python_casefold(name))
+                .collect();
+            for relation in relations {
+                if relation
+                    .columns
+                    .iter()
+                    .any(|column| expanded.contains(&column.folded_name))
+                {
+                    names.insert(python_casefold(&relation.key.1));
+                }
+            }
+        }
+    }
+    relations
+        .iter()
+        .filter(|relation| names.contains(&python_casefold(&relation.key.1)))
+        .map(|relation| relation.key.clone())
+        .collect()
+}
+
+fn schema_table(relation: &FixtureStarRelation) -> SchemaTable {
+    SchemaTable {
+        name: relation.key.1.clone(),
+        schema: None,
+        columns: relation
+            .columns
+            .iter()
+            .map(|column| SchemaColumn {
+                name: column.name.clone(),
+                data_type: column_type(column.sql_type.as_deref()),
+                nullable: None,
+                primary_key: false,
+                unique: false,
+                references: None,
+            })
+            .collect(),
+        aliases: Vec::new(),
+        primary_key: Vec::new(),
+        unique_keys: Vec::new(),
+        foreign_keys: Vec::new(),
+    }
+}
+
+fn column_type(sql_type: Option<&str>) -> String {
+    sql_type
+        .filter(|name| !name.is_empty())
+        .unwrap_or("UNKNOWN")
+        .to_owned()
+}

crates/sqlbuild-analysis/src/planner_fixtures/mod.rs

@@ -7,3 +7,4 @@ pub mod main;
 pub mod models;
 #[cfg(test)]
 mod tests;
+pub mod types;