@@ -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;
 
@@ -0,0 +1 @@
+pub(crate) mod schema_order;
@@ -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))
+}
@@ -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]*$";
@@ -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
+}
@@ -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
+}
@@ -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)
+}
@@ -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,
+    }
+}
@@ -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()
+}
Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,3 @@`
	`1`	`+//! Empty fixture marker pattern.`
	`2`	`+`
	`3`	`+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]*$";`