refactor: read planner inputs natively and build the plan output once

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

@@ -7,7 +7,13 @@ pub fn fixture_schema(columns: &[FixtureColumn], authoritative: bool) -> Fixture
     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()));
+            match expected_types
+                .iter_mut()
+                .find(|(name, _)| *name == column.folded_name)
+            {
+                Some((_, existing)) => existing.clone_from(sql_type),
+                None => expected_types.push((column.folded_name.clone(), sql_type.clone())),
+            }
         }
     }
     FixtureSchema {

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

@@ -58,14 +58,14 @@ pub struct FixtureStarRelation {
 }
 
 /// One upstream typed column identified by its resource and folded name.
-#[derive(Debug)]
+#[derive(Clone, Debug)]
 pub struct FixtureTypeRead {
     pub key: (String, String),
     pub column: String,
 }
 
 /// Direct lineage facts used to propagate fixture types.
-#[derive(Debug)]
+#[derive(Clone, Debug)]
 pub struct FixtureTypeProjection {
     pub output_column: String,
     pub upstream: Vec<FixtureTypeRead>,

crates/sqlbuild-analysis/src/planner_fixtures/tests/test_schema.rs

@@ -2,7 +2,11 @@ use crate::planner_fixtures::main::{
     fixture_schema::fixture_schema, requires_analysis::requires_analysis,
     unknown_columns::unknown_columns,
 };
-use crate::planner_fixtures::tests::{helpers::column, test_types::SchemaTestCase};
+use crate::planner_fixtures::models::FixtureColumn;
+use crate::planner_fixtures::tests::{
+    helpers::column,
+    test_types::{ExpectedTypesTestCase, SchemaTestCase},
+};
 
 #[test]
 fn given_closed_and_open_schemas_when_selecting_then_names_and_diagnostics_agree() {
@@ -86,3 +90,50 @@ fn given_closed_and_open_schemas_when_selecting_then_names_and_diagnostics_agree
         );
     }
 }
+
+#[test]
+fn given_duplicate_folded_columns_when_collecting_types_then_last_declared_type_wins() {
+    let test_cases = [
+        ExpectedTypesTestCase {
+            description: "distinct columns keep declaration order",
+            columns: &[
+                ("order_id", "order_id", "INTEGER"),
+                ("status", "status", "VARCHAR"),
+            ],
+            expected_types: &[("order_id", "INTEGER"), ("status", "VARCHAR")],
+        },
+        ExpectedTypesTestCase {
+            description: "case variants keep the first position and the last type",
+            columns: &[
+                ("order_id", "order_id", "INTEGER"),
+                ("status", "status", "INTEGER"),
+                ("STATUS", "status", "VARCHAR"),
+            ],
+            expected_types: &[("order_id", "INTEGER"), ("status", "VARCHAR")],
+        },
+    ];
+    for test_case in test_cases {
+        let columns: Vec<FixtureColumn> = test_case
+            .columns
+            .iter()
+            .map(|(name, folded_name, sql_type)| FixtureColumn {
+                name: (*name).into(),
+                folded_name: (*folded_name).into(),
+                sql_type: Some((*sql_type).into()),
+                nullable: Some(true),
+            })
+            .collect();
+        let schema = fixture_schema(&columns, false);
+        let collected: Vec<(&str, &str)> = schema
+            .expected_types
+            .iter()
+            .map(|(name, sql_type)| (name.as_str(), sql_type.as_str()))
+            .collect();
+        assert_eq!(
+            collected.as_slice(),
+            test_case.expected_types,
+            "{}",
+            test_case.description
+        );
+    }
+}

crates/sqlbuild-analysis/src/planner_fixtures/tests/test_types.rs

@@ -7,6 +7,13 @@ pub(crate) struct PropagationTestCase {
     pub(crate) expected_type: Option<&'static str>,
 }
 
+#[derive(Debug)]
+pub(crate) struct ExpectedTypesTestCase {
+    pub(crate) description: &'static str,
+    pub(crate) columns: &'static [(&'static str, &'static str, &'static str)],
+    pub(crate) expected_types: &'static [(&'static str, &'static str)],
+}
+
 #[derive(Debug)]
 pub(crate) struct SchemaTestCase {
     pub(crate) description: &'static str,

crates/sqlbuild-python/src/bindings/_helpers/analysis/model_reads.rs

@@ -6,6 +6,19 @@ use crate::bindings::_helpers::compiled_project::compiled_model_view::{
     DeclaredRow, InferredRow, LineageInput, NativeCompiledModel,
 };
 
+/// `(output, transform kind, confidence, [(resource type, resource, column)])` of one lineage output.
+pub(crate) type LineageFactRow = (String, String, String, Vec<(String, String, String)>);
+
+const TRANSFORM_VALUES: [&str; 6] = [
+    "direct",
+    "cast",
+    "expression",
+    "aggregation",
+    "star",
+    "constant",
+];
+const CONFIDENCE_VALUES: [&str; 3] = ["unknown", "high", "medium"];
+
 /// Each inferred output column as `(name, type, nullability)`, `None` when none were inferred.
 pub(crate) fn inferred_rows(model: &Bound<'_, PyAny>) -> PyResult<Option<Vec<InferredRow>>> {
     if let Ok(view) = model.downcast::<NativeCompiledModel>() {
@@ -117,3 +130,110 @@ pub(crate) fn output_proven(model: &Bound<'_, PyAny>) -> PyResult<bool> {
     let proof: Bound<'_, PyAny> = model.getattr("dynamic_column_contract")?;
     Ok(!proof.is_none() && proof.getattr("output_proven")?.is_truthy()?)
 }
+
+/// Each lineage output with its transform kind and confidence values, `None` without lineage.
+pub(crate) fn lineage_facts(model: &Bound<'_, PyAny>) -> PyResult<Option<Vec<LineageFactRow>>> {
+    if let Ok(view) = model.downcast::<NativeCompiledModel>() {
+        let Some(rows) = view.get().lineage_rows(model.py())? else {
+            return Ok(None);
+        };
+        let mut facts: Vec<LineageFactRow> = Vec::with_capacity(rows.len());
+        for row in rows {
+            facts.push((
+                row.output_column,
+                coded(&TRANSFORM_VALUES, row.transform_code)?,
+                coded(&CONFIDENCE_VALUES, row.confidence_code)?,
+                row.sources,
+            ));
+        }
+        return Ok(Some(facts));
+    }
+    let columns: Bound<'_, PyAny> = model.getattr("fast_lineage_columns")?;
+    if columns.is_none() {
+        return Ok(None);
+    }
+    let mut facts: Vec<LineageFactRow> = Vec::new();
+    for output in columns.try_iter()? {
+        let output: Bound<'_, PyAny> = output?;
+        let mut sources: Vec<(String, String, String)> = Vec::new();
+        for source in output.getattr("upstream_columns")?.try_iter()? {
+            let source: Bound<'_, PyAny> = source?;
+            sources.push((
+                source.getattr("resource_type")?.extract()?,
+                source.getattr("resource_name")?.extract()?,
+                source.getattr("column_name")?.extract()?,
+            ));
+        }
+        facts.push((
+            output.getattr("output_column")?.extract()?,
+            output.getattr("transform_kind")?.str()?.extract()?,
+            output.getattr("confidence")?.str()?.extract()?,
+            sources,
+        ));
+    }
+    Ok(Some(facts))
+}
+
+fn coded(values: &[&str], code: u8) -> PyResult<String> {
+    values
+        .get(usize::from(code))
+        .map(|value| (*value).to_owned())
+        .ok_or_else(|| pyo3::exceptions::PyIndexError::new_err("tuple index out of range"))
+}
+
+/// The input relations of a dynamic pivot proof that proved the model's outputs, else `None`.
+pub(crate) fn proven_inputs(model: &Bound<'_, PyAny>) -> PyResult<Option<Vec<String>>> {
+    if let Ok(view) = model.downcast::<NativeCompiledModel>() {
+        return view.get().proven_inputs(model.py());
+    }
+    let proof: Bound<'_, PyAny> = model.getattr("dynamic_column_contract")?;
+    if proof.is_none() || !proof.getattr("output_proven")?.is_truthy()? {
+        return Ok(None);
+    }
+    proof.getattr("input_relations")?.extract().map(Some)
+}
+
+/// The model's name.
+pub(crate) fn model_name(model: &Bound<'_, PyAny>) -> PyResult<String> {
+    if let Ok(view) = model.downcast::<NativeCompiledModel>() {
+        return view.get().name(model.py());
+    }
+    model.getattr("name")?.extract()
+}
+
+/// The model's compiled query SQL.
+pub(crate) fn query_sql<'py>(model: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    if let Ok(view) = model.downcast::<NativeCompiledModel>() {
+        return Ok(view
+            .get()
+            .query_sql(model.py())?
+            .into_bound(model.py())
+            .into_any());
+    }
+    model.getattr("query_sql")
+}
+
+/// The model's `CompiledObjectKey`.
+pub(crate) fn model_key<'py>(model: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    if let Ok(view) = model.downcast::<NativeCompiledModel>() {
+        return Ok(view.get().key(model.py())?.into_bound(model.py()));
+    }
+    model.getattr("key")
+}
+
+/// The model's destination `CompiledRelationLocation`.
+pub(crate) fn destination<'py>(model: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    if let Ok(view) = model.downcast::<NativeCompiledModel>() {
+        return Ok(view.get().destination(model.py())?.into_bound(model.py()));
+    }
+    model.getattr("destination")
+}
+
+/// The model's name as a Python string.
+pub(crate) fn name_object<'py>(model: &Bound<'py, PyAny>) -> PyResult<Bound<'py, PyAny>> {
+    if let Ok(view) = model.downcast::<NativeCompiledModel>() {
+        let name: String = view.get().name(model.py())?;
+        return Ok(pyo3::types::PyString::new(model.py(), &name).into_any());
+    }
+    model.getattr("name")
+}

crates/sqlbuild-python/src/bindings/_helpers/analysis_session/fixture_inferences.rs

@@ -17,7 +17,7 @@ pub(crate) struct FixtureColumnInferences {
 #[pymethods]
 impl FixtureColumnInferences {
     #[new]
-    fn new() -> Self {
+    pub(crate) fn new() -> Self {
         Self {
             by_profile: Mutex::new(Vec::new()),
         }

crates/sqlbuild-python/src/bindings/_helpers/attachments/reference_markers.rs

@@ -32,7 +32,7 @@ fn coded_value_error(py: Python<'_>, message: String, code: &str) -> PyResult<Py
 /// Fail if executable SQL still contains unresolved reference markers or an unclosed construct.
 #[pyfunction]
 #[pyo3(signature = (*, sql, context))]
-fn assert_no_unresolved_sql_markers(
+pub(crate) fn assert_no_unresolved_sql_markers(
     py: Python<'_>,
     sql: &Bound<'_, PyString>,
     context: &str,

crates/sqlbuild-python/src/bindings/_helpers/compile_driver/diff.rs

@@ -12,6 +12,7 @@ use sqlbuild_analysis::diff_selection::errors::SelectorError;
 use sqlbuild_analysis::diff_selection::main::resolve_diff_selection::resolve_diff_selection;
 use sqlbuild_analysis::diff_selection::models::DiffGraph;
 
+use crate::bindings::_helpers::analysis::model_reads;
 use crate::bindings::_helpers::compile_driver::python_calls::{
     KeywordArguments, call_with, imported, keywords,
 };
@@ -165,13 +166,13 @@ fn diff_model_names<'py>(
     let mut names: Vec<Bound<'py, PyAny>> = Vec::new();
     for model in project.getattr("models")?.try_iter()? {
         let model: Bound<'py, PyAny> = model?;
-        let key: Bound<'py, PyAny> = model.getattr("key")?;
+        let key: Bound<'py, PyAny> = model_reads::model_key(&model)?;
         let chosen: bool = match keys.indexes.get_item(&key)? {
             Some(index) => selected.contains(&index.extract::<usize>()?),
             None => false,
         };
         if chosen && key.getattr("resource_type")?.eq(&model_type)? {
-            names.push(model.getattr("name")?);
+            names.push(model_reads::name_object(&model)?);
         }
     }
     PyTuple::new(py, names)

crates/sqlbuild-python/src/bindings/_helpers/compiled_project/compiled_model_view.rs

@@ -72,6 +72,8 @@ pub(crate) struct DestinationFact {
 
 /// What assembly and the semantic checks finished for one compiled model.
 pub(crate) struct CompiledRecord {
+    /// The model's name, its file stem.
+    name: String,
     /// `(resource type, name)` of each dependency.
     deps: Vec<(String, String)>,
     destination: DestinationFact,
@@ -156,6 +158,7 @@ pub(crate) fn compiled_record(
     let config: &Bound<'_, PyAny> = input.config.bind(py);
     let values: Bound<'_, PyDict> = config.getattr(VALUES_ATTRIBUTE)?.downcast_into()?;
     Ok(CompiledRecord {
+        name: name.to_owned(),
         deps,
         destination: DestinationFact {
             database: text_value(&values, DATABASE_KEY)?,
@@ -309,6 +312,25 @@ impl NativeCompiledModel {
         &self.overlay.poisoned
     }
 
+    /// The kept lineage rows with their transform and confidence codes, or `None`.
+    pub(crate) fn lineage_rows(&self, py: Python<'_>) -> PyResult<Option<Vec<LineageRow>>> {
+        self.compiled(py, |compiled, _| {
+            Ok(compiled.analysis.lineage_rows().map(<[LineageRow]>::to_vec))
+        })
+    }
+
+    /// The input relations of a dynamic pivot proof that proved its outputs, else `None`.
+    pub(crate) fn proven_inputs(&self, py: Python<'_>) -> PyResult<Option<Vec<String>>> {
+        self.compiled(py, |compiled, _| {
+            Ok(compiled
+                .analysis
+                .proof
+                .as_ref()
+                .filter(|proof| proof.output_proven)
+                .map(|proof| proof.input_relations.clone()))
+        })
+    }
+
     fn model_file_attribute(&self, py: Python<'_>, name: &str) -> PyResult<Py<PyAny>> {
         self.read(py, |record, py| {
             Ok(record.model_file.bind(py).getattr(name)?.unbind())
@@ -344,7 +366,7 @@ fn compiled_of(record: &ModelInputRecord) -> PyResult<&CompiledRecord> {
 #[pymethods]
 impl NativeCompiledModel {
     #[getter]
-    fn key(&self, py: Python<'_>) -> PyResult<Py<PyAny>> {
+    pub(crate) fn key(&self, py: Python<'_>) -> PyResult<Py<PyAny>> {
         self.read(py, |record, py| {
             let compiled: &CompiledRecord = compiled_of(record)?;
             compiled
@@ -387,7 +409,7 @@ impl NativeCompiledModel {
     }
 
     #[getter]
-    fn query_sql(&self, py: Python<'_>) -> PyResult<Py<PyString>> {
+    pub(crate) fn query_sql(&self, py: Python<'_>) -> PyResult<Py<PyString>> {
         self.read(py, |record, _| Ok(record.query_sql.clone_ref(py)))
     }
 
@@ -397,7 +419,7 @@ impl NativeCompiledModel {
     }
 
     #[getter]
-    fn destination(&self, py: Python<'_>) -> PyResult<Py<PyAny>> {
+    pub(crate) fn destination(&self, py: Python<'_>) -> PyResult<Py<PyAny>> {
         self.compiled(py, |compiled, py| {
             self.projections
                 .destination
@@ -731,6 +753,9 @@ pub(crate) fn model_graph_row(
 }
 
 fn model_name(record: &ModelInputRecord, py: Python<'_>) -> PyResult<String> {
+    if let Some(compiled) = &record.compiled {
+        return Ok(compiled.name.clone());
+    }
     record
         .model_file
         .bind(py)

crates/sqlbuild-python/src/bindings/_helpers/janitor/janitor_old_names.rs

@@ -7,6 +7,7 @@ use pyo3::types::{PyDict, PyDictMethods};
 use sqlbuild_analysis::janitor_planning::main::migration_origin_keys::migration_origin_keys;
 use sqlbuild_analysis::janitor_planning::main::old_name_request_matches::old_name_request_matches;
 
+use crate::bindings::_helpers::analysis::model_reads;
 use crate::bindings::_helpers::compile_driver::python_calls::call_with;
 use crate::bindings::_helpers::janitor::janitor_facts::{
     SchemaRelations, Warehouse, model_destinations,
@@ -228,7 +229,7 @@ fn project_destinations<'py>(project: &Bound<'py, PyAny>) -> PyResult<Bound<'py,
     let py: Python<'py> = project.py();
     let destinations: Bound<'py, PyDict> = PyDict::new(py);
     for (key, model) in model_destinations(project)? {
-        destinations.set_item(key.to_python(py)?, model.getattr("name")?)?;
+        destinations.set_item(key.to_python(py)?, model_reads::name_object(&model)?)?;
     }
     Ok(destinations)
 }
@@ -369,7 +370,7 @@ pub(crate) fn pending_migration_origins(
         if listed.contains(&destination.folded()) {
             continue;
         }
-        let model_name: String = model.getattr("name")?.extract()?;
+        let model_name: String = model_reads::model_name(&model)?;
         for origin in migration_origin_keys(&raw, destination.database.as_deref(), &schemas) {
             let folded: Folded = Key {
                 database: origin.database,