perf(compile): delta-compile dynamic columns, audit factories and model rules

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

@@ -133,6 +133,7 @@ impl AnalysisSession {
                 dependency_ordered: self.dependency_ordered,
                 cyclic: self.cyclic,
                 legacy_models: self.legacy_models.clone(),
+                families: self.request.dynamic_families_by_table.clone(),
             });
         let (schema_additions, analysis_names) = self.catalog.into_changes();
         let mut request: SessionRequest = self.request;

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

@@ -199,6 +199,8 @@ pub struct RelationTables {
     pub cyclic: bool,
     /// Models whose analysis took the legacy path, which reads every relation's facts.
     pub legacy_models: Vec<String>,
+    /// The request's dynamic column families by table, in request order.
+    pub families: Vec<(String, Vec<DynamicFamily>)>,
 }
 
 /// One inferred output column; nullability is Python's `InferredNullability` value.

crates/sqlbuild-cache/src/project_reuse/_helpers/attempt.rs

@@ -30,7 +30,13 @@ use crate::project_snapshot::models::PathStamp;
 const NO_STORED_COMPILE: &str = "no stored compile";
 const REUSE_BYPASSED: &str = "compile reuse is off";
 const OUTPUT_UNREADABLE: &str = "the stored output cannot be replayed";
-const IDENTITY_CHANGED: &str = "the invocation, runtime, environment or settings changed";
+const INVOCATION_CHANGED: &str = "the command invocation changed";
+const RUNTIME_CHANGED: &str = "the Python runtime changed";
+const SEARCH_PATH_CHANGED: &str = "a Python search path entry changed";
+const ENVIRONMENT_NAMES_CHANGED: &str = "the set of tracked environment variables changed";
+const ENVIRONMENT_CHANGED: &str = "a tracked environment variable changed";
+const SETTINGS_CHANGED: &str = "provider settings inputs changed";
+const MODULES_CHANGED: &str = "an imported Python module outside the project changed";
 const TARGET_CHANGED: &str = "the compiled target changed";
 
 fn now_ns() -> i64 {
@@ -115,8 +121,8 @@ pub(crate) fn check(check: ReuseCheck<'_>) -> ReuseAttempt {
         attempt.restamped = restamped_paths(&inputs.project_files, &attempt.snapshot);
         if comparison.unchanged {
             attempt.delta = DeltaCandidate::Unavailable(OUTPUT_UNREADABLE);
-        } else if !comparison.identity_unchanged {
-            attempt.delta = DeltaCandidate::Unavailable(IDENTITY_CHANGED);
+        } else if let Some(reason) = comparison.identity_change {
+            attempt.delta = DeltaCandidate::Unavailable(reason);
         } else if !target_files_unchanged(
             &inputs.target_files,
             &attempt.project_dir,
@@ -146,21 +152,16 @@ pub(crate) fn check(check: ReuseCheck<'_>) -> ReuseAttempt {
 /// Whether a stored compile still matches, with the digests still valid either way.
 struct Comparison {
     unchanged: bool,
-    identity_unchanged: bool,
+    /// The first identity component that differs, if any.
+    identity_change: Option<&'static str>,
     digests: HashMap<String, String>,
     digested_files: usize,
 }
 
 /// Compare every stored identity, carrying forward digests still valid either way.
 fn checked(attempt: &ReuseAttempt, inputs: &StoredInputs, rules: &ReuseRules) -> Comparison {
-    let identity_unchanged: bool = inputs.invocation_digest == attempt.identity.invocation_digest
-        && inputs.runtime == attempt.identity.runtime
-        && inputs.search_path == attempt.search_path
-        && tracked_environment_names(&inputs.environment_names, rules) == inputs.environment_names
-        && environment_digest(&inputs.environment_names, rules) == inputs.environment_digest
-        && settings_inputs_digest(&inputs.settings_inputs, rules) == inputs.settings_digest
-        && module_stamps_unchanged(&inputs.modules);
-    let (files_unchanged, verified, digested_files) = if identity_unchanged {
+    let identity_change: Option<&'static str> = identity_change(attempt, inputs, rules);
+    let (files_unchanged, verified, digested_files) = if identity_change.is_none() {
         compare_project_files(
             &attempt.project_dir,
             &inputs.project_files,
@@ -170,7 +171,7 @@ fn checked(attempt: &ReuseAttempt, inputs: &StoredInputs, rules: &ReuseRules) ->
         (false, HashMap::new(), 0)
     };
     Comparison {
-        identity_unchanged,
+        identity_change,
         unchanged: files_unchanged
             && target_files_unchanged(
                 &inputs.target_files,
@@ -183,6 +184,33 @@ fn checked(attempt: &ReuseAttempt, inputs: &StoredInputs, rules: &ReuseRules) ->
     }
 }
 
+/// The first stored identity component that no longer matches, in checking order.
+fn identity_change(
+    attempt: &ReuseAttempt,
+    inputs: &StoredInputs,
+    rules: &ReuseRules,
+) -> Option<&'static str> {
+    if inputs.invocation_digest != attempt.identity.invocation_digest {
+        Some(INVOCATION_CHANGED)
+    } else if inputs.runtime != attempt.identity.runtime {
+        Some(RUNTIME_CHANGED)
+    } else if inputs.search_path != attempt.search_path {
+        Some(SEARCH_PATH_CHANGED)
+    } else if tracked_environment_names(&inputs.environment_names, rules)
+        != inputs.environment_names
+    {
+        Some(ENVIRONMENT_NAMES_CHANGED)
+    } else if environment_digest(&inputs.environment_names, rules) != inputs.environment_digest {
+        Some(ENVIRONMENT_CHANGED)
+    } else if settings_inputs_digest(&inputs.settings_inputs, rules) != inputs.settings_digest {
+        Some(SETTINGS_CHANGED)
+    } else if !module_stamps_unchanged(&inputs.modules) {
+        Some(MODULES_CHANGED)
+    } else {
+        None
+    }
+}
+
 pub(crate) fn record(
     attempt: &ReuseAttempt,
     record: &CompileRecord,

crates/sqlbuild-python/src/bindings/_helpers/compile_command/constants.rs

@@ -2,6 +2,7 @@
 
 use sqlbuild_core::constants::PROJECT_CONFIG_FILENAME;
 
+use crate::bindings::_helpers::compile_delta::driver::DELTA_VERIFY_ENV_VAR;
 use crate::bindings::_helpers::frontier::stage_capture::CAPTURE_DIR_ENV_VAR;
 
 /// The directory under the compiler cache directory holding the native reuse store.
@@ -16,8 +17,11 @@ pub(crate) const REUSE_DISABLE_VALUE: &str = "1";
 pub(crate) const REUSE_LOGGER_NAME: &str = "sqlbuild.compile.reuse";
 pub(crate) const TRACKED_ENVIRONMENT_PREFIXES: [&str; 2] = ["SQLBUILD_", "SQB_"];
 /// Sorted; tracked-prefix variables that never invalidate a stored compile.
-pub(crate) const UNTRACKED_ENVIRONMENT_NAMES: [&str; 2] =
-    [CAPTURE_DIR_ENV_VAR, REUSE_DISABLE_ENV_VAR];
+pub(crate) const UNTRACKED_ENVIRONMENT_NAMES: [&str; 3] = [
+    CAPTURE_DIR_ENV_VAR,
+    REUSE_DISABLE_ENV_VAR,
+    DELTA_VERIFY_ENV_VAR,
+];
 pub(crate) const MISSING_ENVIRONMENT_VALUE: &str = "\0missing";
 pub(crate) const MISSING_FILE_DIGEST: &str = "\0missing";
 pub(crate) const PROJECT_CONFIG_FILENAMES: [&str; 2] =

crates/sqlbuild-python/src/bindings/_helpers/compile_context/delta.rs

@@ -4,7 +4,9 @@ use std::collections::{BTreeMap, BTreeSet};
 use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
 
 use pyo3::prelude::PyResult;
-use sqlbuild_analysis::assembly::analysis_session::models::{RelationTables, SessionRequest};
+use sqlbuild_analysis::assembly::analysis_session::models::{
+    DynamicFamily, RelationTables, SessionRequest,
+};
 use sqlbuild_cache::project_reuse::types::ImageSections;
 use sqlbuild_discovery::header_changes::types::HeaderKeys;
 use sqlbuild_rules::engine::main::retained_request::retained_request;
@@ -103,6 +105,8 @@ pub(crate) struct ImageCapture {
     pub(crate) environment_names: Vec<String>,
     /// Each discovered model's header keys, by model file path.
     pub(crate) header_keys: BTreeMap<String, HeaderKeys>,
+    /// The audit factories each model file's header attaches, for files attaching any.
+    pub(crate) audit_references: BTreeMap<String, Vec<String>>,
     /// Each model file's SQL lint cache key, in project model order.
     pub(crate) lint_model_keys: Option<Vec<(String, String)>>,
     /// Each other SQL file's lint cache key, in walk order.
@@ -193,6 +197,8 @@ pub(crate) struct DeltaInjection {
     pub(crate) coverage: Option<Vec<ModelCoverage>>,
     /// Every project model in project order, with the models it reads.
     pub(crate) coverage_order: Vec<(String, Vec<String>)>,
+    /// The audit factories each model file's header attaches, for files attaching any.
+    pub(crate) audit_references: BTreeMap<String, Vec<String>>,
 }
 
 static CAPTURE: Mutex<Option<ImageCapture>> = Mutex::new(None);
@@ -328,17 +334,9 @@ pub(crate) fn injected_fact_digest(key: &[String]) -> Option<String> {
 
 /// Give a delta's analysis session the whole project's relation facts and analysis order.
 pub(crate) fn inject_relations(mut request: SessionRequest) -> pyo3::PyResult<SessionRequest> {
-    if !request.dynamic_families_by_table.is_empty() {
-        capture_unsupported("a model declares dynamic column families");
-    }
     let Some(injection) = injection() else {
         return Ok(request);
     };
-    if !request.dynamic_families_by_table.is_empty() {
-        return Err(delta_unsupported(
-            "a model declares dynamic column families",
-        ));
-    }
     let relations: &RelationTables = &injection.relations;
     request.forced_order = Some((relations.dependency_ordered, relations.cyclic));
     request.retain_relations = true;
@@ -347,6 +345,16 @@ pub(crate) fn inject_relations(mut request: SessionRequest) -> pyo3::PyResult<Se
         .iter()
         .map(|model| model.name.as_str())
         .collect();
+    let rerun: BTreeSet<&str> = analysed
+        .iter()
+        .copied()
+        .chain(injection.rerun_names.iter().map(String::as_str))
+        .collect();
+    request.dynamic_families_by_table = injected_families(
+        &request.dynamic_families_by_table,
+        &relations.families,
+        &rerun,
+    )?;
     for (table, injected) in [
         (&mut request.column_types, &relations.types),
         (&mut request.column_nullability, &relations.nullability),
@@ -363,6 +371,32 @@ pub(crate) fn inject_relations(mut request: SessionRequest) -> pyo3::PyResult<Se
     Ok(request)
 }
 
+/// The stored compile's dynamic column families, when no rerun model changed its own.
+fn injected_families(
+    current: &[(String, Vec<DynamicFamily>)],
+    stored: &[(String, Vec<DynamicFamily>)],
+    rerun: &BTreeSet<&str>,
+) -> pyo3::PyResult<Vec<(String, Vec<DynamicFamily>)>> {
+    let families_of = |table: &[(String, Vec<DynamicFamily>)], name: &str| {
+        table
+            .iter()
+            .find(|(table, _)| table == name)
+            .map(|(_, families)| families.clone())
+    };
+    let changed: bool = current
+        .iter()
+        .any(|(name, _)| !rerun.contains(name.as_str()))
+        || rerun
+            .iter()
+            .any(|name| families_of(current, name) != families_of(stored, name));
+    if changed {
+        return Err(delta_unsupported(
+            "a model's dynamic column families changed",
+        ));
+    }
+    Ok(stored.to_vec())
+}
+
 /// What a delta compile's reuse record keeps of the stored compile it patched.
 #[derive(Debug, Default)]
 pub(crate) struct RecordMerge {

crates/sqlbuild-python/src/bindings/_helpers/compile_delta/driver.rs

@@ -34,6 +34,7 @@ use crate::bindings::_helpers::compile_driver::command::{
 };
 use crate::bindings::_helpers::compile_outputs::reuse::NativeReuseAttempt;
 use crate::bindings::_helpers::records::fields::field as native_field;
+use crate::bindings::_helpers::rules::custom_evaluation::stored_custom_rule_refusal;
 use sqlbuild_discovery::header_changes::main::custom_rule_reads::custom_rule_reads;
 use sqlbuild_discovery::header_changes::main::header_change::header_change;
 use sqlbuild_discovery::header_changes::models::HeaderChange;
@@ -57,6 +58,8 @@ const UNKNOWN_SOURCE: &str = "unknown source";
 const SOURCE_KIND: &str = "source";
 /// The upstream kind of a model a model reads.
 const MODEL_KIND: &str = "model";
+/// The custom-rule fact listing every project source.
+const PROJECT_SOURCES_FACT: &str = "project.sources";
 const TESTS_ROOT: &str = "tests/";
 const STDERR: &str = "stderr";
 const SQL_SUFFIX: &str = ".sql";
@@ -321,6 +324,7 @@ fn lent_injection(
             .collect(),
         lint_model_keys: image.lint_model_keys.take(),
         lint_file_keys: image.lint_file_keys.take(),
+        audit_references: image.audit_references.clone(),
         ..DeltaInjection::default()
     };
     (image, injection)
@@ -588,6 +592,17 @@ fn prepared(
         Ok(found) => found,
         Err(reason) => return Ok(Err(reason)),
     };
+    let all_sources: bool = gate
+        .fact_digests
+        .iter()
+        .any(|(key, _)| lists_all_sources(key));
+    if let Some(reason) = stored_custom_rule_refusal(
+        &project_dir,
+        (&rerun.model_paths, &rerun.changed),
+        all_sources,
+    ) {
+        return Ok(Err(reason.to_owned()));
+    }
     let limit: usize = MAX_RERUN_MODELS.max(gate.models.len() * MAX_RERUN_PERCENT / 100);
     if rerun.models.len() > limit {
         return Ok(Err(TOO_MANY_MODELS.to_owned()));
@@ -647,7 +662,11 @@ fn rounds<'py>(
         {
             return stopped("the stored compile reports problems in files the edit leaves alone");
         }
-        rerun.sources = sources_read(&image, &rerun);
+        rerun.sources = if reads_all_sources(&injection) {
+            None
+        } else {
+            sources_read(&image, &rerun)
+        };
         let injection_for_round: DeltaInjection =
             for_rerun(injection, &rerun, rerun.sources.clone());
         let (outcome, back) = round(request, attempt, (injection_for_round, &rerun))?;
@@ -705,6 +724,19 @@ fn rounds<'py>(
     stopped("changed shapes reach too far")
 }
 
+/// Whether a custom rule read the project's whole source list, which a delta must then keep.
+fn reads_all_sources(injection: &DeltaInjection) -> bool {
+    injection
+        .fact_digests
+        .keys()
+        .any(|key| lists_all_sources(key))
+}
+
+/// Whether a custom-rule fact key is the project's whole source list.
+fn lists_all_sources(key: &[String]) -> bool {
+    matches!(key, [name] if name == PROJECT_SOURCES_FACT)
+}
+
 /// The sources the rerun models, functions, audits and rerun SQL tests read; `None` for all.
 fn sources_read(image: &ProjectImage, rerun: &Rerun) -> Option<BTreeSet<String>> {
     if image.diagnostics.iter().any(outside_resources) {
@@ -1162,6 +1194,10 @@ fn patched_image(base: ProjectImage, rerun: &Rerun, capture: ImageCapture) -> Pr
         image.coverage = capture.coverage;
     }
     image.header_keys.extend(capture.header_keys);
+    image
+        .audit_references
+        .retain(|path, _| !rerun.model_paths.contains(path));
+    image.audit_references.extend(capture.audit_references);
     let mut pinned: BTreeSet<String> = std::mem::take(&mut image.pinned_sources)
         .into_iter()
         .collect();