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

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/mod.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/prepare_project_catalog.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/tests/helpers.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/test_query_analysis.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 {