diff --git a/Makefile b/Makefile index 172f1480c1..c5ae1bb499 100644 --- a/Makefile +++ b/Makefile @@ -61,6 +61,7 @@ E2E_DUCKDB_BUILD_CORE_PATHS := \ tests/e2e/src/sqlbuild/cli/commands/main/build/test_audit_failures.py \ tests/e2e/src/sqlbuild/cli/commands/main/build/test_build.py \ tests/e2e/src/sqlbuild/cli/commands/main/build/test_compile_json_behavior.py \ + tests/e2e/src/sqlbuild/cli/commands/main/build/test_dynamic_column_contract.py \ tests/e2e/src/sqlbuild/cli/commands/main/build/test_dag_json_behavior.py \ tests/e2e/src/sqlbuild/cli/commands/main/build/test_enum_contract.py \ tests/e2e/src/sqlbuild/cli/commands/main/build/test_expression_sources.py \ diff --git a/crates/sqlbuild-rules-native/src/constants.rs b/crates/sqlbuild-rules-native/src/constants.rs index 83e162d658..2b7e4de57d 100644 --- a/crates/sqlbuild-rules-native/src/constants.rs +++ b/crates/sqlbuild-rules-native/src/constants.rs @@ -13,6 +13,7 @@ pub(crate) const REFERENCE_KIND: &str = "ref"; pub(crate) const SOURCE_REFERENCE_KIND: &str = "source"; pub(crate) const VIEW_MATERIALIZATION: &str = "view"; pub(crate) const ENFORCED_CONTRACT: &str = "enforced"; +pub(crate) const SELECT_STAR_MODEL_RULE_CODE: &str = "SQBRMODEL102"; pub(crate) const BOOLEAN_TYPE: &str = "BOOLEAN"; pub(crate) const TIMESTAMP_TYPE: &str = "TIMESTAMP"; pub(crate) const DATE_TYPE: &str = "DATE"; diff --git a/crates/sqlbuild-rules-native/src/engine/tests/test_evaluation.rs b/crates/sqlbuild-rules-native/src/engine/tests/test_evaluation.rs index ed97813298..6bddfc24fd 100644 --- a/crates/sqlbuild-rules-native/src/engine/tests/test_evaluation.rs +++ b/crates/sqlbuild-rules-native/src/engine/tests/test_evaluation.rs @@ -544,6 +544,52 @@ fn given_enforced_contract_outputs_when_evaluating_explicit_type_rule_then_retur Ok(()) } +#[test] +fn given_compiler_proven_duckdb_dynamic_pivot_when_evaluating_contract_rules_then_skips_query_parse() +-> Result<(), String> { + let test_cases = [test_types::DynamicPivotEvaluationTestCase { + description: "compiler-proven bare DuckDB pivot", + expected_faults: json!([]), + }]; + for test_case in test_cases { + let project_dir = TempDir::new().map_err(|error| error.to_string())?; + let config = json!({ + "select": ["SQBRMODEL102", "SQBRCONTRACT101", "SQBRCONTRACT105", "SQBRCONTRACT106"], + "cache": {"enabled": false} + }); + let mut request: Value = serde_json::from_str(&helpers::request(&project_dir, &config)) + .map_err(|error| error.to_string())?; + request["dialect"] = json!("duckdb"); + request["models"][0]["query_sql"] = + json!("PIVOT orders ON category USING MAX(amount) GROUP BY customer_id"); + request["models"][0]["authored_sql"] = request["models"][0]["query_sql"].clone(); + request["models"][0]["config"] = json!({"contract": "enforced"}); + request["models"][0]["columns"] = json!([ + {"name": "customer_id", "type": "INTEGER", "type_proven": true} + ]); + request["models"][0]["dynamic_columns"] = json!([{ + "name": "category_amounts", + "pivot_column": "category", + "value_column": "amount", + "aggregate": "MAX", + "type": "DECIMAL(12,2)", + "type_proven": true + }]); + request["models"][0]["dynamic_columns_proven"] = json!(true); + request["models"][0]["bare_dynamic_pivot"] = json!(true); + + let result: Value = serde_json::from_str(&evaluate_json(&request.to_string())?) + .map_err(|error| error.to_string())?; + + assert_eq!( + result["faults"], test_case.expected_faults, + "{}", + test_case.description + ); + } + Ok(()) +} + #[test] fn given_contract_name_type_options_when_evaluating_then_accepts_configured_interfaces() -> Result<(), String> { diff --git a/crates/sqlbuild-rules-native/src/engine/tests/test_types.rs b/crates/sqlbuild-rules-native/src/engine/tests/test_types.rs index 547dd93b27..46a4aa88db 100644 --- a/crates/sqlbuild-rules-native/src/engine/tests/test_types.rs +++ b/crates/sqlbuild-rules-native/src/engine/tests/test_types.rs @@ -6,6 +6,11 @@ pub(crate) struct NativeEvaluationTestCase { pub(crate) expected_faults: Value, } +pub(crate) struct DynamicPivotEvaluationTestCase { + pub(crate) description: &'static str, + pub(crate) expected_faults: Value, +} + pub(crate) struct DeclarationScopeTestCase { pub(crate) description: &'static str, pub(crate) scope: &'static str, diff --git a/crates/sqlbuild-rules-native/src/models.rs b/crates/sqlbuild-rules-native/src/models.rs index bb0d071a21..84500365b2 100644 --- a/crates/sqlbuild-rules-native/src/models.rs +++ b/crates/sqlbuild-rules-native/src/models.rs @@ -185,6 +185,19 @@ pub(crate) struct Column { pub type_proven: bool, } +#[derive(Clone, Debug, Default, Deserialize, Serialize)] +#[serde(default, deny_unknown_fields)] +pub(crate) struct DynamicColumnFamily { + pub name: String, + pub pivot_column: String, + pub value_column: String, + pub aggregate: String, + #[serde(rename = "type", alias = "data_type")] + pub data_type: String, + pub name_pattern: Option, + pub type_proven: bool, +} + #[derive(Clone, Debug, Default, Deserialize, Serialize)] #[serde(default, deny_unknown_fields)] pub(crate) struct DeclarationMember { @@ -497,6 +510,9 @@ pub(crate) struct Model { pub config: BTreeMap, pub references: Vec, pub columns: Vec, + pub dynamic_columns: Vec, + pub dynamic_columns_proven: bool, + pub bare_dynamic_pivot: bool, pub enum_columns: Vec, pub enum_declarations: Vec, pub constant_declarations: Vec, diff --git a/crates/sqlbuild-rules-native/src/rules/_helpers/dynamic_contracts.rs b/crates/sqlbuild-rules-native/src/rules/_helpers/dynamic_contracts.rs new file mode 100644 index 0000000000..2c32435cfc --- /dev/null +++ b/crates/sqlbuild-rules-native/src/rules/_helpers/dynamic_contracts.rs @@ -0,0 +1,120 @@ +use std::collections::BTreeMap; + +use crate::constants::{ENFORCED_CONTRACT, SELECT_STAR_MODEL_RULE_CODE}; +use crate::models::{Fault, Model, RuleMetadata, RulesConfig}; +use crate::rules::_helpers::{contract_name_types, explicit_output_types, typed_contract_columns}; +use crate::rules::models::FaultCollector; +use sqlparser::ast::{PivotValueSource, Query, Select, SetExpr, TableFactor}; + +pub(crate) fn evaluate_without_query( + model: &Model, + config: &RulesConfig, + selected: &BTreeMap, +) -> Option> { + if !model.dynamic_columns_proven || !rules_need_no_query(model, selected) { + return None; + } + let faults = FaultCollector::default(); + if let Some(rule) = selected.get("SQBRCONTRACT101") + && model + .config + .get("contract") + .and_then(serde_json::Value::as_str) + != Some(ENFORCED_CONTRACT) + { + faults.push(fault(model, rule)); + } + if let Some(rule) = selected.get("SQBRCONTRACT106") { + typed_contract_columns::evaluate(model, rule, &faults); + } + if let Some(rule) = selected.get("SQBRCONTRACT105") { + explicit_output_types::evaluate_proven_dynamic(model, rule, &faults); + } + for column in &model.columns { + if let Some((code, message)) = contract_name_types::finding(column, config) + && let Some(rule) = selected.get(code) + { + faults.push(Fault { + code: rule.code.clone(), + path: model.relative_path.clone(), + line: 1, + column: 1, + message, + remediation: rule.remediation.clone(), + }); + } + } + Some(faults.into_inner()) +} + +pub(crate) fn allows_output_star(query: &Query, root: &Query, model: &Model) -> bool { + let Some(select) = root_select(query) else { + return false; + }; + let dynamic_pivot_star = select.projection.len() == 1 + && select.from.len() == 1 + && select.from[0].joins.is_empty() + && matches!( + &select.from[0].relation, + TableFactor::Pivot { + value_source: PivotValueSource::Any(_), + .. + } + ); + dynamic_pivot_star + || (model.dynamic_columns_proven + && std::ptr::eq(query, root) + && root + .with + .as_ref() + .and_then(|with| with.cte_tables.last()) + .is_some_and(|cte| { + sole_table_name(query) + .is_some_and(|name| name.eq_ignore_ascii_case(&cte.alias.name.value)) + })) +} + +fn root_select(query: &Query) -> Option<&Select> { + let SetExpr::Select(select) = query.body.as_ref() else { + return None; + }; + Some(select) +} + +fn sole_table_name(query: &Query) -> Option { + let select = root_select(query)?; + if select.from.len() != 1 || !select.from[0].joins.is_empty() { + return None; + } + let TableFactor::Table { + name, args: None, .. + } = &select.from[0].relation + else { + return None; + }; + Some(name.to_string()) +} + +fn rules_need_no_query(model: &Model, selected: &BTreeMap) -> bool { + selected.values().all(|rule| { + rule.custom + || rule.code.starts_with("SQBRSQL") + || rule.code.starts_with("SQBRCONTRACT10") + || (model.bare_dynamic_pivot && rule.code == SELECT_STAR_MODEL_RULE_CODE) + || matches!( + rule.code.as_str(), + "SQBRDECLARATION201" | "SQBRDECLARATION301" | "SQBRTEST301" + ) + }) +} + +fn fault(model: &Model, rule: &RuleMetadata) -> Fault { + Fault { + code: rule.code.clone(), + path: model.relative_path.clone(), + line: 1, + column: 1, + message: rule.message.clone(), + remediation: rule.remediation.clone(), + } +} diff --git a/crates/sqlbuild-rules-native/src/rules/_helpers/evaluation.rs b/crates/sqlbuild-rules-native/src/rules/_helpers/evaluation.rs index fbeb184d18..c2df5204a3 100644 --- a/crates/sqlbuild-rules-native/src/rules/_helpers/evaluation.rs +++ b/crates/sqlbuild-rules-native/src/rules/_helpers/evaluation.rs @@ -5,15 +5,16 @@ use crate::constants::{ use crate::models::{Declaration, EvaluateRequest, Fault, Model, RuleMetadata, RulesConfig}; use crate::rules::_helpers::domain_layout::folder_layer_details; use crate::rules::_helpers::{ - contract_name_types, explicit_output_types, numeric_decisions, typed_contract_columns, + contract_name_types, dynamic_contracts, explicit_output_types, numeric_decisions, + typed_contract_columns, }; use crate::rules::models::{ FaultCollector, ModelEvaluationRequest, ProjectEvaluationRequest, ResolvedThresholdOverride, }; use globset::{Glob, GlobSetBuilder}; use sqlparser::ast::{ - BinaryOperator, Expr, GroupByExpr, JoinConstraint, JoinOperator, PivotValueSource, Query, - Select, SelectItem, SetExpr, Spanned, Statement, TableFactor, Value, Visit, Visitor, + BinaryOperator, Expr, GroupByExpr, JoinConstraint, JoinOperator, Query, Select, SelectItem, + SetExpr, Spanned, Statement, TableFactor, Value, Visit, Visitor, }; use sqlparser::dialect::{ BigQueryDialect, ClickHouseDialect, DatabricksDialect, Dialect, DuckDbDialect, GenericDialect, @@ -120,6 +121,10 @@ fn evaluate_model_inner(request: ModelEvaluationRequest<'_>) -> Result { fn evaluate_set(&self, set_expr: &SetExpr, scope: OutputScope<'_>) { match set_expr { @@ -97,6 +144,15 @@ impl OutputRule<'_> { item, SelectItem::Wildcard(_) | SelectItem::QualifiedWildcard(_, _) ) { + if self.model.dynamic_columns_proven + && self + .model + .dynamic_columns + .iter() + .all(|family| family.type_proven) + { + continue; + } if proven_dependency_wildcard(select, scope.columns) { continue; } diff --git a/crates/sqlbuild-rules-native/src/rules/_helpers/mod.rs b/crates/sqlbuild-rules-native/src/rules/_helpers/mod.rs index 0c45193930..b2fa6832a4 100644 --- a/crates/sqlbuild-rules-native/src/rules/_helpers/mod.rs +++ b/crates/sqlbuild-rules-native/src/rules/_helpers/mod.rs @@ -1,6 +1,7 @@ pub(crate) mod catalogue; pub(crate) mod contract_name_types; pub(crate) mod domain_layout; +pub(crate) mod dynamic_contracts; pub(crate) mod evaluation; pub(crate) mod explicit_output_types; pub(crate) mod numeric_decisions; diff --git a/crates/sqlbuild-rules-native/src/rules/_helpers/typed_contract_columns.rs b/crates/sqlbuild-rules-native/src/rules/_helpers/typed_contract_columns.rs index 62fcca3435..583740a43c 100644 --- a/crates/sqlbuild-rules-native/src/rules/_helpers/typed_contract_columns.rs +++ b/crates/sqlbuild-rules-native/src/rules/_helpers/typed_contract_columns.rs @@ -23,4 +23,19 @@ pub(crate) fn evaluate(model: &Model, rule: &RuleMetadata, faults: &FaultCollect }); } } + for family in &model.dynamic_columns { + if family.data_type.trim().is_empty() { + faults.push(Fault { + code: rule.code.clone(), + path: model.relative_path.clone(), + line: 1, + column: 1, + message: format!( + "dynamic contract family {:?} has no declared type", + family.name + ), + remediation: rule.remediation.clone(), + }); + } + } } diff --git a/crates/sqlbuild-rules-native/src/sql_lint/_helpers/additional.rs b/crates/sqlbuild-rules-native/src/sql_lint/_helpers/additional.rs index b3f8533dd3..b2aa9e9b4c 100644 --- a/crates/sqlbuild-rules-native/src/sql_lint/_helpers/additional.rs +++ b/crates/sqlbuild-rules-native/src/sql_lint/_helpers/additional.rs @@ -6,7 +6,7 @@ use crate::sql_lint::_helpers::engine::{ direct_indices, is_comment, is_layout, is_query_from, query_end, significant_after, significant_before, token_depths, }; -use crate::sql_lint::models::AdditionalQueryFacts; +use crate::sql_lint::models::{AdditionalFactOptions, AdditionalQueryFacts}; const COUNT_ONE_LITERAL: &str = "1"; const QUALIFIED_REFERENCE_LENGTH: usize = 3; @@ -48,6 +48,7 @@ struct SelectFactContext<'a> { external_identifiers: &'a HashSet, dependency_identifiers: &'a HashSet, allows_ceremonial_select: bool, + allows_dynamic_output_star: bool, } struct ProjectedStarContext<'a> { @@ -59,9 +60,7 @@ struct ProjectedStarContext<'a> { pub(super) fn collect_additional_facts( tokens: &[Token], - external_identifiers: &HashSet, - dependency_identifiers: &HashSet, - allows_ceremonial_select: bool, + options: &AdditionalFactOptions<'_>, ) -> AdditionalQueryFacts { let mut facts = AdditionalQueryFacts::default(); let depths = token_depths(tokens); @@ -166,9 +165,10 @@ pub(super) fn collect_additional_facts( let select_facts = collect_select_additional_facts(&SelectFactContext { tokens, depths: &depths, - external_identifiers, - dependency_identifiers, - allows_ceremonial_select, + external_identifiers: options.external_identifiers, + dependency_identifiers: options.dependency_identifiers, + allows_ceremonial_select: options.allows_ceremonial_select, + allows_dynamic_output_star: options.allows_dynamic_output_star, }); facts.duplicate_output_aliases = select_facts.duplicate_output_aliases; facts.unaliased_calculations = select_facts.unaliased_calculations; @@ -271,6 +271,12 @@ fn collect_select_additional_facts(context: &SelectFactContext<'_>) -> Additiona )); } if !is_dependency_import_select(context, select_index, &direct, from_position) + && !is_compiler_proven_dynamic_output_select( + context, + select_index, + &direct, + from_position, + ) && !has_irreducible_projected_star( &ProjectedStarContext { tokens, @@ -352,6 +358,25 @@ fn collect_select_additional_facts(context: &SelectFactContext<'_>) -> Additiona facts } +fn is_compiler_proven_dynamic_output_select( + context: &SelectFactContext<'_>, + select_index: usize, + direct: &[usize], + from_position: usize, +) -> bool { + if !context.allows_dynamic_output_star + || context.depths[select_index] != 0 + || direct[..from_position].len() != 1 + || context.tokens[direct[0]].token_type != TokenType::Star + { + return false; + } + !context.tokens[select_index + 1..] + .iter() + .zip(&context.depths[select_index + 1..]) + .any(|(token, depth)| token.token_type == TokenType::Select && *depth == 0) +} + fn has_irreducible_projected_star( context: &ProjectedStarContext<'_>, from_position: usize, diff --git a/crates/sqlbuild-rules-native/src/sql_lint/_helpers/engine.rs b/crates/sqlbuild-rules-native/src/sql_lint/_helpers/engine.rs index 60c3df77b4..68ae046abc 100644 --- a/crates/sqlbuild-rules-native/src/sql_lint/_helpers/engine.rs +++ b/crates/sqlbuild-rules-native/src/sql_lint/_helpers/engine.rs @@ -10,7 +10,8 @@ use polyglot_sql::{ use crate::sql_lint::constants::LINT_API_VERSION; use crate::sql_lint::models::{ - LintDiagnostic, LintEdit, LintRequest, LintResponse, LintRuleMetadata, QueryFacts, + AdditionalFactOptions, LintDiagnostic, LintEdit, LintRequest, LintResponse, LintRuleMetadata, + QueryFacts, }; use crate::sql_lint::_helpers::additional::collect_additional_facts; @@ -321,6 +322,7 @@ struct FactBuildOptions<'a> { dependency_identifiers: &'a HashSet, externally_referenced_ctes: &'a HashSet, allows_ceremonial_select: bool, + allows_dynamic_output_star: bool, } pub(crate) fn lint_json_impl(request_json: &str) -> Result { @@ -380,6 +382,7 @@ fn lint(request: LintRequest) -> Result { dependency_identifiers: &dependency_identifiers, externally_referenced_ctes: &externally_referenced_ctes, allows_ceremonial_select: request.allows_ceremonial_select, + allows_dynamic_output_star: request.allows_dynamic_output_star, }; let facts = build_facts(&statements, &tokens, &request.sql, &options); let context = DiagnosticContext { @@ -437,9 +440,12 @@ fn build_facts( facts.null_comparisons = null_comparison_spans(tokens); facts.additional = collect_additional_facts( tokens, - options.external_identifiers, - options.dependency_identifiers, - options.allows_ceremonial_select, + &AdditionalFactOptions { + external_identifiers: options.external_identifiers, + dependency_identifiers: options.dependency_identifiers, + allows_ceremonial_select: options.allows_ceremonial_select, + allows_dynamic_output_star: options.allows_dynamic_output_star, + }, ); for expression in expressions { facts diff --git a/crates/sqlbuild-rules-native/src/sql_lint/models.rs b/crates/sqlbuild-rules-native/src/sql_lint/models.rs index f45c92226d..385588e5f6 100644 --- a/crates/sqlbuild-rules-native/src/sql_lint/models.rs +++ b/crates/sqlbuild-rules-native/src/sql_lint/models.rs @@ -1,3 +1,5 @@ +use std::collections::HashSet; + use polyglot_sql::tokens::Span; use serde::{Deserialize, Serialize}; @@ -67,6 +69,16 @@ pub(crate) struct LintRequest { pub externally_referenced_ctes: Vec, #[serde(default)] pub allows_ceremonial_select: bool, + #[serde(default)] + pub allows_dynamic_output_star: bool, +} + +#[derive(Debug)] +pub(crate) struct AdditionalFactOptions<'a> { + pub external_identifiers: &'a HashSet, + pub dependency_identifiers: &'a HashSet, + pub allows_ceremonial_select: bool, + pub allows_dynamic_output_star: bool, } #[derive(Debug, Serialize)] diff --git a/crates/sqlbuild-rules-native/src/sql_lint/tests/test_sql_lint.rs b/crates/sqlbuild-rules-native/src/sql_lint/tests/test_sql_lint.rs index 56519b4e3f..c344a2dceb 100644 --- a/crates/sqlbuild-rules-native/src/sql_lint/tests/test_sql_lint.rs +++ b/crates/sqlbuild-rules-native/src/sql_lint/tests/test_sql_lint.rs @@ -1563,6 +1563,49 @@ fn given_additional_rule_cases_when_linting_then_findings_and_fixes_match() -> R Ok(()) } +#[test] +fn given_compiler_proven_dynamic_output_when_linting_then_only_terminal_star_is_exempt() +-> Result<(), String> { + let test_cases = [test_types::DynamicOutputStarLintTestCase { + description: "unrelated CTE star remains controlled", + sql: "WITH pivot_input AS (SELECT customer_id, category, amount FROM orders), \ +final AS (SELECT * FROM pivot_input PIVOT(MAX(amount) FOR category IN (ANY))), \ +unrelated AS (SELECT * FROM pivot_input) SELECT * FROM final", + expected_count: 1, + expected_prefix: "unrelated AS (SELECT ", + }]; + for test_case in test_cases { + let response = lint_json( + &json!({ + "version": 1, + "sql": test_case.sql, + "dialect": "snowflake", + "enabled_rules": ["SQBRSQL021"], + "allows_dynamic_output_star": true, + }) + .to_string(), + )?; + let payload: Value = serde_json::from_str(&response).map_err(|error| error.to_string())?; + let diagnostics = payload["diagnostics"] + .as_array() + .ok_or("diagnostics should be an array")?; + + assert_eq!( + diagnostics.len(), + test_case.expected_count, + "{}", + test_case.description + ); + let start = diagnostics[0]["start"].as_u64().unwrap_or_default() as usize; + assert!( + test_case.sql[..start].ends_with(test_case.expected_prefix), + "{}", + test_case.description + ); + } + Ok(()) +} + #[test] fn given_ceremonial_cte_contexts_when_linting_then_only_harness_dependencies_are_reachable() -> Result<(), String> { diff --git a/crates/sqlbuild-rules-native/src/sql_lint/tests/test_types.rs b/crates/sqlbuild-rules-native/src/sql_lint/tests/test_types.rs index c3fdf4f813..db404b8d14 100644 --- a/crates/sqlbuild-rules-native/src/sql_lint/tests/test_types.rs +++ b/crates/sqlbuild-rules-native/src/sql_lint/tests/test_types.rs @@ -42,6 +42,13 @@ pub(crate) struct PlainSqlLintTestCase { pub expected_count: usize, } +pub(crate) struct DynamicOutputStarLintTestCase { + pub(crate) description: &'static str, + pub(crate) sql: &'static str, + pub(crate) expected_count: usize, + pub(crate) expected_prefix: &'static str, +} + pub(crate) struct DependencyImportLintTestCase { pub(crate) description: &'static str, pub(crate) sql: &'static str, diff --git a/docs/contract-adoption.md b/docs/contract-adoption.md index f4837d7992..002c2a2d1f 100644 --- a/docs/contract-adoption.md +++ b/docs/contract-adoption.md @@ -96,7 +96,51 @@ outputs. The cast target must use the declared contract type; an alias with diff not treated as an explicit declaration of that contract. An exact outer cast establishes the Rule's authored type boundary even when general expression inference remains incomplete. Every set-operation branch must cast its outputs before common-type coercion. Wildcards are not an -explicit contract boundary. +explicit contract boundary unless they expose a compiler-proven dynamic family as described below. + +### Runtime-dynamic pivot families + +An enforced contract can remain closed when a native pivot discovers output names from data at +runtime. Declare fixed grouping columns normally and declare each generated family separately: + +```sql +MODEL ( + contract enforced, + columns (customer_id (type INTEGER)), + dynamic_columns ( + category_amounts ( + pivot_column category, + value_column amount, + aggregate MAX, + type "DECIMAL(12,2)" + ) + ), +); + +PIVOT __ref("stg_order_amounts") +ON category +USING MAX(amount) +GROUP BY customer_id +``` + +This is not an open wildcard contract. SQLBuild proves that the complete output boundary is either +one supported runtime-dynamic pivot or an exact wildcard passthrough from an upstream model with the +same families. It derives the fixed columns from the pivot input, matches every pivot aggregate to +one family, and proves the generated type from the authoritative input contract or an explicit cast. +An unproven boundary fails with `K011`; an incompatible family type fails with `K002`; an unknown +family type reports `K003`. `SQBRCONTRACT105` accepts the wildcard only after this compiler proof. + +Snowflake `PIVOT ... IN (ANY ...)` and dynamic subquery pivots are supported. DuckDB and MotherDuck +native dynamic `PIVOT ... ON ... USING ...` statements are supported. A static `IN` list continues +to use ordinary exact column declarations. Dynamically constructed SQL strings are not contract +proof, and adapters without a structurally analyzable native dynamic pivot reject +`dynamic_columns`. + +When more than one family is declared, each requires `name_pattern`, a regular expression matched +against the complete generated column name. Runtime validation requires every non-fixed physical +column to match exactly one family and have that family's declared type. Contract inspection applies +the same classification. Contract generation preserves the family declaration and does not freeze +the current runtime members into `columns`. When `SQBRSQL035` is also selected, the final top-level CTE owns these casts and the terminal `SELECT` remains a plain read from that CTE: diff --git a/src/sqlbuild/compiler/compile/_helpers/analysis/dynamic_pivot.py b/src/sqlbuild/compiler/compile/_helpers/analysis/dynamic_pivot.py new file mode 100644 index 0000000000..fd8385e427 --- /dev/null +++ b/src/sqlbuild/compiler/compile/_helpers/analysis/dynamic_pivot.py @@ -0,0 +1,679 @@ +"""Compiler proof for closed contracts over runtime-dynamic pivot columns.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, cast + +from sqlbuild.compiler.compile.models import ( + DynamicColumnContractProof, + DynamicColumnFamilyProof, + InferredColumn, +) +from sqlbuild.compiler.lineage.types import InferredNullability +from sqlbuild.compiler.sql_analysis.main.import_polyglot_sql import import_polyglot_sql +from sqlbuild.spec.contracts.models import SchemaDynamicColumnFamily + +_DUCKDB_DIALECT: str = "duckdb" +_MOTHERDUCK_DIALECT: str = "motherduck" +_SNOWFLAKE_DIALECT: str = "snowflake" +_PIVOT_AST_KIND: str = "pivot" +_SELECT_AST_KIND: str = "select" +_QUERY_AST_KINDS: frozenset[str] = frozenset({"query", "subquery"}) +_CAST_AST_KIND: str = "cast" +_COLUMN_AST_KIND: str = "column" +_TABLE_AST_KIND: str = "table" +_FUNCTION_AST_KIND: str = "function" +_ALIAS_AST_KIND: str = "alias" +_CTE_AST_KIND: str = "cte" +_STAR_AST_KIND: str = "star" +_DYNAMIC_VALUE_SOURCE_KINDS: frozenset[str] = frozenset({"pivot_any", _SELECT_AST_KIND, "query"}) +_DEPENDENCY_FUNCTION_NAMES: frozenset[str] = frozenset({"__ref", "__source", "__seed"}) +_UNKNOWN_DATA_TYPE: str = "UNKNOWN" +_SIMPLIFIED_PIVOT_DIALECTS: frozenset[str] = frozenset({_DUCKDB_DIALECT, _MOTHERDUCK_DIALECT}) +_SUPPORTED_DIALECTS: frozenset[str] = frozenset({*_SIMPLIFIED_PIVOT_DIALECTS, _SNOWFLAKE_DIALECT}) +_TYPE_PRESERVING_AGGREGATES: frozenset[str] = frozenset({"ANY_VALUE", "MAX", "MIN"}) +type _CteMap = dict[str, dict[str, object] | None] + + +@dataclass(frozen=True) +class _ColumnFact: + name: str + type: str | None + nullability: InferredNullability = InferredNullability.UNKNOWN + + +@dataclass(frozen=True) +class _Boundary: + pivot: dict[str, object] | None = None + passthrough_name: str | None = None + + +def analyze_dynamic_column_contract( + *, + query_sql: str, + dialect: str | None, + families: tuple[SchemaDynamicColumnFamily, ...], + column_types_by_table: dict[str, dict[str, str]], + authoritative_column_types_by_table: dict[str, dict[str, str]], + column_nullability_by_table: dict[str, dict[str, InferredNullability]], + dynamic_families_by_table: dict[str, tuple[SchemaDynamicColumnFamily, ...]], +) -> DynamicColumnContractProof | None: + """Prove that a model output is closed over fixed columns and declared pivot families.""" + + if not families: + return None + normalized_dialect: str = (dialect or "").casefold() + if normalized_dialect not in _SUPPORTED_DIALECTS: + return _failure( + f"adapter dialect '{dialect or 'generic'}' does not support compiler-proven " + "dynamic pivots" + ) + parse_dialect: str = ( + _DUCKDB_DIALECT if normalized_dialect == _MOTHERDUCK_DIALECT else normalized_dialect + ) + polyglot: Any = import_polyglot_sql() + try: + parsed: Any = polyglot.parse_one(query_sql, dialect=parse_dialect) + except polyglot.PolyglotError as error: + return _failure(f"dynamic pivot SQL could not be parsed: {error}") + root: object = parsed.to_dict() + if not isinstance(root, dict): + return _failure("dynamic pivot SQL did not produce a structured query") + ctes: _CteMap = _collect_ctes(root) + boundary: _Boundary | None = _resolve_boundary(node=root, ctes=ctes, seen=frozenset()) + if boundary is None: + return _failure( + "output must be a wildcard projection from one supported dynamic pivot or a proven " + "dynamic-family passthrough" + ) + if boundary.passthrough_name is not None: + return _analyze_passthrough( + name=boundary.passthrough_name, + families=families, + column_types_by_table=column_types_by_table, + column_nullability_by_table=column_nullability_by_table, + dynamic_families_by_table=dynamic_families_by_table, + ) + if boundary.pivot is None: + return _failure("dynamic output boundary did not resolve to a pivot") + return _analyze_pivot( + pivot=boundary.pivot, + dialect=normalized_dialect, + families=families, + ctes=ctes, + column_types_by_table=authoritative_column_types_by_table, + column_nullability_by_table=column_nullability_by_table, + bare_dynamic_pivot=_node_key(root) == _PIVOT_AST_KIND, + ) + + +def _analyze_passthrough( + *, + name: str, + families: tuple[SchemaDynamicColumnFamily, ...], + column_types_by_table: dict[str, dict[str, str]], + column_nullability_by_table: dict[str, dict[str, InferredNullability]], + dynamic_families_by_table: dict[str, tuple[SchemaDynamicColumnFamily, ...]], +) -> DynamicColumnContractProof: + upstream_families: tuple[SchemaDynamicColumnFamily, ...] | None = _casefold_lookup( + mapping=dynamic_families_by_table, key=name + ) + if upstream_families is None or not _families_equal(left=families, right=upstream_families): + return _failure( + f"dynamic-family passthrough from '{name}' must redeclare the upstream families exactly" + ) + source_columns: tuple[_ColumnFact, ...] | None = _external_schema( + name=name, + column_types_by_table=column_types_by_table, + column_nullability_by_table=column_nullability_by_table, + ) + if source_columns is None: + return _failure(f"dynamic-family passthrough source '{name}' has no authoritative schema") + return DynamicColumnContractProof( + output_proven=True, + fixed_columns=tuple( + InferredColumn(name=column.name, type=column.type, nullability=column.nullability) + for column in source_columns + ), + families=tuple( + DynamicColumnFamilyProof(name=family.name, inferred_type=family.type) + for family in families + ), + ) + + +def _analyze_pivot( + *, + pivot: dict[str, object], + dialect: str, + families: tuple[SchemaDynamicColumnFamily, ...], + ctes: _CteMap, + column_types_by_table: dict[str, dict[str, str]], + column_nullability_by_table: dict[str, dict[str, InferredNullability]], + bare_dynamic_pivot: bool, +) -> DynamicColumnContractProof: + if bool(pivot.get("unpivot")): + return _failure("UNPIVOT does not produce a dynamic column family") + pivot_columns, aggregates, dynamic, pivot_columns_valid = _pivot_components( + pivot=pivot, dialect=dialect + ) + if not dynamic: + return _failure("static pivot values must use ordinary exact column declarations") + if len(pivot_columns) != 1: + return _failure("dynamic column contracts currently require exactly one pivot column") + if not pivot_columns_valid: + return _failure("dynamic pivot columns must be direct column references") + source_node: object = pivot.get("this") + source_columns: tuple[_ColumnFact, ...] | None = _relation_schema( + node=source_node, + ctes=ctes, + column_types_by_table=column_types_by_table, + column_nullability_by_table=column_nullability_by_table, + seen=frozenset(), + ) + if source_columns is None: + return _failure("dynamic pivot input does not have an authoritative, explicit schema") + pivot_column: str = pivot_columns[0] + matched_aggregate_indices: set[int] = set() + family_proofs: list[DynamicColumnFamilyProof] = [] + for family in families: + if not family.pivot_column.casefold() == pivot_column.casefold(): + return _failure( + f"dynamic family '{family.name}' declares pivot_column '{family.pivot_column}' " + f"but the output pivot uses '{pivot_column}'" + ) + matches: list[tuple[int, str | None]] = [] + for index, aggregate in enumerate(aggregates): + aggregate_name, value_name, inferred_type = _aggregate_fact( + aggregate=aggregate, source_columns=source_columns + ) + if ( + aggregate_name.casefold() == family.aggregate.casefold() + and value_name is not None + and value_name.casefold() == family.value_column.casefold() + ): + matches.append((index, inferred_type)) + if len(matches) != 1: + return _failure( + f"dynamic family '{family.name}' must match exactly one " + f"{family.aggregate}({family.value_column}) pivot aggregate" + ) + aggregate_index, inferred_type = matches[0] + if aggregate_index in matched_aggregate_indices: + return _failure("one pivot aggregate cannot satisfy multiple dynamic families") + matched_aggregate_indices.add(aggregate_index) + family_proofs.append( + DynamicColumnFamilyProof(name=family.name, inferred_type=inferred_type) + ) + if len(matched_aggregate_indices) != len(aggregates): + return _failure("every dynamic pivot aggregate must have one declared column family") + fixed_names: tuple[str, ...] + if dialect in _SIMPLIFIED_PIVOT_DIALECTS and isinstance(pivot.get("group"), dict): + fixed_names = tuple( + name + for expression in _nested_list(node=pivot.get("group"), keys=("group", "expressions")) + if (name := _column_name(expression)) is not None + ) + else: + excluded: frozenset[str] = frozenset( + {pivot_column.casefold(), *(family.value_column.casefold() for family in families)} + ) + fixed_names = tuple( + column.name for column in source_columns if column.name.casefold() not in excluded + ) + source_by_name: dict[str, _ColumnFact] = { + column.name.casefold(): column for column in source_columns + } + if any(name.casefold() not in source_by_name for name in fixed_names): + return _failure("dynamic pivot grouping columns are not present in the pivot input") + return DynamicColumnContractProof( + output_proven=True, + fixed_columns=tuple( + InferredColumn( + name=name, + type=source_by_name[name.casefold()].type, + nullability=source_by_name[name.casefold()].nullability, + ) + for name in fixed_names + ), + families=tuple(family_proofs), + bare_dynamic_pivot=bare_dynamic_pivot, + ) + + +def _pivot_components( + *, pivot: dict[str, object], dialect: str +) -> tuple[tuple[str, ...], tuple[dict[str, object], ...], bool, bool]: + if dialect in _SIMPLIFIED_PIVOT_DIALECTS and pivot.get("using"): + pivot_expressions: tuple[dict[str, object], ...] = _dict_list(pivot.get("expressions")) + pivot_columns: tuple[str, ...] = tuple( + name + for expression in pivot_expressions + if (name := _column_name(expression)) is not None + ) + return ( + pivot_columns, + _dict_list(pivot.get("using")), + not bool(pivot.get("fields")), + len(pivot_columns) == len(pivot_expressions), + ) + fields: tuple[dict[str, object], ...] = _dict_list(pivot.get("fields")) + pivot_columns = tuple( + name + for field in fields + if (name := _column_name(_nested(node=field, keys=("in", "this")))) is not None + ) + dynamic: bool = _has_dynamic_value_source(fields=fields) + return ( + pivot_columns, + _dict_list(pivot.get("expressions")), + dynamic, + len(pivot_columns) == len(fields), + ) + + +def _has_dynamic_value_source(*, fields: tuple[dict[str, object], ...]) -> bool: + for field in fields: + values: tuple[dict[str, object], ...] = _nested_list(node=field, keys=("in", "expressions")) + if any(_node_key(value) in _DYNAMIC_VALUE_SOURCE_KINDS for value in values): + return True + return False + + +def _aggregate_fact( + *, aggregate: dict[str, object], source_columns: tuple[_ColumnFact, ...] +) -> tuple[str, str | None, str | None]: + aggregate_name: str = _node_key(aggregate).upper() + payload: object = aggregate.get(_node_key(aggregate)) + if not isinstance(payload, dict): + return aggregate_name, None, None + payload_dict: dict[str, object] = cast(dict[str, object], payload) + authored_name: object = payload_dict.get("name") + if isinstance(authored_name, str) and authored_name: + aggregate_name = authored_name.upper() + value: object = payload_dict.get("this") + value_name: str | None = _column_name(value) + inferred_type: str | None = None + if _node_key(value) == _CAST_AST_KIND: + cast_payload: object = ( + cast(dict[str, object], value).get("cast") if isinstance(value, dict) else None + ) + if isinstance(cast_payload, dict): + cast_payload_dict: dict[str, object] = cast(dict[str, object], cast_payload) + value_name = _column_name(cast_payload_dict.get("this")) + if aggregate_name in _TYPE_PRESERVING_AGGREGATES: + inferred_type = _render_type(cast_payload_dict.get("to")) + if inferred_type is None and aggregate_name in _TYPE_PRESERVING_AGGREGATES: + source_by_name: dict[str, _ColumnFact] = { + column.name.casefold(): column for column in source_columns + } + source: _ColumnFact | None = ( + source_by_name.get(value_name.casefold()) if value_name is not None else None + ) + inferred_type = source.type if source is not None else None + return aggregate_name, value_name, inferred_type + + +def _resolve_boundary(*, node: object, ctes: _CteMap, seen: frozenset[str]) -> _Boundary | None: + key: str = _node_key(node) + payload: object = cast(dict[str, object], node).get(key) if isinstance(node, dict) else None + if key == _PIVOT_AST_KIND and isinstance(payload, dict): + return _Boundary(pivot=cast(dict[str, object], payload)) + if key in _QUERY_AST_KINDS and isinstance(payload, dict): + return _resolve_boundary( + node=cast(dict[str, object], payload).get("this"), ctes=ctes, seen=seen + ) + if key != _SELECT_AST_KIND or not isinstance(payload, dict): + return None + payload_dict: dict[str, object] = cast(dict[str, object], payload) + if payload_dict.get("joins") or not _is_single_wildcard(payload_dict.get("expressions")): + return None + relations: tuple[dict[str, object], ...] = _nested_dict_list( + node=payload_dict.get("from"), key="expressions" + ) + if len(relations) != 1: + return None + relation: dict[str, object] = relations[0] + if _node_key(relation) == _PIVOT_AST_KIND: + pivot_payload: object = relation.get("pivot") + return ( + _Boundary(pivot=cast(dict[str, object], pivot_payload)) + if isinstance(pivot_payload, dict) + else None + ) + relation_name: str | None = _relation_name(relation) + if relation_name is None: + return None + normalized_name: str = relation_name.casefold() + if normalized_name in ctes: + cte: dict[str, object] | None = ctes[normalized_name] + if cte is None: + return None + if relation_name.casefold() in seen: + return None + resolved: _Boundary | None = _resolve_boundary( + node=cte, + ctes=ctes, + seen=seen | {relation_name.casefold()}, + ) + if resolved is None: + return None + return resolved + return _Boundary(passthrough_name=relation_name) + + +def _relation_schema( + *, + node: object, + ctes: _CteMap, + column_types_by_table: dict[str, dict[str, str]], + column_nullability_by_table: dict[str, dict[str, InferredNullability]], + seen: frozenset[str], +) -> tuple[_ColumnFact, ...] | None: + relation_name: str | None = _relation_name(node) + if relation_name is None: + return None + normalized_name: str = relation_name.casefold() + if normalized_name not in ctes: + return _external_schema( + name=relation_name, + column_types_by_table=column_types_by_table, + column_nullability_by_table=column_nullability_by_table, + ) + cte: dict[str, object] | None = ctes[normalized_name] + if cte is None or normalized_name in seen: + return None + return _select_schema( + node=cte, + ctes=ctes, + column_types_by_table=column_types_by_table, + column_nullability_by_table=column_nullability_by_table, + seen=seen | {normalized_name}, + ) + + +def _select_schema( + *, + node: object, + ctes: _CteMap, + column_types_by_table: dict[str, dict[str, str]], + column_nullability_by_table: dict[str, dict[str, InferredNullability]], + seen: frozenset[str], +) -> tuple[_ColumnFact, ...] | None: + key: str = _node_key(node) + payload: object = cast(dict[str, object], node).get(key) if isinstance(node, dict) else None + if key in _QUERY_AST_KINDS and isinstance(payload, dict): + return _select_schema( + node=cast(dict[str, object], payload).get("this"), + ctes=ctes, + column_types_by_table=column_types_by_table, + column_nullability_by_table=column_nullability_by_table, + seen=seen, + ) + if key != _SELECT_AST_KIND or not isinstance(payload, dict): + return None + payload_dict: dict[str, object] = cast(dict[str, object], payload) + if payload_dict.get("joins"): + return None + relations: tuple[dict[str, object], ...] = _nested_dict_list( + node=payload_dict.get("from"), key="expressions" + ) + if len(relations) != 1: + return None + source: tuple[_ColumnFact, ...] | None = _relation_schema( + node=relations[0], + ctes=ctes, + column_types_by_table=column_types_by_table, + column_nullability_by_table=column_nullability_by_table, + seen=seen, + ) + if source is None: + return None + expressions: object = payload_dict.get("expressions") + if _is_single_wildcard(expressions): + return source + if not isinstance(expressions, list): + return None + source_by_name: dict[str, _ColumnFact] = {column.name.casefold(): column for column in source} + output: list[_ColumnFact] = [] + for raw_expression in expressions: + if not isinstance(raw_expression, dict): + return None + output_name, expression = _projection(cast(dict[str, object], raw_expression)) + source_name: str | None + expression_type: str | None = None + if _node_key(expression) == _CAST_AST_KIND: + cast_payload: object = expression.get("cast") + if not isinstance(cast_payload, dict): + return None + cast_payload_dict: dict[str, object] = cast(dict[str, object], cast_payload) + source_name = _column_name(cast_payload_dict.get("this")) + expression_type = _render_type(cast_payload_dict.get("to")) + else: + source_name = _column_name(expression) + if output_name is None or source_name is None: + return None + source_column: _ColumnFact | None = source_by_name.get(source_name.casefold()) + if source_column is None: + return None + if expression_type is None: + expression_type = source_column.type + output.append( + _ColumnFact( + name=output_name, + type=expression_type, + nullability=source_column.nullability, + ) + ) + return tuple(output) + + +def _external_schema( + *, + name: str, + column_types_by_table: dict[str, dict[str, str]], + column_nullability_by_table: dict[str, dict[str, InferredNullability]], +) -> tuple[_ColumnFact, ...] | None: + types: dict[str, str] | None = _casefold_lookup(mapping=column_types_by_table, key=name) + if types is None: + return None + nullability: dict[str, InferredNullability] = ( + _casefold_lookup(mapping=column_nullability_by_table, key=name) or {} + ) + nullability_by_name: dict[str, InferredNullability] = { + column_name.casefold(): value for column_name, value in nullability.items() + } + return tuple( + _ColumnFact( + name=column_name, + type=None if data_type == _UNKNOWN_DATA_TYPE else data_type, + nullability=nullability_by_name.get( + column_name.casefold(), InferredNullability.UNKNOWN + ), + ) + for column_name, data_type in types.items() + ) + + +def _collect_ctes(root: dict[str, object]) -> _CteMap: + key: str = _node_key(root) + payload: object = root.get(key) + if not isinstance(payload, dict): + return {} + payload_dict: dict[str, object] = cast(dict[str, object], payload) + with_payload: object = payload_dict.get("with") + if not isinstance(with_payload, dict): + return {} + with_payload_dict: dict[str, object] = cast(dict[str, object], with_payload) + ctes: _CteMap = {} + for cte in _dict_list(with_payload_dict.get("ctes")): + cte_payload: object = cte.get("cte") if _node_key(cte) == _CTE_AST_KIND else cte + if not isinstance(cte_payload, dict): + continue + cte_payload_dict: dict[str, object] = cast(dict[str, object], cte_payload) + name: str | None = _identifier_name(cte_payload_dict.get("alias")) + query: object = cte_payload_dict.get("this") + if name is None: + continue + if cte_payload_dict.get("columns"): + ctes[name.casefold()] = None + elif isinstance(query, dict): + ctes[name.casefold()] = cast(dict[str, object], query) + return ctes + + +def _projection(node: dict[str, object]) -> tuple[str | None, dict[str, object]]: + if _node_key(node) != _ALIAS_AST_KIND: + return _column_name(node), node + payload: object = node.get("alias") + if not isinstance(payload, dict): + return None, node + payload_dict: dict[str, object] = cast(dict[str, object], payload) + expression: object = payload_dict.get("this") + return ( + _identifier_name(payload_dict.get("alias")), + cast(dict[str, object], expression) if isinstance(expression, dict) else {}, + ) + + +def _relation_name(node: object) -> str | None: + key: str = _node_key(node) + payload: object = cast(dict[str, object], node).get(key) if isinstance(node, dict) else None + if not isinstance(payload, dict): + return None + payload_dict: dict[str, object] = cast(dict[str, object], payload) + if key == _TABLE_AST_KIND: + return _identifier_name(payload_dict.get("name")) + if key == _COLUMN_AST_KIND: + return _identifier_name(payload_dict.get("name")) + if ( + key == _FUNCTION_AST_KIND + and str(payload_dict.get("name", "")).casefold() in _DEPENDENCY_FUNCTION_NAMES + ): + arguments: tuple[dict[str, object], ...] = _dict_list(payload_dict.get("args")) + return _column_name(arguments[0]) if arguments else None + return None + + +def _column_name(node: object) -> str | None: + key: str = _node_key(node) + payload: object = cast(dict[str, object], node).get(key) if isinstance(node, dict) else None + if key == _COLUMN_AST_KIND and isinstance(payload, dict): + return _identifier_name(cast(dict[str, object], payload).get("name")) + return None + + +def _identifier_name(node: object) -> str | None: + if isinstance(node, str): + return node + if not isinstance(node, dict): + return None + value: object = cast(dict[str, object], node).get("name") + return value if isinstance(value, str) and value else None + + +def _render_type(node: object) -> str | None: + if not isinstance(node, dict): + return None + node_dict: dict[str, object] = cast(dict[str, object], node) + raw_type: object = node_dict.get("data_type") + if not isinstance(raw_type, str) or not raw_type: + return None + names: dict[str, str] = { + "big_int": "BIGINT", + "bool": "BOOLEAN", + "boolean": "BOOLEAN", + "decimal": "DECIMAL", + "double": "DOUBLE", + "float": "FLOAT", + "int": "INT", + "integer": "INT", + "text": "TEXT", + "timestamp": "TIMESTAMP", + "var_char": "VARCHAR", + "varchar": "VARCHAR", + } + rendered: str = names.get(raw_type.casefold(), raw_type.replace("_", " ").upper()) + precision: object = node_dict.get("precision") + scale: object = node_dict.get("scale") + length: object = node_dict.get("length") + if isinstance(length, int): + return f"{rendered}({length})" + if isinstance(precision, int): + return ( + f"{rendered}({precision}, {scale})" + if isinstance(scale, int) + else f"{rendered}({precision})" + ) + return rendered + + +def _families_equal( + *, + left: tuple[SchemaDynamicColumnFamily, ...], + right: tuple[SchemaDynamicColumnFamily, ...], +) -> bool: + def normalize(family: SchemaDynamicColumnFamily) -> tuple[str, str, str, str, str, str | None]: + return ( + family.name.casefold(), + family.pivot_column.casefold(), + family.value_column.casefold(), + family.aggregate.casefold(), + family.type.casefold(), + family.name_pattern, + ) + + return tuple(map(normalize, left)) == tuple(map(normalize, right)) + + +def _failure(reason: str) -> DynamicColumnContractProof: + return DynamicColumnContractProof(output_proven=False, failure_reason=reason) + + +def _node_key(node: object) -> str: + if not isinstance(node, dict) or len(node) != 1: + return "" + return str(next(iter(node))) + + +def _dict_list(value: object) -> tuple[dict[str, object], ...]: + if not isinstance(value, list): + return () + return tuple(cast(dict[str, object], item) for item in value if isinstance(item, dict)) + + +def _nested(*, node: object, keys: tuple[str, ...]) -> object: + current: object = node + for key in keys: + if not isinstance(current, dict): + return None + current = cast(dict[str, object], current).get(key) + return current + + +def _nested_list(*, node: object, keys: tuple[str, ...]) -> tuple[dict[str, object], ...]: + return _dict_list(_nested(node=node, keys=keys)) + + +def _nested_dict_list(*, node: object, key: str) -> tuple[dict[str, object], ...]: + return _dict_list(cast(dict[str, object], node).get(key) if isinstance(node, dict) else None) + + +def _is_single_wildcard(value: object) -> bool: + expressions: tuple[dict[str, object], ...] = _dict_list(value) + if len(expressions) != 1 or _node_key(expressions[0]) != _STAR_AST_KIND: + return False + payload: object = expressions[0].get("star") + if not isinstance(payload, dict): + return False + star: dict[str, object] = cast(dict[str, object], payload) + return not any(star.get(key) for key in ("except", "replace", "rename", "table")) + + +def _casefold_lookup[ValueT](*, mapping: dict[str, ValueT], key: str) -> ValueT | None: + requested: str = key.casefold() + for candidate, value in mapping.items(): + if candidate.casefold() == requested: + return value + return None diff --git a/src/sqlbuild/compiler/compile/_helpers/assembly/project.py b/src/sqlbuild/compiler/compile/_helpers/assembly/project.py index a94204b1e1..24296acedb 100644 --- a/src/sqlbuild/compiler/compile/_helpers/assembly/project.py +++ b/src/sqlbuild/compiler/compile/_helpers/assembly/project.py @@ -21,8 +21,12 @@ analyze_columns_and_lineage_with_polyglot, get_complete_schema_binding_request, infer_columns_with_sql_analysis, + substitute_placeholder_defaults, table_function_analysis_name, ) +from sqlbuild.compiler.compile._helpers.analysis.dynamic_pivot import ( + analyze_dynamic_column_contract, +) from sqlbuild.compiler.compile._helpers.analysis.validation import ( validate_hook_sql_syntax, validate_sql_syntax, @@ -88,6 +92,7 @@ CompileSqlFunctionInput, CompileSqlScenarioInput, CompileSqlTestInput, + DynamicColumnContractProof, InferredColumn, MacroContext, PolyglotAnalysisResult, @@ -123,6 +128,7 @@ DefaultsConfig, SchemaAuditInstance, SchemaColumn, + SchemaDynamicColumnFamily, SchemaSeedEntry, SourceColumnEntry, SourceEntry, @@ -149,6 +155,12 @@ class _ModelSqlAnalysisRequest: binding_schema: dict[str, dict[str, str]] | None +@dataclass(frozen=True) +class _DynamicContractAnalysisInputs: + families_by_table: dict[str, tuple[SchemaDynamicColumnFamily, ...]] + authoritative_column_types_by_table: dict[str, dict[str, str]] + + def assemble_compiled_project( *, inputs: CompileProjectInputs, @@ -170,7 +182,16 @@ def assemble_compiled_project( _build_column_nullability_by_table(inputs) ) column_types_by_table: dict[str, dict[str, str]] = _build_column_types_by_table(inputs) + dynamic_families_by_table: dict[str, tuple[SchemaDynamicColumnFamily, ...]] = ( + _build_dynamic_families_by_table(inputs) + ) complete_binding_schemas: dict[str, dict[str, str]] = _build_complete_binding_schemas(inputs) + dynamic_contract_analysis_inputs: _DynamicContractAnalysisInputs = ( + _DynamicContractAnalysisInputs( + families_by_table=dynamic_families_by_table, + authoritative_column_types_by_table=complete_binding_schemas, + ) + ) profile: ExpressionInferenceProfile = inference_profile or ExpressionInferenceProfile() allow_compact_analysis: bool = column_lineage_mode == ColumnLineageMode.RICH analysis_cache: AnalysisCacheContext | None = ( @@ -256,6 +277,7 @@ def assemble_compiled_project( seed_names=seed_names, column_nullability_by_table=column_nullability_by_table, column_types_by_table=column_types_by_table, + dynamic_contract_analysis_inputs=dynamic_contract_analysis_inputs, inference_profile=profile, sql_analysis=model_sql_analysis_by_name.get(model_input.model_file.file_path.stem), allow_compact_analysis=allow_compact_analysis, @@ -335,6 +357,7 @@ def _assemble_compiled_model( seed_names: frozenset[str] = frozenset(), column_nullability_by_table: dict[str, dict[str, InferredNullability]] | None = None, column_types_by_table: dict[str, dict[str, str]] | None = None, + dynamic_contract_analysis_inputs: _DynamicContractAnalysisInputs | None = None, inference_profile: ExpressionInferenceProfile | None = None, sql_analysis: _ModelSqlAnalysis | None = None, allow_compact_analysis: bool = False, @@ -402,6 +425,35 @@ def _assemble_compiled_model( placeholders=placeholders, dialect=profile.sql_analysis_dialect, ) + dynamic_column_contract: DynamicColumnContractProof | None = analyze_dynamic_column_contract( + query_sql=( + substitute_placeholder_defaults( + query_sql=analysis_query_sql, + placeholders=placeholders, + ) + if placeholders + else analysis_query_sql + ), + dialect=profile.sql_analysis_dialect, + families=( + model_input.schema_entry.dynamic_columns if model_input.schema_entry is not None else () + ), + column_types_by_table=column_types_by_table or {}, + authoritative_column_types_by_table=( + dynamic_contract_analysis_inputs.authoritative_column_types_by_table + if dynamic_contract_analysis_inputs is not None + else {} + ), + column_nullability_by_table=column_nullability_by_table or {}, + dynamic_families_by_table=( + dynamic_contract_analysis_inputs.families_by_table + if dynamic_contract_analysis_inputs is not None + else {} + ), + ) + if dynamic_column_contract is not None and dynamic_column_contract.output_proven: + inferred_columns = dynamic_column_contract.fixed_columns + fast_lineage_has_star = True return CompiledModel( key=CompiledObjectKey(resource_type=CompiledResourceType.MODEL, name=model_name), deps=model_build_deps(references=model_input.references, seed_names=seed_names), @@ -428,6 +480,7 @@ def _assemble_compiled_model( diagnostics=(polyglot_analysis.binding_diagnostics if sql_analysis_enabled else ()), ), binding_validated=(polyglot_analysis.binding_validated if sql_analysis_enabled else False), + dynamic_column_contract=dynamic_column_contract, ) @@ -955,6 +1008,18 @@ def _build_column_types_by_table(inputs: CompileProjectInputs) -> dict[str, dict return facts +def _build_dynamic_families_by_table( + inputs: CompileProjectInputs, +) -> dict[str, tuple[SchemaDynamicColumnFamily, ...]]: + return { + model_input.schema_entry.name: model_input.schema_entry.dynamic_columns + for model_input in inputs.model_inputs + if model_input.config.values.get("contract") == ContractPolicy.ENFORCED + and model_input.schema_entry is not None + and model_input.schema_entry.dynamic_columns + } + + def _build_complete_binding_schemas( inputs: CompileProjectInputs, ) -> dict[str, dict[str, str]]: @@ -966,6 +1031,7 @@ def _build_complete_binding_schemas( model_input.config.values.get("contract") == ContractPolicy.ENFORCED and model_input.schema_entry is not None and model_input.schema_entry.columns + and not model_input.schema_entry.dynamic_columns ): schemas[_model_name(model_input)] = _typed_schema_columns( model_input.schema_entry.columns diff --git a/src/sqlbuild/compiler/compile/_helpers/attachment/core.py b/src/sqlbuild/compiler/compile/_helpers/attachment/core.py index b659fa6383..9f391f4f09 100644 --- a/src/sqlbuild/compiler/compile/_helpers/attachment/core.py +++ b/src/sqlbuild/compiler/compile/_helpers/attachment/core.py @@ -27,6 +27,9 @@ merge_validated_model_audits, parse_model_header_audit_factories, ) +from sqlbuild.compiler.compile._helpers.config.dynamic_columns import ( + parse_dynamic_column_families, +) from sqlbuild.compiler.compile._helpers.config.model_validation import ( validate_contract_config, validate_custom_materialization_config, @@ -132,6 +135,7 @@ ProjectConfig, SchemaAuditInstance, SchemaColumn, + SchemaDynamicColumnFamily, SchemaModelEntry, SchemaSeedEntry, SettingsConfig, @@ -1490,11 +1494,13 @@ def build_model_header_schema_entry( raw_description: object | None = model_header_values.get("description") raw_columns: object | None = model_header_values.get("columns") + raw_dynamic_columns: object | None = model_header_values.get("dynamic_columns") raw_audits: object | None = model_header_values.get("audits") raw_audit_factories: object | None = model_header_values.get("audit_factories") if ( raw_description is None and raw_columns is None + and raw_dynamic_columns is None and raw_audits is None and raw_audit_factories is None and model_schema_columns is None @@ -1519,6 +1525,11 @@ def build_model_header_schema_entry( model_schema_columns=model_schema_columns, local_columns=local_columns, ) + dynamic_columns: tuple[SchemaDynamicColumnFamily, ...] = parse_dynamic_column_families( + raw_value=raw_dynamic_columns, + model_name=model_name, + file_path=file_path, + ) audits: tuple[SchemaAuditInstance, ...] = _parse_model_header_audits( raw_audits=raw_audits, file_path=file_path, @@ -1537,7 +1548,7 @@ def build_model_header_schema_entry( file_path=file_path, ) type_enforcement: bool | None = ( - True if any(column.type is not None for column in columns) else None + True if any(column.type is not None for column in columns) or dynamic_columns else None ) return SchemaModelEntry( name=model_name, @@ -1545,6 +1556,7 @@ def build_model_header_schema_entry( description=description, type_enforcement=type_enforcement, columns=columns, + dynamic_columns=dynamic_columns, audits=audits, ) diff --git a/src/sqlbuild/compiler/compile/_helpers/config/dynamic_columns.py b/src/sqlbuild/compiler/compile/_helpers/config/dynamic_columns.py new file mode 100644 index 0000000000..620664367a --- /dev/null +++ b/src/sqlbuild/compiler/compile/_helpers/config/dynamic_columns.py @@ -0,0 +1,88 @@ +"""Dynamic column-family configuration parsing for model headers.""" + +from __future__ import annotations + +import re +from pathlib import Path +from typing import cast + +from sqlbuild.compiler.compile.exceptions import CompileInputError +from sqlbuild.spec.contracts.models import SchemaDynamicColumnFamily + + +def parse_dynamic_column_families( + *, raw_value: object | None, model_name: str, file_path: Path +) -> tuple[SchemaDynamicColumnFamily, ...]: + if raw_value is None: + return () + if not isinstance(raw_value, dict) or not raw_value: + raise CompileInputError( + f"model '{model_name}' dynamic_columns must be a non-empty named block in {file_path}" + ) + allowed_keys: frozenset[str] = frozenset( + {"pivot_column", "value_column", "aggregate", "type", "name_pattern"} + ) + families: list[SchemaDynamicColumnFamily] = [] + for family_name, raw_family in raw_value.items(): + if not isinstance(family_name, str) or not family_name: + raise CompileInputError( + f"model '{model_name}' dynamic column family names must be non-empty strings" + ) + if not isinstance(raw_family, dict): + raise CompileInputError( + f"model '{model_name}' dynamic column family '{family_name}' must be a block" + ) + family_values: dict[str, object] = {str(key): value for key, value in raw_family.items()} + unknown_keys: set[str] = set(family_values) - allowed_keys + if unknown_keys: + raise CompileInputError( + f"model '{model_name}' dynamic column family '{family_name}' has unsupported keys: " + f"{', '.join(sorted(unknown_keys))}" + ) + pivot_column: object = family_values.get("pivot_column") + value_column: object = family_values.get("value_column") + aggregate: object = family_values.get("aggregate") + data_type: object = family_values.get("type") + for key, value in ( + ("pivot_column", pivot_column), + ("value_column", value_column), + ("aggregate", aggregate), + ("type", data_type), + ): + if not isinstance(value, str) or not value.strip(): + raise CompileInputError( + f"model '{model_name}' dynamic column family '{family_name}' requires a " + f"non-empty {key}" + ) + name_pattern: object = family_values.get("name_pattern") + if name_pattern is not None and ( + not isinstance(name_pattern, str) or not name_pattern.strip() + ): + raise CompileInputError( + f"model '{model_name}' dynamic column family '{family_name}' name_pattern must be " + "a non-empty regular expression" + ) + if isinstance(name_pattern, str): + try: + re.compile(name_pattern) + except re.error as error: + raise CompileInputError( + f"model '{model_name}' dynamic column family '{family_name}' has invalid " + f"name_pattern: {error}" + ) from error + families.append( + SchemaDynamicColumnFamily( + name=family_name, + pivot_column=cast(str, pivot_column), + value_column=cast(str, value_column), + aggregate=cast(str, aggregate).upper(), + type=cast(str, data_type), + name_pattern=name_pattern, + ) + ) + if len(families) > 1 and any(family.name_pattern is None for family in families): + raise CompileInputError( + f"model '{model_name}' declares multiple dynamic column families; each family requires " + "name_pattern so runtime columns can be classified unambiguously" + ) + return tuple(families) diff --git a/src/sqlbuild/compiler/compile/constants.py b/src/sqlbuild/compiler/compile/constants.py index 761594c3e2..6e90fd0463 100644 --- a/src/sqlbuild/compiler/compile/constants.py +++ b/src/sqlbuild/compiler/compile/constants.py @@ -28,6 +28,7 @@ { "description", "columns", + "dynamic_columns", "model_schema", "audits", "audit_factories", diff --git a/src/sqlbuild/compiler/compile/models.py b/src/sqlbuild/compiler/compile/models.py index f0fabd96cc..cdf94a22da 100644 --- a/src/sqlbuild/compiler/compile/models.py +++ b/src/sqlbuild/compiler/compile/models.py @@ -144,6 +144,25 @@ class InferredColumn: nullability: InferredNullability = InferredNullability.UNKNOWN +@dataclass(frozen=True) +class DynamicColumnFamilyProof: + """Compiler evidence for one declared runtime-generated column family.""" + + name: str + inferred_type: str | None = None + + +@dataclass(frozen=True) +class DynamicColumnContractProof: + """Closed output-shape evidence for a dynamic column contract.""" + + output_proven: bool + fixed_columns: tuple[InferredColumn, ...] = field(default_factory=tuple) + families: tuple[DynamicColumnFamilyProof, ...] = field(default_factory=tuple) + failure_reason: str | None = None + bare_dynamic_pivot: bool = False + + @dataclass(frozen=True) class CteFactResolvers: """Expression resolvers used for conservative CTE fact recovery.""" @@ -740,6 +759,7 @@ class CompiledModel: enum_columns: dict[str, EnumDeclaration] = field(default_factory=dict) binding_diagnostics: tuple[CompilerDiagnostic, ...] = field(default_factory=tuple) binding_validated: bool = False + dynamic_column_contract: DynamicColumnContractProof | None = None @dataclass(frozen=True) diff --git a/src/sqlbuild/compiler/contract_adoption/_helpers/compare.py b/src/sqlbuild/compiler/contract_adoption/_helpers/compare.py index 55c8997968..56fc1845ec 100644 --- a/src/sqlbuild/compiler/contract_adoption/_helpers/compare.py +++ b/src/sqlbuild/compiler/contract_adoption/_helpers/compare.py @@ -11,7 +11,8 @@ from sqlbuild.compiler.compile.types import CompiledResourceType from sqlbuild.compiler.contract_adoption.models import ContractEvidence, ContractFinding from sqlbuild.compiler.contract_adoption.types import ContractFindingKind -from sqlbuild.spec.contracts.models import SourceEntry +from sqlbuild.spec.contracts.main.matching_dynamic_families import matching_dynamic_families +from sqlbuild.spec.contracts.models import SchemaDynamicColumnFamily, SourceEntry def model_evidence(*, model: CompiledModel) -> ContractEvidence: @@ -27,6 +28,9 @@ def model_evidence(*, model: CompiledModel) -> ContractEvidence: ), physical_columns=None, source_path=model.relative_path, + dynamic_columns=( + model.schema_entry.dynamic_columns if model.schema_entry is not None else () + ), ) @@ -85,7 +89,33 @@ def compare_evidence( } for physical in physical_columns: declared: ColumnInfo | None = declared_by_name.get(physical.name.casefold()) - if declared is None or not declared.type: + if declared is None: + matching: tuple[SchemaDynamicColumnFamily, ...] = matching_dynamic_families( + families=item.dynamic_columns, + column_name=physical.name, + ) + if len(matching) == 1: + family: SchemaDynamicColumnFamily = matching[0] + if not types_equal( + left=family.type, + right=physical.type, + dialect=adapter.sql_analysis_dialect(), + ): + findings.append( + ContractFinding( + resource_type=item.resource_type, + resource_name=item.resource_name, + kind=ContractFindingKind.TYPE_MISMATCH, + column_name=physical.name, + declared_type=family.type, + physical_type=physical.type, + message=( + f"dynamic column '{physical.name}' is physically {physical.type} " + f"but family '{family.name}' declares {family.type}" + ), + ) + ) + continue findings.append( ContractFinding( resource_type=item.resource_type, @@ -95,9 +125,20 @@ def compare_evidence( physical_type=physical.type, message=( f"physical column '{physical.name}' ({physical.type}) is not declared" - if declared is None - else f"declared column '{declared.name}' has no type; " - f"physical type is {physical.type}" + ), + ) + ) + elif not declared.type: + findings.append( + ContractFinding( + resource_type=item.resource_type, + resource_name=item.resource_name, + kind=ContractFindingKind.MISSING_DECLARATION, + column_name=physical.name, + physical_type=physical.type, + message=( + f"declared column '{declared.name}' has no type; physical type is " + f"{physical.type}" ), ) ) diff --git a/src/sqlbuild/compiler/contract_adoption/_helpers/repository_edits.py b/src/sqlbuild/compiler/contract_adoption/_helpers/repository_edits.py index 8d4fe04605..6052335d1b 100644 --- a/src/sqlbuild/compiler/contract_adoption/_helpers/repository_edits.py +++ b/src/sqlbuild/compiler/contract_adoption/_helpers/repository_edits.py @@ -13,6 +13,7 @@ get_model_header_spans, ) from sqlbuild.compiler.discovery.models import ModelHeaderColumnSpan, ModelHeaderSpans +from sqlbuild.spec.contracts.main.matching_dynamic_families import matching_dynamic_families from sqlbuild.spec.contracts.models import SchemaColumn, SourceColumnEntry _HEADER_WHITESPACE: str = " \t" @@ -46,6 +47,19 @@ def edit_model( for column in model.schema_entry.columns ): return None, "effective columns are owned by an external schema file" + if model.schema_entry is not None and model.schema_entry.dynamic_columns: + fixed_names: frozenset[str] = frozenset( + column.name.casefold() for column in model.schema_entry.columns + ) + physical_columns = tuple( + column + for column in physical_columns + if column.name.casefold() in fixed_names + or not matching_dynamic_families( + families=model.schema_entry.dynamic_columns, + column_name=column.name, + ) + ) contents: str = model.authored_sql header_spans: ModelHeaderSpans = get_model_header_spans(contents=contents) span_result: tuple[int, int, dict[str, ModelHeaderColumnSpan]] | None = header_spans.columns diff --git a/src/sqlbuild/compiler/contract_adoption/models.py b/src/sqlbuild/compiler/contract_adoption/models.py index 6f6f018ee0..2600c8c5d4 100644 --- a/src/sqlbuild/compiler/contract_adoption/models.py +++ b/src/sqlbuild/compiler/contract_adoption/models.py @@ -8,6 +8,7 @@ from sqlbuild.adapter.contract.models import ColumnInfo from sqlbuild.compiler.compile.types import CompiledResourceType from sqlbuild.compiler.contract_adoption.types import ContractFindingKind +from sqlbuild.spec.contracts.models import SchemaDynamicColumnFamily @dataclass(frozen=True) @@ -31,6 +32,7 @@ class ContractEvidence: declared_columns: tuple[ColumnInfo, ...] physical_columns: tuple[ColumnInfo, ...] | None source_path: Path | None + dynamic_columns: tuple[SchemaDynamicColumnFamily, ...] = field(default_factory=tuple) findings: tuple[ContractFinding, ...] = field(default_factory=tuple) diff --git a/src/sqlbuild/compiler/contracts/_helpers/columns.py b/src/sqlbuild/compiler/contracts/_helpers/columns.py index 1e642b78f4..2503679358 100644 --- a/src/sqlbuild/compiler/contracts/_helpers/columns.py +++ b/src/sqlbuild/compiler/contracts/_helpers/columns.py @@ -11,6 +11,8 @@ from sqlbuild.compiler.compile.models import ( CompiledModel, CompilerDiagnostic, + DynamicColumnContractProof, + DynamicColumnFamilyProof, InferredColumn, RelatedLocation, ) @@ -24,7 +26,11 @@ ) from sqlbuild.compiler.lineage.types import InferredNullability from sqlbuild.compiler.planner.types import ContractPolicy -from sqlbuild.spec.contracts.models import SchemaColumn, SourceLocation +from sqlbuild.spec.contracts.models import ( + SchemaColumn, + SchemaDynamicColumnFamily, + SourceLocation, +) _MISSING_COLUMN_CODE: str = "K001" _TYPE_MISMATCH_CODE: str = "K002" @@ -32,6 +38,7 @@ _NULLABILITY_MISMATCH_CODE: str = "K004" _EXTRA_COLUMN_CODE: str = "K005" _MISSING_DECLARATIONS_CODE: str = "K006" +_DYNAMIC_OUTPUT_NOT_PROVEN_CODE: str = "K011" def collect_model_column_contract_diagnostics( @@ -43,17 +50,20 @@ def collect_model_column_contract_diagnostics( """Collect diagnostics for one compiled model's declared column contract.""" contract_enforced: bool = model.config.values.get("contract") == ContractPolicy.ENFORCED - if model.schema_entry is None or not model.schema_entry.columns: + if model.schema_entry is None or not ( + model.schema_entry.columns or model.schema_entry.dynamic_columns + ): if contract_enforced: return (_missing_declarations_diagnostic(model),) return () + diagnostics: list[CompilerDiagnostic] = [] + diagnostics.extend(_dynamic_column_diagnostics(model=model, dialect=dialect)) if model.inferred_columns is None: - return () + return tuple(diagnostics) inferred_by_name: dict[str, InferredColumn] = { column.name: column for column in model.inferred_columns } - diagnostics: list[CompilerDiagnostic] = [] if contract_enforced: diagnostics.extend(_extra_column_diagnostics(model=model)) declared_column: SchemaColumn @@ -96,6 +106,82 @@ def collect_model_column_contract_diagnostics( return tuple(diagnostics) +def _dynamic_column_diagnostics( + *, model: CompiledModel, dialect: TypeDialect | str | None +) -> tuple[CompilerDiagnostic, ...]: + if model.schema_entry is None or not model.schema_entry.dynamic_columns: + return () + proof: DynamicColumnContractProof | None = model.dynamic_column_contract + if proof is None or not proof.output_proven: + failure_reason: str = ( + proof.failure_reason + if proof is not None and proof.failure_reason + else "no compiler evidence" + ) + return ( + CompilerDiagnostic( + phase=DiagnosticPhase.CONTRACT, + severity=DiagnosticSeverity.ERROR, + code=_DYNAMIC_OUTPUT_NOT_PROVEN_CODE, + message=( + f"model '{model.name}' dynamic column contract is not proven: {failure_reason}" + ), + resource_type=CompiledResourceType.MODEL, + resource_name=model.name, + path=model.relative_path, + help=( + "project one supported runtime-dynamic pivot through a wildcard output " + "boundary " + "and declare every generated family" + ), + ), + ) + proof_by_name: dict[str, DynamicColumnFamilyProof] = { + family.name.casefold(): family for family in proof.families + } + diagnostics: list[CompilerDiagnostic] = [] + family: SchemaDynamicColumnFamily + for family in model.schema_entry.dynamic_columns: + family_proof: DynamicColumnFamilyProof | None = proof_by_name.get(family.name.casefold()) + inferred_type: str | None = family_proof.inferred_type if family_proof is not None else None + if inferred_type is None: + diagnostics.append( + CompilerDiagnostic( + phase=DiagnosticPhase.CONTRACT, + severity=DiagnosticSeverity.WARNING, + code=_UNKNOWN_TYPE_CODE, + message=( + f"dynamic column family '{family.name}' type could not be proven against " + f"declared {family.type}" + ), + resource_type=CompiledResourceType.MODEL, + resource_name=model.name, + path=model.relative_path, + help=( + "establish an authoritative type on the pivot value column or CAST the " + "type-preserving aggregate input explicitly" + ), + ) + ) + elif not types_equal(left=family.type, right=inferred_type, dialect=dialect): + diagnostics.append( + CompilerDiagnostic( + phase=DiagnosticPhase.CONTRACT, + severity=DiagnosticSeverity.ERROR, + code=_TYPE_MISMATCH_CODE, + message=( + f"dynamic column family '{family.name}' inferred as {inferred_type} " + f"but declared type is {family.type}" + ), + resource_type=CompiledResourceType.MODEL, + resource_name=model.name, + path=model.relative_path, + help="correct the family type or the pivot aggregate input type", + ) + ) + return tuple(diagnostics) + + def _missing_declarations_diagnostic(model: CompiledModel) -> CompilerDiagnostic: return CompilerDiagnostic( phase=DiagnosticPhase.CONTRACT, @@ -105,7 +191,10 @@ def _missing_declarations_diagnostic(model: CompiledModel) -> CompilerDiagnostic resource_type=CompiledResourceType.MODEL, resource_name=model.name, path=model.relative_path, - help="add MODEL(columns (...)) or set contract none for this model", + help=( + "add MODEL(columns (...)), declare a proven MODEL(dynamic_columns (...)) family, " + "or set contract none for this model" + ), ) diff --git a/src/sqlbuild/compiler/manifest/_helpers/model_nodes.py b/src/sqlbuild/compiler/manifest/_helpers/model_nodes.py index c965027bfe..2ded851ef4 100644 --- a/src/sqlbuild/compiler/manifest/_helpers/model_nodes.py +++ b/src/sqlbuild/compiler/manifest/_helpers/model_nodes.py @@ -122,6 +122,29 @@ def _extract_meta( sqlbuild_meta["authored_value"] = authored_sqlbuild_meta sqlbuild_meta["lifecycle_hooks"] = lifecycle_hooks meta["sqlbuild"] = sqlbuild_meta + if model.schema_entry is not None and model.schema_entry.dynamic_columns: + authored_sqlbuild_meta = meta.get("sqlbuild") + sqlbuild_meta: dict[str, object] = {} + if isinstance(authored_sqlbuild_meta, dict): + authored_key: object + authored_value: object + for authored_key, authored_value in authored_sqlbuild_meta.items(): + if isinstance(authored_key, str): + sqlbuild_meta[authored_key] = authored_value + if authored_sqlbuild_meta is not None and not isinstance(authored_sqlbuild_meta, dict): + sqlbuild_meta["authored_value"] = authored_sqlbuild_meta + sqlbuild_meta["dynamic_columns"] = [ + { + "name": family.name, + "pivot_column": family.pivot_column, + "value_column": family.value_column, + "aggregate": family.aggregate, + "type": family.type, + "name_pattern": family.name_pattern, + } + for family in model.schema_entry.dynamic_columns + ] + meta["sqlbuild"] = sqlbuild_meta return meta diff --git a/src/sqlbuild/compiler/planner/_helpers/changes/detect.py b/src/sqlbuild/compiler/planner/_helpers/changes/detect.py index 0c5a9fc5ef..952ffd0c1b 100644 --- a/src/sqlbuild/compiler/planner/_helpers/changes/detect.py +++ b/src/sqlbuild/compiler/planner/_helpers/changes/detect.py @@ -211,6 +211,9 @@ def detect_model_changes( warehouse_columns=warehouse_columns, type_enforcement=type_enforcement, inferred_schema_complete=not model.fast_lineage_has_star, + dynamic_columns=( + model.schema_entry.dynamic_columns if model.schema_entry is not None else () + ), ) if schema_findings: raw_policy = get_config_str(values=model.config.values, key="replay_on_change") diff --git a/src/sqlbuild/compiler/planner/_helpers/changes/schema.py b/src/sqlbuild/compiler/planner/_helpers/changes/schema.py index 22bebe6242..90b1d831dc 100644 --- a/src/sqlbuild/compiler/planner/_helpers/changes/schema.py +++ b/src/sqlbuild/compiler/planner/_helpers/changes/schema.py @@ -7,6 +7,8 @@ from sqlbuild.compiler.compile.models import InferredColumn from sqlbuild.compiler.planner.models import SchemaFinding from sqlbuild.compiler.planner.types import SchemaChangeKind, SchemaColumnSource +from sqlbuild.spec.contracts.main.matching_dynamic_families import matching_dynamic_families +from sqlbuild.spec.contracts.models import SchemaDynamicColumnFamily def detect_schema_changes( @@ -16,6 +18,7 @@ def detect_schema_changes( warehouse_columns: tuple[ColumnInfo, ...], type_enforcement: bool, inferred_schema_complete: bool, + dynamic_columns: tuple[SchemaDynamicColumnFamily, ...] = (), ) -> tuple[SchemaFinding, ...]: """Compare yml and inferred columns against warehouse columns and return findings.""" @@ -65,6 +68,28 @@ def detect_schema_changes( for col_ne in yml_columns: seen_names.add(col_ne.name) + for column_name, column_type in warehouse_map.items(): + if column_name in seen_names: + continue + matching: tuple[SchemaDynamicColumnFamily, ...] = matching_dynamic_families( + families=dynamic_columns, + column_name=column_name, + ) + if len(matching) != 1: + continue + family: SchemaDynamicColumnFamily = matching[0] + seen_names.add(column_name) + if not types_equal(left=family.type, right=column_type, dialect=None): + findings.append( + SchemaFinding( + kind=SchemaChangeKind.COLUMN_TYPE_CHANGED, + column_name=column_name, + source=SchemaColumnSource.YML, + expected_type=family.type, + actual_type=column_type, + ) + ) + if type_enforcement or inferred_schema_complete: col_name: str col_type: str diff --git a/src/sqlbuild/compiler/planner/_helpers/identity/model_metadata.py b/src/sqlbuild/compiler/planner/_helpers/identity/model_metadata.py index 2f132c713f..3df39762f1 100644 --- a/src/sqlbuild/compiler/planner/_helpers/identity/model_metadata.py +++ b/src/sqlbuild/compiler/planner/_helpers/identity/model_metadata.py @@ -12,17 +12,30 @@ def contract_output_signature(*, model: CompiledModel) -> dict[str, object] | No """Build required model output metadata that participates in execution identity.""" schema_entry: SchemaModelEntry | None = model.schema_entry - if schema_entry is None or not schema_entry.columns: + if schema_entry is None or not (schema_entry.columns or schema_entry.dynamic_columns): return None enforced: bool = model.config.values.get("contract") == ContractPolicy.ENFORCED if not enforced and schema_entry.model_schema is None: return None - return { + signature: dict[str, object] = { "enforced": enforced, "columns": [ _column_output_signature(model=model, column=column) for column in schema_entry.columns ], } + if schema_entry.dynamic_columns: + signature["dynamic_columns"] = [ + { + "name": family.name, + "pivot_column": family.pivot_column, + "value_column": family.value_column, + "aggregate": family.aggregate, + "type": family.type, + "name_pattern": family.name_pattern, + } + for family in schema_entry.dynamic_columns + ] + return signature def _column_output_signature(*, model: CompiledModel, column: SchemaColumn) -> dict[str, object]: diff --git a/src/sqlbuild/compiler/planner/_helpers/output/plan_entry.py b/src/sqlbuild/compiler/planner/_helpers/output/plan_entry.py index d633b63a8a..c73d9649b3 100644 --- a/src/sqlbuild/compiler/planner/_helpers/output/plan_entry.py +++ b/src/sqlbuild/compiler/planner/_helpers/output/plan_entry.py @@ -133,6 +133,7 @@ LocalConfig, ProjectConfig, SchemaColumn, + SchemaDynamicColumnFamily, SourceEntry, StartCursorsConfig, ) @@ -678,6 +679,9 @@ def plan_model_from_change( declared_columns: tuple[ColumnInfo, ...] = _get_declared_columns(model) contract_enforced: bool = model.config.values.get("contract") == ContractPolicy.ENFORCED contract_columns: tuple[ColumnInfo, ...] = _get_contract_columns(model) + contract_dynamic_columns: tuple[SchemaDynamicColumnFamily, ...] = ( + model.schema_entry.dynamic_columns if model.schema_entry is not None else () + ) unique_key: tuple[str, ...] = _get_unique_key(model) warehouse_columns: tuple[ColumnInfo, ...] = snapshot.existing_columns.get(model.name, ()) @@ -881,6 +885,7 @@ def plan_model_from_change( declared_columns=declared_columns, contract_enforced=contract_enforced, contract_columns=contract_columns, + contract_dynamic_columns=contract_dynamic_columns, pre_hooks=pre_hooks, post_hooks=post_hooks, previous_query_sql=previous_query_sql, diff --git a/src/sqlbuild/compiler/planner/models.py b/src/sqlbuild/compiler/planner/models.py index e78974da28..f838b88f59 100644 --- a/src/sqlbuild/compiler/planner/models.py +++ b/src/sqlbuild/compiler/planner/models.py @@ -74,6 +74,7 @@ FutureCursorsConfig, LocalConfig, ProjectConfig, + SchemaDynamicColumnFamily, SeedCsvSettings, SourceEntry, StartCursorsConfig, @@ -1001,6 +1002,7 @@ class ModelPlanEntry: declared_columns: tuple[ColumnInfo, ...] = field(default_factory=tuple) contract_enforced: bool = False contract_columns: tuple[ColumnInfo, ...] = field(default_factory=tuple) + contract_dynamic_columns: tuple[SchemaDynamicColumnFamily, ...] = field(default_factory=tuple) pre_hooks: object = None post_hooks: object = None previous_query_sql: str | None = None diff --git a/src/sqlbuild/executor/run/_helpers/validation/contracts.py b/src/sqlbuild/executor/run/_helpers/validation/contracts.py index 7f0354fef2..b9a2dcfc4c 100644 --- a/src/sqlbuild/executor/run/_helpers/validation/contracts.py +++ b/src/sqlbuild/executor/run/_helpers/validation/contracts.py @@ -13,6 +13,8 @@ RUNTIME_CONTRACT_MISSING_DECLARATIONS_CODE, RUNTIME_CONTRACT_TYPE_MISMATCH_CODE, ) +from sqlbuild.spec.contracts.main.matching_dynamic_families import matching_dynamic_families +from sqlbuild.spec.contracts.models import SchemaDynamicColumnFamily def validate_runtime_contract( @@ -25,7 +27,7 @@ def validate_runtime_contract( if not entry.contract_enforced: return - if not entry.contract_columns: + if not entry.contract_columns and not entry.contract_dynamic_columns: raise ExecutorInputError( f"model '{entry.name}' has contract enforced but declares no columns", code=RUNTIME_CONTRACT_MISSING_DECLARATIONS_CODE, @@ -49,8 +51,19 @@ def validate_runtime_contract( code=RUNTIME_CONTRACT_MISSING_COLUMN_CODE, ) + extra_columns: tuple[ColumnInfo, ...] = tuple( + column for column in actual_columns if column.name.lower() not in declared_by_name + ) extra_names: tuple[str, ...] = tuple( - column.name for column in actual_columns if column.name.lower() not in declared_by_name + column.name + for column in extra_columns + if len( + matching_dynamic_families( + families=entry.contract_dynamic_columns, + column_name=column.name, + ) + ) + != 1 ) if extra_names: raise ExecutorInputError( @@ -58,9 +71,26 @@ def validate_runtime_contract( code=RUNTIME_CONTRACT_EXTRA_COLUMN_CODE, ) + for column in extra_columns: + matching: tuple[SchemaDynamicColumnFamily, ...] = matching_dynamic_families( + families=entry.contract_dynamic_columns, + column_name=column.name, + ) + if len(matching) != 1: + continue + family: SchemaDynamicColumnFamily = matching[0] + if not types_equal(left=family.type, right=column.type, dialect=dialect): + raise ExecutorInputError( + f"model '{entry.name}' runtime dynamic column '{column.name}' has type " + f"{column.type} but family '{family.name}' declares {family.type}", + code=RUNTIME_CONTRACT_TYPE_MISMATCH_CODE, + ) + actual_column: ColumnInfo for actual_column in actual_columns: - declared_column: ColumnInfo = declared_by_name[actual_column.name.lower()] + declared_column: ColumnInfo | None = declared_by_name.get(actual_column.name.lower()) + if declared_column is None: + continue if not declared_column.type or not actual_column.type: continue if types_equal(left=actual_column.type, right=declared_column.type, dialect=dialect): diff --git a/src/sqlbuild/lint/_helpers/expansion.py b/src/sqlbuild/lint/_helpers/expansion.py index 5b4c88c57b..68e76d1cf5 100644 --- a/src/sqlbuild/lint/_helpers/expansion.py +++ b/src/sqlbuild/lint/_helpers/expansion.py @@ -132,6 +132,7 @@ def prepare_lint_body( context: SqlExpansionContext, external_identifiers: tuple[str, ...] = (), allows_ceremonial_select: bool = False, + allows_dynamic_output_star: bool = False, ) -> LintBody: """Expand one authored body and neutralize whatever interpolation remains.""" @@ -189,6 +190,7 @@ def prepare_lint_body( dependency_identifiers=dependency_identifiers, externally_referenced_ctes=externally_referenced_ctes, allows_ceremonial_select=allows_ceremonial_select, + allows_dynamic_output_star=allows_dynamic_output_star, ) diff --git a/src/sqlbuild/lint/_helpers/native_sql.py b/src/sqlbuild/lint/_helpers/native_sql.py index 9bd8f1f974..da91aa2672 100644 --- a/src/sqlbuild/lint/_helpers/native_sql.py +++ b/src/sqlbuild/lint/_helpers/native_sql.py @@ -71,6 +71,7 @@ tuple[str, ...], tuple[str, ...], bool, + bool, ] type _NativeResult = dict[str, Any] | NativeLintError @@ -106,6 +107,8 @@ def run_native_sql_lint( payload["externally_referenced_ctes"] = list(body.externally_referenced_ctes) if body.allows_ceremonial_select: payload["allows_ceremonial_select"] = True + if body.allows_dynamic_output_star: + payload["allows_dynamic_output_star"] = True requests[cache_key] = payload response_cache: dict[_NativeCacheKey, _NativeResult] = _native_responses(requests=requests) @@ -145,6 +148,7 @@ def _native_cache_key(*, body: LintBody, config: LintConfig) -> _NativeCacheKey: body.dependency_identifiers, body.externally_referenced_ctes, body.allows_ceremonial_select, + body.allows_dynamic_output_star, ) diff --git a/src/sqlbuild/lint/main/run_lint.py b/src/sqlbuild/lint/main/run_lint.py index d189417b4b..afd632134c 100644 --- a/src/sqlbuild/lint/main/run_lint.py +++ b/src/sqlbuild/lint/main/run_lint.py @@ -29,6 +29,7 @@ def run_lint( value_renderer: TypedSqlValueRenderer | None = None, selected_paths: frozenset[Path] | None = None, discovered_inputs: DiscoveredProjectInputs | None = None, + dynamic_output_paths: frozenset[Path] = frozenset(), ) -> LintRunResult: """Lint all DSL files in the project without modifying anything.""" @@ -73,6 +74,7 @@ def run_lint( headers=headers, context=context, project_dir=project_dir, + allows_dynamic_output_star=file_path.resolve() in dynamic_output_paths, ) ) @@ -117,6 +119,7 @@ def _prepared_bodies( headers: tuple[HeaderSpan, ...], context: SqlExpansionContext, project_dir: Path, + allows_dynamic_output_star: bool, ) -> tuple[LintBody, ...]: bodies: list[LintBody] = [] external_identifiers: tuple[str, ...] = external_identifiers_for_headers( @@ -143,6 +146,7 @@ def _prepared_bodies( context=context, external_identifiers=external_identifiers, allows_ceremonial_select=allows_ceremonial_select, + allows_dynamic_output_star=allows_dynamic_output_star, ) ) return tuple(bodies) diff --git a/src/sqlbuild/lint/models.py b/src/sqlbuild/lint/models.py index 0f2a5b12a1..a611835cf0 100644 --- a/src/sqlbuild/lint/models.py +++ b/src/sqlbuild/lint/models.py @@ -45,6 +45,7 @@ class LintBody: dependency_identifiers: tuple[str, ...] = () externally_referenced_ctes: tuple[str, ...] = () allows_ceremonial_select: bool = False + allows_dynamic_output_star: bool = False @dataclass(frozen=True) diff --git a/src/sqlbuild/rule_engine/_helpers/engine/native.py b/src/sqlbuild/rule_engine/_helpers/engine/native.py index bf8a3f8e18..4eaa3b15cc 100644 --- a/src/sqlbuild/rule_engine/_helpers/engine/native.py +++ b/src/sqlbuild/rule_engine/_helpers/engine/native.py @@ -23,6 +23,7 @@ CompiledProject, CompiledSqlScenario, CompiledSqlTest, + DynamicColumnContractProof, InferredColumn, ) from sqlbuild.compiler.compile.types import SqlTestMode @@ -382,6 +383,37 @@ def _model_payload( ) ) columns.append(payload) + dynamic_columns: list[dict[str, object]] = [] + dynamic_columns_proven: bool = False + if model.schema_entry is not None and model.schema_entry.dynamic_columns: + proof: DynamicColumnContractProof | None = model.dynamic_column_contract + dynamic_columns_proven = bool(proof is not None and proof.output_proven) + proof_types: dict[str, str | None] = ( + {family.name.casefold(): family.inferred_type for family in proof.families} + if proof is not None + else {} + ) + for family in model.schema_entry.dynamic_columns: + inferred_type: str | None = proof_types.get(family.name.casefold()) + dynamic_columns.append( + { + "name": family.name, + "pivot_column": family.pivot_column, + "value_column": family.value_column, + "aggregate": family.aggregate, + "type": family.type, + "name_pattern": family.name_pattern, + "type_proven": bool( + include_type_proof + and inferred_type + and types_equal( + left=family.type, + right=inferred_type, + dialect=dialect, + ) + ), + } + ) return { "name": model.name, "relative_path": model.relative_path.as_posix(), @@ -397,6 +429,12 @@ def _model_payload( for reference in model.references ], "columns": columns, + "dynamic_columns": dynamic_columns, + "dynamic_columns_proven": dynamic_columns_proven, + "bare_dynamic_pivot": bool( + model.dynamic_column_contract is not None + and model.dynamic_column_contract.bare_dynamic_pivot + ), "enum_columns": list(model.enum_columns), "enum_declarations": [ _enum_payload(declaration) for declaration in model.enum_declarations diff --git a/src/sqlbuild/rule_engine/_helpers/run/rules.py b/src/sqlbuild/rule_engine/_helpers/run/rules.py index 63cd79e122..b03fb82f8b 100644 --- a/src/sqlbuild/rule_engine/_helpers/run/rules.py +++ b/src/sqlbuild/rule_engine/_helpers/run/rules.py @@ -6,7 +6,7 @@ import json import os import time -from dataclasses import dataclass, replace +from dataclasses import asdict, dataclass, replace from importlib.metadata import version from pathlib import Path from typing import Any @@ -25,7 +25,7 @@ from sqlbuild.rule_engine.exceptions import RulesError from sqlbuild.rule_engine.models import Finding, Rule, RulesConfig, RulesResult, RulesRunResult -_SQL_RULE_CACHE_VERSION: str = "sql-rules-v1" +_SQL_RULE_CACHE_VERSION: str = "sql-rules-v2" _SQLBUILD_VERSION: str = version("sqlbuild") _SQL_RULE_SUPPRESSION_CODE: str = "SQBRSQL000" @@ -195,6 +195,12 @@ def _run_sql_rules( config=LintConfig(dialect=dialect, enabled_native_rules=codes), selected_paths=frozenset(selected_paths), discovered_inputs=discovered_inputs, + dynamic_output_paths=frozenset( + (project_dir / model.relative_path).resolve() + for model in models_by_path.values() + if model.dynamic_column_contract is not None + and model.dynamic_column_contract.output_proven + ), ) if selected_paths else None @@ -247,6 +253,15 @@ def _sql_rule_identity(*, model: CompiledModel, codes: tuple[str, ...], dialect: digest.update("\0".join(codes).encode()) digest.update(model.authored_sql.encode()) digest.update(model.query_sql.encode()) + if model.dynamic_column_contract is not None: + digest.update( + json.dumps( + asdict(model.dynamic_column_contract), + sort_keys=True, + separators=(",", ":"), + default=str, + ).encode() + ) return digest.hexdigest() diff --git a/src/sqlbuild/spec/contracts/_helpers/dynamic_columns.py b/src/sqlbuild/spec/contracts/_helpers/dynamic_columns.py new file mode 100644 index 0000000000..c7c053acf7 --- /dev/null +++ b/src/sqlbuild/spec/contracts/_helpers/dynamic_columns.py @@ -0,0 +1,27 @@ +"""Shared matching helpers for dynamic column contract families.""" + +from __future__ import annotations + +import re + +from sqlbuild.spec.contracts.models import SchemaDynamicColumnFamily + + +def dynamic_family_accepts_name(*, family: SchemaDynamicColumnFamily, column_name: str) -> bool: + """Return whether a runtime column name belongs to a declared family.""" + + if family.name_pattern is None: + return True + return re.fullmatch(family.name_pattern, column_name) is not None + + +def matching_dynamic_families( + *, families: tuple[SchemaDynamicColumnFamily, ...], column_name: str +) -> tuple[SchemaDynamicColumnFamily, ...]: + """Return every family whose optional naming constraint accepts a column.""" + + return tuple( + family + for family in families + if dynamic_family_accepts_name(family=family, column_name=column_name) + ) diff --git a/src/sqlbuild/spec/contracts/main/matching_dynamic_families.py b/src/sqlbuild/spec/contracts/main/matching_dynamic_families.py new file mode 100644 index 0000000000..7f1823b328 --- /dev/null +++ b/src/sqlbuild/spec/contracts/main/matching_dynamic_families.py @@ -0,0 +1,12 @@ +"""Public compiler entry point for dynamic-family name matching.""" + +from sqlbuild.spec.contracts._helpers.dynamic_columns import matching_dynamic_families as _match +from sqlbuild.spec.contracts.models import SchemaDynamicColumnFamily + + +def matching_dynamic_families( + *, families: tuple[SchemaDynamicColumnFamily, ...], column_name: str +) -> tuple[SchemaDynamicColumnFamily, ...]: + """Return declared dynamic families accepting a physical column name.""" + + return _match(families=families, column_name=column_name) diff --git a/src/sqlbuild/spec/contracts/models.py b/src/sqlbuild/spec/contracts/models.py index 6e4c5bee93..af8ec6e662 100644 --- a/src/sqlbuild/spec/contracts/models.py +++ b/src/sqlbuild/spec/contracts/models.py @@ -475,6 +475,18 @@ class SchemaColumn: location: SourceLocation | None = None +@dataclass(frozen=True) +class SchemaDynamicColumnFamily: + """One declared family of columns generated by a runtime-dynamic pivot.""" + + name: str + pivot_column: str + value_column: str + aggregate: str + type: str + name_pattern: str | None = None + + @dataclass(frozen=True) class SchemaModelEntry: """One model metadata entry normalized from MODEL(...).""" @@ -485,6 +497,7 @@ class SchemaModelEntry: type_enforcement: bool | None = None meta: dict[str, object] = field(default_factory=dict) columns: tuple[SchemaColumn, ...] = field(default_factory=tuple) + dynamic_columns: tuple[SchemaDynamicColumnFamily, ...] = field(default_factory=tuple) audits: tuple[SchemaAuditInstance, ...] = field(default_factory=tuple) tags: tuple[str, ...] = field(default_factory=tuple) diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/build/_test_types.py b/tests/e2e/src/sqlbuild/cli/commands/main/build/_test_types.py index 86ec75d9d1..2c0ff1aba7 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/build/_test_types.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/build/_test_types.py @@ -1046,3 +1046,10 @@ class RemoveColumnSemanticsBuildE2ETestCase: expected_backfill_action: str expected_backfill_duration: str | None expected_warning_count: int + + +@dataclass(frozen=True) +class DynamicColumnContractE2ETestCase: + description: str + project_name: str + expected_rows: tuple[tuple[object, ...], ...] diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/build/test_dynamic_column_contract.py b/tests/e2e/src/sqlbuild/cli/commands/main/build/test_dynamic_column_contract.py new file mode 100644 index 0000000000..fefccc6727 --- /dev/null +++ b/tests/e2e/src/sqlbuild/cli/commands/main/build/test_dynamic_column_contract.py @@ -0,0 +1,103 @@ +"""E2E coverage for governed runtime-dynamic pivot columns.""" + +from __future__ import annotations + +import subprocess +from pathlib import Path + +import pytest + +from tests.e2e.src.sqlbuild.cli.commands.main.build._test_types import ( + DynamicColumnContractE2ETestCase, +) +from tests.e2e.src.sqlbuild.cli.commands.shared.helpers import ( + prepare_inline_project, + query_duckdb, + run_sqb, +) + + +@pytest.mark.parametrize( + "test_case", + ( + DynamicColumnContractE2ETestCase( + description="generated category columns are enforced", + project_name="dynamic_category_amounts", + expected_rows=((1, "10.25", "20.50"), (2, "7.00", None)), + ), + ), + ids=lambda case: case.description, +) +def test_given_dynamic_pivot_contract_when_building_then_generated_columns_are_type_checked( + test_case: DynamicColumnContractE2ETestCase, + tmp_path: Path, +) -> None: + project_dir: Path = prepare_inline_project( + tmp_path=tmp_path, + project_name=test_case.project_name, + repo_files={ + "sqlbuild_project.toml": ( + 'name = "dynamic_category_amounts"\n' + 'adapter = "duckdb"\n\n' + "[connection]\n" + 'database = "warehouse.duckdb"\n\n' + "[defaults]\n" + 'contract = "enforced"\n\n' + "[rules]\n" + 'select = ["SQBRMODEL102", "SQBRCONTRACT101", "SQBRCONTRACT105", ' + '"SQBRCONTRACT106"]\n' + ), + "models/stg_order_amounts.sql": ( + "MODEL (\n" + " materialized table,\n" + " columns (\n" + " customer_id (type INTEGER),\n" + " category (type VARCHAR),\n" + ' amount (type "DECIMAL(12,2)"),\n' + " ),\n" + ");\n" + "SELECT CAST(1 AS INTEGER) AS customer_id, " + "CAST('books' AS VARCHAR) AS category, " + "CAST(10.25 AS DECIMAL(12,2)) AS amount\n" + "UNION ALL\n" + "SELECT CAST(1 AS INTEGER), CAST('games' AS VARCHAR), " + "CAST(20.50 AS DECIMAL(12,2))\n" + "UNION ALL\n" + "SELECT CAST(2 AS INTEGER), CAST('books' AS VARCHAR), " + "CAST(7.00 AS DECIMAL(12,2))\n" + ), + "models/customer_category_amounts.sql": ( + "MODEL (\n" + " materialized table,\n" + " columns (customer_id (type INTEGER)),\n" + " dynamic_columns (\n" + " category_amounts (\n" + " pivot_column category,\n" + " value_column amount,\n" + " aggregate MAX,\n" + ' type "DECIMAL(12,2)"\n' + " )\n" + " ),\n" + ");\n" + 'PIVOT __ref("stg_order_amounts")\n' + "ON category\n" + "USING MAX(amount)\n" + "GROUP BY customer_id\n" + ), + }, + ) + + result: subprocess.CompletedProcess[str] = run_sqb( + command=("--no-color", "build"), + project_dir=project_dir, + ) + assert result.returncode == 0, result.stdout + result.stderr + rows: list[tuple[object, ...]] = query_duckdb( + db_path=project_dir / "warehouse.duckdb", + sql=( + "SELECT customer_id, CAST(books AS VARCHAR), CAST(games AS VARCHAR) " + "FROM main.customer_category_amounts ORDER BY customer_id" + ), + ) + + assert tuple(rows) == test_case.expected_rows 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 88aa107dbd..1858d72b77 100644 --- a/tests/integration/src/sqlbuild/cli/commands/main/_test_types.py +++ b/tests/integration/src/sqlbuild/cli/commands/main/_test_types.py @@ -46,6 +46,15 @@ class TypedContractRuleIntegrationTestCase: expected_contract_106_findings: int +@dataclass(frozen=True) +class DynamicPivotRulesIntegrationTestCase: + description: str + expected_valid_exit_code: int + expected_invalid_exit_code: int + expected_invalid_model_findings: int + expected_invalid_sql_findings: int + + @dataclass(frozen=True) class SnowflakeCompileIntegrationTestCase: """One Snowflake SQL compatibility expectation through compile.""" 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 3a6a977d57..732bd5e6d8 100644 --- a/tests/integration/src/sqlbuild/cli/commands/main/test_rules.py +++ b/tests/integration/src/sqlbuild/cli/commands/main/test_rules.py @@ -11,6 +11,7 @@ from sqlbuild.cli.commands.main.entrypoint.entry import main from tests.integration.src.sqlbuild.cli.commands.main._test_types import ( + DynamicPivotRulesIntegrationTestCase, ExplicitContractOutputRuleIntegrationTestCase, RulePassIntegrationTestCase, RulesIntegrationTestCase, @@ -241,6 +242,89 @@ def test_given_enforced_contract_when_compiling_explicit_output_rule_then_enforc assert diagnostic_codes.count("SQBRCONTRACT105") == test_case.expected_rule_findings +@pytest.mark.parametrize( + "test_case", + ( + DynamicPivotRulesIntegrationTestCase( + description="proven family governs only output wildcards", + expected_valid_exit_code=0, + expected_invalid_exit_code=1, + expected_invalid_model_findings=1, + expected_invalid_sql_findings=1, + ), + ), + ids=lambda case: case.description, +) +def test_given_proven_dynamic_pivot_when_compiling_contract_rules_then_wildcard_is_governed( + test_case: DynamicPivotRulesIntegrationTestCase, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + (tmp_path / "sqlbuild_project.toml").write_text( + 'name = "orders"\nadapter = "snowflake"\n\n[rules]\n' + 'select = ["SQBRMODEL102", "SQBRSQL021", "SQBRCONTRACT101", ' + '"SQBRCONTRACT105", "SQBRCONTRACT106"]\n\n' + "[defaults]\n" + 'contract = "enforced"\n', + encoding="utf-8", + ) + models_dir: Path = tmp_path / "models" + models_dir.mkdir() + (models_dir / "stg_order_amounts.sql").write_text( + "MODEL (database analytics, schema analytics, columns (customer_id (type INTEGER), " + "category (type VARCHAR), " + 'amount (type "DECIMAL(12,2)")));\n' + "SELECT CAST(1 AS INTEGER) AS customer_id, " + "CAST('books' AS VARCHAR) AS category, " + "CAST(10.25 AS DECIMAL(12,2)) AS amount\n", + encoding="utf-8", + ) + dynamic_model: Path = models_dir / "customer_category_amounts.sql" + dynamic_model.write_text( + "MODEL (\n" + " database analytics,\n" + " schema analytics,\n" + " columns (customer_id (type INTEGER)),\n" + " dynamic_columns (category_amounts (pivot_column category, value_column amount, " + 'aggregate MAX, type "DECIMAL(12,2)")),\n' + ");\n" + "WITH pivot_input AS (\n" + " SELECT customer_id, category, amount\n" + ' FROM __ref("stg_order_amounts")\n' + ")\n" + "SELECT *\n" + "FROM pivot_input PIVOT(MAX(amount) FOR category IN (ANY ORDER BY category))\n", + encoding="utf-8", + ) + + exit_code: int = main(["--project-dir", str(tmp_path), "compile", "--json", "--no-cache"]) + result: dict[str, object] = json.loads(capsys.readouterr().out) + diagnostics: list[dict[str, object]] = cast(list[dict[str, object]], result["diagnostics"]) + + assert exit_code == test_case.expected_valid_exit_code + assert diagnostics == [] + + dynamic_model.write_text( + dynamic_model.read_text(encoding="utf-8").replace( + ")\nSELECT *\nFROM pivot_input PIVOT", + "),\nunrelated AS (SELECT * FROM pivot_input)\nSELECT *\nFROM pivot_input PIVOT", + ), + encoding="utf-8", + ) + invalid_exit_code: int = main( + ["--project-dir", str(tmp_path), "compile", "--json", "--no-cache"] + ) + invalid_result: dict[str, object] = json.loads(capsys.readouterr().out) + invalid_diagnostics: list[dict[str, object]] = cast( + list[dict[str, object]], invalid_result["diagnostics"] + ) + invalid_codes: tuple[object, ...] = tuple(item["code"] for item in invalid_diagnostics) + + assert invalid_exit_code == test_case.expected_invalid_exit_code + assert invalid_codes.count("SQBRMODEL102") == test_case.expected_invalid_model_findings + assert invalid_codes.count("SQBRSQL021") == test_case.expected_invalid_sql_findings + + @pytest.mark.parametrize( "test_case", ( diff --git a/tests/unit/src/sqlbuild/compiler/compile/_helpers/_test_types.py b/tests/unit/src/sqlbuild/compiler/compile/_helpers/_test_types.py index a532095bea..28a99f7382 100644 --- a/tests/unit/src/sqlbuild/compiler/compile/_helpers/_test_types.py +++ b/tests/unit/src/sqlbuild/compiler/compile/_helpers/_test_types.py @@ -324,6 +324,13 @@ class ModelSchemaCursorTestCase: expected_column_names: tuple[str, ...] +@dataclass(frozen=True) +class DynamicModelSchemaTestCase: + description: str + expected_codes: tuple[str, ...] = () + expected_fixed_columns: tuple[str, ...] = () + + @dataclass(frozen=True) class ExtractSqlTestCtesTestCase: description: str diff --git a/tests/unit/src/sqlbuild/compiler/compile/_helpers/analysis/_test_types.py b/tests/unit/src/sqlbuild/compiler/compile/_helpers/analysis/_test_types.py new file mode 100644 index 0000000000..35719bacd6 --- /dev/null +++ b/tests/unit/src/sqlbuild/compiler/compile/_helpers/analysis/_test_types.py @@ -0,0 +1,15 @@ +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass(frozen=True) +class DynamicPivotAnalysisTestCase: + description: str + dialect: str + query_sql: str + expected_proven: bool + expected_fixed_columns: tuple[str, ...] = () + expected_failure_fragment: str = "" + expected_family_types: tuple[str | None, ...] = ("DECIMAL(12,2)",) + aggregate: str = "MAX" diff --git a/tests/unit/src/sqlbuild/compiler/compile/_helpers/analysis/test_dynamic_pivot.py b/tests/unit/src/sqlbuild/compiler/compile/_helpers/analysis/test_dynamic_pivot.py new file mode 100644 index 0000000000..cb323f3930 --- /dev/null +++ b/tests/unit/src/sqlbuild/compiler/compile/_helpers/analysis/test_dynamic_pivot.py @@ -0,0 +1,177 @@ +from __future__ import annotations + +import pytest + +from sqlbuild.compiler.compile._helpers.analysis.dynamic_pivot import ( + analyze_dynamic_column_contract, +) +from sqlbuild.compiler.compile.models import DynamicColumnContractProof +from sqlbuild.spec.contracts.models import SchemaDynamicColumnFamily +from tests.unit.src.sqlbuild.compiler.compile._helpers.analysis._test_types import ( + DynamicPivotAnalysisTestCase, +) + + +@pytest.mark.parametrize( + "test_case", + ( + DynamicPivotAnalysisTestCase( + description="Snowflake ANY pivot is compiler proven", + dialect="snowflake", + query_sql=( + "WITH pivot_input AS (" + "SELECT customer_id, category, amount, country FROM order_amounts" + ") SELECT * FROM pivot_input " + "PIVOT(MAX(amount) FOR category IN (ANY ORDER BY category))" + ), + expected_proven=True, + expected_fixed_columns=("customer_id", "country"), + ), + DynamicPivotAnalysisTestCase( + description="static pivot list requires exact columns", + dialect="snowflake", + query_sql=( + "SELECT * FROM order_amounts PIVOT(MAX(amount) FOR category IN ('books', 'games'))" + ), + expected_proven=False, + expected_failure_fragment="static pivot values", + expected_family_types=(), + ), + DynamicPivotAnalysisTestCase( + description="unsupported adapter rejects dynamic family", + dialect="bigquery", + query_sql="SELECT * FROM order_amounts", + expected_proven=False, + expected_failure_fragment="does not support compiler-proven dynamic pivots", + expected_family_types=(), + ), + DynamicPivotAnalysisTestCase( + description="ordinary dependency wildcard is not a pivot family", + dialect="snowflake", + query_sql='SELECT * FROM __ref("order_amounts")', + expected_proven=False, + expected_failure_fragment="must redeclare the upstream families exactly", + expected_family_types=(), + ), + DynamicPivotAnalysisTestCase( + description="partial upstream declaration cannot prove fixed shape", + dialect="snowflake", + query_sql=( + "SELECT * FROM partial_order_amounts PIVOT(MAX(amount) FOR category IN (ANY))" + ), + expected_proven=False, + expected_failure_fragment="authoritative, explicit schema", + expected_family_types=(), + ), + DynamicPivotAnalysisTestCase( + description="star replacement cannot masquerade as passthrough", + dialect="snowflake", + query_sql=( + "WITH pivoted AS (SELECT * FROM order_amounts " + "PIVOT(MAX(amount) FOR category IN (ANY))) " + "SELECT * REPLACE (CAST(99 AS VARCHAR) AS customer_id) FROM pivoted" + ), + expected_proven=False, + expected_failure_fragment="wildcard projection", + expected_family_types=(), + ), + DynamicPivotAnalysisTestCase( + description="CTE output aliases cannot silently rename fixed columns", + dialect="snowflake", + query_sql=( + "WITH order_amounts(customer_id, category, amount, extra) AS (" + "SELECT customer_id, category, amount, country FROM order_amounts" + ") SELECT * FROM order_amounts " + "PIVOT(MAX(amount) FOR category IN (ANY))" + ), + expected_proven=False, + expected_failure_fragment="authoritative, explicit schema", + expected_family_types=(), + ), + DynamicPivotAnalysisTestCase( + description="explicit cast in pivot input establishes value type", + dialect="snowflake", + query_sql=( + "WITH pivot_input AS (" + "SELECT customer_id, category, CAST(amount AS DECIMAL(12,2)) AS amount, country " + "FROM order_amounts" + ") SELECT * FROM pivot_input " + "PIVOT(MAX(amount) FOR category IN (ANY))" + ), + expected_proven=True, + expected_fixed_columns=("customer_id", "country"), + expected_family_types=("DECIMAL(12, 2)",), + ), + DynamicPivotAnalysisTestCase( + description="widening aggregate input cast does not prove output type", + dialect="duckdb", + query_sql=( + "PIVOT order_amounts ON category USING SUM(CAST(amount AS INTEGER)) " + "GROUP BY customer_id" + ), + expected_proven=True, + expected_fixed_columns=("customer_id",), + expected_family_types=(None,), + aggregate="SUM", + ), + DynamicPivotAnalysisTestCase( + description="compound pivot expression cannot disappear from shape proof", + dialect="duckdb", + query_sql=( + "PIVOT order_amounts ON category, UPPER(country) USING MAX(amount) " + "GROUP BY customer_id" + ), + expected_proven=False, + expected_failure_fragment="direct column references", + expected_family_types=(), + ), + ), + ids=lambda case: case.description, +) +def test_given_dynamic_family_when_analyzing_output_then_requires_supported_closed_shape( + test_case: DynamicPivotAnalysisTestCase, +) -> None: + proof: DynamicColumnContractProof | None = analyze_dynamic_column_contract( + query_sql=test_case.query_sql, + dialect=test_case.dialect, + families=( + SchemaDynamicColumnFamily( + name="category_amounts", + pivot_column="category", + value_column="amount", + aggregate=test_case.aggregate, + type="DECIMAL(12,2)", + ), + ), + column_types_by_table={ + "order_amounts": { + "customer_id": "INTEGER", + "category": "VARCHAR", + "amount": "DECIMAL(12,2)", + "country": "VARCHAR", + }, + "partial_order_amounts": { + "customer_id": "INTEGER", + "category": "VARCHAR", + "amount": "DECIMAL(12,2)", + }, + }, + authoritative_column_types_by_table={ + "order_amounts": { + "customer_id": "INTEGER", + "category": "VARCHAR", + "amount": "DECIMAL(12,2)", + "country": "VARCHAR", + } + }, + column_nullability_by_table={}, + dynamic_families_by_table={}, + ) + + assert proof is not None + assert proof.output_proven is test_case.expected_proven + assert tuple(column.name for column in proof.fixed_columns) == test_case.expected_fixed_columns + assert tuple(family.inferred_type for family in proof.families) == ( + test_case.expected_family_types + ) + assert test_case.expected_failure_fragment in (proof.failure_reason or "") diff --git a/tests/unit/src/sqlbuild/compiler/compile/_helpers/test_model_schemas.py b/tests/unit/src/sqlbuild/compiler/compile/_helpers/test_model_schemas.py index 91245459f5..e7a9979353 100644 --- a/tests/unit/src/sqlbuild/compiler/compile/_helpers/test_model_schemas.py +++ b/tests/unit/src/sqlbuild/compiler/compile/_helpers/test_model_schemas.py @@ -7,6 +7,7 @@ import pytest +from sqlbuild.adapters.duckdb.classes.duckdb_adapter import DuckDbAdapter from sqlbuild.compiler.compile.exceptions import CompileInputError from sqlbuild.compiler.compile.main._assemble_project import assemble_project from sqlbuild.compiler.compile.main._build_compile_inputs import build_compile_inputs @@ -25,6 +26,7 @@ ) from sqlbuild.spec.contracts.models import SchemaAuditInstance, SourceLocation from tests.unit.src.sqlbuild.compiler.compile._helpers._test_types import ( + DynamicModelSchemaTestCase, ModelSchemaCompilationTestCase, ModelSchemaContractDiagnosticTestCase, ModelSchemaCursorTestCase, @@ -314,7 +316,8 @@ def test_given_named_schema_contract_mismatch_when_validating_then_reports_expec discovered_inputs=discovered, adapter_context=DUCKDB_COMPILE_ADAPTER_CONTEXT, run_id="test_run", - ) + ), + inference_profile=DuckDbAdapter().expression_inference_profile(), ) diagnostics: tuple[CompilerDiagnostic, ...] = evaluate_model_contracts( @@ -326,6 +329,283 @@ def test_given_named_schema_contract_mismatch_when_validating_then_reports_expec ) +@pytest.mark.parametrize( + "test_case", + ( + DynamicModelSchemaTestCase( + description="proven dynamic pivot family", + expected_fixed_columns=("customer_id",), + ), + ), + ids=lambda case: case.description, +) +def test_given_dynamic_pivot_family_when_compiling_then_preserves_closed_contract_metadata( + test_case: DynamicModelSchemaTestCase, + tmp_path: Path, + write_repo_files: Callable[[Path, dict[str, str]], None], +) -> None: + write_repo_files( + tmp_path, + { + "sqlbuild_project.toml": _PROJECT_FILE, + "models/stg_order_amounts.sql": """ +MODEL ( + contract enforced, + columns ( + customer_id (type INTEGER), + category (type VARCHAR), + amount (type "DECIMAL(12,2)"), + ), +); +SELECT + CAST(1 AS INTEGER) AS customer_id, + CAST('books' AS VARCHAR) AS category, + CAST(10.25 AS DECIMAL(12,2)) AS amount +""", + "models/customer_category_amounts.sql": """ +MODEL ( + contract enforced, + columns (customer_id (type INTEGER)), + dynamic_columns ( + category_amounts ( + pivot_column category, + value_column amount, + aggregate MAX, + type "DECIMAL(12,2)" + ) + ), +); +PIVOT __ref("stg_order_amounts") +ON category +USING MAX(amount) +GROUP BY customer_id +""", + "models/customer_category_amounts_copy.sql": """ +MODEL ( + contract enforced, + columns (customer_id (type INTEGER)), + dynamic_columns ( + category_amounts ( + pivot_column category, + value_column amount, + aggregate MAX, + type "DECIMAL(12,2)" + ) + ), +); +SELECT * FROM __ref("customer_category_amounts") +""", + }, + ) + project: CompiledProject = assemble_project( + inputs=build_compile_inputs( + discovered_inputs=discover_project_inputs(project_dir=tmp_path), + adapter_context=DUCKDB_COMPILE_ADAPTER_CONTEXT, + run_id="test_run", + ), + inference_profile=DuckDbAdapter().expression_inference_profile(), + ) + models_by_name: dict[str, CompiledModel] = {item.name: item for item in project.models} + model: CompiledModel = models_by_name["customer_category_amounts"] + + assert model.schema_entry is not None + assert tuple(column.name for column in model.inferred_columns or ()) == ( + test_case.expected_fixed_columns + ) + assert model.dynamic_column_contract is not None + assert model.dynamic_column_contract.output_proven is True + assert model.dynamic_column_contract.families[0].inferred_type == "DECIMAL(12,2)" + assert evaluate_model_contracts(project=project, dialect="duckdb").diagnostics == () + identity: dict[str, object] = json.loads( + build_model_version_identity_metadata_json(model=model) + ) + contract_identity: dict[str, object] = cast( + dict[str, object], identity["execution_signature"]["contract"] + ) + assert contract_identity["dynamic_columns"] == [ + { + "name": "category_amounts", + "pivot_column": "category", + "value_column": "amount", + "aggregate": "MAX", + "type": "DECIMAL(12,2)", + "name_pattern": None, + } + ] + manifest_node: dict[str, object] = build_model_node( + model=model, + plan_entry=None, + project_name="demo", + ) + manifest_meta: dict[str, object] = cast(dict[str, object], manifest_node["meta"]) + sqlbuild_meta: dict[str, object] = cast(dict[str, object], manifest_meta["sqlbuild"]) + assert sqlbuild_meta["dynamic_columns"] == contract_identity["dynamic_columns"] + passthrough: CompiledModel = models_by_name["customer_category_amounts_copy"] + assert passthrough.dynamic_column_contract is not None + assert passthrough.dynamic_column_contract.output_proven is True + assert tuple(column.name for column in passthrough.inferred_columns or ()) == ("customer_id",) + + +@pytest.mark.parametrize( + "test_case", + ( + DynamicModelSchemaTestCase( + description="partial upstream schema", + expected_codes=("K011",), + ), + ), + ids=lambda case: case.description, +) +def test_given_partial_upstream_schema_when_compiling_dynamic_pivot_then_closure_is_rejected( + test_case: DynamicModelSchemaTestCase, + tmp_path: Path, + write_repo_files: Callable[[Path, dict[str, str]], None], +) -> None: + write_repo_files( + tmp_path, + { + "sqlbuild_project.toml": _PROJECT_FILE, + "models/order_amounts.sql": """ +MODEL ( + contract none, + columns ( + customer_id (type INTEGER), + category (type VARCHAR), + amount (type INTEGER), + ), +); +SELECT + CAST(1 AS INTEGER) AS customer_id, + CAST('books' AS VARCHAR) AS category, + CAST(10 AS INTEGER) AS amount, + CAST(99 AS INTEGER) AS unrelated +""", + "models/customer_category_amounts.sql": """ +MODEL ( + contract enforced, + columns (customer_id (type INTEGER)), + dynamic_columns ( + category_amounts ( + pivot_column category, + value_column amount, + aggregate MAX, + type INTEGER + ) + ), +); +PIVOT __ref("order_amounts") +ON category +USING MAX(amount) +GROUP BY customer_id +""", + }, + ) + project: CompiledProject = assemble_project( + inputs=build_compile_inputs( + discovered_inputs=discover_project_inputs(project_dir=tmp_path), + adapter_context=DUCKDB_COMPILE_ADAPTER_CONTEXT, + run_id="test_run", + ), + inference_profile=DuckDbAdapter().expression_inference_profile(), + ) + + diagnostics: tuple[CompilerDiagnostic, ...] = evaluate_model_contracts( + project=project, + dialect="duckdb", + ).diagnostics + + assert tuple(diagnostic.code for diagnostic in diagnostics) == test_case.expected_codes + assert "authoritative, explicit schema" in diagnostics[0].message + + +@pytest.mark.parametrize( + "test_case", + ( + DynamicModelSchemaTestCase( + description="unenforced dynamic upstream", + expected_codes=("K011",), + ), + ), + ids=lambda case: case.description, +) +def test_given_unenforced_dynamic_upstream_when_compiling_passthrough_then_closure_is_rejected( + test_case: DynamicModelSchemaTestCase, + tmp_path: Path, + write_repo_files: Callable[[Path, dict[str, str]], None], +) -> None: + write_repo_files( + tmp_path, + { + "sqlbuild_project.toml": _PROJECT_FILE, + "models/order_amounts.sql": """ +MODEL ( + contract enforced, + columns ( + customer_id (type INTEGER), + category (type VARCHAR), + amount (type INTEGER), + unrelated (type INTEGER), + ), +); +SELECT + CAST(1 AS INTEGER) AS customer_id, + CAST('books' AS VARCHAR) AS category, + CAST(10 AS INTEGER) AS amount, + CAST(99 AS INTEGER) AS unrelated +""", + "models/open_category_amounts.sql": """ +MODEL ( + contract none, + columns (customer_id (type INTEGER)), + dynamic_columns ( + category_amounts ( + pivot_column category, + value_column amount, + aggregate MAX, + type INTEGER + ) + ), +); +PIVOT __ref("order_amounts") +ON category +USING MAX(amount) +GROUP BY customer_id +""", + "models/closed_category_amounts.sql": """ +MODEL ( + contract enforced, + columns (customer_id (type INTEGER)), + dynamic_columns ( + category_amounts ( + pivot_column category, + value_column amount, + aggregate MAX, + type INTEGER + ) + ), +); +SELECT * FROM __ref("open_category_amounts") +""", + }, + ) + project: CompiledProject = assemble_project( + inputs=build_compile_inputs( + discovered_inputs=discover_project_inputs(project_dir=tmp_path), + adapter_context=DUCKDB_COMPILE_ADAPTER_CONTEXT, + run_id="test_run", + ), + inference_profile=DuckDbAdapter().expression_inference_profile(), + ) + + diagnostics: tuple[CompilerDiagnostic, ...] = evaluate_model_contracts( + project=project, + dialect="duckdb", + ).diagnostics + + assert tuple(diagnostic.code for diagnostic in diagnostics) == test_case.expected_codes + assert "must redeclare the upstream families exactly" in diagnostics[0].message + + @pytest.mark.parametrize( "test_case", [ diff --git a/tests/unit/src/sqlbuild/compiler/contract_adoption/_helpers/_test_types.py b/tests/unit/src/sqlbuild/compiler/contract_adoption/_helpers/_test_types.py index 817b3d75e3..8efe294dff 100644 --- a/tests/unit/src/sqlbuild/compiler/contract_adoption/_helpers/_test_types.py +++ b/tests/unit/src/sqlbuild/compiler/contract_adoption/_helpers/_test_types.py @@ -7,3 +7,11 @@ class DeclaredTypeSpanTestCase: contents: str declared_type: str expected_value: str + + +@dataclass(frozen=True) +class DynamicContractAdoptionTestCase: + description: str + expected_reason: str | None = None + expected_matching_findings: int = 0 + expected_mismatching_findings: int = 0 diff --git a/tests/unit/src/sqlbuild/compiler/contract_adoption/_helpers/test_repository_edits.py b/tests/unit/src/sqlbuild/compiler/contract_adoption/_helpers/test_repository_edits.py index b0e73ee18d..914534f838 100644 --- a/tests/unit/src/sqlbuild/compiler/contract_adoption/_helpers/test_repository_edits.py +++ b/tests/unit/src/sqlbuild/compiler/contract_adoption/_helpers/test_repository_edits.py @@ -1,10 +1,22 @@ import pytest -from sqlbuild.compiler.contract_adoption._helpers.repository_edits import _declared_type_span +from sqlbuild.adapter.contract.models import ColumnInfo +from sqlbuild.adapters.duckdb.classes.duckdb_adapter import DuckDbAdapter +from sqlbuild.compiler.compile.models import CompiledModel +from sqlbuild.compiler.compile.types import CompiledResourceType +from sqlbuild.compiler.contract_adoption._helpers.compare import compare_evidence +from sqlbuild.compiler.contract_adoption._helpers.repository_edits import ( + _declared_type_span, + edit_model, +) +from sqlbuild.compiler.contract_adoption.models import ContractEvidence from sqlbuild.compiler.discovery.models import ModelHeaderColumnSpan +from sqlbuild.spec.contracts.models import SchemaDynamicColumnFamily from tests.unit.src.sqlbuild.compiler.contract_adoption._helpers._test_types import ( DeclaredTypeSpanTestCase, + DynamicContractAdoptionTestCase, ) +from tests.unit.src.sqlbuild.compiler.contracts.helpers import make_contract_project @pytest.mark.parametrize( @@ -39,3 +51,115 @@ def test_given_nested_type_key_when_locating_declared_type_then_returns_top_leve ) assert result == (expected_start, expected_start + len(test_case.expected_value)) + + +@pytest.mark.parametrize( + "test_case", + (DynamicContractAdoptionTestCase(description="preserve dynamic family generation"),), + ids=lambda case: case.description, +) +def test_given_dynamic_family_when_generating_contract_then_does_not_freeze_runtime_members( + test_case: DynamicContractAdoptionTestCase, +) -> None: + authored_sql: str = """MODEL ( + contract enforced, + columns (customer_id (type INTEGER)), + dynamic_columns ( + category_amounts ( + pivot_column category, + value_column amount, + aggregate MAX, + type "DECIMAL(12,2)" + ) + ), +); +SELECT * FROM category_amounts +""" + model: CompiledModel = make_contract_project( + declared_columns=(("customer_id", "INTEGER"),), + inferred_columns=(("customer_id", "INTEGER"),), + type_enforcement=True, + contract="enforced", + dynamic_columns=( + SchemaDynamicColumnFamily( + name="category_amounts", + pivot_column="category", + value_column="amount", + aggregate="MAX", + type="DECIMAL(12,2)", + ), + ), + authored_sql=authored_sql, + ).models[0] + + edited, reason = edit_model( + model=model, + physical_columns=( + ColumnInfo(name="customer_id", type="INTEGER"), + ColumnInfo(name="books", type="DECIMAL(12,2)"), + ColumnInfo(name="games", type="DECIMAL(12,2)"), + ), + overwrite=False, + ) + + assert reason == test_case.expected_reason + assert edited == authored_sql + assert "books (" not in edited + assert "games (" not in edited + + +@pytest.mark.parametrize( + "test_case", + ( + DynamicContractAdoptionTestCase( + description="compare dynamic member types", + expected_matching_findings=0, + expected_mismatching_findings=1, + ), + ), + ids=lambda case: case.description, +) +def test_given_dynamic_family_when_inspecting_contract_then_compares_generated_member_types( + test_case: DynamicContractAdoptionTestCase, +) -> None: + evidence: ContractEvidence = ContractEvidence( + resource_type=CompiledResourceType.MODEL, + resource_name="customer_category_amounts", + database=None, + schema="main", + relation="customer_category_amounts", + declared_columns=(ColumnInfo(name="customer_id", type="INTEGER"),), + physical_columns=None, + source_path=None, + dynamic_columns=( + SchemaDynamicColumnFamily( + name="category_amounts", + pivot_column="category", + value_column="amount", + aggregate="MAX", + type="DECIMAL(12,2)", + ), + ), + ) + + matching: ContractEvidence = compare_evidence( + item=evidence, + physical_columns=( + ColumnInfo(name="customer_id", type="INTEGER"), + ColumnInfo(name="books", type="DECIMAL(12,2)"), + ), + adapter=DuckDbAdapter(), + ) + mismatching: ContractEvidence = compare_evidence( + item=evidence, + physical_columns=( + ColumnInfo(name="customer_id", type="INTEGER"), + ColumnInfo(name="books", type="VARCHAR"), + ), + adapter=DuckDbAdapter(), + ) + + assert len(matching.findings) == test_case.expected_matching_findings + assert len(mismatching.findings) == test_case.expected_mismatching_findings + assert mismatching.findings[0].column_name == "books" + assert mismatching.findings[0].declared_type == "DECIMAL(12,2)" diff --git a/tests/unit/src/sqlbuild/compiler/contracts/_test_types.py b/tests/unit/src/sqlbuild/compiler/contracts/_test_types.py index 1c9a7ad6dc..dedb4ff39b 100644 --- a/tests/unit/src/sqlbuild/compiler/contracts/_test_types.py +++ b/tests/unit/src/sqlbuild/compiler/contracts/_test_types.py @@ -41,3 +41,14 @@ class ContractMissingDeclarationsTestCase: contract: str | None expected_codes: tuple[str, ...] expected_messages: tuple[str, ...] + + +@dataclass(frozen=True) +class DynamicContractValidationTestCase: + description: str + declared_type: str + output_proven: bool + inferred_type: str | None + failure_reason: str | None + expected_code: str + expected_message: str diff --git a/tests/unit/src/sqlbuild/compiler/contracts/helpers.py b/tests/unit/src/sqlbuild/compiler/contracts/helpers.py index fc5730082b..9da4604558 100644 --- a/tests/unit/src/sqlbuild/compiler/contracts/helpers.py +++ b/tests/unit/src/sqlbuild/compiler/contracts/helpers.py @@ -9,6 +9,7 @@ CompiledProject, CompiledRelationLocation, CompileModelConfig, + DynamicColumnContractProof, InferredColumn, ) from sqlbuild.compiler.compile.types import CompiledResourceType @@ -16,6 +17,7 @@ from sqlbuild.spec.contracts.models import ( SchemaAuditInstance, SchemaColumn, + SchemaDynamicColumnFamily, SchemaModelEntry, SettingsConfig, SourceLocation, @@ -37,6 +39,8 @@ def make_contract_project( inferred_nullability_by_column: dict[str, InferredNullability] | None = None, fast_lineage_has_star: bool = False, authored_sql: str = "", + dynamic_columns: tuple[SchemaDynamicColumnFamily, ...] = (), + dynamic_column_contract: DynamicColumnContractProof | None = None, ) -> CompiledProject: """Build a compiled project for contract validation tests.""" @@ -85,6 +89,7 @@ def make_contract_project( ) for name, column_type in declared_columns ), + dynamic_columns=dynamic_columns, ), inferred_columns=( None, @@ -101,6 +106,7 @@ def make_contract_project( )[inferred_columns is not None], fast_lineage_has_star=fast_lineage_has_star, authored_sql=authored_sql, + dynamic_column_contract=dynamic_column_contract, ), ), ) diff --git a/tests/unit/src/sqlbuild/compiler/contracts/test_validate.py b/tests/unit/src/sqlbuild/compiler/contracts/test_validate.py index 3b9573efa9..18b847af75 100644 --- a/tests/unit/src/sqlbuild/compiler/contracts/test_validate.py +++ b/tests/unit/src/sqlbuild/compiler/contracts/test_validate.py @@ -5,14 +5,19 @@ import pytest from sqlbuild.adapter.contract.types import TypeDialect +from sqlbuild.compiler.compile.models import ( + DynamicColumnContractProof, + DynamicColumnFamilyProof, +) from sqlbuild.compiler.contracts.main.validate import evaluate_model_contracts from sqlbuild.compiler.contracts.models import ContractValidationResult from sqlbuild.compiler.lineage.types import InferredNullability -from sqlbuild.spec.contracts.models import SourceLocation +from sqlbuild.spec.contracts.models import SchemaDynamicColumnFamily, SourceLocation from tests.unit.src.sqlbuild.compiler.contracts._test_types import ( ContractLocationTestCase, ContractMissingDeclarationsTestCase, ContractValidationTestCase, + DynamicContractValidationTestCase, ) from tests.unit.src.sqlbuild.compiler.contracts.helpers import make_contract_project @@ -401,3 +406,68 @@ def test_given_contract_without_declared_columns_when_validating_then_returns_ex tuple(diagnostic.message for diagnostic in result.diagnostics) == test_case.expected_messages ) + + +@pytest.mark.parametrize( + "test_case", + ( + DynamicContractValidationTestCase( + description="unproven family rejects open wildcard", + declared_type="DECIMAL(12,2)", + output_proven=False, + inferred_type=None, + failure_reason="output is not a dynamic pivot boundary", + expected_code="K011", + expected_message=( + "model 'orders' dynamic column contract is not proven: " + "output is not a dynamic pivot boundary" + ), + ), + DynamicContractValidationTestCase( + description="incompatible family type reports mismatch", + declared_type="VARCHAR", + output_proven=True, + inferred_type="DECIMAL(12,2)", + failure_reason=None, + expected_code="K002", + expected_message=( + "dynamic column family 'category_amounts' inferred as DECIMAL(12,2) " + "but declared type is VARCHAR" + ), + ), + ), + ids=lambda case: case.description, +) +def test_given_dynamic_contract_when_validating_then_reports_expected_diagnostic( + test_case: DynamicContractValidationTestCase, +) -> None: + family: SchemaDynamicColumnFamily = SchemaDynamicColumnFamily( + name="category_amounts", + pivot_column="category", + value_column="amount", + aggregate="MAX", + type=test_case.declared_type, + ) + result: ContractValidationResult = evaluate_model_contracts( + project=make_contract_project( + declared_columns=(("customer_id", "INTEGER"),), + inferred_columns=(("customer_id", "INTEGER"),), + type_enforcement=True, + contract="enforced", + dynamic_columns=(family,), + dynamic_column_contract=DynamicColumnContractProof( + output_proven=test_case.output_proven, + families=( + DynamicColumnFamilyProof( + name="category_amounts", + inferred_type=test_case.inferred_type, + ), + ), + failure_reason=test_case.failure_reason, + ), + ), + dialect=TypeDialect.DUCKDB, + ) + + assert tuple(diagnostic.code for diagnostic in result.diagnostics) == (test_case.expected_code,) + assert result.diagnostics[0].message == test_case.expected_message diff --git a/tests/unit/src/sqlbuild/compiler/planner/_helpers/changes/_test_types.py b/tests/unit/src/sqlbuild/compiler/planner/_helpers/changes/_test_types.py index 08c1e05196..42d945f677 100644 --- a/tests/unit/src/sqlbuild/compiler/planner/_helpers/changes/_test_types.py +++ b/tests/unit/src/sqlbuild/compiler/planner/_helpers/changes/_test_types.py @@ -4,6 +4,7 @@ from sqlbuild.compiler.compile.models import InferredColumn from sqlbuild.compiler.planner.models import BackfillResult, SchemaFinding from sqlbuild.compiler.planner.types import BackfillAction, ChangeKind +from sqlbuild.spec.contracts.models import SchemaDynamicColumnFamily @dataclass(frozen=True) @@ -55,6 +56,7 @@ class DetectSchemaChangesTestCase: type_enforcement: bool expected_findings: tuple[SchemaFinding, ...] inferred_schema_complete: bool = True + dynamic_columns: tuple[SchemaDynamicColumnFamily, ...] = () @dataclass(frozen=True) diff --git a/tests/unit/src/sqlbuild/compiler/planner/_helpers/changes/test_schema.py b/tests/unit/src/sqlbuild/compiler/planner/_helpers/changes/test_schema.py index 2f8b19b4f0..c883deafb9 100644 --- a/tests/unit/src/sqlbuild/compiler/planner/_helpers/changes/test_schema.py +++ b/tests/unit/src/sqlbuild/compiler/planner/_helpers/changes/test_schema.py @@ -7,6 +7,7 @@ from sqlbuild.compiler.planner._helpers.changes.schema import detect_schema_changes from sqlbuild.compiler.planner.models import SchemaFinding from sqlbuild.compiler.planner.types import SchemaChangeKind, SchemaColumnSource +from sqlbuild.spec.contracts.models import SchemaDynamicColumnFamily from tests.unit.src.sqlbuild.compiler.planner._helpers.changes._test_types import ( DetectSchemaChangesTestCase, ) @@ -267,6 +268,28 @@ ), ), ), + DetectSchemaChangesTestCase( + description="typed dynamic members match their family", + yml_columns=(ColumnInfo(name="customer_id", type="INTEGER"),), + inferred_columns=(InferredColumn(name="customer_id", type="INTEGER"),), + warehouse_columns=( + ColumnInfo(name="customer_id", type="INTEGER"), + ColumnInfo(name="books", type="DECIMAL(12,2)"), + ColumnInfo(name="games", type="DECIMAL(12,2)"), + ), + type_enforcement=True, + inferred_schema_complete=False, + dynamic_columns=( + SchemaDynamicColumnFamily( + name="category_amounts", + pivot_column="category", + value_column="amount", + aggregate="MAX", + type="DECIMAL(12,2)", + ), + ), + expected_findings=(), + ), ], ids=lambda case: case.description, ) @@ -279,6 +302,7 @@ def test_given_columns_when_detecting_schema_changes_then_returns_expected_findi warehouse_columns=test_case.warehouse_columns, type_enforcement=test_case.type_enforcement, inferred_schema_complete=test_case.inferred_schema_complete, + dynamic_columns=test_case.dynamic_columns, ) assert result == test_case.expected_findings diff --git a/tests/unit/src/sqlbuild/executor/run/_helpers/_test_types.py b/tests/unit/src/sqlbuild/executor/run/_helpers/_test_types.py index 70dec009fb..6abdb62087 100644 --- a/tests/unit/src/sqlbuild/executor/run/_helpers/_test_types.py +++ b/tests/unit/src/sqlbuild/executor/run/_helpers/_test_types.py @@ -3,6 +3,7 @@ from sqlbuild.adapter.contract.models import ColumnInfo, LifeCycleEvent from sqlbuild.executor.run.types import AuditGateReuseReason, AuditGateStatus, ExecutionPhase +from sqlbuild.spec.contracts.models import SchemaDynamicColumnFamily @dataclass(frozen=True) @@ -495,5 +496,6 @@ class RuntimeContractValidationTestCase: contract_columns: tuple[ColumnInfo, ...] actual_columns: tuple[ColumnInfo, ...] expected_valid: bool + contract_dynamic_columns: tuple[SchemaDynamicColumnFamily, ...] = () expected_error_fragment: str | None = None expected_error_code: str | None = None diff --git a/tests/unit/src/sqlbuild/executor/run/_helpers/helpers.py b/tests/unit/src/sqlbuild/executor/run/_helpers/helpers.py index d879afc697..a0a7713b2e 100644 --- a/tests/unit/src/sqlbuild/executor/run/_helpers/helpers.py +++ b/tests/unit/src/sqlbuild/executor/run/_helpers/helpers.py @@ -21,6 +21,7 @@ from sqlbuild.compiler.planner.types import MaterializationType, PlanAction, PlanReason from sqlbuild.executor.auditing.models import AuditExecutionResult from sqlbuild.executor.run.models import HookContext +from sqlbuild.spec.contracts.models import SchemaDynamicColumnFamily def insert_snapshot_hook_log(ctx: HookContext, phase: str) -> None: @@ -55,7 +56,10 @@ def build_result_model_plan_entry() -> ModelPlanEntry: def build_contract_model_plan_entry( - *, contract_enforced: bool, contract_columns: tuple[ColumnInfo, ...] + *, + contract_enforced: bool, + contract_columns: tuple[ColumnInfo, ...], + contract_dynamic_columns: tuple[SchemaDynamicColumnFamily, ...] = (), ) -> ModelPlanEntry: entry: ModelPlanEntry = build_result_model_plan_entry() return ModelPlanEntry( @@ -71,6 +75,7 @@ def build_contract_model_plan_entry( logical_ddl=entry.logical_ddl, contract_enforced=contract_enforced, contract_columns=contract_columns, + contract_dynamic_columns=contract_dynamic_columns, ) diff --git a/tests/unit/src/sqlbuild/executor/run/_helpers/test_contracts.py b/tests/unit/src/sqlbuild/executor/run/_helpers/test_contracts.py index 9ed9b892f4..63cefeb15a 100644 --- a/tests/unit/src/sqlbuild/executor/run/_helpers/test_contracts.py +++ b/tests/unit/src/sqlbuild/executor/run/_helpers/test_contracts.py @@ -6,6 +6,7 @@ from sqlbuild.compiler.planner.models import ModelPlanEntry from sqlbuild.errors.contracts.exceptions import ExecutorInputError from sqlbuild.executor.run._helpers.validation.contracts import validate_runtime_contract +from sqlbuild.spec.contracts.models import SchemaDynamicColumnFamily from tests.unit.src.sqlbuild.executor.run._helpers._test_types import ( RuntimeContractValidationTestCase, ) @@ -25,6 +26,26 @@ ), expected_valid=True, ), + RuntimeContractValidationTestCase( + description="allows runtime columns covered by one typed dynamic family", + contract_enforced=True, + contract_columns=(ColumnInfo(name="customer_id", type="INTEGER"),), + contract_dynamic_columns=( + SchemaDynamicColumnFamily( + name="category_amounts", + pivot_column="category", + value_column="amount", + aggregate="MAX", + type="DECIMAL(12,2)", + ), + ), + actual_columns=( + ColumnInfo(name="customer_id", type="INTEGER"), + ColumnInfo(name="books", type="DECIMAL(12,2)"), + ColumnInfo(name="games", type="DECIMAL(12,2)"), + ), + expected_valid=True, + ), RuntimeContractValidationTestCase( description="allows exact enforced contract with compatible types", contract_enforced=True, @@ -54,6 +75,7 @@ def test_given_valid_runtime_contract_when_validating_then_passes( entry: ModelPlanEntry = build_contract_model_plan_entry( contract_enforced=test_case.contract_enforced, contract_columns=test_case.contract_columns, + contract_dynamic_columns=test_case.contract_dynamic_columns, ) validate_runtime_contract(entry=entry, actual_columns=test_case.actual_columns) @@ -106,6 +128,27 @@ def test_given_valid_runtime_contract_when_validating_then_passes( expected_error_fragment="has type VARCHAR but contract declares INTEGER", expected_error_code="K010", ), + RuntimeContractValidationTestCase( + description="rejects runtime dynamic column with incompatible family type", + contract_enforced=True, + contract_columns=(ColumnInfo(name="customer_id", type="INTEGER"),), + contract_dynamic_columns=( + SchemaDynamicColumnFamily( + name="category_amounts", + pivot_column="category", + value_column="amount", + aggregate="MAX", + type="DECIMAL(12,2)", + ), + ), + actual_columns=( + ColumnInfo(name="customer_id", type="INTEGER"), + ColumnInfo(name="books", type="VARCHAR"), + ), + expected_valid=False, + expected_error_fragment="dynamic column 'books' has type VARCHAR", + expected_error_code="K010", + ), ], ids=lambda case: case.description, ) @@ -115,6 +158,7 @@ def test_given_invalid_runtime_contract_when_validating_then_raises( entry: ModelPlanEntry = build_contract_model_plan_entry( contract_enforced=test_case.contract_enforced, contract_columns=test_case.contract_columns, + contract_dynamic_columns=test_case.contract_dynamic_columns, ) assert test_case.expected_error_fragment is not None