Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -552,6 +552,7 @@ pub(crate) fn decode_outcome(
binding_validated,
},
cleaned_sql: String::new(),
deferred_cleaning: None,
validated_schema: schema.clone(),
fused_binding_validated: true,
};
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -31,9 +31,9 @@ use crate::assembly::analysis_session::_helpers::mappings::{
use crate::assembly::analysis_session::_helpers::mappings::{producers, waves};
use crate::assembly::analysis_session::constants::{UNKNOWN_NULLABILITY, UNKNOWN_TYPE};
use crate::assembly::analysis_session::models::{
AnalysisSession, ColumnFact, ContractProof, FinishedSession, LineageFacts, LineageRow,
ModelAnalysis, ModelOutcome, ModelRequest, Phase, PivotOutcome, PivotTables, SessionModelFacts,
SessionOutcome, SessionRequest, SessionStep,
AnalysisSession, ColumnFact, ContractProof, DeferredCleaning, FinishedSession, LineageFacts,
LineageRow, ModelAnalysis, ModelOutcome, ModelRequest, Phase, PivotOutcome, PivotTables,
SessionModelFacts, SessionOutcome, SessionRequest, SessionStep,
};
use crate::assembly::analysis_session::types::{Pairs, Shapes};
use crate::semantic_validation::models::ProjectCatalog;
Expand Down Expand Up @@ -63,6 +63,7 @@ impl AnalysisSession {
available_nullability: ShapeTable::from_shapes(&request.column_nullability),
complete_shapes,
outcomes: vec![None; request.models.len()],
pending_cleaned: vec![false; request.models.len()],
waves: if dependency_ordered || request.models.is_empty() {
waves
} else {
Expand Down Expand Up @@ -112,13 +113,26 @@ impl AnalysisSession {
}
let dynamic_contracts: Vec<PivotOutcome> = self.dynamic_contracts()?;
let pool: Result<Arc<ThreadPool>, String> = self.catalog.native.analysis_pool();
let models: Vec<ModelOutcome> = self
let mut models: Vec<ModelOutcome> = self
.outcomes
.into_iter()
.collect::<Option<_>>()
.ok_or("the session left a model unanalysed")?;
let (schema_additions, analysis_names) = self.catalog.into_changes();
let request: SessionRequest = self.request;
let mut request: SessionRequest = self.request;
for ((outcome, model), pending) in models
.iter_mut()
.zip(request.models.iter_mut())
.zip(self.pending_cleaned)
{
if pending {
outcome.deferred_cleaning = Some(DeferredCleaning {
dialect: request.dialect.clone(),
query_sql: std::mem::take(&mut model.query_sql),
placeholders: std::mem::take(&mut model.placeholders),
});
}
}
let mut model_facts: HashMap<String, SessionModelFacts> = HashMap::new();
let mut dynamic_proofs: HashMap<String, ContractProof> = HashMap::new();
for ((model, outcome), proof) in request.models.iter().zip(&models).zip(&dynamic_contracts)
Expand Down Expand Up @@ -247,6 +261,7 @@ impl AnalysisSession {
self.outcomes[*model] = Some(ModelOutcome {
analysis,
cleaned_sql: result.cleaned_sql,
deferred_cleaning: None,
validated_schema: schema,
fused_binding_validated: true,
});
Expand Down Expand Up @@ -394,8 +409,8 @@ impl AnalysisSession {
/// Python's deferred binding validation, then the dataflow's shape publication.
fn finish_wave(&mut self, wave: usize) -> Result<(), String> {
let models: Vec<usize> = self.waves[wave].clone();
let mut validations: Vec<(usize, String, Shapes)> = Vec::new();
let validated: &[usize] = if self.cyclic { &[] } else { &models };
let mut revalidated: Vec<(usize, Shapes)> = Vec::new();
for model in validated {
let schema: Shapes = self.binding_schema(&self.request.models[*model]);
let outcome: &ModelOutcome = self.outcomes[*model]
Expand All @@ -404,9 +419,18 @@ impl AnalysisSession {
if !outcome.fused_binding_validated
|| !same_relations(&outcome.validated_schema, &schema)
{
validations.push((*model, outcome.cleaned_sql.clone(), schema));
revalidated.push((*model, schema));
}
}
let pending: Vec<usize> = revalidated.iter().map(|(model, _)| *model).collect();
self.ensure_cleaned(&pending)?;
let mut validations: Vec<(usize, String, Shapes)> = Vec::new();
for (model, schema) in revalidated {
let outcome: &ModelOutcome = self.outcomes[model]
.as_ref()
.ok_or("a wave model was not analysed")?;
validations.push((model, outcome.cleaned_sql.clone(), schema));
}
if !validations.is_empty() {
let requests: Vec<(&str, &Shapes)> = validations
.iter()
Expand All @@ -421,6 +445,9 @@ impl AnalysisSession {
}
}
if self.dependency_ordered {
if self.request.case_sensitive_shapes {
self.ensure_cleaned(&models)?;
}
for model in &models {
self.publish(*model)?;
}
Expand Down Expand Up @@ -500,26 +527,11 @@ impl AnalysisSession {
None => misses.push(position),
}
}
let members: Vec<BatchMember<'_>> = hits
.iter()
.map(|(position, _)| {
batch_member(&self.request.models[models[*position]], &schemas[*position])
})
.collect();
let cleaned: Vec<Result<String, String>> = self
.catalog
.normalize_each(&self.request.dialect, &members)
.unwrap_or_else(|error| members.iter().map(|_| Err(error.clone())).collect());
for ((position, mut outcome), sql) in hits.into_iter().zip(cleaned) {
for (position, outcome) in hits {
let model: usize = models[position];
match sql {
Ok(sql) => {
outcome.cleaned_sql = sql;
self.outcomes[model] = Some(outcome);
cache.hits[model] = true;
}
Err(_) => misses.push(position),
}
self.outcomes[model] = Some(outcome);
self.pending_cleaned[model] = true;
cache.hits[model] = true;
}
misses.sort_unstable();
cache.stats.hits += models.len() - misses.len();
Expand All @@ -528,6 +540,38 @@ impl AnalysisSession {
Ok(misses)
}

/// Normalize the cleaned SQL of the cache hits among `models` that still lack it.
fn ensure_cleaned(&mut self, models: &[usize]) -> Result<(), String> {
let pending: Vec<usize> = models
.iter()
.copied()
.filter(|model| self.pending_cleaned[*model])
.collect();
if pending.is_empty() {
return Ok(());
}
let schemas: Vec<Shapes> = pending
.iter()
.map(|model| self.binding_schema(&self.request.models[*model]))
.collect();
let members: Vec<BatchMember<'_>> = pending
.iter()
.zip(&schemas)
.map(|(model, schema)| batch_member(&self.request.models[*model], schema))
.collect();
let cleaned: Vec<Result<String, String>> = self
.catalog
.normalize_each(&self.request.dialect, &members)?;
for (model, sql) in pending.into_iter().zip(cleaned) {
let sql: String = sql.map_err(|reason| {
native_failure(&model_context(&self.request.models[model]), &reason)
})?;
self.outcome_mut(model)?.cleaned_sql = sql;
self.pending_cleaned[model] = false;
}
Ok(())
}

/// Store each finished native outcome of `models` the cache did not answer.
fn store_cached(&mut self, models: &[usize]) {
let Some(cache) = self.cache.as_mut() else {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ pub mod prove_dynamic_contracts;
pub mod prove_finished_dynamic_contracts;
pub mod query_columns;
pub mod recover_cte_facts;
pub mod resolved_cleaned_sql;
pub mod run_analysis_session;
pub mod session_sharing;
pub mod sql_test_helper_reads;
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,29 @@
//! A finished model outcome's cleaned analysis SQL, normalized on first read for cache hits.

use std::borrow::Cow;
use std::collections::HashMap;

use sqlbuild_core::panics::main::catch_compiler_panic::catch_compiler_panic;

use crate::assembly::analysis_session::models::{DeferredCleaning, ModelOutcome};
use crate::semantic_validation::main::normalize::normalize_analysis_sql;
use crate::semantic_validation::models::NormalizationInput;

/// The outcome's cleaned SQL, normalizing a cache hit's deferred input when it has one.
pub fn resolved_cleaned_sql(outcome: &ModelOutcome) -> Result<Cow<'_, str>, String> {
let Some(DeferredCleaning {
dialect,
query_sql,
placeholders,
}) = outcome.deferred_cleaning.as_ref()
else {
return Ok(Cow::Borrowed(&outcome.cleaned_sql));
};
let request: NormalizationInput = NormalizationInput {
sql: query_sql.clone(),
dialect: dialect.clone(),
stubs: HashMap::new(),
placeholders: placeholders.iter().cloned().collect(),
};
catch_compiler_panic(|| normalize_analysis_sql(request)).map(Cow::Owned)
}
12 changes: 12 additions & 0 deletions crates/sqlbuild-analysis/src/assembly/analysis_session/models.rs
Original file line number Diff line number Diff line change
Expand Up @@ -213,11 +213,21 @@ pub struct ModelAnalysis {
pub binding_validated: bool,
}

/// The inputs a cache hit normalizes into its cleaned SQL only when a reader needs it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DeferredCleaning {
pub(crate) dialect: String,
pub(crate) query_sql: String,
pub(crate) placeholders: Pairs,
}

/// Python's `ModelSqlAnalysis` for one model, without the placeholders Python keeps.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ModelOutcome {
pub analysis: ModelAnalysis,
/// Empty while `deferred_cleaning` is set; read through `resolved_cleaned_sql`.
pub cleaned_sql: String,
pub deferred_cleaning: Option<DeferredCleaning>,
pub validated_schema: Shapes,
pub fused_binding_validated: bool,
}
Expand Down Expand Up @@ -307,6 +317,8 @@ pub struct AnalysisSession {
pub(crate) available_nullability: ShapeTable,
pub(crate) complete_shapes: ShapeTable,
pub(crate) outcomes: Vec<Option<ModelOutcome>>,
/// Cache hits whose cleaned SQL is still unnormalized.
pub(crate) pending_cleaned: Vec<bool>,
pub(crate) phase: Phase,
pub(crate) publications: Shapes,
pub(crate) failures: Vec<String>,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -83,11 +83,17 @@ pub(crate) fn rename_ctes(
preserves_tokens((sql, &tokens), (&renamed, &renamed_tokens), &replaced).then_some(renamed)
}

/// Case-folded names of every CTE a statement defines at any depth; `None` when unscannable.
pub(crate) fn defined_cte_keys(sql: &str, dialect: SliceDialect) -> Option<Vec<String>> {
let tokens = match tokens(sql, dialect) {
Ok(tokens) => tokens,
Err(_) => return None,
/// Case-folded names of every CTE a statement defines at any depth; empty without `with`.
pub(crate) fn defined_cte_keys(sql: &str, dialect: SliceDialect) -> Vec<String> {
if !sql
.as_bytes()
.windows(4)
.any(|window| window.eq_ignore_ascii_case(b"with"))
{
return Vec::new();
}
let Ok(tokens) = tokens(sql, dialect) else {
return Vec::new();
};
let code: Vec<&Token> = tokens.iter().filter(|token| !token.comment).collect();
let text = |index: usize| code.get(index).map(|token| &sql[token.start..token.end]);
Expand Down Expand Up @@ -138,7 +144,7 @@ pub(crate) fn defined_cte_keys(sql: &str, dialect: SliceDialect) -> Option<Vec<S
keys.push(key.to_string());
}
}
Some(keys)
keys
}

/// Whether each token sits directly in a WITH list, so a comma there separates CTEs.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -228,7 +228,7 @@ impl ScopeGraph {
request: &ScopeRequest<'_>,
) -> Result<(), String> {
for (label, sql) in readers {
for key in defined_cte_keys(sql, request.slice_dialect).unwrap_or_default() {
for key in defined_cte_keys(sql, request.slice_dialect) {
if let Some(index) = self.helper_indexes.get(&key) {
return Err(compile_error(&format!(
"SQL test '{}' defines CTE '{}' inside CTE '{label}', which redefines \
Expand Down
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
//! Coarse project-level planning for SQL-native test artifacts.

use std::collections::{BTreeMap, HashMap, HashSet};
use std::sync::{Arc, Mutex, OnceLock};
use std::sync::{Arc, Mutex, OnceLock, PoisonError};
use std::time::Instant;

use polyglot_sql::{Dialect, Expression};
Expand Down Expand Up @@ -91,6 +91,8 @@ struct ProjectContext {
patterns: SqlTestPatterns,
render_dialect: Arc<Dialect>,
rejects_nested_with: bool,
/// Each model query's reserved CTE name, since every test reading the model checks it.
reserved_ctes: Arc<Mutex<HashMap<String, Option<String>>>>,
}

#[derive(Clone)]
Expand Down Expand Up @@ -542,6 +544,7 @@ pub(crate) fn plan_batch(request: SqlTestPlanBatch) -> Result<SqlTestPlanBatchOu
patterns: SqlTestPatterns::new(request.lexical_syntax)?,
render_dialect,
rejects_nested_with,
reserved_ctes: Arc::new(Mutex::new(HashMap::new())),
};
let workers = request.workers.clamp(1, MAX_WORKERS);
let pool = rayon::ThreadPoolBuilder::new()
Expand Down Expand Up @@ -695,6 +698,27 @@ fn plan_direct_test(
})
}

impl ProjectContext {
fn reserved_cte_name(&self, query_sql: &str) -> Option<String> {
let cached: Option<Option<String>> = self
.reserved_ctes
.lock()
.unwrap_or_else(PoisonError::into_inner)
.get(query_sql)
.cloned();
if let Some(name) = cached {
return name;
}
let name: Option<String> =
reserved_cte_name(query_sql, SliceDialect::new(Some(&self.dialect)));
self.reserved_ctes
.lock()
.unwrap_or_else(PoisonError::into_inner)
.insert(query_sql.to_owned(), name.clone());
name
}
}

/// The first CTE a model defines under a name reserved for generated test-query CTEs.
fn reserved_cte_name(query_sql: &str, dialect: SliceDialect) -> Option<String> {
let reserved = MOCK_CTE_PREFIXES.iter().copied().chain([
Expand All @@ -705,7 +729,6 @@ fn reserved_cte_name(query_sql: &str, dialect: SliceDialect) -> Option<String> {
]);
let prefixes: Vec<&str> = reserved.collect();
defined_cte_keys(query_sql, dialect)
.unwrap_or_default()
.into_iter()
.find(|key| is_reserved_cte_key(key, &prefixes))
}
Expand Down Expand Up @@ -769,8 +792,7 @@ fn plan_model_test(
.model_query_overrides
.get(model_name)
.unwrap_or(&model.query_sql);
if let Some(name) = reserved_cte_name(query_sql, SliceDialect::new(Some(&context.dialect)))
{
if let Some(name) = context.reserved_cte_name(query_sql) {
return Err(compile_error(&format!(
"SQL test '{}' reads model '{model_name}', which defines CTE '{name}'; names \
starting with a SQLBuild test prefix such as __helper__ or __ref__ are reserved \
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -449,7 +449,7 @@ impl RenderCteState {
.collect();
let keys: HashSet<String> = top_level.iter().map(|(name, _, _)| cte_key(name)).collect();
for (_, body, origin) in &top_level {
let nested = defined_cte_keys(body, self.dialect).unwrap_or_default();
let nested = defined_cte_keys(body, self.dialect);
if let Some(name) = nested.iter().find(|name| keys.contains(*name)) {
return Err(format!(
"{origin} defines CTE '{name}', which is also a CTE SQLBuild generates at the \
Expand Down
2 changes: 1 addition & 1 deletion crates/sqlbuild-analysis/src/compiler/tests/helpers.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1793,7 +1793,7 @@ fn assert_names_isolated(sql: &str, dialect: &str, helper_name: &str, label: &st
});
assert!(!helper_at_top_level, "{label}: {sql}");
for cte in &split.ctes {
let nested = defined_cte_keys(cte.body, slice_dialect).expect("body scans");
let nested = defined_cte_keys(cte.body, slice_dialect);
assert!(
nested.iter().all(|name| !top_level.contains(name)),
"{label}: {} nests {nested:?}: {sql}",
Expand Down
Loading
Loading