refactor: host migration, cursor and fixture planning in Rust

crates/sqlbuild-analysis/src/lib.rs

@@ -20,8 +20,10 @@ pub mod lineage_command;
 pub mod lineage_graph;
 pub mod manifest;
 pub mod planner_changes;
+pub mod planner_fixtures;
 pub mod planner_identity;
 pub mod planner_migrations;
+pub mod planner_scenario;
 pub mod planner_scope;
 pub mod planner_selection;
 pub mod planner_strategy;

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

@@ -0,0 +1 @@
+pub(crate) mod schema_order;

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

@@ -0,0 +1,15 @@
+//! Fixture alignment phases.
+
+use crate::planner_fixtures::models::AlignmentInput;
+
+pub(crate) fn supplied_in_schema_order(input: &AlignmentInput<'_>) -> bool {
+    input
+        .supplied
+        .iter()
+        .map(|column| &column.folded_name)
+        .eq(input
+            .metadata
+            .iter()
+            .take(input.supplied.len())
+            .map(|column| &column.folded_name))
+}

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

@@ -0,0 +1,3 @@
+//! Empty fixture marker pattern.
+
+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]*$";

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

@@ -0,0 +1,96 @@
+//! Align supplied columns to their authoritative schema without rendering adapter SQL.
+
+use std::collections::{BTreeSet, HashMap, HashSet};
+
+use crate::planner_fixtures::_helpers::schema_order::supplied_in_schema_order;
+use crate::planner_fixtures::models::{Alignment, AlignmentInput, RequiredColumn};
+
+/// Resolve missing columns, diagnostics and star-prefix ordering for one fixture.
+pub fn align_columns(label: &str, input: AlignmentInput<'_>) -> Alignment {
+    let missing = missing_columns(&input);
+    if missing.is_empty() && !input.typed_nulls {
+        return Alignment::default();
+    }
+    let metadata: HashMap<&str, usize> = input
+        .metadata
+        .iter()
+        .enumerate()
+        .map(|(index, column)| (column.folded_name.as_str(), index))
+        .collect();
+    let mut unknown: Vec<&str> = Vec::new();
+    let mut non_nullable: Vec<&str> = Vec::new();
+    let mut readers: BTreeSet<&str> = BTreeSet::new();
+    let mut completed: Vec<usize> = Vec::new();
+    for required in missing {
+        match metadata.get(required.folded_name.as_str()).copied() {
+            Some(index) if input.metadata[index].sql_type.is_some() => {
+                if input.metadata[index].nullable == Some(false) {
+                    non_nullable.push(required.name.as_str());
+                } else {
+                    completed.push(index);
+                }
+            }
+            _ => {
+                unknown.push(required.name.as_str());
+                readers.extend(required.readers.iter().map(String::as_str));
+            }
+        }
+    }
+    let diagnostics = missing_diagnostics(label, &unknown, &non_nullable, readers);
+    if !diagnostics.is_empty() {
+        return Alignment {
+            diagnostics,
+            completed: None,
+        };
+    }
+    if input.star && !supplied_in_schema_order(&input) {
+        return Alignment {
+            diagnostics: vec![format!(
+                "{label} supplies columns out of authoritative order for a SELECT * closure; provide the leading relation columns in schema order"
+            )],
+            completed: None,
+        };
+    }
+    completed.sort_unstable();
+    Alignment {
+        diagnostics,
+        completed: Some(completed),
+    }
+}
+
+fn missing_columns<'a>(input: &AlignmentInput<'a>) -> Vec<&'a RequiredColumn> {
+    let supplied: HashSet<&str> = input
+        .supplied
+        .iter()
+        .map(|column| column.folded_name.as_str())
+        .collect();
+    let mut missing: Vec<_> = input
+        .required
+        .iter()
+        .filter(|column| !supplied.contains(column.folded_name.as_str()))
+        .collect();
+    missing.sort_by(|left, right| left.name.cmp(&right.name));
+    missing
+}
+
+fn missing_diagnostics(
+    label: &str,
+    unknown: &[&str],
+    non_nullable: &[&str],
+    readers: BTreeSet<&str>,
+) -> Vec<String> {
+    let mut diagnostics: Vec<String> = Vec::new();
+    if !unknown.is_empty() {
+        diagnostics.push(format!(
+            "{label} is missing required columns: {}; provide explicit values because their types are not authoritative (read by: {})",
+            unknown.join(", "), readers.into_iter().collect::<Vec<_>>().join(", ")
+        ));
+    }
+    if !non_nullable.is_empty() {
+        diagnostics.push(format!(
+            "{label} must provide required non-nullable columns: {}",
+            non_nullable.join(", ")
+        ));
+    }
+    diagnostics
+}

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

@@ -0,0 +1,25 @@
+//! Authoritative schema requirements for empty fixtures.
+
+use crate::planner_fixtures::models::FixtureColumn;
+
+/// Return the first schema failure before rendering any empty-fixture column.
+pub fn empty_fixture_error(authoritative: bool, columns: &[FixtureColumn]) -> Option<String> {
+    if !authoritative {
+        return Some("uses __empty_fixture() but its column schema is not authoritative".into());
+    }
+    let unknown: Vec<_> = columns
+        .iter()
+        .filter(|column| column.sql_type.is_none())
+        .map(|column| column.name.as_str())
+        .collect();
+    if !unknown.is_empty() {
+        return Some(format!(
+            "uses __empty_fixture() but columns have unknown types: {}",
+            unknown.join(", ")
+        ));
+    }
+    if columns.is_empty() {
+        return Some("uses __empty_fixture() but its schema has no columns".into());
+    }
+    None
+}

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

@@ -0,0 +1,15 @@
+//! Exact empty-fixture marker recognition.
+
+use crate::planner_fixtures::constants::EMPTY_PATTERN;
+use regex::Regex;
+use std::sync::OnceLock;
+
+/// Whether SQL consists solely of the empty-fixture marker.
+pub fn empty_fixture_marker(sql: &str) -> Result<bool, regex::Error> {
+    static PATTERN: OnceLock<Result<Regex, regex::Error>> = OnceLock::new();
+    PATTERN
+        .get_or_init(|| Regex::new(EMPTY_PATTERN))
+        .as_ref()
+        .map(|pattern| pattern.is_match(sql))
+        .map_err(Clone::clone)
+}

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

@@ -0,0 +1,22 @@
+//! Derive fixture schema facts from relation metadata.
+
+use crate::planner_fixtures::models::{FixtureColumn, FixtureSchema};
+
+/// Collect authoritative folded names and last-declared known column types.
+pub fn fixture_schema(columns: &[FixtureColumn], authoritative: bool) -> FixtureSchema {
+    let mut expected_types: Vec<(String, String)> = Vec::new();
+    for column in columns {
+        if let Some(sql_type) = &column.sql_type {
+            expected_types.push((column.folded_name.clone(), sql_type.clone()));
+        }
+    }
+    FixtureSchema {
+        authoritative_names: authoritative.then(|| {
+            columns
+                .iter()
+                .map(|column| column.folded_name.clone())
+                .collect()
+        }),
+        expected_types,
+    }
+}

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

@@ -0,0 +1,35 @@
+//! Merge inferred model columns with declared schema metadata.
+
+use std::collections::HashMap;
+
+use crate::planner_fixtures::models::FixtureColumn;
+
+/// Merge declarations over inferred columns while preserving inferred order.
+pub fn merge_metadata(
+    inferred: Vec<FixtureColumn>,
+    declared: Vec<FixtureColumn>,
+) -> Vec<FixtureColumn> {
+    let mut order: Vec<String> = Vec::new();
+    let mut columns: HashMap<String, FixtureColumn> = HashMap::new();
+    for column in inferred {
+        order.push(column.folded_name.clone());
+        columns.insert(column.folded_name.clone(), column);
+    }
+    for mut column in declared {
+        if let Some(inferred) = columns.get(&column.folded_name) {
+            if column.sql_type.as_ref().is_none_or(String::is_empty) {
+                column.sql_type.clone_from(&inferred.sql_type);
+            }
+            if column.nullable.is_none() {
+                column.nullable = inferred.nullable;
+            }
+        } else {
+            order.push(column.folded_name.clone());
+        }
+        columns.insert(column.folded_name.clone(), column);
+    }
+    order
+        .into_iter()
+        .filter_map(|name| columns.get(&name).cloned())
+        .collect()
+}

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

@@ -0,0 +1,10 @@
+pub mod align_columns;
+pub mod empty_fixture_error;
+pub mod empty_fixture_marker;
+pub mod fixture_schema;
+pub mod merge_metadata;
+pub mod propagate_types;
+pub mod render_completed;
+pub mod render_empty;
+pub mod requires_analysis;
+pub mod unknown_columns;

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

@@ -0,0 +1,39 @@
+//! Ordered direct-lineage fixture type propagation.
+
+use std::collections::BTreeSet;
+
+use crate::planner_fixtures::models::{FixtureTypeModel, FixtureTypeRead, FixtureTypeState};
+
+/// Infer a missing output type only when known upstream types agree exactly.
+pub fn propagate_types(
+    mut state: FixtureTypeState,
+    models: &[FixtureTypeModel],
+) -> FixtureTypeState {
+    for model in models {
+        for projection in &model.projections {
+            let upstream_types: BTreeSet<String> = projection
+                .upstream
+                .iter()
+                .filter_map(|read| state.known.get(&read.key)?.get(&read.column).cloned())
+                .collect();
+            if upstream_types.len() != 1 {
+                continue;
+            }
+            let Some(sql_type) = upstream_types.into_iter().next() else {
+                continue;
+            };
+            let model_types = state.known.entry(model.key.clone()).or_default();
+            if !model_types.contains_key(&projection.output_column) {
+                model_types.insert(projection.output_column.clone(), sql_type.clone());
+                state.additions.push((
+                    FixtureTypeRead {
+                        key: model.key.clone(),
+                        column: projection.output_column.clone(),
+                    },
+                    sql_type,
+                ));
+            }
+        }
+    }
+    state
+}

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

@@ -0,0 +1,40 @@
+//! Adapter-independent completion SQL with identifier callbacks.
+
+use crate::planner_fixtures::models::FixtureSqlInput;
+
+/// Render completion SQL while preserving identifier callback order.
+pub fn render_completed<E>(
+    input: FixtureSqlInput<'_>,
+    mut render: impl FnMut(&str, bool) -> Result<String, E>,
+) -> Result<String, E> {
+    let alias = render("__sqlbuild_partial_fixture", false)?;
+    let mut projections: Vec<String> = Vec::new();
+    if input.typed_nulls {
+        for column in input.supplied {
+            let name = if column.quoted {
+                render(&column.name, true)?
+            } else {
+                column.name.clone()
+            };
+            let source = format!("{alias}.{name}");
+            projections.push(match &column.expected_type {
+                Some(sql_type) => format!("CAST({source} AS {sql_type}) AS {name}"),
+                None => source,
+            });
+        }
+    } else {
+        projections.push(format!("{alias}.*"));
+    }
+    for column in input.completed {
+        let name = render(&column.name, false)?;
+        projections.push(format!(
+            "CAST(NULL AS {}) AS {name}",
+            column.sql_type.as_deref().unwrap_or("None")
+        ));
+    }
+    Ok(format!(
+        "SELECT\n  {}\nFROM (\n{}\n) AS {alias}",
+        projections.join(",\n  "),
+        input.sql
+    ))
+}

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

@@ -0,0 +1,22 @@
+//! Adapter-independent empty-fixture projection SQL.
+
+use crate::planner_fixtures::models::FixtureColumn;
+
+/// Render the validated authoritative empty-fixture schema.
+pub fn render_empty<E>(
+    columns: &[FixtureColumn],
+    mut render: impl FnMut(&str) -> Result<String, E>,
+) -> Result<String, E> {
+    let mut projections: Vec<String> = Vec::new();
+    for column in columns {
+        let name = render(&column.name)?;
+        projections.push(format!(
+            "CAST(NULL AS {}) AS {name}",
+            column.sql_type.as_deref().unwrap_or("None")
+        ));
+    }
+    Ok(format!(
+        "SELECT\n  {}\nWHERE FALSE",
+        projections.join(",\n  ")
+    ))
+}

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

@@ -0,0 +1,17 @@
+//! Select incomplete or open fixture schemas for read analysis.
+
+use std::collections::BTreeSet;
+
+use crate::planner_fixtures::models::FixtureColumn;
+
+/// Require analysis unless every authoritative name is supplied.
+pub fn requires_analysis(
+    supplied: &[FixtureColumn],
+    authoritative: Option<&BTreeSet<String>>,
+) -> bool {
+    authoritative.is_none_or(|expected| {
+        let supplied_names: BTreeSet<_> =
+            supplied.iter().map(|column| &column.folded_name).collect();
+        expected.iter().any(|name| !supplied_names.contains(name))
+    })
+}

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

@@ -0,0 +1,22 @@
+//! Diagnose supplied names outside a closed relation schema.
+
+use std::collections::BTreeSet;
+
+use crate::planner_fixtures::models::FixtureColumn;
+
+/// Report sorted unknown names while preserving duplicate supplied names.
+pub fn unknown_columns(
+    label: &str,
+    supplied: &[FixtureColumn],
+    authoritative: Option<&BTreeSet<String>>,
+) -> Option<String> {
+    let expected = authoritative?;
+    let mut unknown: Vec<_> = supplied
+        .iter()
+        .filter(|column| !expected.contains(&column.folded_name))
+        .map(|column| column.name.as_str())
+        .collect();
+    unknown.sort_unstable();
+    (!unknown.is_empty())
+        .then(|| format!("{label} supplies unknown columns: {}", unknown.join(", ")))
+}