perf(compile): analyze models as soon as their producers finish

Replace wave-by-wave model analysis with a dependency-driven scheduler. A
model is analyzed once every producer shape it binds against is final, so a
slow model no longer holds back unrelated models in the next topological
level. Worker threads serialize Python preparation and projection under one
lock, while self-contained native analysis releases it, so the next ready
batch is prepared while earlier batches run natively. The calling thread
never takes that lock: it only starts and joins the workers, and on an
interrupt or start failure it cancels and joins every started worker before
re-raising.

Each compile catalog now owns one native analysis pool that every compact
analysis and binding batch reuses, instead of building a pool per call.
Compact analysis splits into a catalog-resolving prepare step and a run step
that no longer borrows the catalog, so catalog updates for later batches
proceed while earlier batches run. Spawned drainer tasks replace the pool
broadcast, so a batch no longer needs every pool thread to finish.

Every model sees exactly the producer shapes, cache entries, and changed
upstream signatures it saw under wave scheduling, so compile output does not
depend on completion order. An ordinary analysis error replays the wave
schedule to raise the same first error; an interrupt or other non-Exception
failure always wins and propagates without a replay.

Author: kvlonge

crates/sqlbuild-rules-native/src/bindings/_helpers/functions.rs

@@ -515,6 +515,7 @@ pub(crate) fn register(module: &Bound<'_, PyModule>) -> PyResult<()> {
             .get_type::<crate::bindings::_helpers::panics::NativeCompilerError>(),
     )?;
     module.add_class::<crate::semantic_validation::models::ProjectCatalog>()?;
+    module.add_class::<crate::semantic_validation::models::CompactAnalysisJob>()?;
     module.add_class::<crate::semantic_validation::models::BindingPositions>()?;
     module.add_function(wrap_pyfunction!(normalize_analysis_sql, module)?)?;
     module.add_function(wrap_pyfunction!(normalize_analysis_sqls, module)?)?;

crates/sqlbuild-rules-native/src/query_analysis/_helpers/engine.rs

@@ -323,6 +323,24 @@ pub(crate) fn analyze_project_compact_with_catalog(
     request_json: &str,
     catalog: Option<&crate::semantic_validation::models::ProjectCatalog>,
 ) -> Result<String, String> {
+    analyze_resolved_compact_batch(resolve_compact_batch(request_json, catalog)?, catalog)
+}
+
+/// Resolve catalog schemas now and return the analysis to run later without catalog tables.
+pub(crate) fn prepare_project_compact_with_catalog(
+    request_json: &str,
+    catalog: &crate::semantic_validation::models::ProjectCatalog,
+) -> Result<crate::semantic_validation::types::PreparedCompactAnalysis, String> {
+    let request = resolve_compact_batch(request_json, Some(catalog))?;
+    Ok(Box::new(move |view| {
+        analyze_resolved_compact_batch(request, Some(view))
+    }))
+}
+
+fn resolve_compact_batch(
+    request_json: &str,
+    catalog: Option<&crate::semantic_validation::models::ProjectCatalog>,
+) -> Result<CompactProjectAnalysisBatchRequest, String> {
     let mut request: CompactProjectAnalysisBatchRequest =
         serde_json::from_str(request_json).map_err(|error| error.to_string())?;
     for query in &mut request.queries {
@@ -349,6 +367,14 @@ pub(crate) fn analyze_project_compact_with_catalog(
             }
         }
     }
+    Ok(request)
+}
+
+/// Analyse a resolved batch; a catalog supplies options, probes, and its compile-wide pool.
+fn analyze_resolved_compact_batch(
+    request: CompactProjectAnalysisBatchRequest,
+    catalog: Option<&crate::semantic_validation::models::ProjectCatalog>,
+) -> Result<String, String> {
     let workers = request.workers.clamp(1, MAX_WORKERS);
     if request
         .templates
@@ -400,12 +426,16 @@ pub(crate) fn analyze_project_compact_with_catalog(
         })
         .collect();
     let sql_bytes: usize = request.queries.iter().map(|query| query.sql.len()).sum();
-    let analysis_workers = workers.min(unique_query_count.max(1));
-    let pool = rayon::ThreadPoolBuilder::new()
-        .num_threads(analysis_workers)
-        .stack_size(ANALYSIS_WORKER_STACK_BYTES)
-        .build()
-        .map_err(|error| error.to_string())?;
+    let pool = match catalog {
+        Some(catalog) => catalog.analysis_pool()?,
+        None => std::sync::Arc::new(
+            rayon::ThreadPoolBuilder::new()
+                .num_threads(workers.min(unique_query_count.max(1)))
+                .stack_size(ANALYSIS_WORKER_STACK_BYTES)
+                .build()
+                .map_err(|error| error.to_string())?,
+        ),
+    };
     let query_work: Vec<CompactQueryWork> = request
         .queries
         .into_iter()
@@ -450,7 +480,7 @@ pub(crate) fn analyze_project_compact_with_catalog(
     serde_json::to_string(&response).map_err(|error| error.to_string())
 }
 
-/// Analyse a batch largest SQL first so no late wide query becomes its critical path.
+/// Analyse largest SQL first on spawned drainers rather than a broadcast to every pool thread.
 fn analyze_largest_first(
     pool: &rayon::ThreadPool,
     batch: Vec<CompactQueryWork>,
@@ -463,21 +493,33 @@ fn analyze_largest_first(
         .map(|work| Mutex::new(Some(work)))
         .collect();
     let next = AtomicUsize::new(0);
-    let finished: Vec<Vec<(usize, CompiledQueryWorkResult)>> = pool.broadcast(|_| {
-        let mut finished: Vec<(usize, CompiledQueryWorkResult)> = Vec::new();
-        while let Some(index) = order.get(next.fetch_add(1, Ordering::Relaxed)).copied() {
-            let work = slots[index]
-                .lock()
-                .unwrap_or_else(PoisonError::into_inner)
-                .take();
-            if let Some(work) = work {
-                finished.push((index, analyze_compact_query_work(work, catalog)));
-            }
+    let finished: Mutex<Vec<(usize, CompiledQueryWorkResult)>> = Mutex::new(Vec::new());
+    let drainers = pool.current_num_threads().min(slots.len());
+    pool.scope(|scope| {
+        for _ in 0..drainers {
+            scope.spawn(|_| {
+                let mut drained: Vec<(usize, CompiledQueryWorkResult)> = Vec::new();
+                while let Some(index) = order.get(next.fetch_add(1, Ordering::Relaxed)).copied() {
+                    let work = slots[index]
+                        .lock()
+                        .unwrap_or_else(PoisonError::into_inner)
+                        .take();
+                    if let Some(work) = work {
+                        drained.push((index, analyze_compact_query_work(work, catalog)));
+                    }
+                }
+                finished
+                    .lock()
+                    .unwrap_or_else(PoisonError::into_inner)
+                    .extend(drained);
+            });
         }
-        finished
     });
     let mut results: Vec<Option<CompiledQueryWorkResult>> = slots.iter().map(|_| None).collect();
-    for (index, result) in finished.into_iter().flatten() {
+    for (index, result) in finished
+        .into_inner()
+        .unwrap_or_else(PoisonError::into_inner)
+    {
         results[index] = Some(result);
     }
     results.into_iter().flatten().collect()
@@ -497,7 +539,7 @@ fn analyze_compact_query_work(
         && let Some(validation) = result.validation.take()
     {
         let local_probes = crate::semantic_validation::models::FunctionProbes::default();
-        let probes = catalog.map_or(&local_probes, |catalog| &catalog.function_probes);
+        let probes = catalog.map_or(&local_probes, |catalog| catalog.function_probes.as_ref());
         result.validation = Some(validation.and_then(|value| {
             crate::semantic_validation::main::map_diagnostics::map_diagnostics(
                 &sql, dialect, value, probes,

crates/sqlbuild-rules-native/src/query_analysis/main/analyze_project_catalog.rs

@@ -1,4 +1,4 @@
 pub(crate) mod analyze;
 pub(crate) mod analyze_project;
-pub(crate) mod analyze_project_catalog;
 pub(crate) mod analyze_project_compact;
+pub(crate) mod prepare_project_catalog;

crates/sqlbuild-rules-native/src/query_analysis/main/mod.rs

@@ -0,0 +1,6 @@
+pub(crate) fn prepare_project_compact_with_catalog(
+    request_json: &str,
+    catalog: &crate::semantic_validation::models::ProjectCatalog,
+) -> Result<crate::semantic_validation::types::PreparedCompactAnalysis, String> {
+    crate::query_analysis::engine::prepare_project_compact_with_catalog(request_json, catalog)
+}

crates/sqlbuild-rules-native/src/query_analysis/main/prepare_project_catalog.rs

@@ -578,3 +578,64 @@ pub(crate) fn borrowed_facts_preserve_union_dependencies_and_join_nullability()
     );
     true
 }
+
+pub(crate) fn catalog_views_reuse_one_analysis_pool() -> bool {
+    let catalog = crate::semantic_validation::models::ProjectCatalog::with_options(
+        polyglot_sql::DialectType::DuckDB,
+        polyglot_sql::SchemaValidationOptions::default(),
+        false,
+    );
+    let first = catalog.analysis_pool().expect("test assumption must hold");
+    let second = catalog.analysis_pool().expect("test assumption must hold");
+    let from_view = catalog
+        .analysis_view()
+        .analysis_pool()
+        .expect("test assumption must hold");
+    std::sync::Arc::ptr_eq(&first, &second)
+        && std::sync::Arc::ptr_eq(&first, &from_view)
+        && first.current_num_threads() == 4
+}
+
+pub(crate) fn concurrent_batches_match_serial_batches() -> bool {
+    let requests: Vec<String> = (0..12)
+        .map(|index| {
+            json!({
+                "queries": (0..=index % 5).map(|query| json!({
+                    "sql": format!(
+                        "WITH ranked AS (SELECT {query} AS order_id, {index} AS amount) \
+                         SELECT order_id, amount * {query} AS amount FROM ranked"
+                    ),
+                    "dialect": "duckdb"
+                })).collect::<Vec<Value>>(),
+                "templates": (0..=index % 5).map(|query| json!({
+                    "queryIndex": query, "recoverCteFacts": true
+                })).collect::<Vec<Value>>(),
+                "projections": (0..=index % 5).map(|query| json!({
+                    "templateIndex": query
+                })).collect::<Vec<Value>>(),
+                "workers": 4
+            })
+            .to_string()
+        })
+        .collect();
+    let serial: Vec<String> = requests
+        .iter()
+        .map(|request| analyze_project_compact_json(request).expect("test assumption must hold"))
+        .collect();
+    let concurrent: Vec<String> = std::thread::scope(|scope| {
+        let handles: Vec<_> = requests
+            .iter()
+            .map(|request| scope.spawn(move || analyze_project_compact_json(request)))
+            .collect();
+        handles
+            .into_iter()
+            .map(|handle| {
+                handle
+                    .join()
+                    .expect("test assumption must hold")
+                    .expect("test assumption must hold")
+            })
+            .collect()
+    });
+    concurrent == serial
+}

crates/sqlbuild-rules-native/src/query_analysis/tests/helpers.rs

@@ -5,10 +5,10 @@ use crate::query_analysis::tests::helpers::{
     borrowed_facts_preserve_named_outputs_and_terminal_sources,
     borrowed_facts_preserve_union_dependencies_and_join_nullability,
     bound_lineage_survives_type_recovery, canonical_queries_reuse_semantics_and_project_resources,
-    combined_queries_preserve_standalone_binding,
+    catalog_views_reuse_one_analysis_pool, combined_queries_preserve_standalone_binding,
     compact_project_query_interns_repeated_lineage_strings,
-    interleaved_query_templates_preserve_template_order, larger_later_queries_keep_request_order,
-    native_compatibility_types_preserve_result_semantics,
+    concurrent_batches_match_serial_batches, interleaved_query_templates_preserve_template_order,
+    larger_later_queries_keep_request_order, native_compatibility_types_preserve_result_semantics,
     repeated_project_facts_intern_complete_facts,
     unannotated_types_require_complete_nested_evidence,
     widening_aggregates_require_compatibility_recovery,
@@ -86,6 +86,16 @@ fn given_compact_query_cases_when_analyzing_projects_then_expected_behavior_hold
             run: larger_later_queries_keep_request_order,
             expected_success: true,
         },
+        CompactQueryAnalysisTestCase {
+            description: "catalog views reuse one analysis pool",
+            run: catalog_views_reuse_one_analysis_pool,
+            expected_success: true,
+        },
+        CompactQueryAnalysisTestCase {
+            description: "concurrent batches match serial batches",
+            run: concurrent_batches_match_serial_batches,
+            expected_success: true,
+        },
     ];
 
     for test_case in test_cases {