diff --git a/crates/sqlbuild-rules-native/src/query_analysis/_helpers/borrowed_facts.rs b/crates/sqlbuild-rules-native/src/query_analysis/_helpers/borrowed_facts.rs index f7ae2b0367..36bd0f9c60 100644 --- a/crates/sqlbuild-rules-native/src/query_analysis/_helpers/borrowed_facts.rs +++ b/crates/sqlbuild-rules-native/src/query_analysis/_helpers/borrowed_facts.rs @@ -1,11 +1,12 @@ //! Immutable CTE output graph over one borrowed syntax tree. +use std::cell::Cell; use std::collections::{BTreeSet, HashMap, HashSet}; use std::rc::Rc; use polyglot_sql::expressions::{Column, Join, JoinKind, Select, Star, Values, With}; use polyglot_sql::{ - DialectType, Expression, ProjectionNullability, TransformKind, ValidationSchema, + DialectType, Expression, ExpressionWalk, ProjectionNullability, TransformKind, ValidationSchema, }; const SQL_WILDCARD: &str = "*"; @@ -51,11 +52,64 @@ struct Sources { } type Aliases = HashMap>; +struct BindingProof { + enabled: bool, + complete: Cell, + lateral_aliases: bool, +} + +impl BindingProof { + fn require(&self, condition: bool) { + if self.enabled && !condition { + self.complete.set(false); + } + } +} + pub(super) fn infer( expression: &Expression, schema: Option<&ValidationSchema>, dialect: DialectType, ) -> Vec { + infer_with_binding_proof(expression, schema, dialect, false).0 +} + +/// Prove references while deriving facts; an incomplete proof requires normal validation. +pub(super) fn infer_bound( + expression: &Expression, + schema: Option<&ValidationSchema>, + dialect: DialectType, +) -> Option> { + if expression.dfs().any(|node| { + matches!( + node, + Expression::Union(_) | Expression::Intersect(_) | Expression::Except(_) + ) + }) { + return None; + } + let (outputs, complete) = infer_with_binding_proof(expression, schema, dialect, true); + complete.then_some(outputs) +} + +fn infer_with_binding_proof( + expression: &Expression, + schema: Option<&ValidationSchema>, + dialect: DialectType, + enabled: bool, +) -> (Vec, bool) { + let proof = BindingProof { + enabled, + complete: Cell::new(true), + lateral_aliases: matches!( + dialect, + DialectType::Snowflake + | DialectType::DuckDB + | DialectType::Redshift + | DialectType::Spark + | DialectType::Databricks + ), + }; let mut relations = Relations::new(); if let Some(schema) = schema { for table in &schema.tables { @@ -74,32 +128,32 @@ pub(super) fn infer( transform: TransformKind::Direct, }) .collect(); - relations.insert( + let previous = relations.insert( table.name.to_lowercase(), Source::new(outputs, Some(table.name.clone())), ); + proof.require(previous.is_none()); } } - query( - expression, - &relations, - matches!( - dialect, - DialectType::Snowflake - | DialectType::DuckDB - | DialectType::Redshift - | DialectType::Spark - | DialectType::Databricks - ), - ) + let outputs = query(expression, &relations, &proof); + (outputs, proof.complete.get()) } -fn query(expression: &Expression, inherited: &Relations, lateral_aliases: bool) -> Vec { +fn query(expression: &Expression, inherited: &Relations, proof: &BindingProof) -> Vec { let expression = unwrapped(expression); + proof.require(matches!( + expression, + Expression::Select(_) | Expression::Subquery(_) + )); let mut relations = inherited.clone(); if let Some(with) = query_with(expression) { + let mut names: HashSet = HashSet::new(); for cte in &with.ctes { - let mut outputs = query(&cte.this, &relations, lateral_aliases); + if proof.enabled { + proof.require(!with.recursive && names.insert(cte.alias.name.to_lowercase())); + } + let mut outputs = query(&cte.this, &relations, proof); + proof.require(cte.columns.len() <= outputs.len()); for (output, alias) in outputs.iter_mut().zip(&cte.columns) { output.name = alias.name.clone(); } @@ -107,25 +161,32 @@ fn query(expression: &Expression, inherited: &Relations, lateral_aliases: bool) } } match expression { - Expression::Select(select) => select_outputs(select, &relations, lateral_aliases), - Expression::Values(values) => values_outputs(values), + Expression::Select(select) => select_outputs(select, &relations, proof), + Expression::Values(values) => { + let outputs = values_outputs(values); + proof.require(outputs.iter().all(|fact| fact.resolved)); + outputs + } Expression::Union(set) => set_outputs( - query(&set.left, &relations, lateral_aliases), - query(&set.right, &relations, lateral_aliases), + query(&set.left, &relations, proof), + query(&set.right, &relations, proof), set.by_name, ), Expression::Intersect(set) => set_outputs( - query(&set.left, &relations, lateral_aliases), - query(&set.right, &relations, lateral_aliases), + query(&set.left, &relations, proof), + query(&set.right, &relations, proof), set.by_name, ), Expression::Except(set) => set_outputs( - query(&set.left, &relations, lateral_aliases), - query(&set.right, &relations, lateral_aliases), + query(&set.left, &relations, proof), + query(&set.right, &relations, proof), set.by_name, ), - Expression::Subquery(subquery) => query(&subquery.this, &relations, lateral_aliases), - _ => Vec::new(), + Expression::Subquery(subquery) => query(&subquery.this, &relations, proof), + _ => { + proof.require(false); + Vec::new() + } } } @@ -166,25 +227,61 @@ fn set_outputs( left } -fn select_outputs( - select: &Select, - relations: &Relations, - lateral_aliases: bool, -) -> Vec { +fn select_outputs(select: &Select, relations: &Relations, proof: &BindingProof) -> Vec { + if proof.enabled { + proof.require( + select.lateral_views.is_empty() + && select.prewhere.is_none() + && select.distribute_by.is_none() + && select.cluster_by.is_none() + && select.sort_by.is_none() + && select.limit.is_none() + && select.offset.is_none() + && select.limit_by.is_none() + && select.fetch.is_none() + && select.distinct_on.is_none() + && select.top.is_none() + && select.sample.is_none() + && select.settings.is_none() + && select.format.is_none() + && select.windows.is_none() + && select.connect.is_none() + && select.into.is_none() + && select.locks.is_empty() + && select.for_xml.is_empty() + && select.for_json.is_empty() + && select.exclude.is_none() + && select.kind.is_none() + && select.operation_modifiers.is_empty() + && select.hint.is_none() + && select.option.is_none(), + ); + } let mut sources = Sources::default(); for expression in select.from.iter().flat_map(|from| &from.expressions) { - if let Some(source) = source(expression, relations, &sources, lateral_aliases) { + if let Some(source) = source(expression, relations, &sources, proof) { + proof.require( + !sources + .items + .iter() + .any(|(name, _)| name.eq_ignore_ascii_case(&source.0)), + ); sources.items.push(source); } } for join in &select.joins { - sources = joined_sources(sources, join, relations, lateral_aliases); + sources = joined_sources(sources, join, relations, proof); } let mut outputs: Vec = Vec::new(); let mut aliases = Aliases::new(); for projection in &select.expressions { let projection = unwrapped(projection); if let Some(star) = projection_star(projection) { + proof.require(star.except.is_none() && star.replace.is_none() && star.rename.is_none()); + if proof.enabled { + let expanded = star_outputs(&star, &sources); + proof.require(!expanded.is_empty() && expanded.iter().all(|fact| fact.resolved)); + } outputs.extend(star_outputs(&star, &sources)); continue; } @@ -197,6 +294,9 @@ fn select_outputs( _ => projection, }; let mut fact = expression_fact(expression, &sources, &aliases); + if proof.enabled { + proof.require(fact.resolved && scalar_binding_expression(expression)); + } fact.name = name.to_string(); if let Expression::Column(column) = unwrapped(expression) && select @@ -206,30 +306,69 @@ fn select_outputs( { fact.nullability = ProjectionNullability::NonNull; } - if lateral_aliases && matches!(projection, Expression::Alias(_)) { + if proof.lateral_aliases && matches!(projection, Expression::Alias(_)) { let key = name.to_lowercase(); let value = (!aliases.contains_key(&key)).then(|| fact.clone()); aliases.insert(key, value); } outputs.push(fact); } + if proof.enabled { + for clause in select + .where_clause + .iter() + .map(|value| &value.this) + .chain(select.having.iter().map(|value| &value.this)) + .chain(select.qualify.iter().map(|value| &value.this)) + .chain(select.group_by.iter().flat_map(|value| &value.expressions)) + .chain( + select + .order_by + .iter() + .flat_map(|value| &value.expressions) + .map(|value| &value.this), + ) + { + proof.require( + scalar_binding_expression(clause) + && expression_fact(clause, &sources, &aliases).resolved, + ); + } + } outputs } +fn scalar_binding_expression(expression: &Expression) -> bool { + !expression.dfs().any(|node| { + matches!( + node, + Expression::Select(_) + | Expression::Union(_) + | Expression::Intersect(_) + | Expression::Except(_) + | Expression::Subquery(_) + | Expression::Window(_) + | Expression::WindowFunction(_) + ) + }) +} + fn source( expression: &Expression, relations: &Relations, preceding: &Sources, - lateral_aliases: bool, + proof: &BindingProof, ) -> Option<(String, Source)> { match unwrapped(expression) { Expression::Table(table) => { + proof.require(relations.contains_key(&table.name.name.to_lowercase())); let name = table.alias.as_ref().unwrap_or(&table.name).name.clone(); let mut source = relations .get(&table.name.name.to_lowercase()) .cloned() .unwrap_or_else(|| Source::new(Vec::new(), Some(table.name.name.clone()))); if !table.column_aliases.is_empty() { + proof.require(table.column_aliases.len() <= source.outputs.len()); let mut outputs = source.outputs.as_ref().clone(); for (output, alias) in outputs.iter_mut().zip(&table.column_aliases) { output.name = alias.name.clone(); @@ -239,7 +378,9 @@ fn source( Some((name, source)) } Expression::Subquery(subquery) => { - let mut outputs = query(&subquery.this, relations, lateral_aliases); + proof.require(subquery.alias.is_some()); + let mut outputs = query(&subquery.this, relations, proof); + proof.require(subquery.column_aliases.len() <= outputs.len()); for (output, alias) in outputs.iter_mut().zip(&subquery.column_aliases) { output.name = alias.name.clone(); } @@ -248,15 +389,19 @@ fn source( Source::new(outputs, None), )) } - Expression::Values(values) => Some(( - values - .alias - .as_ref() - .map_or(String::new(), |alias| alias.name.clone()), - Source::new(values_outputs(values), None), - )), + Expression::Values(values) => { + proof.require(false); + Some(( + values + .alias + .as_ref() + .map_or(String::new(), |alias| alias.name.clone()), + Source::new(values_outputs(values), None), + )) + } Expression::Lateral(lateral) if matches!(unwrapped(&lateral.this), Expression::Function(function) if function.name.eq_ignore_ascii_case("FLATTEN")) => { + proof.require(false); let fact = expression_fact(&lateral.this, preceding, &Aliases::new()); let names: Vec = if lateral.column_aliases.is_empty() { ["SEQ", "KEY", "PATH", "INDEX", "VALUE", "THIS"] @@ -281,7 +426,10 @@ fn source( ), )) } - _ => None, + _ => { + proof.require(false); + None + } } } @@ -310,11 +458,27 @@ fn joined_sources( mut sources: Sources, join: &Join, relations: &Relations, - lateral_aliases: bool, + proof: &BindingProof, ) -> Sources { - let Some((name, mut right)) = source(&join.this, relations, &sources, lateral_aliases) else { + proof.require( + matches!( + join.kind, + JoinKind::Inner | JoinKind::Left | JoinKind::Right | JoinKind::Full + ) && !join.deferred_condition + && join.match_condition.is_none() + && join.pivots.is_empty() + && (join.on.is_some() || !join.using.is_empty()), + ); + proof.require(join.using.iter().all(|identifier| !identifier.quoted)); + let Some((name, mut right)) = source(&join.this, relations, &sources, proof) else { return sources; }; + proof.require( + !sources + .items + .iter() + .any(|(prior, _)| prior.eq_ignore_ascii_case(&name)), + ); let right_scope = Sources { items: vec![(name.clone(), right.clone())], ..Default::default() @@ -331,6 +495,10 @@ fn joined_sources( }; let left = column_fact(&column, &sources, &Aliases::new()); let right = column_fact(&column, &right_scope, &Aliases::new()); + proof.require( + left.as_ref().is_some_and(|fact| fact.resolved) + && right.as_ref().is_some_and(|fact| fact.resolved), + ); let output = match (left, right) { (Some(mut left), Some(right)) if join.kind == JoinKind::Full => { Rc::make_mut(&mut left.upstream).extend(right.upstream.iter().cloned()); @@ -362,6 +530,14 @@ fn joined_sources( ); sources.items.push((name, right)); sources.using_columns.extend(joined); + if proof.enabled + && let Some(on) = &join.on + { + proof.require( + scalar_binding_expression(on) + && expression_fact(on, &sources, &Aliases::new()).resolved, + ); + } sources } diff --git a/crates/sqlbuild-rules-native/src/query_analysis/_helpers/compatibility_types.rs b/crates/sqlbuild-rules-native/src/query_analysis/_helpers/compatibility_types.rs index b254e14938..640b985b8f 100644 --- a/crates/sqlbuild-rules-native/src/query_analysis/_helpers/compatibility_types.rs +++ b/crates/sqlbuild-rules-native/src/query_analysis/_helpers/compatibility_types.rs @@ -1,5 +1,6 @@ //! Borrowed-tree type evaluation with immutable lexical CTE outputs. +use std::cell::Cell; use std::collections::{HashMap, HashSet}; use std::rc::Rc; @@ -38,6 +39,26 @@ pub(super) fn infer_outputs( functions: &HashMap, dialect: DialectType, ) -> Outputs { + evaluate_outputs(expression, schema, functions, dialect).0 +} + +/// Return outputs only when evaluation never consults mutable AST type annotations. +pub(super) fn infer_unannotated_outputs( + expression: &Expression, + schema: Option<&ValidationSchema>, + functions: &HashMap, + dialect: DialectType, +) -> Option { + let (outputs, used_annotations) = evaluate_outputs(expression, schema, functions, dialect); + (!used_annotations).then_some(outputs) +} + +fn evaluate_outputs( + expression: &Expression, + schema: Option<&ValidationSchema>, + functions: &HashMap, + dialect: DialectType, +) -> (Outputs, bool) { let mut relations = Relations::new(); if let Some(schema) = schema { for table in &schema.tables { @@ -53,13 +74,19 @@ pub(super) fn infer_outputs( ); } } - let context = TypeContext { functions, dialect }; - context.query(expression, &relations) + let context = TypeContext { + functions, + dialect, + used_annotations: Cell::new(false), + }; + let outputs = context.query(expression, &relations); + (outputs, context.used_annotations.get()) } struct TypeContext<'a> { functions: &'a HashMap, dialect: DialectType, + used_annotations: Cell, } impl TypeContext<'_> { @@ -222,6 +249,7 @@ impl TypeContext<'_> { Some(NULL_TYPE.to_string()) } else { self.expression(expression, aliases).or_else(|| { + self.used_annotations.set(true); expression .inferred_type() .filter(|data_type| **data_type != DataType::Unknown) diff --git a/crates/sqlbuild-rules-native/src/query_analysis/_helpers/engine.rs b/crates/sqlbuild-rules-native/src/query_analysis/_helpers/engine.rs index b7332a3de4..aaf19f56f1 100644 --- a/crates/sqlbuild-rules-native/src/query_analysis/_helpers/engine.rs +++ b/crates/sqlbuild-rules-native/src/query_analysis/_helpers/engine.rs @@ -461,12 +461,26 @@ fn try_borrowed_query( }) { return Err(Box::new(work)); } + let bound_facts = work + .query + .schema + .as_ref() + .filter(|schema| { + plain_binding_schema(schema) + && work.query.binding_schema.as_ref().is_none_or(|binding| { + plain_binding_schema(binding) && same_binding_columns(schema, binding) + }) + }) + .and_then(|schema| super::borrowed_facts::infer_bound(&expression, Some(schema), dialect)); let validation = work .query .binding_schema .as_ref() .or(work.query.schema.as_ref()) .map(|schema| { + if bound_facts.is_some() { + return Ok(ValidationResult::with_errors(Vec::new())); + } let result = polyglot_sql::validation::validate_parsed_with_schema( vec![expression.clone()], dialect, @@ -493,7 +507,19 @@ fn try_borrowed_query( if !matches!(validation, Some(Ok(ref result)) if result.valid) { return Err(Box::new(work)); } - { + let unannotated_outputs: Option> = work + .projections + .iter() + .map(|(_, projection)| { + super::compatibility_types::infer_unannotated_outputs( + &expression, + work.query.schema.as_ref(), + &projection.function_return_types, + dialect, + ) + }) + .collect(); + if unannotated_outputs.is_none() { let schema = work.query.schema.as_ref().map(|schema| { polyglot_sql::validation::mapping_schema_from_validation_schema_with_dialect( schema, dialect, @@ -508,7 +534,9 @@ fn try_borrowed_query( Some(dialect), ); } - let facts = super::borrowed_facts::infer(&expression, work.query.schema.as_ref(), dialect); + let facts = bound_facts.unwrap_or_else(|| { + super::borrowed_facts::infer(&expression, work.query.schema.as_ref(), dialect) + }); if facts.is_empty() || facts .iter() @@ -523,13 +551,16 @@ fn try_borrowed_query( return Err(Box::new(work)); } } + let mut unannotated_outputs = unannotated_outputs.unwrap_or_default().into_iter(); for (index, projection) in &work.projections { - let mut outputs = super::compatibility_types::infer_outputs( - &expression, - work.query.schema.as_ref(), - &projection.function_return_types, - dialect, - ); + let mut outputs = unannotated_outputs.next().unwrap_or_else(|| { + super::compatibility_types::infer_outputs( + &expression, + work.query.schema.as_ref(), + &projection.function_return_types, + dialect, + ) + }); { let types: HashMap<_, _> = outputs.into_iter().collect(); outputs = facts @@ -603,6 +634,49 @@ fn try_borrowed_query( }) } +fn same_binding_columns(left: &ValidationSchema, right: &ValidationSchema) -> bool { + if left.tables.len() != right.tables.len() { + return false; + } + for table in &left.tables { + let Some(other) = right + .tables + .iter() + .find(|other| other.name.eq_ignore_ascii_case(&table.name)) + else { + return false; + }; + if table.columns.len() != other.columns.len() + || !table + .columns + .iter() + .zip(&other.columns) + .all(|(left, right)| left.name.eq_ignore_ascii_case(&right.name)) + { + return false; + } + } + true +} + +fn plain_binding_schema(schema: &ValidationSchema) -> bool { + for table in &schema.tables { + if table.schema.is_some() + || !table.aliases.is_empty() + || !table.foreign_keys.is_empty() + || !table.primary_key.is_empty() + || !table.unique_keys.is_empty() + || table + .columns + .iter() + .any(|column| column.references.is_some() || column.primary_key || column.unique) + { + return false; + } + } + true +} + fn has_case_sensitive_binding(expression: &polyglot_sql::Expression) -> bool { use polyglot_sql::Expression; match expression { diff --git a/crates/sqlbuild-rules-native/src/query_analysis/tests/helpers.rs b/crates/sqlbuild-rules-native/src/query_analysis/tests/helpers.rs index 4e849059da..cdfd37eea8 100644 --- a/crates/sqlbuild-rules-native/src/query_analysis/tests/helpers.rs +++ b/crates/sqlbuild-rules-native/src/query_analysis/tests/helpers.rs @@ -71,7 +71,27 @@ pub(crate) fn combined_queries_preserve_standalone_binding() -> bool { "WITH orders AS (SELECT * FROM orders), typed AS (SELECT CAST(CAST(o.order_id AS INT) AS NUMBER(38, 0)) AS order_id FROM orders o) SELECT order_id FROM typed", "WITH combined AS (SELECT order_id FROM orders UNION ALL BY NAME SELECT customer_id AS order_id FROM customers), typed AS (SELECT CAST(order_id AS BIGINT) AS order_id FROM combined) SELECT order_id FROM typed", "SELECT order_id FROM orders QUALIFY missing > 0", + "SELECT order_id FROM orders WHERE missing > 0", + "SELECT order_id FROM orders GROUP BY missing", + "SELECT order_id FROM orders HAVING missing > 0", + "SELECT order_id FROM orders ORDER BY missing", + "SELECT orders.order_id FROM orders JOIN customers ON orders.order_id = customers.missing", + "WITH unused AS (SELECT missing FROM orders) SELECT CAST(1 AS BIGINT) AS order_id", + "SELECT order_id FROM orders WHERE EXISTS (SELECT missing FROM customers)", + "SELECT order_id AS selected_id FROM orders WHERE selected_id > 0", + "SELECT selected_id AS result, order_id AS selected_id FROM orders", + "WITH selected AS (SELECT missing.* FROM orders) SELECT CAST(1 AS BIGINT) AS order_id FROM selected", + "WITH selected AS (SELECT * EXCLUDE (missing) FROM orders) SELECT order_id FROM selected", + "SELECT order_id, ROW_NUMBER() OVER (ORDER BY missing) AS position FROM orders", + "SELECT order_id FROM orders LIMIT 1", + "SELECT a.order_id FROM orders a JOIN orders a ON a.order_id = a.order_id", "SELECT orders.order_id FROM orders JOIN customers USING (order_id)", + "SELECT orders.order_id FROM orders CROSS JOIN customers", + "SELECT orders.order_id FROM orders JOIN customers", + "WITH combined AS (SELECT order_id FROM orders UNION ALL SELECT missing FROM customers) SELECT order_id FROM combined", + "WITH combined AS (SELECT order_id FROM orders UNION ALL SELECT customer_id, customer_id AS extra FROM customers) SELECT order_id FROM combined", + "WITH combined AS (SELECT order_id FROM orders UNION ALL BY NAME SELECT customer_id FROM customers) SELECT order_id FROM combined", + "SELECT order_id FROM orders UNION ALL SELECT customer_id FROM customers ORDER BY missing", "SELECT FROM", "SELECT order_id FROM orders; SELECT customer_id FROM customers", ] { @@ -113,6 +133,37 @@ pub(crate) fn combined_queries_preserve_standalone_binding() -> bool { true } +pub(crate) fn binding_warnings_preserve_declared_relationship_evidence() -> bool { + let schema = json!({"strict": true, "tables": [ + {"name": "orders", "columns": [ + {"name": "order_id", "type": "BIGINT"}, + {"name": "customer_id", "type": "BIGINT", "references": {"table": "customers", "column": "customer_id"}} + ]}, + {"name": "customers", "columns": [{"name": "customer_id", "type": "BIGINT"}]} + ]}); + let sql = "SELECT o.order_id FROM orders o JOIN customers c ON o.order_id = c.customer_id"; + let standalone: Value = serde_json::from_str(&crate::semantic_validation::main::validation_json( + &json!({"sql": sql, "dialect": "duckdb", "schema": schema, + "options": {"check_types": false, "check_references": true, "strict": true, "semantic": false, "strict_syntax": false}}).to_string() + ).expect("validation succeeds")).expect("valid JSON"); + assert!( + standalone["errors"] + .as_array() + .expect("diagnostics") + .iter() + .any(|item| item["code"] == "W221") + ); + let response: Value = serde_json::from_str(&analyze_project_compact_json(&json!({ + "queries": [{"sql": sql, "dialect": "duckdb", "schema": schema, "binding_schema": schema}], + "templates": [{"queryIndex": 0, "recoverCteFacts": true, "references": { + "orders": {"resourceType": "model", "resourceName": "orders"}, + "customers": {"resourceType": "model", "resourceName": "customers"} + }}], "projections": [{"templateIndex": 0}] + }).to_string()).expect("analysis succeeds")).expect("valid JSON"); + assert_eq!(response["validations"][0], standalone); + true +} + pub(crate) fn widening_aggregates_require_compatibility_recovery() -> bool { for expression in [ "AVG(order_id)", @@ -320,6 +371,45 @@ pub(crate) fn canonical_queries_reuse_semantics_and_project_resources() -> bool assert_eq!(strings[second_resource_index], "archived_orders"); true } +pub(crate) fn unannotated_types_require_complete_nested_evidence() -> bool { + let dialect = polyglot_sql::DialectType::DuckDB; + let functions = std::collections::HashMap::new(); + let expression = polyglot_sql::parse_one( + "WITH orders AS (SELECT CAST(1 AS INTEGER) AS order_id) SELECT order_id FROM orders", + dialect, + ) + .expect("valid SQL"); + let outputs = crate::query_analysis::compatibility_types::infer_unannotated_outputs( + &expression, + None, + &functions, + dialect, + ) + .expect("all nested outputs have independently known types"); + let mut annotated = expression.clone(); + polyglot_sql::optimizer::annotate_types::annotate_types(&mut annotated, None, Some(dialect)); + assert_eq!( + outputs, + crate::query_analysis::compatibility_types::infer_outputs( + &annotated, None, &functions, dialect, + ) + ); + let expression = polyglot_sql::parse_one( + "WITH orders AS (SELECT unknown_function() AS amount) SELECT CAST(amount AS INTEGER) AS amount FROM orders", + dialect, + ).expect("valid SQL"); + assert!( + crate::query_analysis::compatibility_types::infer_unannotated_outputs( + &expression, + None, + &functions, + dialect, + ) + .is_none() + ); + true +} + pub(crate) fn native_compatibility_types_preserve_result_semantics() -> bool { let expression = polyglot_sql::parse_one( "WITH orders AS (SELECT CAST(1 AS INTEGER) AS order_id), values AS (SELECT CASE WHEN order_id > 0 THEN CAST(order_id AS BIGINT) ELSE NULL END AS result, order_id > 0 AS positive, SUM(order_id) AS total FROM orders GROUP BY order_id) SELECT result, positive, total FROM values", diff --git a/crates/sqlbuild-rules-native/src/query_analysis/tests/test_query_analysis.rs b/crates/sqlbuild-rules-native/src/query_analysis/tests/test_query_analysis.rs index cd78e51a87..d12bf69441 100644 --- a/crates/sqlbuild-rules-native/src/query_analysis/tests/test_query_analysis.rs +++ b/crates/sqlbuild-rules-native/src/query_analysis/tests/test_query_analysis.rs @@ -1,6 +1,7 @@ use crate::query_analysis::main::analyze::analyze_json; use crate::query_analysis::main::analyze_project::analyze_project_json; use crate::query_analysis::tests::helpers::{ + binding_warnings_preserve_declared_relationship_evidence, borrowed_facts_preserve_named_outputs_and_terminal_sources, borrowed_facts_preserve_union_dependencies_and_join_nullability, canonical_queries_reuse_semantics_and_project_resources, @@ -9,6 +10,7 @@ use crate::query_analysis::tests::helpers::{ interleaved_query_templates_preserve_template_order, native_compatibility_types_preserve_result_semantics, repeated_project_facts_intern_complete_facts, + unannotated_types_require_complete_nested_evidence, widening_aggregates_require_compatibility_recovery, }; use crate::query_analysis::tests::test_types::{ @@ -19,6 +21,16 @@ use serde_json::{Value, json}; #[test] fn given_compact_query_cases_when_analyzing_projects_then_expected_behavior_holds() { let test_cases = [ + CompactQueryAnalysisTestCase { + description: "binding warnings preserve declared relationship evidence", + run: binding_warnings_preserve_declared_relationship_evidence, + expected_success: true, + }, + CompactQueryAnalysisTestCase { + description: "unannotated types require complete nested evidence", + run: unannotated_types_require_complete_nested_evidence, + expected_success: true, + }, CompactQueryAnalysisTestCase { description: "borrowed facts preserve named outputs and actual terminal columns", run: borrowed_facts_preserve_named_outputs_and_terminal_sources, diff --git a/docs/dense-compile-performance.md b/docs/dense-compile-performance.md index 9570a7c6d6..a8938ec2e4 100644 --- a/docs/dense-compile-performance.md +++ b/docs/dense-compile-performance.md @@ -26,7 +26,7 @@ The guard checks: | 1,000 | 14s | 1.5GiB | | 3,000 | 34s | 2.75GiB | | 5,000 | 55s | 3.25GiB | -| 10,000 | 115s | 4GiB | +| 10,000 | 145s | 4GiB | These are regression ceilings for the implemented compiler, with headroom for host variation. They are not a claim of sub-10-second compilation at every size. @@ -47,6 +47,32 @@ analysis results in batches of 64. Smaller requests retain a single batch. Both use up to four workers; local resolution views avoid copying CTE definitions that are already available through the source catalogue. These limits change scheduling, not analysis or fallback semantics. +Projects with at least 10,000 model projections and 48MiB of prepared SQL use one +native analysis worker to leave memory headroom for the isolated custom-rule host +after analysis. Smaller SQL payloads retain parallel analysis. +The 10k cold ceiling includes headroom for this memory-bounded scheduling; warm +cache hits and small edit batches keep their existing scheduling. + +The compiler reuses borrowed syntax-tree facts when it can prove complete reference +binding and infer types without consulting mutable type annotations. Unsupported +clauses, incomplete schemas and unresolved expressions retain the normal validation +and analysis paths. + +For cold projects with 128–5,000 models and selected rules, artifact rendering can overlap rule +evaluation after graph and contract analysis. Rendering uses one background thread +and a disposable directory outside the project. The normal write phase publishes +the staged files only after its existing diagnostic gates pass. Unchanged files +retain their timestamps, stale files are removed using the normal policy, and +cached compilation keeps its normal artifact-cache path. Temporary-storage failure +falls back to ordinary artifact writing. +Larger projects retain sequential rendering to bound combined compiler and rule-host +memory; projects without selected rules avoid staging's extra file I/O. + +Detailed phase timings can overlap; physical-write time includes any staging work. +Use whole-process wall time for performance acceptance rather than summing phases. +Custom rules continue to run in their isolated host. Disabled rules caches avoid +unused fact fingerprints; enabled caches retain their normal identities and +invalidation behavior. This fixture protects dense query analysis, declaration expansion, rules and artifact generation. It is not an exact reproduction of every application: diff --git a/src/sqlbuild/cli/commands/_helpers/compile/pipeline.py b/src/sqlbuild/cli/commands/_helpers/compile/pipeline.py index 8f82c27a20..1fe21cd808 100644 --- a/src/sqlbuild/cli/commands/_helpers/compile/pipeline.py +++ b/src/sqlbuild/cli/commands/_helpers/compile/pipeline.py @@ -19,6 +19,7 @@ ) from sqlbuild.cli.commands._helpers.compile.target_writer import write_static_compile_target from sqlbuild.cli.commands._helpers.runtime.adapters import resolve_adapter +from sqlbuild.cli.commands.classes.prepared_compile_artifacts import PreparedCompileArtifacts from sqlbuild.cli.commands.types import CompileLineageMode from sqlbuild.cli.compile.models import ( CompileAnalysis, @@ -48,7 +49,7 @@ from sqlbuild.rule_engine.classes.early_sql_lint import EarlySqlLint from sqlbuild.rule_engine.main.load_config import load_rules_config from sqlbuild.rule_engine.main.run_rules import run_rules -from sqlbuild.rule_engine.models import RulesRunResult +from sqlbuild.rule_engine.models import RulesConfig, RulesRunResult from sqlbuild.runtime.observability.classes.operation_lifecycle import OperationLifecycle from sqlbuild.spec.contracts.main.resolve_effective_adapter_name import ( resolve_effective_adapter_name, @@ -59,18 +60,19 @@ def analyze_compile_project( *, project_dir: Path, no_sql_validation: bool, - no_cache: bool, selected_target: str | None, lineage_mode: CompileLineageMode, cli_vars: dict[str, object] | None, profile_flags: CompileProfileFlags, - select: tuple[str, ...], - exclude: tuple[str, ...], + analysis_selection: CompileAnalysisSelection, status: TransientStatusReporter | None, + prepared_artifacts: PreparedCompileArtifacts | None = None, ) -> CompileAnalysis: """Discover, compile, and validate the project into one analysis result.""" - with EarlySqlLint(enabled=not (select or exclude)) as early_lint: + with EarlySqlLint( + enabled=not (analysis_selection.select or analysis_selection.exclude) + ) as early_lint: return _analyze_compile_project( project_dir=project_dir, no_sql_validation=no_sql_validation, @@ -78,11 +80,10 @@ def analyze_compile_project( lineage_mode=lineage_mode, cli_vars=cli_vars, profile_flags=profile_flags, - analysis_selection=CompileAnalysisSelection( - select=select, exclude=exclude, no_cache=no_cache - ), + analysis_selection=analysis_selection, status=status, early_lint=early_lint, + prepared_artifacts=prepared_artifacts, ) @@ -97,6 +98,7 @@ def _analyze_compile_project( analysis_selection: CompileAnalysisSelection, status: TransientStatusReporter | None, early_lint: EarlySqlLint, + prepared_artifacts: PreparedCompileArtifacts | None = None, ) -> CompileAnalysis: select: tuple[str, ...] = analysis_selection.select exclude: tuple[str, ...] = analysis_selection.exclude @@ -186,11 +188,18 @@ def _analyze_compile_project( *contract_result.diagnostics, ) if not any(diagnostic.is_error for diagnostic in graph.project.diagnostics): + rules_config: RulesConfig = load_rules_config(project_dir=project_dir) + if ( + prepared_artifacts is not None + and rules_config.select + and not any(item.is_error for item in core_diagnostics) + ): + prepared_artifacts.start(project=graph.project, adapter=adapter) _ = start_compile_phase(status=status, message="Evaluating built-in and custom rules...") rules_result = run_rules( graph=graph, discovered_inputs=discovered_inputs, - config=load_rules_config(project_dir=project_dir), + config=rules_config, project_dir=project_dir, dialect=adapter.sql_analysis_dialect() or "generic", selected_keys=selected_keys if select or exclude else None, @@ -310,6 +319,7 @@ def write_compile_artifacts( analysis: CompileAnalysis, manifest_payload: dict[str, object] | None, status: TransientStatusReporter | None, + prepared_artifacts: PreparedCompileArtifacts | None = None, ) -> CompileWriteResult: """Write compiled artifacts to target/ and report the written counts.""" @@ -327,11 +337,20 @@ def write_compile_artifacts( ) else: _ = start_compile_phase(status=status, message="Writing compiled artifacts...") - written = write_static_compile_target( - target_dir=target_dir, - adapter=analysis.adapter, - project=analysis.graph.project, - manifest=manifest_payload, + staged: WrittenTarget | None = ( + prepared_artifacts.publish(target_dir=target_dir, manifest=manifest_payload) + if prepared_artifacts is not None + else None + ) + written = ( + staged + if staged is not None + else write_static_compile_target( + target_dir=target_dir, + adapter=analysis.adapter, + project=analysis.graph.project, + manifest=manifest_payload, + ) ) write_ms: int = elapsed_ms(write_start) if not profile_skip_write: diff --git a/src/sqlbuild/cli/commands/_helpers/compile/target_writer.py b/src/sqlbuild/cli/commands/_helpers/compile/target_writer.py index 4f52b6f1dc..b1dddd963c 100644 --- a/src/sqlbuild/cli/commands/_helpers/compile/target_writer.py +++ b/src/sqlbuild/cli/commands/_helpers/compile/target_writer.py @@ -155,6 +155,39 @@ def write_static_compile_target( ) +def publish_static_compile_target( + *, prepared: WrittenTarget, target_dir: Path, manifest: dict[str, object] | None +) -> WrittenTarget: + """Publish staged files with the same unchanged-file and stale-file semantics.""" + + compiled_dir: Path = target_dir / _COMPILED_DIR + staged_dir: Path = prepared.target_dir / _COMPILED_DIR + check_existing: bool = compiled_dir.is_dir() + managed_paths: set[Path] = set() + target_dir.mkdir(parents=True, exist_ok=True) + for source in sorted(staged_dir.rglob("*")): + if not source.is_file(): + continue + path: Path = compiled_dir / source.relative_to(staged_dir) + _write_bytes_if_changed( + path=path, contents=source.read_bytes(), check_existing=check_existing + ) + managed_paths.add(path) + if check_existing: + with record_compile_timing("stale_traversal_ms"): + _remove_stale_compiled_files(target_dir=target_dir, managed_paths=managed_paths) + if manifest is not None: + _write_manifest(target_dir=target_dir, manifest=manifest) + return WrittenTarget( + model_count=prepared.model_count, + seed_count=prepared.seed_count, + function_count=prepared.function_count, + audit_count=prepared.audit_count, + test_count=prepared.test_count, + target_dir=target_dir, + ) + + def _write_models(*, target_dir: Path, plan_output: PlanOutput, check_existing: bool) -> set[Path]: """Write model resolved SQL.""" diff --git a/src/sqlbuild/cli/commands/classes/prepared_compile_artifacts.py b/src/sqlbuild/cli/commands/classes/prepared_compile_artifacts.py new file mode 100644 index 0000000000..9e53343942 --- /dev/null +++ b/src/sqlbuild/cli/commands/classes/prepared_compile_artifacts.py @@ -0,0 +1,85 @@ +"""Own staged cold compile artifacts and publish them only at the normal write phase.""" + +from __future__ import annotations + +from concurrent.futures import Future, ThreadPoolExecutor +from contextvars import Context, copy_context +from pathlib import Path +from tempfile import TemporaryDirectory +from types import TracebackType + +from sqlbuild.adapter.contract.classes.base_adapter import BaseAdapter +from sqlbuild.cli.commands._helpers.compile.target_writer import ( + publish_static_compile_target, + write_static_compile_target, +) +from sqlbuild.cli.output.models import WrittenTarget +from sqlbuild.compiler.compile.models import CompiledProject + +_MIN_PREPARED_ARTIFACT_MODELS: int = 128 +_MAX_PREPARED_ARTIFACT_MODELS: int = 5000 + + +class PreparedCompileArtifacts: + """Own one bounded preparation task and its disposable output directory.""" + + def __init__(self, *, enabled: bool) -> None: + self.enabled: bool = enabled + self._directory: TemporaryDirectory[str] | None = None + self._executor: ThreadPoolExecutor | None = None + self._future: Future[WrittenTarget] | None = None + + def __enter__(self) -> PreparedCompileArtifacts: + return self + + def __exit__( + self, + exception_type: type[BaseException] | None, + exception: BaseException | None, + traceback: TracebackType | None, + ) -> None: + try: + if self._executor is not None: + self._executor.shutdown(wait=True, cancel_futures=True) + finally: + if self._directory is not None: + self._directory.cleanup() + + def start(self, *, project: CompiledProject, adapter: BaseAdapter) -> None: + if ( + not self.enabled + or project.compile_cache_dir is not None + or len(project.models) < _MIN_PREPARED_ARTIFACT_MODELS + or len(project.models) > _MAX_PREPARED_ARTIFACT_MODELS + ): + return + try: + self._directory = TemporaryDirectory(prefix="sqlbuild-compile-") + except OSError: + return + self._executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="sqlbuild-artifacts") + context: Context = copy_context() + target_dir: Path = Path(self._directory.name) + + def prepare() -> WrittenTarget: + return context.run( + write_static_compile_target, + target_dir=target_dir, + adapter=adapter, + project=project, + ) + + self._future = self._executor.submit(prepare) + + def publish( + self, *, target_dir: Path, manifest: dict[str, object] | None + ) -> WrittenTarget | None: + if self._future is None: + return None + try: + prepared: WrittenTarget = self._future.result() + except OSError: + return None + return publish_static_compile_target( + prepared=prepared, target_dir=target_dir, manifest=manifest + ) diff --git a/src/sqlbuild/cli/commands/main/project/_compile.py b/src/sqlbuild/cli/commands/main/project/_compile.py index 3a0d041409..a69aeea2fc 100644 --- a/src/sqlbuild/cli/commands/main/project/_compile.py +++ b/src/sqlbuild/cli/commands/main/project/_compile.py @@ -17,12 +17,14 @@ write_compile_dag_artifact, ) from sqlbuild.cli.commands._helpers.compile.status import elapsed_ms, start_compile_status +from sqlbuild.cli.commands.classes.prepared_compile_artifacts import PreparedCompileArtifacts from sqlbuild.cli.commands.types import CompileLineageMode from sqlbuild.cli.compile.models import ( CompileAnalysis, CompileCommandRequest, CompileWriteResult, ) +from sqlbuild.compiler.compile.models import CompileAnalysisSelection from sqlbuild.compiler.compile.types import DiagnosticPhase from sqlbuild.compiler.profiling.main.collect import collect_compile_timings from sqlbuild.compiler.profiling.models import CompileTimingCollector @@ -42,12 +44,18 @@ def run_compile(request: CompileCommandRequest) -> int: no_color=request.no_color, ) try: - with collect_compile_timings() as detailed_timings: + with ( + collect_compile_timings() as detailed_timings, + PreparedCompileArtifacts( + enabled=not request.profile_flags.skip_write + ) as prepared_artifacts, + ): return _run_compile_with_status( request=effective_request, total_start=total_start, status=status, detailed_timings=detailed_timings, + prepared_artifacts=prepared_artifacts, ) finally: if status is not None: @@ -60,6 +68,7 @@ def _run_compile_with_status( total_start: float, status: TransientStatusReporter | None, detailed_timings: CompileTimingCollector, + prepared_artifacts: PreparedCompileArtifacts, ) -> int: """Execute compile after the optional interactive status reporter is initialized.""" @@ -71,14 +80,15 @@ def _run_compile_with_status( analysis: CompileAnalysis = analyze_compile_project( project_dir=project_dir, no_sql_validation=request.no_sql_validation, - no_cache=request.no_cache, selected_target=request.selected_target, lineage_mode=lineage_mode, cli_vars=request.cli_vars, profile_flags=request.profile_flags, - select=request.select, - exclude=request.exclude, + analysis_selection=CompileAnalysisSelection( + select=request.select, exclude=request.exclude, no_cache=request.no_cache + ), status=status, + prepared_artifacts=prepared_artifacts, ) rules_failed: bool = any( diagnostic.is_error and diagnostic.phase is DiagnosticPhase.RULE @@ -101,6 +111,7 @@ def _run_compile_with_status( analysis=analysis, manifest_payload=manifest_payload, status=status, + prepared_artifacts=prepared_artifacts, ) timings_ms: dict[str, int] = { "discover_ms": analysis.discover_ms, diff --git a/src/sqlbuild/compiler/compile/_helpers/analysis/columns.py b/src/sqlbuild/compiler/compile/_helpers/analysis/columns.py index a65089ce9b..36d1ccf721 100644 --- a/src/sqlbuild/compiler/compile/_helpers/analysis/columns.py +++ b/src/sqlbuild/compiler/compile/_helpers/analysis/columns.py @@ -405,9 +405,10 @@ def _analyze_columns_and_lineage_from_polyglot_ast( expression=inner, context=non_null_filter_context, ) - else cte_passthrough_nullability.get( - output_column, - ( + else ( + cte_passthrough_nullability[output_column] + if output_column in cte_passthrough_nullability + else ( _infer_polyglot_nullability( expression=inner, alias_nullability=alias_nullability, @@ -419,7 +420,7 @@ def _analyze_columns_and_lineage_from_polyglot_ast( expression=inner, inference_profile=inference_profile, ) - ), + ) ) ) columns.append(InferredColumn(name=output_column, type=col_type, nullability=nullability)) diff --git a/src/sqlbuild/compiler/compile/_helpers/analysis/compact.py b/src/sqlbuild/compiler/compile/_helpers/analysis/compact.py index 1cb0db36fc..c30951ec00 100644 --- a/src/sqlbuild/compiler/compile/_helpers/analysis/compact.py +++ b/src/sqlbuild/compiler/compile/_helpers/analysis/compact.py @@ -170,6 +170,8 @@ from sqlbuild.diagnostics.main.log_debug_event import log_debug_event _DEBUG_LOGGER: logging.Logger = logging.getLogger("sqlbuild.compile") +_LARGE_COMPACT_PROJECT_MODELS: int = 10_000 +_LARGE_COMPACT_SQL_BYTES: int = 48 * 1024 * 1024 _NATIVE_LEGACY_FALLBACK: str = "native project type recovery requires legacy fallback" _REF_PATTERN: re.Pattern[str] = quoted_reference_call_pattern(SqlReferenceKind.REF) _SEED_PATTERN: re.Pattern[str] = quoted_reference_call_pattern(SqlReferenceKind.SEED) @@ -623,6 +625,14 @@ def _prepare_compact_analysis_batch( def _run_compact_analysis_batch(*, preparation: CompactBatchPreparation) -> object: + """Bound resident analysis heaps before the isolated custom-rule host starts.""" + + large_project: bool = ( + len(preparation.projections) >= _LARGE_COMPACT_PROJECT_MODELS + and sum(len(sql.encode("utf-8")) for sql in preparation.cleaned_sql) + >= _LARGE_COMPACT_SQL_BYTES + ) + workers: int = 1 if large_project else 4 return orjson.loads( cast(NativeQueryAnalysisModule, _native).analyze_project_queries_compact_json( orjson.dumps( @@ -630,7 +640,7 @@ def _run_compact_analysis_batch(*, preparation: CompactBatchPreparation) -> obje "queries": preparation.queries, "templates": preparation.templates, "projections": preparation.projections, - "workers": 4, + "workers": workers, }, option=orjson.OPT_SORT_KEYS, ).decode() diff --git a/src/sqlbuild/rule_engine/_helpers/engine/native.py b/src/sqlbuild/rule_engine/_helpers/engine/native.py index fdbf9e72ca..5fd3a7e7b6 100644 --- a/src/sqlbuild/rule_engine/_helpers/engine/native.py +++ b/src/sqlbuild/rule_engine/_helpers/engine/native.py @@ -101,7 +101,10 @@ def evaluate_native( "initial_findings": [_finding_payload(finding) for finding in initial_findings], "defer_suppressions": defer_suppressions, "custom_rules": _custom_rule_payloads( - catalogue=catalogue, project=project, project_dir=project_dir + catalogue=catalogue, + project=project, + project_dir=project_dir, + cache_enabled=config.cache.enabled, ), } custom_host_input: Path | None = None @@ -529,7 +532,11 @@ def _typed_value_payload(value: SqlValue) -> object: def _custom_rule_payloads( - *, catalogue: tuple[Rule, ...], project: CompiledProject, project_dir: Path + *, + catalogue: tuple[Rule, ...], + project: CompiledProject, + project_dir: Path, + cache_enabled: bool, ) -> list[dict[str, object]]: fingerprints: dict[frozenset[str], str] = {} closures: dict[str, tuple[Path, ...]] = {} @@ -543,11 +550,11 @@ def _custom_rule_payloads( attributes: frozenset[str] = custom_rule_project_fact_attributes( rule=rule, project_dir=project_dir, import_closure=closure ) - if attributes not in fingerprints: + if cache_enabled and attributes not in fingerprints: fingerprints[attributes] = _custom_fact_fingerprint( project=project, attributes=attributes ) - fact_fingerprint: str = fingerprints[attributes] + fact_fingerprint: str = fingerprints[attributes] if cache_enabled else "" payloads.append( _custom_rule_payload( rule=rule, diff --git a/src/sqlbuild/runtime/output_capture/_helpers/text.py b/src/sqlbuild/runtime/output_capture/_helpers/text.py index ed30e9fe3c..6bd41a1887 100644 --- a/src/sqlbuild/runtime/output_capture/_helpers/text.py +++ b/src/sqlbuild/runtime/output_capture/_helpers/text.py @@ -18,6 +18,13 @@ def strip_ansi(text: str) -> str: def chunk_text(*, text: str, max_bytes: int) -> tuple[str, ...]: """Split text deterministically without splitting Unicode code points.""" + if text.isascii() and max_bytes > 0: + if len(text) <= max_bytes: + return (text,) + return tuple(text[index : index + max_bytes] for index in range(0, len(text), max_bytes)) + if len(text) <= max_bytes and len(text.encode("utf-8", "surrogateescape")) <= max_bytes: + return (text,) + chunks: list[str] = [] current: list[str] = [] current_bytes: int = 0 diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_dense_compile_performance.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_dense_compile_performance.py index 17cf08cd1f..5fc15dbfbc 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_dense_compile_performance.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_dense_compile_performance.py @@ -59,7 +59,7 @@ DenseCompileGuardTestCase( "dense_models_10000_all_rules", 10000, - 115.0, + 145.0, 4 * _GIB, "8125ddde6d3ae172bb15878f95048d66398453c516bbd837b1fd83c04c5982b5", ), diff --git a/tests/integration/src/sqlbuild/cli/commands/main/_test_types.py b/tests/integration/src/sqlbuild/cli/commands/main/_test_types.py index 9f27e2bdfa..4ff7fd72b7 100644 --- a/tests/integration/src/sqlbuild/cli/commands/main/_test_types.py +++ b/tests/integration/src/sqlbuild/cli/commands/main/_test_types.py @@ -1,10 +1,20 @@ from __future__ import annotations +from collections.abc import Callable from dataclasses import dataclass, field +from tempfile import TemporaryDirectory from sqlbuild.compiler.planner.models import CursorOverrides +@dataclass(frozen=True) +class PreparedArtifactsCompileTestCase: + description: str + model_count: int + temporary_directory_factory: Callable[..., TemporaryDirectory[str]] + expected_exit_code: int = 0 + + @dataclass(frozen=True) class BoundProjectionCompileTestCase: description: str diff --git a/tests/integration/src/sqlbuild/cli/commands/main/helpers.py b/tests/integration/src/sqlbuild/cli/commands/main/helpers.py index acec82cc50..3edfb27989 100644 --- a/tests/integration/src/sqlbuild/cli/commands/main/helpers.py +++ b/tests/integration/src/sqlbuild/cli/commands/main/helpers.py @@ -1,10 +1,15 @@ from __future__ import annotations from pathlib import Path +from tempfile import TemporaryDirectory import duckdb +def unavailable_artifact_directory(*, prefix: str) -> TemporaryDirectory[str]: + raise OSError("temporary storage unavailable") + + def write_compile_startup_project(project_dir: Path) -> None: """Write the minimal project used by fresh-process import tests.""" diff --git a/tests/integration/src/sqlbuild/cli/commands/main/test_compile_prepared_artifacts.py b/tests/integration/src/sqlbuild/cli/commands/main/test_compile_prepared_artifacts.py new file mode 100644 index 0000000000..f3c0e4990d --- /dev/null +++ b/tests/integration/src/sqlbuild/cli/commands/main/test_compile_prepared_artifacts.py @@ -0,0 +1,101 @@ +"""Real cold compilation preserves publication gates and unchanged artifact files.""" + +import json +from pathlib import Path +from tempfile import TemporaryDirectory + +import duckdb +import pytest + +from sqlbuild.cli.commands.classes import prepared_compile_artifacts +from sqlbuild.cli.commands.main.entrypoint.entry import main +from tests.integration.src.sqlbuild.cli.commands.main._test_types import ( + PreparedArtifactsCompileTestCase, +) +from tests.integration.src.sqlbuild.cli.commands.main.helpers import unavailable_artifact_directory + + +@pytest.mark.parametrize( + "test_case", + ( + PreparedArtifactsCompileTestCase("cold staged artifacts", 128, TemporaryDirectory), + PreparedArtifactsCompileTestCase( + "unavailable temporary storage", 128, unavailable_artifact_directory + ), + ), + ids=lambda case: case.description, +) +def test_given_existing_artifacts_when_rules_fail_then_staged_changes_are_not_published( + test_case: PreparedArtifactsCompileTestCase, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr( + prepared_compile_artifacts, "TemporaryDirectory", test_case.temporary_directory_factory + ) + config_path: Path = tmp_path / "sqlbuild_project.toml" + config: str = 'name = "orders"\nadapter = "duckdb"\n[rules]\nselect = ["SQBRSQL021"]\n' + config_path.write_text(config, encoding="utf-8") + models: Path = tmp_path / "models" + models.mkdir() + header: str = ( + "MODEL (materialized table, contract enforced, columns (order_id (type INTEGER)));\n" + ) + for index in range(test_case.model_count): + (models / f"orders_{index:03}.sql").write_text( + header + "SELECT CAST(1 AS INTEGER) AS order_id\n", encoding="utf-8" + ) + arguments: list[str] = [ + "--project-dir", + str(tmp_path), + "compile", + "--no-cache", + "--json", + "--manifest", + ] + assert main(arguments) == test_case.expected_exit_code + first: dict[str, object] = json.loads(capsys.readouterr().out) + assert first["has_errors"] is False + compiled: Path = tmp_path / "target" / "compiled" + before: dict[Path, bytes] = {path: path.read_bytes() for path in compiled.rglob("*.sql")} + modified_times: dict[Path, int] = {path: path.stat().st_mtime_ns for path in before} + assert len(before) == test_case.model_count + stale: Path = compiled / "models" / "obsolete.sql" + stale.write_text("SELECT 0\n", encoding="utf-8") + (models / "orders_000.sql").write_text( + header + "SELECT CAST(2 AS INTEGER) AS order_id\n", encoding="utf-8" + ) + rules: Path = tmp_path / "rules" + rules.mkdir() + (rules / "orders.py").write_text( + "from sqlbuild.rules import Finding, Model, RuleContext, rule\n" + '@rule(code="XSQBRARCH001", message="Model rejected", remediation="Update the model.")\n' + "def check_orders(*, model: Model, ctx: RuleContext) -> list[Finding]:\n" + ' return [ctx.finding(subject=model)] if model.name == "orders_000" else []\n', + encoding="utf-8", + ) + config_path.write_text( + config.replace('select = ["SQBRSQL021"]', 'select = ["XSQBRARCH001"]'), encoding="utf-8" + ) + assert main(arguments) == 1 + failed: dict[str, object] = json.loads(capsys.readouterr().out) + assert failed["has_errors"] is True + assert {path: path.read_bytes() for path in before} == before + assert stale.exists() + + config_path.write_text(config, encoding="utf-8") + assert main(arguments) == 0 + final: dict[str, object] = json.loads(capsys.readouterr().out) + assert final["has_errors"] is False + assert not stale.exists() + changed: Path = compiled / "models" / "orders_000.sql" + for path in modified_times.keys() - {changed}: + assert path.stat().st_mtime_ns == modified_times[path] + assert path.read_bytes() == before[path] + with duckdb.connect() as connection: + assert connection.execute(changed.read_text(encoding="utf-8")).fetchall() == [(2,)] + + +if __name__ == "__main__": + raise SystemExit(pytest.main([__file__, "-n", "auto", "--dist", "loadfile"])) diff --git a/tests/integration/src/sqlbuild/cli/commands/main/test_rules.py b/tests/integration/src/sqlbuild/cli/commands/main/test_rules.py index 670c9fefdd..7d9c69d8e4 100644 --- a/tests/integration/src/sqlbuild/cli/commands/main/test_rules.py +++ b/tests/integration/src/sqlbuild/cli/commands/main/test_rules.py @@ -598,6 +598,20 @@ def final_directory(*, model: Model, ctx: RuleContext) -> list[Finding]: assert first_exit == second_exit == test_case.expected_exit_code assert first["diagnostics"][0]["code"] == test_case.expected_code assert second["compile_timings"]["rule_cache_hits"] >= 1 + config_path: Path = tmp_path / "sqlbuild_project.toml" + original_config: str = config_path.read_text(encoding="utf-8") + config_path.write_text(original_config + "\n[rules.cache]\nenabled = false\n", encoding="utf-8") + cold_exit: int = main(["--project-dir", str(tmp_path), "compile", "--json", "--no-cache"]) + cold: dict[str, object] = json.loads(capsys.readouterr().out) + config_path.write_text(original_config, encoding="utf-8") + warm_exit: int = main(["--project-dir", str(tmp_path), "compile", "--json"]) + warm: dict[str, object] = json.loads(capsys.readouterr().out) + + assert cold_exit == warm_exit == test_case.expected_exit_code + assert cold["diagnostics"] == warm["diagnostics"] == first["diagnostics"] + assert cold["compile_timings"]["rule_cache_hits"] == 0 + assert warm["compile_timings"]["rule_cache_hits"] >= 1 + assert not (tmp_path / "target" / "compiled").exists() host_inputs: Path = tmp_path / "target" / "rules-cache" / "host-inputs" assert not tuple(host_inputs.glob("*.pickle")) diff --git a/tests/unit/src/sqlbuild/runtime/output_capture/classes/test_output_capture_dispatcher.py b/tests/unit/src/sqlbuild/runtime/output_capture/classes/test_output_capture_dispatcher.py index b02ea43b66..e4a907bdb5 100644 --- a/tests/unit/src/sqlbuild/runtime/output_capture/classes/test_output_capture_dispatcher.py +++ b/tests/unit/src/sqlbuild/runtime/output_capture/classes/test_output_capture_dispatcher.py @@ -82,6 +82,9 @@ def test_given_interleaved_partial_writes_when_closed_then_complete_lines_have_g ( ChunkingTestCase("ascii_sql", "SELECT 123456789\n", 8, ("SELECT 1", "23456789", "\n")), ChunkingTestCase("unicode_sql", "ééé\n", 4, ("éé", "é\n")), + ChunkingTestCase("ascii_exact_boundary", "SELECT 1\n", 9, ("SELECT 1\n",)), + ChunkingTestCase("unicode_single_chunk", "é\n", 4, ("é\n",)), + ChunkingTestCase("multibyte_boundary", "🙂\n", 4, ("🙂", "\n")), ), ids=lambda case: case.description, )