feat(native): batch dynamic pivot proofs and opt-out syntax checks through the native session

The finished analysis session is kept on CompiledProject.native_session and proves pivot models outside it from its own tables; without a session one batched standalone call sends the tables once. The per-model native proof path is deleted. SQL-analysis opt-out syntax checks use the native checker under the preview engine. Rule hosts drop the session before pickling.

Author: kvlonge

crates/sqlbuild-analysis/src/assembly/analysis_session/_helpers/dynamic_pivot.rs

@@ -1,8 +1,11 @@
 //! Python's `analyze_dynamic_column_contract` over the crate's parse of the model SQL.
 
 use std::collections::{HashMap, HashSet};
+use std::panic::{AssertUnwindSafe, catch_unwind};
 
 use polyglot_sql::{ComplexityGuardOptions, Dialect, Expression, ParseOptions};
+use rayon::ThreadPool;
+use rayon::iter::{IntoParallelRefIterator, ParallelIterator};
 use serde_json::{Map, Value, json};
 use sqlbuild_core::panics::main::catch_compiler_panic::catch_compiler_panic;
 
@@ -17,7 +20,7 @@ use crate::assembly::analysis_session::constants::{
     UNKNOWN_NULLABILITY, UNKNOWN_TYPE,
 };
 use crate::assembly::analysis_session::models::{
-    ColumnFact, ContractProof, DynamicFamily, PivotOutcome,
+    ColumnFact, ContractProof, DynamicFamily, PivotModel, PivotOutcome, PivotTables,
 };
 use crate::assembly::analysis_session::types::{Pairs, Shapes};
 
@@ -33,6 +36,47 @@ pub(crate) struct PivotFacts<'a> {
     pub(crate) families_by_table: &'a [(String, Vec<DynamicFamily>)],
 }
 
+/// Each model's proof in order; a model whose proof panics is deferred to Python.
+pub(crate) fn pivot_outcomes(models: &[PivotModel], tables: &PivotTables) -> Vec<PivotOutcome> {
+    let facts = pivot_facts(tables);
+    models
+        .iter()
+        .map(|model| guarded_outcome(model, &facts))
+        .collect()
+}
+
+/// Each model's proof in order on `pool`; a model whose proof panics is deferred to Python.
+pub(crate) fn pooled_pivot_outcomes(
+    pool: &ThreadPool,
+    models: &[PivotModel],
+    tables: &PivotTables,
+) -> Vec<PivotOutcome> {
+    let facts = pivot_facts(tables);
+    pool.install(|| {
+        models
+            .par_iter()
+            .map(|model| guarded_outcome(model, &facts))
+            .collect()
+    })
+}
+
+fn pivot_facts(tables: &PivotTables) -> PivotFacts<'_> {
+    PivotFacts {
+        dialect: &tables.dialect,
+        column_types: &tables.column_types,
+        authoritative_types: &tables.authoritative_types,
+        column_nullability: &tables.column_nullability,
+        families_by_table: &tables.families_by_table,
+    }
+}
+
+fn guarded_outcome(model: &PivotModel, facts: &PivotFacts<'_>) -> PivotOutcome {
+    catch_unwind(AssertUnwindSafe(|| {
+        pivot_outcome(&model.sql, &model.families, facts)
+    }))
+    .unwrap_or(PivotOutcome::Deferred)
+}
+
 /// One CTE body by folded name; None for a CTE with column aliases.
 type CteMap<'a> = HashMap<String, Option<&'a Value>>;
 

crates/sqlbuild-analysis/src/assembly/analysis_session/_helpers/session.rs

@@ -1,7 +1,9 @@
 //! The session's phases: Python's uncached model analysis, wave by wave.
 
 use std::collections::HashMap;
+use std::sync::Arc;
 
+use rayon::ThreadPool;
 use rayon::iter::{IntoParallelRefIterator, ParallelIterator};
 
 use crate::assembly::analysis_session::_helpers::catalog_state::SessionCatalog;
@@ -21,8 +23,9 @@ use crate::assembly::analysis_session::constants::{
     DEFERRAL_ANALYSIS, DEFERRAL_ENRICHMENT, UNKNOWN_NULLABILITY, UNKNOWN_TYPE,
 };
 use crate::assembly::analysis_session::models::{
-    AnalysisSession, Awaiting, ColumnFact, Deferral, DeferredAnalysis, LineageFacts, ModelAnalysis,
-    ModelOutcome, ModelRequest, Phase, PivotOutcome, SessionOutcome, SessionRequest, SessionStep,
+    AnalysisSession, Awaiting, ColumnFact, Deferral, DeferredAnalysis, FinishedSession,
+    LineageFacts, ModelAnalysis, ModelOutcome, ModelRequest, Phase, PivotOutcome, PivotTables,
+    SessionOutcome, SessionRequest, SessionStep,
 };
 use crate::assembly::analysis_session::types::{Pairs, Shapes};
 use crate::semantic_validation::models::ProjectCatalog;
@@ -148,23 +151,36 @@ impl AnalysisSession {
     }
 
     /// Every model's outcome and the binding catalog changes Python records.
-    pub(crate) fn finish(self) -> Result<SessionOutcome, String> {
+    pub(crate) fn finish(self) -> Result<(SessionOutcome, FinishedSession), String> {
         if !matches!(self.phase, Phase::Done) {
             return Err("the session has not finished".to_owned());
         }
         let dynamic_contracts: Vec<PivotOutcome> = self.dynamic_contracts()?;
+        let pool: Result<Arc<ThreadPool>, String> = self.catalog.native.analysis_pool();
         let models: Vec<ModelOutcome> = self
             .outcomes
             .into_iter()
             .collect::<Option<_>>()
             .ok_or("the session left a model unanalysed")?;
         let (schema_additions, analysis_names) = self.catalog.into_changes();
-        Ok(SessionOutcome {
+        let request: SessionRequest = self.request;
+        let finished = FinishedSession {
+            tables: PivotTables {
+                dialect: request.dialect,
+                column_types: request.column_types,
+                authoritative_types: request.complete_schemas,
+                column_nullability: request.column_nullability,
+                families_by_table: request.dynamic_families_by_table,
+            },
+            pool,
+        };
+        let outcome = SessionOutcome {
             models,
             schema_additions,
             analysis_names,
             dynamic_contracts,
-        })
+        };
+        Ok((outcome, finished))
     }
 
     /// Python's dynamic pivot proof of each model over the relation facts before analysis.

crates/sqlbuild-analysis/src/assembly/analysis_session/main/finish_analysis_session.rs

@@ -1,8 +1,10 @@
 //! Collect a finished native model analysis session's outcomes.
 
-use crate::assembly::analysis_session::models::{AnalysisSession, SessionOutcome};
+use crate::assembly::analysis_session::models::{AnalysisSession, FinishedSession, SessionOutcome};
 
-/// Every model's outcome and the binding catalog changes Python records.
-pub fn finish_analysis_session(session: AnalysisSession) -> Result<SessionOutcome, String> {
+/// Every model's outcome and catalog changes, and the session kept for later pivot proofs.
+pub fn finish_analysis_session(
+    session: AnalysisSession,
+) -> Result<(SessionOutcome, FinishedSession), String> {
     session.finish()
 }

crates/sqlbuild-analysis/src/assembly/analysis_session/main/mod.rs

@@ -1,6 +1,7 @@
 pub mod expression_shapes;
 pub mod finish_analysis_session;
-pub mod prove_dynamic_contract;
+pub mod prove_dynamic_contracts;
+pub mod prove_finished_dynamic_contracts;
 pub mod provide_deferred_analyses;
 pub mod run_analysis_session;
 pub mod start_analysis_session;

crates/sqlbuild-analysis/src/assembly/analysis_session/main/prove_dynamic_contract.rs

@@ -0,0 +1,17 @@
+//! Python's `analyze_dynamic_column_contract` for models no analysis session analysed.
+
+use crate::assembly::analysis_session::_helpers::dynamic_pivot::pivot_outcomes;
+use crate::assembly::analysis_session::constants::PIVOT_WORKER_STACK_BYTES;
+use crate::assembly::analysis_session::models::{PivotBatchRequest, PivotOutcome};
+
+/// Each model's proof in order; a model whose proof panics is deferred to Python.
+pub fn prove_dynamic_contracts(request: &PivotBatchRequest) -> Result<Vec<PivotOutcome>, String> {
+    std::thread::scope(|scope| {
+        std::thread::Builder::new()
+            .stack_size(PIVOT_WORKER_STACK_BYTES)
+            .spawn_scoped(scope, || pivot_outcomes(&request.models, &request.tables))
+            .map_err(|error| error.to_string())?
+            .join()
+            .map_err(|_| "the native dynamic pivot proofs panicked".to_owned())
+    })
+}

crates/sqlbuild-analysis/src/assembly/analysis_session/main/prove_dynamic_contracts.rs

@@ -0,0 +1,13 @@
+//! Python's `analyze_dynamic_column_contract` for models outside a finished analysis session.
+
+use crate::assembly::analysis_session::_helpers::dynamic_pivot::pooled_pivot_outcomes;
+use crate::assembly::analysis_session::models::{FinishedSession, PivotModel, PivotOutcome};
+
+/// Each model's proof in order from the session's own tables; a panicking proof is deferred.
+pub fn prove_finished_dynamic_contracts(
+    session: &FinishedSession,
+    models: &[PivotModel],
+) -> Result<Vec<PivotOutcome>, String> {
+    let pool = session.pool.as_ref().map_err(Clone::clone)?;
+    Ok(pooled_pivot_outcomes(pool, models, &session.tables))
+}

crates/sqlbuild-analysis/src/assembly/analysis_session/main/prove_finished_dynamic_contracts.rs

@@ -1,6 +1,9 @@
 //! Plain-data request, deferrals and outcomes of the native model analysis session.
 
 use std::collections::HashMap;
+use std::sync::Arc;
+
+use rayon::ThreadPool;
 
 use crate::assembly::analysis_session::_helpers::catalog_state::SessionCatalog;
 use crate::assembly::analysis_session::_helpers::mappings::ShapeTable;
@@ -61,18 +64,37 @@ pub struct ContractProof {
     pub bare_dynamic_pivot: bool,
 }
 
-/// One model's dynamic pivot proof inputs, for a model the session does not analyse.
+/// The relation facts every dynamic pivot proof reads: Python's tables before analysis.
 #[derive(Debug, Clone, Default)]
-pub struct PivotRequest {
+pub struct PivotTables {
     pub dialect: String,
     pub column_types: Shapes,
     pub authoritative_types: Shapes,
     pub column_nullability: Shapes,
     pub families_by_table: Vec<(String, Vec<DynamicFamily>)>,
+}
+
+/// One model's pivot SQL and declared families.
+#[derive(Debug, Clone, Default)]
+pub struct PivotModel {
     pub sql: String,
     pub families: Vec<DynamicFamily>,
 }
 
+/// Dynamic pivot proofs for models no session analysed.
+#[derive(Debug, Clone, Default)]
+pub struct PivotBatchRequest {
+    pub tables: PivotTables,
+    pub models: Vec<PivotModel>,
+}
+
+/// A finished session's pivot tables and analysis pool, kept to prove models outside it.
+#[derive(Debug)]
+pub struct FinishedSession {
+    pub(crate) tables: PivotTables,
+    pub(crate) pool: Result<Arc<ThreadPool>, String>,
+}
+
 /// A model's dynamic pivot proof: none declared, left to Python, or proven natively.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum PivotOutcome {

crates/sqlbuild-analysis/src/assembly/analysis_session/models.rs

@@ -5,8 +5,8 @@ use crate::assembly::analysis_session::main::run_analysis_session::run_analysis_
 use crate::assembly::analysis_session::main::start_analysis_session::start_analysis_session;
 use crate::assembly::analysis_session::models::{
     AnalysisSession, ColumnFact, ContractProof, Deferral, DeferredAnalysis, DynamicFamily,
-    ModelOutcome, ModelReference, ModelRequest, PivotOutcome, PivotRequest, SessionRequest,
-    SessionStep,
+    FinishedSession, ModelOutcome, ModelReference, ModelRequest, PivotBatchRequest, PivotModel,
+    PivotOutcome, PivotTables, SessionRequest, SessionStep,
 };
 use crate::assembly::analysis_session::tests::test_types::ModelSpec;
 use crate::assembly::analysis_session::types::{Pairs, Shapes};
@@ -131,7 +131,7 @@ pub(crate) fn completed(
         })
     })
     .collect();
-    let outcome = finish_analysis_session(session).expect("the session finished");
+    let (outcome, _) = finish_analysis_session(session).expect("the session finished");
     (steps, outcome.models)
 }
 
@@ -215,12 +215,12 @@ pub(crate) fn session_lines(models: &[ModelSpec]) -> (Vec<Vec<String>>, Vec<Vec<
     )
 }
 
-/// A pivot proof request over typed `raw_orders`, whose upstream `status_amounts` is a pivot.
+/// Proofs over typed `raw_orders` and pivot `status_amounts`: the model, then a family-less one.
 pub(crate) fn pivot_request(
     dialect: &str,
     sql: &str,
     family: Option<(&str, &str, &str)>,
-) -> PivotRequest {
+) -> PivotBatchRequest {
     let raw_orders: Shapes = shapes(&[(
         "raw_orders",
         &[
@@ -237,17 +237,35 @@ pub(crate) fn pivot_request(
         })
         .into_iter()
         .collect();
-    PivotRequest {
-        dialect: dialect.to_owned(),
-        column_types: [raw_orders.clone(), upstream].concat(),
-        authoritative_types: raw_orders,
-        column_nullability: shapes(&[("raw_orders", &[("customer_id", "non_null")])]),
-        families_by_table: vec![(
-            "status_amounts".to_owned(),
-            vec![amounts("status", "amount", "MAX")],
-        )],
-        sql: sql.to_owned(),
-        families,
+    PivotBatchRequest {
+        tables: PivotTables {
+            dialect: dialect.to_owned(),
+            column_types: [raw_orders.clone(), upstream].concat(),
+            authoritative_types: raw_orders,
+            column_nullability: shapes(&[("raw_orders", &[("customer_id", "non_null")])]),
+            families_by_table: vec![(
+                "status_amounts".to_owned(),
+                vec![amounts("status", "amount", "MAX")],
+            )],
+        },
+        models: vec![
+            PivotModel {
+                sql: sql.to_owned(),
+                families,
+            },
+            PivotModel {
+                sql: "SELECT 1".to_owned(),
+                families: Vec::new(),
+            },
+        ],
+    }
+}
+
+/// A finished session holding `tables` and an analysis pool.
+pub(crate) fn finished_session(tables: PivotTables) -> FinishedSession {
+    FinishedSession {
+        tables,
+        pool: catalog("duckdb", &Vec::new()).analysis_pool(),
     }
 }
 

crates/sqlbuild-analysis/src/assembly/analysis_session/tests/helpers.rs

@@ -1,8 +1,10 @@
-use crate::assembly::analysis_session::main::prove_dynamic_contract::prove_dynamic_contract;
+use crate::assembly::analysis_session::main::prove_dynamic_contracts::prove_dynamic_contracts;
+use crate::assembly::analysis_session::main::prove_finished_dynamic_contracts::prove_finished_dynamic_contracts;
 use crate::assembly::analysis_session::main::start_analysis_session::start_analysis_session;
 use crate::assembly::analysis_session::models::PivotOutcome;
 use crate::assembly::analysis_session::tests::helpers::{
-    catalog, failed, model_requests, orders_request, pivot_request, proven, session_lines,
+    catalog, failed, finished_session, model_requests, orders_request, pivot_request, proven,
+    session_lines,
 };
 use crate::assembly::analysis_session::tests::test_types::{
     PivotTestCase, SessionTestCase, UnscheduledTestCase,
@@ -281,11 +283,21 @@ fn given_dynamic_pivots_when_proving_then_matches_python_or_defers() {
     ];
     for test_case in test_cases {
         let request = pivot_request(test_case.dialect, test_case.sql, test_case.family);
+        let session = finished_session(request.tables.clone());
 
-        let outcome = prove_dynamic_contract(&request).expect("the proof runs");
+        let standalone = prove_dynamic_contracts(&request).expect("the proofs run");
+        let in_session =
+            prove_finished_dynamic_contracts(&session, &request.models).expect("the proofs run");
 
         assert_eq!(
-            outcome, test_case.expected_outcome,
+            standalone,
+            vec![test_case.expected_outcome.clone(), PivotOutcome::Absent],
+            "{}",
+            test_case.description
+        );
+        assert_eq!(
+            in_session,
+            vec![test_case.expected_outcome, PivotOutcome::Absent],
             "{}",
             test_case.description
         );