perf(compile): batch analysis preparation normalization

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

@@ -38,18 +38,24 @@ fn normalize_dialect_sql(py: Python<'_>, sql: &str, dialect: &str) -> PyResult<S
     .map_err(value_error)
 }
 
+/// Normalize a batch in one detached call; a failed member is returned as its exception.
 #[pyfunction]
 fn normalize_analysis_sqls(
     py: Python<'_>,
     dialect: &str,
-    requests: Vec<(String, std::collections::HashMap<String, String>)>,
-) -> PyResult<Vec<String>> {
-    py.compiler_detach(|| {
-        crate::semantic_validation::main::normalize_batch::normalize_analysis_sqls(
-            dialect, requests,
+    requests: Vec<crate::semantic_validation::types::NormalizationRequest>,
+) -> PyResult<Vec<Py<PyAny>>> {
+    let results = py.compiler_detach(|| {
+        Ok(
+            crate::semantic_validation::main::normalize_batch::normalize_analysis_sqls(
+                dialect, requests, None,
+            ),
         )
-    })
-    .map_err(value_error)
+    });
+    crate::bindings::main::normalization_results::normalization_results(
+        py,
+        results.map_err(value_error)?,
+    )
 }
 
 #[pyfunction]

crates/sqlbuild-rules-native/src/bindings/main/catch_compiler_panic.rs

@@ -0,0 +1,5 @@
+pub(crate) fn catch_compiler_panic<T>(
+    operation: impl FnOnce() -> Result<T, String>,
+) -> Result<T, String> {
+    crate::bindings::_helpers::panics::catch_compiler_panic(operation)
+}

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

@@ -1,3 +1,5 @@
+pub(crate) mod catch_compiler_panic;
 pub(crate) mod compiler_error;
 pub(crate) mod compiler_guard;
+pub(crate) mod normalization_results;
 pub mod register;

crates/sqlbuild-rules-native/src/bindings/main/normalization_results.rs

@@ -0,0 +1,17 @@
+use pyo3::prelude::{IntoPyObject, Py, PyAny, PyResult, Python};
+
+/// Return normalized SQL in order, carrying each failure as the exception a single call raises.
+pub(crate) fn normalization_results(
+    py: Python<'_>,
+    results: Vec<Result<String, String>>,
+) -> PyResult<Vec<Py<PyAny>>> {
+    results
+        .into_iter()
+        .map(|result| match result {
+            Ok(sql) => Ok(sql.into_pyobject(py)?.into_any().unbind()),
+            Err(error) => Ok(crate::bindings::_helpers::panics::compiler_error(error)
+                .into_value(py)
+                .into_any()),
+        })
+        .collect()
+}

crates/sqlbuild-rules-native/src/semantic_validation/_helpers/catalog.rs

@@ -1,14 +1,19 @@
 //! Compile-owned schema catalog and native, batched binding requests.
 
 use crate::bindings::main::compiler_error::compiler_error;
+use crate::bindings::main::normalization_results::normalization_results;
 use crate::bindings::types::CompilerDetach;
+use crate::semantic_validation::main::normalize_batch;
 use std::collections::HashMap;
+use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::{Arc, Mutex, PoisonError};
 
 use crate::semantic_validation::models::{
     CatalogInput, Columns, CompactAnalysisJob, FunctionProbes, ProjectCatalog,
 };
-use crate::semantic_validation::types::{BindingRequest, DiagnosticRow, Relations};
+use crate::semantic_validation::types::{
+    BindingRequest, DiagnosticRow, NormalizationRequest, Relations,
+};
 use crate::semantic_validation::{
     _helpers::{diagnostics, identifiers},
     main as validation,
@@ -19,7 +24,7 @@ use polyglot_sql::{
     ValidationSchema,
 };
 use pyo3::exceptions::PyValueError;
-use pyo3::prelude::{Bound, FromPyObject, PyAny, PyAnyMethods, PyResult, Python};
+use pyo3::prelude::{Bound, FromPyObject, Py, PyAny, PyAnyMethods, PyResult, Python};
 use pyo3::pymethods;
 use pyo3::types::{PyBytes, PyDict, PyDictMethods};
 use rayon::iter::{IntoParallelIterator, ParallelIterator};
@@ -137,6 +142,7 @@ impl ProjectCatalog {
             analysis_tables: HashMap::new(),
             function_probes: Arc::new(FunctionProbes::default()),
             analysis_pool: Arc::default(),
+            running_analyses: Arc::default(),
         };
         catalog.update_relations(relations);
         Ok(catalog)
@@ -158,6 +164,7 @@ impl ProjectCatalog {
             analysis_tables: self.analysis_tables.clone(),
             function_probes: Arc::new(FunctionProbes::default()),
             analysis_pool: Arc::clone(&self.analysis_pool),
+            running_analyses: Arc::clone(&self.running_analyses),
         };
         catalog.update_relations(relations);
         catalog
@@ -234,6 +241,7 @@ impl ProjectCatalog {
     ) -> PyResult<Vec<Vec<DiagnosticRow>>> {
         py.compiler_detach(|| {
             let pool = self.analysis_pool()?;
+            let _running = RunningAnalysis::start(&self.running_analyses);
             pool.install(|| {
                 requests
                     .into_par_iter()
@@ -253,6 +261,44 @@ impl ProjectCatalog {
         })
         .map_err(compiler_error)
     }
+
+    /// Normalize a preparation batch on the idle analysis pool, else in turn, in one detached call.
+    fn normalize_analysis_sqls(
+        &self,
+        py: Python<'_>,
+        dialect: &str,
+        requests: Vec<NormalizationRequest>,
+    ) -> PyResult<Vec<Py<PyAny>>> {
+        let results = py
+            .compiler_detach(|| {
+                let pool = (self.running_analyses.load(Ordering::Acquire) == 0)
+                    .then(|| self.analysis_pool())
+                    .transpose()?;
+                Ok(normalize_batch::normalize_analysis_sqls(
+                    dialect,
+                    requests,
+                    pool.as_deref(),
+                ))
+            })
+            .map_err(compiler_error)?;
+        normalization_results(py, results)
+    }
+}
+
+/// Count one analysis batch as running on the catalog's pool until dropped.
+struct RunningAnalysis<'a>(&'a AtomicUsize);
+
+impl<'a> RunningAnalysis<'a> {
+    fn start(running: &'a AtomicUsize) -> Self {
+        running.fetch_add(1, Ordering::AcqRel);
+        Self(running)
+    }
+}
+
+impl Drop for RunningAnalysis<'_> {
+    fn drop(&mut self) {
+        self.0.fetch_sub(1, Ordering::AcqRel);
+    }
 }
 
 #[pymethods]
@@ -265,7 +311,10 @@ impl CompactAnalysisJob {
             .take()
             .ok_or_else(|| PyValueError::new_err("compact analysis job already ran"))?;
         let result = py
-            .compiler_detach(|| analysis(&self.catalog))
+            .compiler_detach(|| {
+                let _running = RunningAnalysis::start(&self.catalog.running_analyses);
+                analysis(&self.catalog)
+            })
             .map_err(compiler_error)?;
         Ok(PyBytes::new(py, result.as_bytes()))
     }
@@ -283,6 +332,7 @@ impl ProjectCatalog {
             analysis_tables: HashMap::new(),
             function_probes: Arc::clone(&self.function_probes),
             analysis_pool: Arc::clone(&self.analysis_pool),
+            running_analyses: Arc::clone(&self.running_analyses),
         }
     }
 
@@ -321,6 +371,7 @@ impl ProjectCatalog {
             analysis_tables: HashMap::new(),
             function_probes: Arc::new(FunctionProbes::default()),
             analysis_pool: Arc::default(),
+            running_analyses: Arc::default(),
         }
     }
     pub(crate) fn analysis_schema(

crates/sqlbuild-rules-native/src/semantic_validation/main/normalize_batch.rs

@@ -1,19 +1,27 @@
-use std::collections::HashMap;
+use rayon::ThreadPool;
+use rayon::iter::{IntoParallelIterator, ParallelIterator};
 
+use crate::bindings::main::catch_compiler_panic::catch_compiler_panic;
+use crate::semantic_validation::types::NormalizationRequest;
+
+/// Normalize every request, keeping each failure in place so callers raise the first in order.
 pub(crate) fn normalize_analysis_sqls(
     dialect: &str,
-    requests: Vec<(String, HashMap<String, String>)>,
-) -> Result<Vec<String>, String> {
-    let mut results: Vec<String> = Vec::with_capacity(requests.len());
-    for (sql, placeholders) in requests {
-        results.push(
+    requests: Vec<NormalizationRequest>,
+    pool: Option<&ThreadPool>,
+) -> Vec<Result<String, String>> {
+    let normalize = |(sql, stubs, placeholders): NormalizationRequest| {
+        catch_compiler_panic(|| {
             crate::semantic_validation::_helpers::normalization::normalize_analysis_sql(
                 &sql,
                 dialect,
-                HashMap::new(),
+                stubs,
                 placeholders,
-            )?,
-        );
+            )
+        })
+    };
+    match pool {
+        Some(pool) => pool.install(|| requests.into_par_iter().map(normalize).collect()),
+        None => requests.into_iter().map(normalize).collect(),
     }
-    Ok(results)
 }

crates/sqlbuild-rules-native/src/semantic_validation/models.rs

@@ -6,6 +6,7 @@ use polyglot_sql::{DialectType, SchemaValidationOptions};
 use pyo3::{FromPyObject, pyclass};
 use rayon::ThreadPool;
 use std::collections::HashMap;
+use std::sync::atomic::AtomicUsize;
 use std::sync::{Arc, Mutex};
 
 #[derive(Default, Debug)]
@@ -57,6 +58,8 @@ pub(crate) struct ProjectCatalog {
     pub(crate) function_probes: Arc<FunctionProbes>,
     /// One analysis pool per compile catalog, built on first use and shared by its views.
     pub(super) analysis_pool: Arc<Mutex<Option<Arc<ThreadPool>>>>,
+    /// Analysis batches running on the pool; preparation only borrows the pool while none run.
+    pub(super) running_analyses: Arc<AtomicUsize>,
 }
 
 /// One resolved compact analysis batch that runs without borrowing its project catalog.

crates/sqlbuild-rules-native/src/semantic_validation/tests.rs

@@ -1,5 +1,7 @@
 #[path = "tests/test_function_names.rs"]
 mod function_names;
+#[path = "tests/test_normalize_batch.rs"]
+mod normalize_batch;
 #[path = "tests/test_positions.rs"]
 mod positions;
 #[path = "tests/test_semantic_validation.rs"]

crates/sqlbuild-rules-native/src/semantic_validation/tests/test_normalize_batch.rs

@@ -0,0 +1,89 @@
+use std::collections::HashMap;
+
+use crate::semantic_validation::main::normalize_batch::normalize_analysis_sqls;
+use crate::semantic_validation::tests::test_types::NormalizationBatchTestCase;
+use crate::semantic_validation::types::NormalizationRequest;
+
+#[test]
+fn given_batch_when_normalizing_on_pool_or_in_turn_then_matches_single_calls_in_order() {
+    let test_cases = [
+        NormalizationBatchTestCase {
+            description: "stubbed references and placeholders",
+            dialect: "snowflake",
+            requests: &[
+                (
+                    r#"SELECT * FROM __ref("orders") JOIN __ref("customers") USING (id)"#,
+                    &[("orders", "__sqb_rel_0"), ("customers", "__sqb_rel_1")],
+                ),
+                ("SELECT payload:item[0] AS item FROM inventory", &[]),
+                ("SELECT 1 LIMIT @@@limit", &[]),
+            ],
+            expected_failures: &[],
+        },
+        NormalizationBatchTestCase {
+            description: "failures stay at their own positions",
+            dialect: "snowflake",
+            requests: &[
+                ("SELECT 'unterminated FROM orders", &[]),
+                ("SELECT id FROM products", &[]),
+                ("SELECT /* unterminated FROM customers", &[]),
+            ],
+            expected_failures: &[0, 2],
+        },
+    ];
+    let pool = rayon::ThreadPoolBuilder::new()
+        .num_threads(3)
+        .build()
+        .expect("test pool must build");
+
+    for test_case in test_cases {
+        let requests = || -> Vec<NormalizationRequest> {
+            test_case
+                .requests
+                .iter()
+                .map(|(sql, stubs)| {
+                    (
+                        (*sql).to_owned(),
+                        stubs
+                            .iter()
+                            .map(|(name, stub)| ((*name).to_owned(), (*stub).to_owned()))
+                            .collect(),
+                        HashMap::from([("limit".to_owned(), "10".to_owned())]),
+                    )
+                })
+                .collect()
+        };
+        let expected: Vec<Result<String, String>> = requests()
+            .into_iter()
+            .map(|(sql, stubs, placeholders)| {
+                crate::semantic_validation::_helpers::normalization::normalize_analysis_sql(
+                    &sql,
+                    test_case.dialect,
+                    stubs,
+                    placeholders,
+                )
+            })
+            .collect();
+        let failures: Vec<usize> = (0..expected.len())
+            .filter(|index| expected[*index].is_err())
+            .collect();
+
+        assert_eq!(
+            failures, test_case.expected_failures,
+            "{}",
+            test_case.description
+        );
+        assert_eq!(
+            normalize_analysis_sqls(test_case.dialect, requests(), Some(&pool)),
+            expected,
+            "{}",
+            test_case.description
+        );
+        assert_eq!(
+            normalize_analysis_sqls(test_case.dialect, requests(), None),
+            expected,
+            "{}",
+            test_case.description
+        );
+    }
+}

crates/sqlbuild-rules-native/src/semantic_validation/tests/test_types.rs

@@ -45,3 +45,10 @@ pub(crate) struct TypeLocationTestCase {
     pub(crate) findings: &'static [(&'static str, &'static str)],
     pub(crate) expected_starts: &'static [&'static str],
 }
+
+pub(crate) struct NormalizationBatchTestCase {
+    pub(crate) description: &'static str,
+    pub(crate) dialect: &'static str,
+    pub(crate) requests: &'static [(&'static str, &'static [(&'static str, &'static str)])],
+    pub(crate) expected_failures: &'static [usize],
+}

crates/sqlbuild-rules-native/src/semantic_validation/types.rs

@@ -3,6 +3,8 @@ use std::collections::HashMap;
 
 pub(crate) type Relations = HashMap<String, Columns>;
 pub(crate) type BindingRequest = (String, Vec<(String, bool)>, Relations);
+/// One analysis SQL with its relation stubs and placeholder defaults.
+pub(crate) type NormalizationRequest = (String, HashMap<String, String>, HashMap<String, String>);
 pub(crate) type DiagnosticRow = (
     String,
     String,