perf(compile): reuse per-file facts on warm compiles

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

@@ -22,7 +22,9 @@ use fensu_policy::lifecycle::models::{
     RuntimeIdentity, ScopedIgnore,
 };
 use fensu_policy::{apply_suppressions, evaluate_batch, run_custom_host};
-use rayon::iter::{IntoParallelIterator, ParallelIterator};
+use rayon::iter::{
+    IndexedParallelIterator, IntoParallelIterator, IntoParallelRefIterator, ParallelIterator,
+};
 use sha2::{Digest, Sha256};
 use std::collections::{BTreeMap, BTreeSet};
 use std::path::{Path, PathBuf};
@@ -98,20 +100,20 @@ fn evaluate_native_models(
     let mut result = NativeModelEvaluation::default();
     let mut model_results: Vec<Option<Vec<Fault>>> = vec![None; model_limit];
     let mut pending_models: Vec<PendingModelRule<'_>> = Vec::new();
+    let mut identities = if cache.is_some() {
+        model_cache_identities(
+            &models[..model_limit],
+            ruleset_fingerprint,
+            project_fingerprint,
+        )?
+        .into_iter()
+        .map(Some)
+        .collect()
+    } else {
+        vec![None; model_limit]
+    };
     for (index, model) in models.iter().take(model_limit).enumerate() {
-        let identity = cache
-            .map(|_| {
-                model_cache_identity(
-                    model,
-                    ruleset_fingerprint,
-                    if index == 0 {
-                        project_fingerprint
-                    } else {
-                        None
-                    },
-                )
-            })
-            .transpose()?;
+        let identity = identities[index].take();
         if let Some(identity) = &identity
             && let Some(entry) = bucket
                 .entries
@@ -546,20 +548,28 @@ fn custom_local_identity(model_identity: &str, rule: &str, project: Option<&str>
 fn custom_model_fact_identities(
     models: &[crate::models::Model],
 ) -> Result<BTreeMap<String, String>, String> {
-    models
-        .iter()
-        .map(|model| {
-            let mut digest = Sha256::new();
-            digest.update(b"custom-rule-model-facts-v1");
-            digest.update(ANALYSIS_BATCH_SCHEMA_VERSION.to_le_bytes());
-            digest.update(env!("CARGO_PKG_VERSION").as_bytes());
-            digest.update(serde_json::to_vec(model).map_err(|error| error.to_string())?);
-            Ok((
-                model.relative_path.clone(),
-                format!("{:x}", digest.finalize()),
-            ))
-        })
-        .collect()
+    let identity = |model: &crate::models::Model| -> Result<(String, String), String> {
+        let mut digest = Sha256::new();
+        digest.update(b"custom-rule-model-facts-v1");
+        digest.update(ANALYSIS_BATCH_SCHEMA_VERSION.to_le_bytes());
+        digest.update(env!("CARGO_PKG_VERSION").as_bytes());
+        digest.update(serde_json::to_vec(model).map_err(|error| error.to_string())?);
+        Ok((
+            model.relative_path.clone(),
+            format!("{:x}", digest.finalize()),
+        ))
+    };
+    let results: Vec<Result<(String, String), String>> = if models.len() <= 1 {
+        models.iter().map(identity).collect()
+    } else {
+        let pool = rayon::ThreadPoolBuilder::new()
+            .num_threads(NATIVE_RULE_WORKERS.min(models.len()))
+            .stack_size(NATIVE_RULE_STACK_BYTES)
+            .build()
+            .map_err(|error| error.to_string())?;
+        pool.install(|| models.par_iter().map(identity).collect())
+    };
+    results.into_iter().collect()
 }
 
 fn custom_rule_identity(rule: &RuleMetadata, request: &EvaluateRequest) -> Result<String, String> {
@@ -581,6 +591,27 @@ fn custom_rule_identity(rule: &RuleMetadata, request: &EvaluateRequest) -> Resul
     Ok(format!("{:x}", digest.finalize()))
 }
 
+fn model_cache_identities(
+    models: &[&crate::models::Model],
+    ruleset: &str,
+    project: Option<&str>,
+) -> Result<Vec<String>, String> {
+    let identity = |(index, model): (usize, &&crate::models::Model)| {
+        model_cache_identity(model, ruleset, if index == 0 { project } else { None })
+    };
+    let results: Vec<Result<String, String>> = if models.len() <= 1 {
+        models.iter().enumerate().map(identity).collect()
+    } else {
+        let pool = rayon::ThreadPoolBuilder::new()
+            .num_threads(NATIVE_RULE_WORKERS.min(models.len()))
+            .stack_size(NATIVE_RULE_STACK_BYTES)
+            .build()
+            .map_err(|error| error.to_string())?;
+        pool.install(|| models.par_iter().enumerate().map(identity).collect())
+    };
+    results.into_iter().collect()
+}
+
 fn model_cache_identity(
     model: &crate::models::Model,
     ruleset: &str,

fensu.toml

@@ -159,6 +159,12 @@ path = "src/sqlbuild/runtime/event_exporting/main/command_output_sink.py"
 symbols = ["command_output_sink"]
 reason = "Bare command-output sink decorators are evaluated as decorator(function), requiring positional function calling."
 
+[[targets.python.rule_exceptions]]
+rule = "FFS120"
+path = "src/sqlbuild/compiler/fact_cache/_helpers/restricted_pickle.py"
+symbols = ["_RestrictedUnpickler.find_class"]
+reason = "This override preserves pickle.Unpickler's positional find_class(module, name) protocol."
+
 [[targets.python.rule_exceptions]]
 rule = "FFR402"
 path = "src/sqlbuild/integrations/dagster/__init__.py"

src/sqlbuild/cli/commands/_helpers/compile/pipeline.py

@@ -42,7 +42,7 @@
 from sqlbuild.compiler.contracts.main.validate import evaluate_model_contracts
 from sqlbuild.compiler.contracts.models import ContractValidationResult
 from sqlbuild.compiler.discovery.main.discover import discover_project_inputs
-from sqlbuild.compiler.discovery.models import DiscoveredProjectInputs
+from sqlbuild.compiler.discovery.models import DiscoveredProjectInputs, DiscoveryCacheRequest
 from sqlbuild.compiler.lineage.models import ProjectColumnLineage
 from sqlbuild.compiler.pipeline.main.selected_graph import (
     build_project_graph_with_analysis_selection,
@@ -116,6 +116,9 @@ def _analyze_compile_project(
         sql_analysis_enabled_override=(
             False if profile_flags.skip_discovery_sql_analysis else None
         ),
+        cache_request=DiscoveryCacheRequest(
+            selected_target=selected_target, no_cache=analysis_selection.no_cache
+        ),
     )
     discover_ms: int = elapsed_ms(discover_start)
     _ = complete_compile_phase(

src/sqlbuild/compiler/compile/_helpers/attachment/declaration_scope.py

@@ -5,6 +5,7 @@
 import inspect
 from collections.abc import Mapping
 from dataclasses import replace
+from pathlib import Path
 from types import CodeType
 
 from sqlbuild.compiler.compile._helpers.attachment.scope_relationships import (
@@ -28,7 +29,10 @@
 
 
 def build_declaration_scope(
-    *, discovered_inputs: DiscoveredProjectInputs, loaded_macros: dict[str, LoadedMacro]
+    *,
+    discovered_inputs: DiscoveredProjectInputs,
+    loaded_macros: dict[str, LoadedMacro],
+    compile_cache_dir: Path | None = None,
 ) -> DeclarationScopeBuild:
     """Build one canonical index and validate it before SQL expansion."""
 
@@ -46,7 +50,11 @@ def build_declaration_scope(
         for declaration in index.declarations
     )
     relationships: ScopeRelationshipBuild = (
-        build_scope_relationship_grants(discovered_inputs=discovered_inputs, index=index)
+        build_scope_relationship_grants(
+            discovered_inputs=discovered_inputs,
+            index=index,
+            compile_cache_dir=compile_cache_dir,
+        )
         if has_scoped_relationship_declarations
         and (discovered_inputs.test_files or discovered_inputs.scenario_files)
         else ScopeRelationshipBuild()

src/sqlbuild/compiler/compile/_helpers/attachment/scope_relationships.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from pathlib import Path
+from typing import cast
 
 from sqlbuild.compiler.compile._helpers.render.macros import (
     find_macro_call_names,
@@ -15,14 +16,19 @@
     extract_sql_test_expected_model_names,
     extract_unclassified_sql_test_ctes,
 )
-from sqlbuild.compiler.compile.constants import MACRO_ACTUAL_TEST_CTE_NAME
+from sqlbuild.compiler.compile.constants import (
+    MACRO_ACTUAL_TEST_CTE_NAME,
+    SQL_TEST_EXPECTED_MODELS_FACT_ALGORITHM,
+    SQL_TEST_FACT_CACHE_NAMESPACE,
+)
 from sqlbuild.compiler.compile.models import (
     CompileSqlTestCte,
     ScopeRelationshipBuild,
     ScopeRelationshipFault,
 )
 from sqlbuild.compiler.compile.types import SqlTestMode
-from sqlbuild.compiler.discovery.models import DiscoveredProjectInputs
+from sqlbuild.compiler.discovery.models import DiscoveredProjectInputs, DiscoveredSqlTestFile
+from sqlbuild.compiler.fact_cache.classes.fact_cache_store import FactCacheStore
 from sqlbuild.compiler.scopes.main._resolve_scope_path_visibility import (
     resolve_scope_path_visibility,
 )
@@ -44,17 +50,26 @@
 
 
 def build_scope_relationship_grants(
-    *, discovered_inputs: DiscoveredProjectInputs, index: ScopeIndex
+    *,
+    discovered_inputs: DiscoveredProjectInputs,
+    index: ScopeIndex,
+    compile_cache_dir: Path | None = None,
 ) -> ScopeRelationshipBuild:
     """Return expected-model grants while retaining independent extraction faults."""
 
     lookup: ScopeLookup = build_scope_lookup(index=index)
     shared_declarations: _SharedDeclarations = _shared_declarations_by_directory(lookup=lookup)
-    test_grants, test_faults = _test_relationship_grants(
-        discovered_inputs=discovered_inputs,
-        lookup=lookup,
-        shared_declarations=shared_declarations,
-    )
+    with FactCacheStore(
+        root=compile_cache_dir,
+        namespace=SQL_TEST_FACT_CACHE_NAMESPACE,
+        algorithm=SQL_TEST_EXPECTED_MODELS_FACT_ALGORITHM,
+    ) as fact_cache:
+        test_grants, test_faults = _test_relationship_grants(
+            discovered_inputs=discovered_inputs,
+            lookup=lookup,
+            shared_declarations=shared_declarations,
+            fact_cache=fact_cache,
+        )
     scenario_grants, scenario_faults = _scenario_relationship_grants(
         discovered_inputs=discovered_inputs,
         lookup=lookup,
@@ -71,17 +86,43 @@ def _test_relationship_grants(
     discovered_inputs: DiscoveredProjectInputs,
     lookup: ScopeLookup,
     shared_declarations: _SharedDeclarations,
+    fact_cache: FactCacheStore,
 ) -> tuple[tuple[GrantRecord, ...], tuple[ScopeRelationshipFault, ...]]:
     grants: list[GrantRecord] = []
     faults: list[ScopeRelationshipFault] = []
-    for test_file in discovered_inputs.test_files:
-        for block in test_file.blocks:
+    cache_keys: dict[int, str] = (
+        {
+            file_index: _expected_models_fact_key(test_file=test_file, fact_cache=fact_cache)
+            for file_index, test_file in enumerate(discovered_inputs.test_files)
+        }
+        if fact_cache.enabled
+        else {}
+    )
+    cached_names: dict[str, object] = fact_cache.read_many(
+        tuple(
+            (_expected_models_fact_slot(discovered_inputs.test_files[file_index]), cache_key)
+            for file_index, cache_key in cache_keys.items()
+        )
+    )
+    for file_index, test_file in enumerate(discovered_inputs.test_files):
+        cache_key: str | None = cache_keys.get(file_index)
+        cached_file_names: tuple[tuple[str, ...], ...] | None = _cached_expected_names(
+            value=None if cache_key is None else cached_names.get(cache_key),
+            block_count=len(test_file.blocks),
+        )
+        extracted_names: list[tuple[str, ...]] = []
+        for block_index, block in enumerate(test_file.blocks):
             try:
-                expected_names: tuple[str, ...] = extract_sql_test_expected_model_names(
-                    sql=block.sql_body,
-                    file_label=str(test_file.relative_path),
-                    mode=block.mode,
+                expected_names: tuple[str, ...] = (
+                    cached_file_names[block_index]
+                    if cached_file_names is not None
+                    else extract_sql_test_expected_model_names(
+                        sql=block.sql_body,
+                        file_label=str(test_file.relative_path),
+                        mode=block.mode,
+                    )
                 )
+                extracted_names.append(expected_names)
                 grants.extend(
                     _expected_model_grants(
                         lookup=lookup,
@@ -106,9 +147,43 @@ def _test_relationship_grants(
                     )
             except Exception as error:
                 faults.append(ScopeRelationshipFault(test_file.relative_path, str(error)))
+        if (
+            cache_key is not None
+            and cached_file_names is None
+            and len(extracted_names) == len(test_file.blocks)
+        ):
+            fact_cache.stage(
+                key=cache_key,
+                slot=_expected_models_fact_slot(test_file),
+                value=tuple(extracted_names),
+            )
     return tuple(grants), tuple(faults)
 
 
+def _expected_models_fact_key(
+    *, test_file: DiscoveredSqlTestFile, fact_cache: FactCacheStore
+) -> str:
+    parts: list[str] = [str(test_file.relative_path)]
+    for block in test_file.blocks:
+        parts.extend((block.sql_body, block.mode.value))
+    return fact_cache.key(*parts)
+
+
+def _cached_expected_names(
+    *, value: object, block_count: int
+) -> tuple[tuple[str, ...], ...] | None:
+    if not isinstance(value, tuple) or len(value) != block_count:
+        return None
+    for names in value:
+        if not isinstance(names, tuple) or not all(isinstance(name, str) for name in names):
+            return None
+    return cast(tuple[tuple[str, ...], ...], value)
+
+
+def _expected_models_fact_slot(test_file: DiscoveredSqlTestFile) -> str:
+    return f"expected:{test_file.relative_path}"
+
+
 def _scenario_relationship_grants(
     *,
     discovered_inputs: DiscoveredProjectInputs,

src/sqlbuild/compiler/compile/_helpers/attachment/sql_tests.py

@@ -32,7 +32,9 @@
     extract_assertion_target_model_names,
     extract_sql_test_ctes,
 )
-from sqlbuild.compiler.compile._helpers.sql_tests.native import extract_expanded_sql_tests
+from sqlbuild.compiler.compile._helpers.sql_tests.native import (
+    extract_expanded_sql_tests_cached,
+)
 from sqlbuild.compiler.compile.exceptions import CompileInputError
 from sqlbuild.compiler.compile.models import (
     AuthoredSqlExpansionResult,
@@ -102,7 +104,6 @@ def build_test_inputs_with_cache(
     """Build SQL test inputs inside the compile timing boundary."""
 
     with record_compile_timing("test_input_compile_ms"):
-        del compile_cache_dir
         return build_test_inputs(
             discovered_inputs=discovered_inputs,
             effective_vars=effective_vars,
@@ -111,6 +112,7 @@ def build_test_inputs_with_cache(
             declaration_expansion=declaration_expansion,
             external_sql_reference_resolver=external_sql_reference_resolver,
             sql_function_inputs=sql_function_inputs,
+            compile_cache_dir=compile_cache_dir,
         )
 
 
@@ -123,6 +125,7 @@ def build_test_inputs(
     declaration_expansion: DeclarationExpansionContext,
     external_sql_reference_resolver: ExternalSqlReferenceResolver | None = None,
     sql_function_inputs: tuple[CompileSqlFunctionInput, ...] = (),
+    compile_cache_dir: Path | None = None,
 ) -> tuple[CompileSqlTestInput, ...]:
     """Build compile-time test inputs from discovered SQL-native test blocks."""
 
@@ -276,10 +279,11 @@ def build_test_inputs(
                         test_case=test_case,
                     )
                 )
-    test_ctes_batch: tuple[CompileSqlTestCtes, ...] = extract_expanded_sql_tests(
-        tuple(
+    test_ctes_batch: tuple[CompileSqlTestCtes, ...] = extract_expanded_sql_tests_cached(
+        tests=tuple(
             (test.sql_body, str(test.test_file.relative_path), test.mode) for test in expanded_tests
-        )
+        ),
+        cache_root=compile_cache_dir,
     )
     for test, test_ctes in zip(expanded_tests, test_ctes_batch, strict=True):
         validate_test_ctes(