From 813d01fb875a7946503560c9a0fbb3efea98955d Mon Sep 17 00:00:00 2001 From: Gareth Price Date: Sun, 13 Sep 2026 22:54:26 +0100 Subject: [PATCH] =?UTF-8?q?feat:=20contract=20diffs=20+=20breaking-change?= =?UTF-8?q?=20gates=20=E2=80=94=20structured=20contract=20analysis,=20decl?= =?UTF-8?q?ared=20renames,=20persisted=20contracts?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bundle 7 on top of shared state/cache. Turns the opaque contract_hash change into a structured, machine-readable contract diff and gates promotion on breaking contract changes, computed live. - contracts.rs: contract_diff classifies SAFE/REVIEW/BREAKING per change — contract added/dropped, enforcement on/off, column added/removed/renamed, type changes (numeric widening review, incompatible breaking), nullability tighten/relax. - Declared renames: [model..renames] new = "old" resolves a removed+added pair into a reviewable rename in both contract and physical schema diffs. Stale or ambiguous declarations resolve nothing — they cannot hide a real removal or addition. - MaterializedRecord persists the full contract (contract_json, SQLite + Postgres with migration), so base/target comparisons work against what the environment actually recorded. - BranchDiffReport gains contract_changes and lineage-aware impacts naming the downstream models/tests a breaking change affects. - Promotion: contract breaks join physical schema breaks in the schema gate, computed live against the target's recorded contracts — a contract edited after `diff` cannot slip past on a stale artifact. --allow-breaking-schema waives both. - plan/explain reasons name the changed columns and breaking count instead of "contract or assertions changed". --- SPEC.md | 11 +- crates/phlo-transform-cli/src/main.rs | 523 +++++++++++++- crates/phlo-transform-core/src/compile.rs | 10 +- crates/phlo-transform-core/src/config.rs | 3 + crates/phlo-transform-core/src/discovery.rs | 3 +- crates/phlo-transform-core/src/semantic.rs | 13 +- crates/phlo-transform-core/tests/analysis.rs | 2 + .../phlo-transform-core/tests/compilation.rs | 19 + .../phlo-transform-engine/src/branch_diff.rs | 166 ++++- crates/phlo-transform-engine/src/contracts.rs | 673 ++++++++++++++++++ crates/phlo-transform-engine/src/diff.rs | 195 ++++- crates/phlo-transform-engine/src/lib.rs | 9 +- crates/phlo-transform-engine/src/plan.rs | 64 +- crates/phlo-transform-engine/src/run.rs | 7 + crates/phlo-transform-engine/src/state.rs | 204 +++++- .../src/state_postgres.rs | 65 +- crates/phlo-transform-engine/tests/engine.rs | 249 ++++++- .../tests/nessie_wap_e2e.rs | 1 + .../phlo-transform-trino/tests/trino_e2e.rs | 2 + .../tests/trino_incremental_e2e.rs | 1 + docs/diff.md | 62 +- docs/state.md | 15 +- docs/wap.md | 16 +- fixtures/contract-renames/phlo.toml | 12 + .../workflows/assay/transforms/results.sql | 1 + 25 files changed, 2248 insertions(+), 78 deletions(-) create mode 100644 crates/phlo-transform-engine/src/contracts.rs create mode 100644 fixtures/contract-renames/phlo.toml create mode 100644 fixtures/contract-renames/workflows/assay/transforms/results.sql diff --git a/SPEC.md b/SPEC.md index 6789af3..350fb43 100644 --- a/SPEC.md +++ b/SPEC.md @@ -811,6 +811,15 @@ type = "DOUBLE" nullable = false ``` +Column renames are declared alongside the contract so diffs can classify a +removed+added pair as a named rename — a breaking change for consumers +still selecting the old name — rather than an unexplained removal: + +```toml +[model.assay_results.renames] +result = "legacy_result" +``` + Explicit contract files remain optional. ## 32. Inferred tests @@ -1072,7 +1081,7 @@ phlo transform promote feature/new-assay --to main phlo transform promote --from feature/new-assay --to main ``` -Promotion is authorised by named gates, reported identically in human and JSON output: `run` (latest candidate run passed and validated the exact commit being promoted — a passed run is bound to the candidate head **after** its writes land, and a candidate that advanced since its run, or a run recorded before commit binding, fails), `tests` (no failed tests), `blocked` (no blocked/cancelled model, seed or test work), `schema` (a fresh audited diff inspected this pair at these commits and found no unwaived breaking changes — absent or stale evidence fails closed, never reading "no evidence" as "no changes"), `data_diff` (when `--require-diff` is set: a passing `--full` audited diff bound to this candidate→target pair at the commits being promoted — the artifact records both refs' resolved heads and is rejected when they no longer match — still fresh per recorded versions, and never a self-comparison; a single-model `diff.json` is never promotion evidence), `base` (the target still equals the commit the evidence was established against — the hash-bound artifact's recorded base, else the candidate's `created_from` provenance; unknown provenance fails rather than redefining the base as the current head, and the merge asserts the evaluated target hash) and `conflicts` (the merge check is clean). A candidate that advanced between gate evaluation and merge is refused rather than promoted unaudited. `--check` evaluates gates without merging; a passing promotion merges and persists a `PromotionRecord` (refs, hashes, plan/run ids, gate results, timestamp) in the state store. +Promotion is authorised by named gates, reported identically in human and JSON output: `run` (latest candidate run passed and validated the exact commit being promoted — a passed run is bound to the candidate head **after** its writes land, and a candidate that advanced since its run, or a run recorded before commit binding, fails), `tests` (no failed tests), `blocked` (no blocked/cancelled model, seed or test work), `schema` (a fresh audited diff inspected this pair at these commits and found no unwaived breaking changes — physical breaks come from the audited diff; contract breaks are computed live from the workspace's desired contracts against the target's recorded contracts, so a post-`diff` contract edit cannot slip past — and absent or stale evidence fails closed, never reading "no evidence" as "no changes"), `data_diff` (when `--require-diff` is set: a passing `--full` audited diff bound to this candidate→target pair at the commits being promoted — the artifact records both refs' resolved heads and is rejected when they no longer match — still fresh per recorded versions, and never a self-comparison; a single-model `diff.json` is never promotion evidence), `base` (the target still equals the commit the evidence was established against — the hash-bound artifact's recorded base, else the candidate's `created_from` provenance; unknown provenance fails rather than redefining the base as the current head, and the merge asserts the evaluated target hash) and `conflicts` (the merge check is clean). A candidate that advanced between gate evaluation and merge is refused rather than promoted unaudited. `--check` evaluates gates without merging; a passing promotion merges and persists a `PromotionRecord` (refs, hashes, plan/run ids, gate results, timestamp) in the state store. Preconditions may include successful plan, successful execution, required tests passing, no blocking schema changes, no stale state and optional approval. diff --git a/crates/phlo-transform-cli/src/main.rs b/crates/phlo-transform-cli/src/main.rs index 2be0e96..82b6bd4 100644 --- a/crates/phlo-transform-cli/src/main.rs +++ b/crates/phlo-transform-cli/src/main.rs @@ -24,11 +24,11 @@ use phlo_transform_engine::{ adapter_default_schema, branch_diff, changed_models, collect_source_states, diff, diff_reasons, ensure_environment, evaluate_gates, materialized_for_environment, model_keys, promote, relation_for_source, retarget, seeds_for_environment, Adapter, ArtifactWriter, - BranchDiffReport, BranchDiffRequest, CancelHandle, DatasetKind, DiffPolicy, DiffRequest, - DiffStrategy, EnvironmentArtifact, EnvironmentSetup, EnvironmentSpec, ExecutionStatus, - GateInput, Membership, Plan, PlanAction, PlanOptions, PlanReason, Planner, PostgresStateStore, - PromotionRequest, ReasonKind, RetryPolicy, RunOptions, RunResult, Runner, SqliteStateStore, - StateStore, SCHEMA_VERSION, + BranchDiffReport, BranchDiffRequest, CancelHandle, ContractSafety, DatasetKind, DiffPolicy, + DiffRequest, DiffStrategy, EnvironmentArtifact, EnvironmentSetup, EnvironmentSpec, + ExecutionStatus, GateInput, Membership, Plan, PlanAction, PlanOptions, PlanReason, Planner, + PostgresStateStore, PromotionRequest, ReasonKind, RetryPolicy, RunOptions, RunResult, Runner, + SqliteStateStore, StateStore, SCHEMA_VERSION, }; use phlo_transform_nessie::{NessieClient, NessieConfig, NessieRestClient}; use phlo_transform_trino::{TrinoAdapter, TrinoConfig}; @@ -538,6 +538,7 @@ async fn run(cli: &Cli) -> Result { } => { run_promote( cli, + &compilation, candidate.as_deref(), to, *check, @@ -1128,8 +1129,10 @@ fn build_nessie(cli: &Cli) -> Result, String> { Ok(Arc::new(client)) } +#[allow(clippy::too_many_arguments)] async fn run_promote( cli: &Cli, + compilation: &Compilation, candidate: Option<&str>, to: &str, check: bool, @@ -1186,7 +1189,7 @@ async fn run_promote( }; let environment = read_environment_for(cli, candidate); - let audit = audited_diff( + let mut audit = audited_diff( cli, state.as_deref(), candidate, @@ -1194,6 +1197,17 @@ async fn run_promote( Some(&candidate_reference.hash), Some(&target.hash), ); + // Contract breaks are computed live — the workspace's desired contracts + // against what the target environment last recorded — so a contract + // edited after `diff` cannot sneak a break past the gate on a stale + // artifact's analysis. + audit + .breaking_schema_changes + .extend(contract_breaking_changes( + state.as_deref(), + compilation, + to, + )?); let merge_check = nessie.can_merge(candidate, to).await.ok(); // The `base` gate needs a target commit the evidence was established @@ -1346,8 +1360,58 @@ struct AuditEvidence { schema_audited: bool, } -/// Read the audited diff artifact (`branch_diff.json` preferred, single-model -/// `diff.json` as fallback) and derive the evidence it carries for this +/// Live contract analysis for the promotion gate: the workspace's desired +/// contracts against the contracts the target environment last recorded. +/// Returns the breaking subset in the same `model.column: detail` shape as +/// physical schema breaks. +fn contract_breaking_changes( + state: Option<&dyn StateStore>, + compilation: &Compilation, + to: &str, +) -> Result, String> { + let Some(state) = state else { + return Ok(Vec::new()); + }; + // The recorded base contracts are promotion evidence — a store error + // must fail promotion, never read as "no contracts recorded". + let base = materialized_for_environment(state, to).map_err(|error| error.to_string())?; + let mut breaking = Vec::new(); + for model in &compilation.models { + let name = model.id.logical_name(); + let record = base.get(&name); + let mut changes = phlo_transform_engine::contract_diff( + record.and_then(|record| record.contract.as_ref()), + model.contract.as_ref(), + ); + // The effective key — incremental `key` columns and unique-assertion + // columns collapse onto the same identity concept — is compared + // against the key the target's materialisation persisted. Changing + // or dropping it is breaking; a record that cannot prove its + // historical key fails closed. + if let Some(change) = phlo_transform_engine::key_change( + &record + .map(|record| record.recorded_key()) + .unwrap_or(phlo_transform_engine::RecordedKey::Known(None)), + phlo_transform_engine::effective_key(model).as_deref(), + ) { + changes.push(change); + } + for change in changes { + if change.safety == ContractSafety::Breaking { + let subject = if change.column.is_empty() { + name.clone() + } else { + format!("{name}.{}", change.column) + }; + breaking.push(format!("{subject}: contract {}", change.detail)); + } + } + } + Ok(breaking) +} + +/// Read the audited `branch_diff.json` artifact and derive the evidence it +/// carries for this /// promotion: the diff verdict, why the artifact cannot be used, the breaking /// schema changes it recorded, and whether a schema audit genuinely ran. /// @@ -1428,10 +1492,29 @@ fn audited_diff( // dataset that had no materialisation at diff time has one now (it // stopped being `removed`/`absent` since the audit). `main` folds in // the default environment, matching how the diff was produced. - let candidate_models = materialized_for_environment(state, candidate).unwrap_or_default(); - let candidate_seeds = seeds_for_environment(state, candidate).unwrap_or_default(); - let base_models = materialized_for_environment(state, to).unwrap_or_default(); - let base_seeds = seeds_for_environment(state, to).unwrap_or_default(); + // State reads are promotion evidence: a store error cannot masquerade + // as "nothing recorded" — the artifact is rejected rather than + // trusted against an empty map. + let (candidate_models, candidate_seeds) = match ( + materialized_for_environment(state, candidate), + seeds_for_environment(state, candidate), + ) { + (Ok(models), Ok(seeds)) => (models, seeds), + (Err(error), _) | (_, Err(error)) => { + stale(format!("cannot confirm the diff is current: {error}")); + (BTreeMap::new(), BTreeMap::new()) + } + }; + let (base_models, base_seeds) = match ( + materialized_for_environment(state, to), + seeds_for_environment(state, to), + ) { + (Ok(models), Ok(seeds)) => (models, seeds), + (Err(error), _) | (_, Err(error)) => { + stale(format!("cannot confirm the diff is current: {error}")); + (BTreeMap::new(), BTreeMap::new()) + } + }; for dataset in &report.datasets { let (current_candidate, current_base) = match dataset.kind { DatasetKind::Model => ( @@ -1849,6 +1932,28 @@ fn run_state(cli: &Cli, action: &StateAction) -> Result { println!(" sources: {}", record.version.source_state_hash); println!(" compiler: {}", record.version.compiler_version); println!(" target: {}", record.version.target_hash); + if let Some(contract) = &record.contract { + let columns = contract + .columns + .iter() + .map(|column| column.name.clone()) + .collect::>() + .join(", "); + println!( + "Contract: {} ({} columns{})", + if contract.enforced { + "enforced" + } else { + "advisory" + }, + contract.columns.len(), + if columns.is_empty() { + String::new() + } else { + format!(": {columns}") + } + ); + } } } } @@ -3295,6 +3400,11 @@ async fn run_diff( strategy, policy: diff_policy(compiled.config.diff.as_ref()), sample_fraction: sample, + renames: compiled + .contract + .as_ref() + .map(|contract| contract.renames.clone()) + .unwrap_or_default(), }; match diff(adapter, &request).await { @@ -3480,6 +3590,24 @@ fn print_branch_diff_human(report: &BranchDiffReport) { } } } + if !report.contract_changes.is_empty() { + println!(); + println!("Contracts"); + for diff in &report.contract_changes { + for change in &diff.changes { + let subject = if change.column.is_empty() { + diff.model.clone() + } else { + format!("{}.{}", diff.model, change.column) + }; + println!( + " {subject}: {} [{}]", + change.detail, + change.safety.as_str() + ); + } + } + } if !report.rows.is_empty() { println!(); println!("Rows"); @@ -4115,7 +4243,7 @@ async fn run_explain( mod tests { use super::*; use phlo_transform_engine::{ - BranchDiffReport, DatasetDiff, DatasetKind, DatasetStatus, MaterializedRecord, + BranchDiffReport, DatasetDiff, DatasetKind, DatasetStatus, EngineError, MaterializedRecord, ModelSchemaDiff, SchemaChange, SqliteStateStore, }; @@ -4137,6 +4265,8 @@ mod tests { base_hash: None, datasets: Vec::new(), schema_changes: Vec::new(), + contract_changes: Vec::new(), + impacts: Vec::new(), rows: Vec::new(), diffs: Vec::new(), deep, @@ -4287,6 +4417,8 @@ mod tests { incremental_key: None, adapter: None, output_identity: None, + contract: None, + effective_key: None, run_id: "run-1".to_string(), materialized_at: "t".to_string(), }) @@ -4332,6 +4464,9 @@ mod tests { incremental_key: None, adapter: None, output_identity: None, + contract: None, + // A current-era record: recorded keyless, not legacy-unknown. + effective_key: Some(Vec::new()), run_id: "run-1".to_string(), materialized_at: "t".to_string(), } @@ -4536,4 +4671,366 @@ mod tests { // Nor may a degenerate ref collapse onto the single-slot artifact. assert_ne!(environment_artifact_name("///"), "environment.json"); } + + fn contract(names: &[&str]) -> phlo_transform_core::ModelContract { + phlo_transform_core::ModelContract { + enforced: true, + columns: names + .iter() + .map(|name| phlo_transform_core::ColumnContract { + name: (*name).to_string(), + data_type: None, + nullable: None, + }) + .collect(), + renames: Default::default(), + } + } + + #[test] + fn contract_breaking_changes_reads_recorded_contracts() { + let state = SqliteStateStore::in_memory().expect("state"); + let mut main_record = record("main.base", Some("main"), "v1"); + main_record.contract = Some(contract(&["id", "legacy"])); + state.record_materialized(&main_record).expect("record"); + + // The workspace now declares a contract without `legacy` — removing + // it is a breaking change against the recorded base contract. + let mut base = phlo_transform_core::SemanticModel::in_memory( + ModelId::parse("main.base").expect("model id"), + "select 1 as id, 2 as legacy", + ); + base.contract = Some(contract(&["id"])); + let compilation = compile(&phlo_transform_core::SemanticProject::in_memory(vec![base])); + assert!(compilation.is_ok(), "{:?}", compilation.diagnostics); + + let breaks = contract_breaking_changes(Some(&state), &compilation, "main").expect("breaks"); + assert_eq!(breaks.len(), 1, "{breaks:?}"); + assert!(breaks[0].contains("legacy"), "{breaks:?}"); + + // The default environment folds into `main`, so a default-env + // record feeds the same gate. + let state = SqliteStateStore::in_memory().expect("state"); + let mut default_record = record("main.base", None, "v1"); + default_record.contract = Some(contract(&["id", "legacy"])); + state.record_materialized(&default_record).expect("record"); + let breaks = contract_breaking_changes(Some(&state), &compilation, "main").expect("breaks"); + assert_eq!(breaks.len(), 1, "{breaks:?}"); + } + + struct FailingStore; + impl StateStore for FailingStore { + fn start_run( + &self, + _run: &phlo_transform_engine::RunRecord, + _plan: &phlo_transform_engine::StoredPlan, + ) -> Result<(), EngineError> { + unimplemented!() + } + fn reopen_run(&self, _run_id: &str) -> Result<(), EngineError> { + unimplemented!() + } + fn bind_run_reference_hash( + &self, + _run_id: &str, + _reference_hash: &str, + ) -> Result<(), EngineError> { + unimplemented!() + } + fn finish_run( + &self, + _run_id: &str, + _status: phlo_transform_engine::ExecutionStatus, + _finished_at: &str, + _failed_count: usize, + ) -> Result<(), EngineError> { + unimplemented!() + } + fn record_model( + &self, + _record: &phlo_transform_engine::ModelRunRecord, + ) -> Result<(), EngineError> { + unimplemented!() + } + fn record_seed_run( + &self, + _record: &phlo_transform_engine::SeedRunRecord, + ) -> Result<(), EngineError> { + unimplemented!() + } + fn record_test( + &self, + _record: &phlo_transform_engine::TestRunRecord, + ) -> Result<(), EngineError> { + unimplemented!() + } + fn runs(&self) -> Result, EngineError> { + unimplemented!() + } + fn latest_run( + &self, + _environment: Option<&str>, + ) -> Result, EngineError> { + unimplemented!() + } + fn run( + &self, + _run_id: &str, + ) -> Result, EngineError> { + unimplemented!() + } + fn find_runs( + &self, + _prefix: &str, + ) -> Result, EngineError> { + unimplemented!() + } + fn model_runs( + &self, + _run_id: &str, + ) -> Result, EngineError> { + unimplemented!() + } + fn seed_runs( + &self, + _run_id: &str, + ) -> Result, EngineError> { + unimplemented!() + } + fn test_runs( + &self, + _run_id: &str, + ) -> Result, EngineError> { + unimplemented!() + } + fn record_materialized(&self, _record: &MaterializedRecord) -> Result<(), EngineError> { + unimplemented!() + } + fn materialized_version( + &self, + _model_id: &str, + _environment: Option<&str>, + ) -> Result, EngineError> { + unimplemented!() + } + fn materialized_by_hash( + &self, + _version_hash: &str, + ) -> Result, EngineError> { + unimplemented!() + } + // The one method the promotion evidence path exercises — it fails. + fn materialized_in( + &self, + _environment: Option<&str>, + ) -> Result, EngineError> { + Err(EngineError::State("state store unreachable".to_string())) + } + fn record_promotion( + &self, + _record: &phlo_transform_engine::PromotionRecord, + ) -> Result<(), EngineError> { + unimplemented!() + } + fn promotions(&self) -> Result, EngineError> { + unimplemented!() + } + fn set_watermark( + &self, + _model_id: &str, + _environment: Option<&str>, + _value: &str, + _run_id: &str, + ) -> Result<(), EngineError> { + unimplemented!() + } + fn watermark( + &self, + _model_id: &str, + _environment: Option<&str>, + ) -> Result, EngineError> { + unimplemented!() + } + fn record_seed( + &self, + _record: &phlo_transform_engine::SeedRecord, + ) -> Result<(), EngineError> { + unimplemented!() + } + fn seed_state( + &self, + _name: &str, + _environment: Option<&str>, + ) -> Result, EngineError> { + unimplemented!() + } + fn seeds_in( + &self, + _environment: Option<&str>, + ) -> Result, EngineError> { + unimplemented!() + } + } + + #[test] + fn contract_breaking_changes_fails_closed_on_state_error() { + // A store whose read fails must fail promotion — never read as + // "no contracts recorded". + let model = phlo_transform_core::SemanticModel::in_memory( + ModelId::parse("main.base").expect("model id"), + "select 1 as id", + ); + let compilation = compile(&phlo_transform_core::SemanticProject::in_memory(vec![ + model, + ])); + let result = contract_breaking_changes(Some(&FailingStore), &compilation, "main"); + let error = result.expect_err("a state error must fail promotion"); + assert!(error.contains("unreachable"), "{error}"); + } + + #[test] + fn contract_breaking_changes_reports_key_changes() { + // A legacy record — written before effective keys were persisted — + // still proves its incremental key: `batch_id` recorded → `id` + // desired is a breaking key change, not a keyless base. + let state = SqliteStateStore::in_memory().expect("state"); + let mut main_record = record("main.base", Some("main"), "v1"); + main_record.incremental_strategy = Some("key".to_string()); + main_record.incremental_key = Some("batch_id".to_string()); + main_record.effective_key = None; // predates the column + state.record_materialized(&main_record).expect("record"); + + let mut base = phlo_transform_core::SemanticModel::in_memory( + ModelId::parse("main.base").expect("model id"), + "select 1 as id", + ); + base.config.incremental = Some(phlo_transform_core::IncrementalStrategy::Key { + columns: vec!["id".to_string()], + }); + let compilation = compile(&phlo_transform_core::SemanticProject::in_memory(vec![base])); + assert!(compilation.is_ok(), "{:?}", compilation.diagnostics); + + let breaks = contract_breaking_changes(Some(&state), &compilation, "main").expect("breaks"); + assert!( + breaks.iter().any(|change| change.contains("key changed")), + "{breaks:?}" + ); + } + + #[test] + fn contract_breaking_changes_reports_persisted_key_changes() { + // The persisted effective key — not the incremental fields — is the + // historical key: a `unique(sample_id)` assertion alone materialised + // with `effective_key = [[sample_id]]`, no incremental strategy. + let mut keyed = phlo_transform_core::SemanticModel::in_memory( + ModelId::parse("main.base").expect("model id"), + "select 1 as id", + ); + keyed.directives.keys.push("batch_id".to_string()); + let compilation = compile(&phlo_transform_core::SemanticProject::in_memory(vec![ + keyed, + ])); + assert!(compilation.is_ok(), "{:?}", compilation.diagnostics); + + // Changed: recorded [[sample_id]] vs desired [[batch_id]]. + let state = SqliteStateStore::in_memory().expect("state"); + let mut main_record = record("main.base", Some("main"), "v1"); + main_record.effective_key = Some(vec![vec!["sample_id".to_string()]]); + state.record_materialized(&main_record).expect("record"); + let breaks = contract_breaking_changes(Some(&state), &compilation, "main").expect("breaks"); + assert!( + breaks.iter().any(|change| change.contains("key changed")), + "{breaks:?}" + ); + + // Removed: the workspace model declares no key at all. + let keyless = phlo_transform_core::SemanticModel::in_memory( + ModelId::parse("main.base").expect("model id"), + "select 1 as id", + ); + let compilation = compile(&phlo_transform_core::SemanticProject::in_memory(vec![ + keyless, + ])); + let breaks = contract_breaking_changes(Some(&state), &compilation, "main").expect("breaks"); + assert!( + breaks + .iter() + .any(|change| change.contains("key") && change.contains("dropped")), + "{breaks:?}" + ); + } + + #[test] + fn contract_breaking_changes_treats_incremental_and_unique_keys_equally() { + // The base materialised with `key = "id"` — a legacy record whose + // only key evidence is the incremental fields; the workspace now + // claims the same identity through a `unique(id)` assertion — + // same effective key, no change. + let state = SqliteStateStore::in_memory().expect("state"); + let mut main_record = record("main.base", Some("main"), "v1"); + main_record.incremental_strategy = Some("key".to_string()); + main_record.incremental_key = Some("id".to_string()); + main_record.effective_key = None; // predates the column + state.record_materialized(&main_record).expect("record"); + + let mut asserted = phlo_transform_core::SemanticModel::in_memory( + ModelId::parse("main.base").expect("model id"), + "select 1 as id", + ); + asserted.directives.keys.push("id".to_string()); + let compilation = compile(&phlo_transform_core::SemanticProject::in_memory(vec![ + asserted, + ])); + assert!(compilation.is_ok(), "{:?}", compilation.diagnostics); + + let breaks = contract_breaking_changes(Some(&state), &compilation, "main").expect("breaks"); + assert!( + !breaks.iter().any(|change| change.contains("key")), + "equivalent key declared differently must not report: {breaks:?}" + ); + } + + #[test] + fn contract_breaking_changes_fails_closed_on_unverifiable_key() { + // A legacy record with neither a persisted effective key nor an + // incremental key cannot prove the base had no key — an unknown + // historical key must fail closed, not read as keyless. + let state = SqliteStateStore::in_memory().expect("state"); + let mut main_record = record("main.base", Some("main"), "v1"); + main_record.effective_key = None; // predates the column + state.record_materialized(&main_record).expect("record"); + + // Desired has a key: the record might be hiding a change. + let mut keyed = phlo_transform_core::SemanticModel::in_memory( + ModelId::parse("main.base").expect("model id"), + "select 1 as id", + ); + keyed.directives.keys.push("id".to_string()); + let compilation = compile(&phlo_transform_core::SemanticProject::in_memory(vec![ + keyed, + ])); + let breaks = contract_breaking_changes(Some(&state), &compilation, "main").expect("breaks"); + assert!( + breaks + .iter() + .any(|change| change.contains("key") && change.contains("cannot be verified")), + "unverifiable record must block, got: {breaks:?}" + ); + + // Desired has no key either: the record might be hiding a removal. + let keyless = phlo_transform_core::SemanticModel::in_memory( + ModelId::parse("main.base").expect("model id"), + "select 1 as id", + ); + let compilation = compile(&phlo_transform_core::SemanticProject::in_memory(vec![ + keyless, + ])); + let breaks = contract_breaking_changes(Some(&state), &compilation, "main").expect("breaks"); + assert!( + breaks + .iter() + .any(|change| change.contains("key") && change.contains("cannot be verified")), + "unverifiable record must block even against a keyless model: {breaks:?}" + ); + } } diff --git a/crates/phlo-transform-core/src/compile.rs b/crates/phlo-transform-core/src/compile.rs index a7f5741..f51d5f1 100644 --- a/crates/phlo-transform-core/src/compile.rs +++ b/crates/phlo-transform-core/src/compile.rs @@ -420,10 +420,16 @@ fn version_inputs( ) }) .collect(); + let renames: Vec = contract + .renames + .iter() + .map(|(new, old)| format!("{new}={old}")) + .collect(); format!( - "enforced={};columns={}", + "enforced={};columns={};renames={}", contract.enforced, - columns.join(",") + columns.join(","), + renames.join(",") ) }) .unwrap_or_default(); diff --git a/crates/phlo-transform-core/src/config.rs b/crates/phlo-transform-core/src/config.rs index ca061ee..fbf5004 100644 --- a/crates/phlo-transform-core/src/config.rs +++ b/crates/phlo-transform-core/src/config.rs @@ -70,6 +70,9 @@ pub struct ModelContractConfig { pub incremental: IncrementalModelConfig, /// `[model..diff]`. pub diff: DiffConfig, + /// `[model..renames]` — column renames the model declares + /// (`new name = "old name"`). + pub renames: BTreeMap, } /// The `[model..diff]` settings. diff --git a/crates/phlo-transform-core/src/discovery.rs b/crates/phlo-transform-core/src/discovery.rs index 018d9b3..605d43f 100644 --- a/crates/phlo-transform-core/src/discovery.rs +++ b/crates/phlo-transform-core/src/discovery.rs @@ -206,7 +206,7 @@ fn normalized_sections(config: &PhloConfig) -> BTreeMap Option { - if !section.contract.enforced && section.columns.is_empty() { + if !section.contract.enforced && section.columns.is_empty() && section.renames.is_empty() { return None; } Some(ModelContract { @@ -223,6 +223,7 @@ fn contract_from_section(section: &ModelContractConfig) -> Option nullable: column.nullable, }) .collect(), + renames: section.renames.clone(), }) } diff --git a/crates/phlo-transform-core/src/semantic.rs b/crates/phlo-transform-core/src/semantic.rs index 8f69f02..b6f7a5c 100644 --- a/crates/phlo-transform-core/src/semantic.rs +++ b/crates/phlo-transform-core/src/semantic.rs @@ -7,7 +7,7 @@ use crate::identity::{ModelId, SourceId}; /// A SQL scalar or nested type as understood by the compiler. -#[derive(Clone, Debug, PartialEq, Eq)] +#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] pub enum DataType { Boolean, TinyInt, @@ -176,7 +176,7 @@ impl std::fmt::Display for DataType { } /// Nullability of a column. `Unknown` is distinct from `Nullable`. -#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] pub enum Nullability { NotNull, Nullable, @@ -392,7 +392,7 @@ impl Assertion { } /// An explicit column-level contract. -#[derive(Clone, Debug, PartialEq, Eq)] +#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] pub struct ColumnContract { pub name: String, pub data_type: Option, @@ -400,10 +400,15 @@ pub struct ColumnContract { } /// An explicit model contract. -#[derive(Clone, Debug, Default, PartialEq, Eq)] +#[derive(Clone, Debug, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)] pub struct ModelContract { pub enforced: bool, pub columns: Vec, + /// Column renames the model declares (`new name` → `old name`), from + /// `[model..renames]`: a removed+added pair that is a declared + /// rename classifies as a rename, not a breaking removal. + #[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")] + pub renames: std::collections::BTreeMap, } /// Numeric tolerance for a diffed column. diff --git a/crates/phlo-transform-core/tests/analysis.rs b/crates/phlo-transform-core/tests/analysis.rs index c3672b8..56cb130 100644 --- a/crates/phlo-transform-core/tests/analysis.rs +++ b/crates/phlo-transform-core/tests/analysis.rs @@ -330,6 +330,7 @@ fn enforced_contract_violation_is_an_error() { data_type: None, nullable: None, }], + renames: Default::default(), }); let compilation = compile(vec![model]); assert!(!compilation.is_ok()); @@ -349,6 +350,7 @@ fn contract_type_mismatch_is_reported() { data_type: Some(DataType::Varchar), nullable: Some(false), }], + renames: Default::default(), }); let compilation = compile(vec![model]); assert!(!compilation.is_ok()); diff --git a/crates/phlo-transform-core/tests/compilation.rs b/crates/phlo-transform-core/tests/compilation.rs index 1b522f4..7314738 100644 --- a/crates/phlo-transform-core/tests/compilation.rs +++ b/crates/phlo-transform-core/tests/compilation.rs @@ -484,6 +484,25 @@ fn discovery_attaches_contracts_and_generates_tests() { })); } +/// `[model..renames]` lands on the contract so schema and contract +/// diffs can classify a remove+add pair as a rename. +#[test] +fn discovery_attaches_declared_renames() { + let compilation = compile_fixture("contract-renames"); + assert!(compilation.is_ok(), "{:?}", compilation.diagnostics); + + let contract = compilation + .model(&ModelId::parse("assay.results").unwrap()) + .expect("model") + .contract + .as_ref() + .expect("contract attached"); + assert_eq!( + contract.renames.get("result_value").map(String::as_str), + Some("concentration") + ); +} + #[test] fn models_expose_workflow_ownership() { let compilation = compile_fixture("cross-workflow"); diff --git a/crates/phlo-transform-engine/src/branch_diff.rs b/crates/phlo-transform-engine/src/branch_diff.rs index ceb7b8e..34bb909 100644 --- a/crates/phlo-transform-engine/src/branch_diff.rs +++ b/crates/phlo-transform-engine/src/branch_diff.rs @@ -36,6 +36,9 @@ use crate::error::{AdapterError, EngineError}; use crate::state::{MaterializedRecord, SeedRecord, StateStore}; use crate::util::now_rfc3339; +/// Empty rename map shared by datasets whose model declares no renames. +static EMPTY_RENAMES: BTreeMap = BTreeMap::new(); + /// What a dataset is. #[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] @@ -86,6 +89,32 @@ pub struct ModelSchemaDiff { pub changes: Vec, } +/// Contract-level changes for one model — what promotion would apply to +/// the base environment's recorded contract. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct ModelContractDiff { + pub model: String, + pub changes: Vec, +} + +/// The lineage impact of one breaking change — which downstream models and +/// tests consume the thing that breaks. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct BreakingImpact { + pub model: String, + /// The broken column; `None` for model-level breaks. + #[serde(skip_serializing_if = "Option::is_none")] + pub column: Option, + /// What changed, in the schema/contract diff's own words. + pub change: String, + /// Where the break came from: `schema` or `contract`. + pub source: String, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub downstream_models: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub tests: Vec, +} + /// Row-count comparison for one dataset. #[derive(Clone, Debug, Serialize, Deserialize)] pub struct ModelRowDiff { @@ -140,6 +169,13 @@ pub struct BranchDiffReport { /// Column-level changes for datasets present on at least one side. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub schema_changes: Vec, + /// Contract changes each model would promote onto the base + /// environment's recorded contract — declared-rename aware. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub contract_changes: Vec, + /// Lineage impact for every breaking schema or contract change. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub impacts: Vec, /// Row counts for datasets present on at least one side. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub rows: Vec, @@ -350,7 +386,15 @@ pub async fn branch_diff( (Some(relation), true) => adapter.relation_columns(relation).await.unwrap_or_default(), _ => Vec::new(), }; - let changes = compare_column_lists(&candidate_columns, &base_columns); + let renames = compilation + .model_by_name(&dataset.dataset) + .and_then(|model| model.contract.as_ref()) + .map(|contract| &contract.renames); + let changes = compare_column_lists( + &candidate_columns, + &base_columns, + renames.unwrap_or(&EMPTY_RENAMES), + ); if !changes.is_empty() { schema_changes.push(ModelSchemaDiff { model: dataset.dataset.clone(), @@ -432,6 +476,11 @@ pub async fn branch_diff( strategy: DiffStrategy::Keyed, policy: diff_policy(model.config.diff.as_ref()), sample_fraction: None, + renames: model + .contract + .as_ref() + .map(|contract| contract.renames.clone()) + .unwrap_or_default(), }, ) .await?, @@ -441,6 +490,69 @@ pub async fn branch_diff( diffs.sort_by(|left, right| left.model.cmp(&right.model)); let passed = diffs.iter().all(|report| report.passed); + // Contract diffs: the workspace's declared contract against what the + // base environment last recorded. A model without a base record reports + // its contract as newly declared; a model whose base record predates + // contract persistence compares against `None` the same way. + let mut contract_changes = Vec::new(); + for model in &compilation.models { + let name = model.id.logical_name(); + let record = base_records.get(&name); + let mut changes = crate::contracts::contract_diff( + record.and_then(|record| record.contract.as_ref()), + model.contract.as_ref(), + ); + // The effective key collapses incremental `key` columns and + // unique-assertion columns onto one identity concept; compare the + // workspace's against the key the base materialisation persisted. + // A model never materialised on the base has no key to protect; a + // record that predates key persistence fails closed. + if let Some(change) = crate::contracts::key_change( + &record + .map(|record| record.recorded_key()) + .unwrap_or(crate::contracts::RecordedKey::Known(None)), + crate::contracts::effective_key(model).as_deref(), + ) { + changes.push(change); + } + if !changes.is_empty() { + contract_changes.push(ModelContractDiff { + model: name, + changes, + }); + } + } + + // Lineage impact of every breaking change — what promoting this branch + // would break downstream. + let mut impacts = Vec::new(); + for schema_diff in &schema_changes { + for change in &schema_diff.changes { + if matches!(change.safety.as_str(), "error" | "full_rebuild_required") { + impacts.push(breaking_impact( + compilation, + &schema_diff.model, + Some(change.column.as_str()), + "schema", + &change.detail, + )); + } + } + } + for contract_diff in &contract_changes { + for change in &contract_diff.changes { + if change.safety == crate::contracts::ContractSafety::Breaking { + impacts.push(breaking_impact( + compilation, + &contract_diff.model, + (!change.column.is_empty()).then_some(change.column.as_str()), + "contract", + &change.detail, + )); + } + } + } + Ok(BranchDiffReport { candidate_ref: request.candidate_ref.clone(), base_ref: request.base_ref.clone(), @@ -448,6 +560,8 @@ pub async fn branch_diff( base_hash: request.base_hash.clone(), datasets, schema_changes, + contract_changes, + impacts, rows, diffs, deep: request.deep, @@ -526,6 +640,56 @@ async fn snapshot(adapter: &dyn Adapter, relation: Option<&Relation>) -> Option< } } +/// Resolve one breaking change's downstream blast radius through the +/// lineage graph: column-level breaks use the column impact walk, +/// model-level breaks report the whole model's downstream. +fn breaking_impact( + compilation: &Compilation, + model_name: &str, + column: Option<&str>, + source: &str, + detail: &str, +) -> BreakingImpact { + let (downstream_models, tests) = match compilation.model_by_name(model_name) { + Some(model) => match column { + Some(column) => { + let report = compilation.impact_report( + &phlo_transform_core::semantic::ColumnRef::model(model.id.clone(), column), + ); + (report.downstream_models, report.tests) + } + None => { + let mut models = Vec::new(); + for node in compilation.lineage.downstream_transitive( + &phlo_transform_core::lineage::LineageNode::Model(model.id.clone()), + ) { + if let phlo_transform_core::lineage::LineageNode::Model(id) = node { + models.push(id.logical_name()); + } + } + let tests = compilation + .lineage + .tests_for_dataset(&phlo_transform_core::lineage::DatasetId::model(&model.id)) + .into_iter() + .map(|test| test.to_string()) + .collect(); + (models, tests) + } + }, + // The breaking dataset is not a compiled model (a seed, or a model + // dropped from the workspace) — no lineage to walk. + None => (Vec::new(), Vec::new()), + }; + BreakingImpact { + model: model_name.to_string(), + column: column.map(str::to_string), + change: detail.to_string(), + source: source.to_string(), + downstream_models, + tests, + } +} + /// The default (unlabeled) environment — `apply`/`run` with no `--ref` /// records under it while physically writing the default catalog, which is /// `main`'s view. When `main` is asked for, its records count too. diff --git a/crates/phlo-transform-engine/src/contracts.rs b/crates/phlo-transform-engine/src/contracts.rs new file mode 100644 index 0000000..6d6ebba --- /dev/null +++ b/crates/phlo-transform-engine/src/contracts.rs @@ -0,0 +1,673 @@ +//! Candidate/base contract comparison. +//! +//! `contract_diff` turns the opaque `contract_hash` change into a structured, +//! machine-readable report: which columns the contract added, removed, +//! renamed, retightened or relaxed, and whether enforcement changed. Safety +//! uses the same three-level vocabulary as the physical schema diff so +//! promotion gates can treat both uniformly. + +use phlo_transform_core::semantic::{DataType, ModelContract}; +use serde::{Deserialize, Serialize}; +use std::collections::BTreeMap; + +/// How dangerous a contract change is for downstream consumers. +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ContractSafety { + /// Consumption stays valid without action. + Safe, + /// Deserves human review but doesn't inherently break consumers. + Review, + /// Existing consumers, tests or downstream models can break. + Breaking, +} + +impl ContractSafety { + pub fn as_str(&self) -> &'static str { + match self { + Self::Safe => "safe", + Self::Review => "review", + Self::Breaking => "breaking", + } + } +} + +/// One entry in a contract diff. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct ContractChange { + /// Column the change applies to. For `renamed`, this is the new name. + pub column: String, + /// `contract_added`, `contract_removed`, `enforcement_changed`, `added`, + /// `removed`, `renamed`, `rename_ambiguous`, `type_changed`, + /// `nullability_changed`, `key_added`, `key_removed` or `key_changed`. + pub kind: String, + /// Human-readable detail. + pub detail: String, + pub safety: ContractSafety, +} + +/// Diff a previous contract against a desired contract. +/// +/// `previous` may be `None` (no contract recorded on the base side) and +/// `desired` may be `None` (the model dropped its contract) — both are +/// reported as changes rather than being silently ignored. +pub fn contract_diff( + previous: Option<&ModelContract>, + desired: Option<&ModelContract>, +) -> Vec { + let mut changes = Vec::new(); + let (previous, desired) = match (previous, desired) { + (None, None) => return changes, + (None, Some(desired)) => { + // A contract appearing where none existed tightens guarantees — + // safe for consumers, worth noting. + changes.push(ContractChange { + column: String::new(), + kind: "contract_added".to_string(), + detail: format!("contract declared ({} columns)", desired.columns.len()), + safety: ContractSafety::Safe, + }); + return changes; + } + (Some(previous), None) => { + // The contract disappeared entirely — guarantees are gone. + changes.push(ContractChange { + column: String::new(), + kind: "contract_removed".to_string(), + detail: format!("contract dropped ({} columns)", previous.columns.len()), + safety: ContractSafety::Breaking, + }); + return changes; + } + (Some(previous), Some(desired)) => (previous, desired), + }; + + if previous.enforced != desired.enforced { + changes.push(ContractChange { + column: String::new(), + kind: "enforcement_changed".to_string(), + detail: if desired.enforced { + "contract now enforced — violations become errors".to_string() + } else { + "contract no longer enforced — violations downgrade to warnings".to_string() + }, + safety: if desired.enforced { + ContractSafety::Review + } else { + ContractSafety::Breaking + }, + }); + } + + // Declared renames resolve a removed+added pair into a single rename — + // but only when both names actually exist (`new` in the desired + // contract, `old` in the previous contract). A stale rename declaration + // resolves nothing and must not suppress a real removal or addition. + let previous_columns: BTreeMap<&str, &phlo_transform_core::semantic::ColumnContract> = previous + .columns + .iter() + .map(|column| (column.name.as_str(), column)) + .collect(); + let desired_columns: BTreeMap<&str, &phlo_transform_core::semantic::ColumnContract> = desired + .columns + .iter() + .map(|column| (column.name.as_str(), column)) + .collect(); + // Rename resolution is one-to-one: an `old` name claimed by two new + // columns is ambiguous — resolve none of them so the underlying + // removal and additions report normally instead of half the claim + // silently vanishing. + let mut claims: BTreeMap<&str, Vec<&str>> = BTreeMap::new(); + for (new, old) in &desired.renames { + claims.entry(old.as_str()).or_default().push(new.as_str()); + } + let ambiguous: std::collections::BTreeSet<&str> = claims + .iter() + .filter(|(_, news)| news.len() > 1) + .map(|(old, _)| *old) + .collect(); + let resolved_renames: BTreeMap<&str, &str> = desired + .renames + .iter() + .filter(|(new, old)| { + !ambiguous.contains(old.as_str()) + && desired_columns.contains_key(new.as_str()) + && !previous_columns.contains_key(new.as_str()) + && previous_columns.contains_key(old.as_str()) + }) + .map(|(new, old)| (new.as_str(), old.as_str())) + .collect(); + for old in &ambiguous { + if previous_columns.contains_key(old) { + changes.push(ContractChange { + column: (*old).to_string(), + kind: "rename_ambiguous".to_string(), + detail: format!( + "declared renames for `{old}` are ambiguous: {}", + claims[*old].join(", ") + ), + safety: ContractSafety::Review, + }); + } + } + + for (new, old) in &resolved_renames { + // A rename the previous contract already declared is acknowledged + // history — only newly declared renames are changes. + if previous.renames.get(*new).map(String::as_str) != Some(*old) { + changes.push(ContractChange { + column: (*new).to_string(), + kind: "renamed".to_string(), + detail: format!("renamed from `{old}`"), + // A declared rename proves intent, not compatibility — + // consumers selecting the old name break. Gated like a + // removal; a waiver is the acknowledgement. + safety: ContractSafety::Breaking, + }); + } + } + + for column in &previous.columns { + let renamed_away = resolved_renames + .values() + .any(|old| *old == column.name.as_str()); + if !desired_columns.contains_key(column.name.as_str()) && !renamed_away { + changes.push(ContractChange { + column: column.name.clone(), + kind: "removed".to_string(), + detail: "contract column removed".to_string(), + safety: ContractSafety::Breaking, + }); + } + } + + for column in &desired.columns { + // A resolved rename compares the new column against its old-name + // contract so a type or nullability change cannot hide behind the + // rename; otherwise compare same-name columns. + let previous_column = resolved_renames + .get(column.name.as_str()) + .and_then(|old| previous_columns.get(*old).copied()) + .or_else(|| previous_columns.get(column.name.as_str()).copied()); + match previous_column { + None => changes.push(ContractChange { + column: column.name.clone(), + kind: "added".to_string(), + detail: format!("contract column added ({})", describe_column(column)), + safety: ContractSafety::Safe, + }), + Some(previous_column) => { + compare_column_contracts(&mut changes, &column.name, previous_column, column) + } + } + } + + changes +} + +/// Type and nullability changes between two contract columns, emitted under +/// the desired column's name (the rename target when renamed). +fn compare_column_contracts( + changes: &mut Vec, + name: &str, + previous: &phlo_transform_core::semantic::ColumnContract, + desired: &phlo_transform_core::semantic::ColumnContract, +) { + if let (Some(previous_type), Some(desired_type)) = (&previous.data_type, &desired.data_type) { + if previous_type != desired_type { + // Mirror `classify_schema_change`: numeric widening is + // reviewable, anything else breaks consumers. + let widening = previous_type.is_known() + && desired_type.is_known() + && previous_type.is_numeric() + && desired_type.is_numeric() + && DataType::widen(desired_type, previous_type) == *desired_type; + changes.push(ContractChange { + column: name.to_string(), + kind: "type_changed".to_string(), + detail: format!("contract type `{previous_type}` → `{desired_type}`"), + safety: if widening { + ContractSafety::Review + } else { + ContractSafety::Breaking + }, + }); + } + } + if let (Some(was_nullable), Some(is_nullable)) = (previous.nullable, desired.nullable) { + if was_nullable != is_nullable { + changes.push(ContractChange { + column: name.to_string(), + kind: "nullability_changed".to_string(), + detail: if is_nullable { + "contract relaxed to nullable".to_string() + } else { + "contract tightened to non-null".to_string() + }, + // Downstream direction: relaxing a NOT NULL guarantee breaks + // consumers that relied on it; tightening can reject producer + // data but doesn't break downstream compatibility. + safety: if is_nullable { + ContractSafety::Breaking + } else { + ContractSafety::Review + }, + }); + } + } +} + +/// The effective row-identity key of a model — the semantic concept that +/// incremental `key` columns and native unique assertions collapse onto. +/// Each claim contributes one column set: a `Key` strategy's columns, and +/// each `unique` assertion's columns. Sets stay distinct — `unique(a) ∧ +/// unique(b)` is a stronger identity claim than `unique(a,b)`. `None` when +/// the model declares no key at all. +pub fn effective_key(model: &phlo_transform_core::CompiledModel) -> Option>> { + use phlo_transform_core::semantic::Assertion; + let mut key: Vec> = Vec::new(); + if let Some(phlo_transform_core::IncrementalStrategy::Key { columns }) = + &model.config.incremental + { + let mut columns = columns.clone(); + columns.sort(); + key.push(columns); + } + for assertion in &model.assertions { + if let Assertion::Unique { columns } = assertion { + let mut columns = columns.clone(); + columns.sort(); + key.push(columns); + } + } + key.sort(); + key.dedup(); + (!key.is_empty()).then_some(key) +} + +/// What a materialisation record can prove about the key it was written +/// with. `MaterializedRecord::recorded_key` produces this from the persisted +/// columns. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum RecordedKey { + /// The key is known: `Some` claims, or `None` meaning the record + /// definitely proves the materialised model had no key. + Known(Option>>), + /// The row predates effective-key persistence and carries no + /// incremental key either — its historical key cannot be told from + /// "no key". Promotion evidence must not read this as keyless. + Unknown, +} + +impl RecordedKey { + /// The claims the record proves — `None` for an unverifiable record. + pub fn as_deref(&self) -> Option<&[Vec]> { + match self { + RecordedKey::Known(key) => key.as_deref(), + RecordedKey::Unknown => None, + } + } +} + +/// Compare a recorded key against the desired effective key. Changing or +/// dropping the key is breaking: it changes row identity for merges and any +/// consumer relying on the uniqueness guarantee. A newly declared key is a +/// new guarantee — safe direction, worth reviewing. A record whose +/// historical key cannot be verified fails closed: it may be hiding a +/// changed or dropped key. +pub fn key_change( + recorded: &RecordedKey, + desired: Option<&[Vec]>, +) -> Option { + let render = |key: &[Vec]| { + key.iter() + .map(|columns| columns.join("+")) + .collect::>() + .join(", ") + }; + let known = match recorded { + RecordedKey::Unknown => { + return Some(ContractChange { + column: String::new(), + kind: "key_unverifiable".to_string(), + detail: "the base materialisation predates key persistence — its \ + historical key cannot be verified; rematerialise the base \ + environment to bind it" + .to_string(), + safety: ContractSafety::Breaking, + }); + } + RecordedKey::Known(key) => key.as_deref(), + }; + match (known, desired) { + (Some(was), Some(now)) if was != now => Some(ContractChange { + column: String::new(), + kind: "key_changed".to_string(), + detail: format!("key changed: `{}` → `{}`", render(was), render(now)), + safety: ContractSafety::Breaking, + }), + (Some(was), None) => Some(ContractChange { + column: String::new(), + kind: "key_removed".to_string(), + detail: format!("key `{}` dropped", render(was)), + safety: ContractSafety::Breaking, + }), + (None, Some(now)) => Some(ContractChange { + column: String::new(), + kind: "key_added".to_string(), + detail: format!("key `{}` declared", render(now)), + safety: ContractSafety::Review, + }), + _ => None, + } +} + +/// The breaking subset of a contract diff — what promotion must gate on. +pub fn breaking_contract_changes(changes: &[ContractChange]) -> Vec { + changes + .iter() + .filter(|change| change.safety == ContractSafety::Breaking) + .cloned() + .collect() +} + +fn describe_column(column: &phlo_transform_core::semantic::ColumnContract) -> String { + match (&column.data_type, column.nullable) { + (Some(data_type), Some(nullable)) => format!( + "{data_type}, {}", + if nullable { "nullable" } else { "non-null" } + ), + (Some(data_type), None) => data_type.to_string(), + (None, Some(nullable)) => { + if nullable { + "nullable".to_string() + } else { + "non-null".to_string() + } + } + (None, None) => "untyped".to_string(), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use phlo_transform_core::semantic::ColumnContract; + + fn column(name: &str, data_type: Option, nullable: Option) -> ColumnContract { + ColumnContract { + name: name.to_string(), + data_type, + nullable, + } + } + + fn contract(enforced: bool, columns: Vec) -> ModelContract { + ModelContract { + enforced, + columns, + renames: BTreeMap::new(), + } + } + + fn kind<'a>(changes: &'a [ContractChange], column: &str) -> Option<&'a ContractChange> { + changes.iter().find(|change| change.column == column) + } + + #[test] + fn identical_contracts_diff_empty() { + let contract = contract( + true, + vec![column("id", Some(DataType::BigInt), Some(false))], + ); + assert!(contract_diff(Some(&contract), Some(&contract.clone())).is_empty()); + } + + #[test] + fn added_column_is_safe() { + let previous = contract(true, vec![column("id", None, None)]); + let desired = contract( + true, + vec![column("id", None, None), column("note", None, Some(true))], + ); + let changes = contract_diff(Some(&previous), Some(&desired)); + let change = kind(&changes, "note").unwrap(); + assert_eq!(change.kind, "added"); + assert_eq!(change.safety, ContractSafety::Safe); + } + + #[test] + fn removed_column_is_breaking() { + let previous = contract( + true, + vec![column("id", None, None), column("legacy", None, None)], + ); + let desired = contract(true, vec![column("id", None, None)]); + let changes = contract_diff(Some(&previous), Some(&desired)); + let change = kind(&changes, "legacy").unwrap(); + assert_eq!(change.kind, "removed"); + assert_eq!(change.safety, ContractSafety::Breaking); + assert_eq!(breaking_contract_changes(&changes).len(), 1); + } + + #[test] + fn declared_rename_is_breaking() { + // Declaring a rename proves intent, not compatibility — consumers + // selecting `old_name` break, so the gate must see it. + let previous = contract( + true, + vec![column("id", None, None), column("old_name", None, None)], + ); + let mut desired = contract( + true, + vec![column("id", None, None), column("new_name", None, None)], + ); + desired + .renames + .insert("new_name".to_string(), "old_name".to_string()); + let changes = contract_diff(Some(&previous), Some(&desired)); + let change = kind(&changes, "new_name").unwrap(); + assert_eq!(change.kind, "renamed"); + assert_eq!(change.safety, ContractSafety::Breaking); + assert!(kind(&changes, "old_name").is_none()); + assert_eq!(breaking_contract_changes(&changes).len(), 1); + } + + #[test] + fn ambiguous_renames_resolve_nothing() { + // Two new columns claiming the same old name cannot both be the + // rename — resolve neither: the removal stays breaking and both + // additions report. + let previous = contract( + true, + vec![column("id", None, None), column("old_name", None, None)], + ); + let mut desired = contract( + true, + vec![ + column("id", None, None), + column("new_a", None, None), + column("new_b", None, None), + ], + ); + desired + .renames + .insert("new_a".to_string(), "old_name".to_string()); + desired + .renames + .insert("new_b".to_string(), "old_name".to_string()); + let changes = contract_diff(Some(&previous), Some(&desired)); + let removed = changes + .iter() + .find(|change| change.column == "old_name" && change.kind == "removed") + .expect("old_name removal"); + assert_eq!(removed.safety, ContractSafety::Breaking); + assert_eq!(kind(&changes, "new_a").unwrap().kind, "added"); + assert_eq!(kind(&changes, "new_b").unwrap().kind, "added"); + let ambiguous = changes + .iter() + .find(|change| change.kind == "rename_ambiguous") + .expect("ambiguity must be reported"); + assert_eq!(ambiguous.safety, ContractSafety::Review); + } + + #[test] + fn numeric_widening_is_review_incompatible_is_breaking() { + let previous = contract( + true, + vec![ + column("wide", Some(DataType::Integer), None), + column("kind", Some(DataType::BigInt), None), + ], + ); + let desired = contract( + true, + vec![ + column("wide", Some(DataType::BigInt), None), + column("kind", Some(DataType::Varchar), None), + ], + ); + let changes = contract_diff(Some(&previous), Some(&desired)); + assert_eq!( + kind(&changes, "wide").unwrap().safety, + ContractSafety::Review + ); + assert_eq!( + kind(&changes, "kind").unwrap().safety, + ContractSafety::Breaking + ); + } + + #[test] + fn nullability_relax_breaks_tighten_reviews() { + // Downstream direction: losing the NOT NULL guarantee breaks + // consumers; gaining it can only reject producer data. + let previous = contract( + true, + vec![ + column("tight", None, Some(true)), + column("loose", None, Some(false)), + ], + ); + let desired = contract( + true, + vec![ + column("tight", None, Some(false)), + column("loose", None, Some(true)), + ], + ); + let changes = contract_diff(Some(&previous), Some(&desired)); + assert_eq!( + kind(&changes, "tight").unwrap().safety, + ContractSafety::Review + ); + assert_eq!( + kind(&changes, "loose").unwrap().safety, + ContractSafety::Breaking + ); + } + + #[test] + fn enforcement_off_is_breaking_on_is_review() { + let enforced = contract(true, vec![column("id", None, None)]); + let relaxed = contract(false, vec![column("id", None, None)]); + let off = contract_diff(Some(&enforced), Some(&relaxed)); + assert_eq!(off[0].kind, "enforcement_changed"); + assert_eq!(off[0].safety, ContractSafety::Breaking); + let on = contract_diff(Some(&relaxed), Some(&enforced)); + assert_eq!(on[0].kind, "enforcement_changed"); + assert_eq!(on[0].safety, ContractSafety::Review); + } + + #[test] + fn dropped_contract_is_breaking_new_contract_is_safe() { + let existing = contract(true, vec![column("id", None, None)]); + let dropped = contract_diff(Some(&existing), None); + assert_eq!(dropped[0].kind, "contract_removed"); + assert_eq!(dropped[0].safety, ContractSafety::Breaking); + let declared = contract_diff(None, Some(&existing)); + assert_eq!(declared[0].kind, "contract_added"); + assert_eq!(declared[0].safety, ContractSafety::Safe); + assert!(contract_diff(None, None).is_empty()); + } + + #[test] + fn stale_rename_is_ignored() { + // A rename whose new name is not in the desired contract cannot + // explain the old column's removal. + let previous = contract(true, vec![column("old_name", None, None)]); + let mut desired = contract(true, Vec::new()); + desired + .renames + .insert("not_present".to_string(), "old_name".to_string()); + let changes = contract_diff(Some(&previous), Some(&desired)); + assert_eq!(kind(&changes, "old_name").unwrap().kind, "removed"); + } + + #[test] + fn rename_to_an_existing_column_does_not_hide_removal() { + // `new_name` was already a contract column before — claiming + // `old_name` became it is ambiguous, so the removal stays breaking. + let previous = contract( + true, + vec![ + column("id", None, None), + column("new_name", None, None), + column("old_name", None, None), + ], + ); + let mut desired = contract( + true, + vec![column("id", None, None), column("new_name", None, None)], + ); + desired + .renames + .insert("new_name".to_string(), "old_name".to_string()); + let changes = contract_diff(Some(&previous), Some(&desired)); + let change = kind(&changes, "old_name").unwrap(); + assert_eq!(change.kind, "removed"); + assert_eq!(change.safety, ContractSafety::Breaking); + } + + #[test] + fn key_changes_classify() { + let key = |columns: &[&str]| -> RecordedKey { + RecordedKey::Known(Some(vec![columns.iter().map(|c| c.to_string()).collect()])) + }; + let was = key(&["id"]); + let now = key(&["batch_id"]); + let same = key(&["id"]); + let none = RecordedKey::Known(None); + + assert!(key_change(&was, same.as_deref()).is_none()); + assert!(key_change(&none, None).is_none()); + + let changed = key_change(&was, now.as_deref()).expect("key_changed"); + assert_eq!(changed.kind, "key_changed"); + assert_eq!(changed.safety, ContractSafety::Breaking); + + let removed = key_change(&was, None).expect("key_removed"); + assert_eq!(removed.kind, "key_removed"); + assert_eq!(removed.safety, ContractSafety::Breaking); + + let added = key_change(&none, now.as_deref()).expect("key_added"); + assert_eq!(added.kind, "key_added"); + assert_eq!(added.safety, ContractSafety::Review); + + // A composite key split into separate unique claims is a different + // identity — it reports, not silently equal. + let composite = key(&["a", "b"]); + let separate = RecordedKey::Known(Some(vec![vec!["a".to_string()], vec!["b".to_string()]])); + assert!(key_change(&composite, separate.as_deref()).is_some()); + + // A record that cannot prove its historical key fails closed — + // whether the workspace declares a key or not, an unverifiable + // record may be hiding a change or a removal. + for desired in [None, now.as_deref()] { + let unverifiable = + key_change(&RecordedKey::Unknown, desired).expect("key_unverifiable"); + assert_eq!(unverifiable.kind, "key_unverifiable"); + assert_eq!(unverifiable.safety, ContractSafety::Breaking); + } + } +} diff --git a/crates/phlo-transform-engine/src/diff.rs b/crates/phlo-transform-engine/src/diff.rs index cf4693a..eb269ed 100644 --- a/crates/phlo-transform-engine/src/diff.rs +++ b/crates/phlo-transform-engine/src/diff.rs @@ -66,6 +66,9 @@ pub struct DiffRequest { pub strategy: DiffStrategy, pub policy: DiffPolicy, pub sample_fraction: Option, + /// Declared column renames (`candidate name` → `base name`) — turns a + /// removed+added schema pair into a reviewable rename. + pub renames: std::collections::BTreeMap, } /// Row-level summary. @@ -436,7 +439,13 @@ fn partition_delta( } async fn schema_changes(adapter: &dyn Adapter, request: &DiffRequest) -> Vec { - compare_columns(adapter, &request.candidate_relation, &request.base_relation).await + compare_columns( + adapter, + &request.candidate_relation, + &request.base_relation, + &request.renames, + ) + .await } /// Column-level comparison of two relations: added/removed columns, type @@ -446,6 +455,7 @@ pub(crate) async fn compare_columns( adapter: &dyn Adapter, candidate: &phlo_transform_core::Relation, base: &phlo_transform_core::Relation, + renames: &BTreeMap, ) -> Vec { let Ok(candidate_columns) = adapter.relation_columns(candidate).await else { return Vec::new(); @@ -453,14 +463,18 @@ pub(crate) async fn compare_columns( let Ok(base_columns) = adapter.relation_columns(base).await else { return Vec::new(); }; - compare_column_lists(&candidate_columns, &base_columns) + compare_column_lists(&candidate_columns, &base_columns, renames) } /// Column-level comparison of two column lists — usable when only one side -/// exists (pass an empty list for the absent side). +/// exists (pass an empty list for the absent side). `renames` maps a +/// candidate column name to the base column it was renamed from, so a +/// remove+add pair that is a declared rename reports as `renamed` instead of +/// a breaking removal. pub(crate) fn compare_column_lists( candidate_columns: &[crate::adapter::ColumnInfo], base_columns: &[crate::adapter::ColumnInfo], + renames: &BTreeMap, ) -> Vec { if candidate_columns.is_empty() && base_columns.is_empty() { return Vec::new(); @@ -486,14 +500,79 @@ pub(crate) fn compare_column_lists( .collect(); let mut changes = Vec::new(); + // Renames that actually resolve: the new name exists only on the + // candidate side and the old name exists on the base side. A stale + // declaration (`new` absent, `old` absent, or `new` already a base + // column) explains nothing and must not suppress a real addition or + // removal. Renames are one-to-one: an `old` name claimed by two new + // columns is ambiguous — resolve none of them so the removal and both + // additions report normally. + let mut claims: BTreeMap<&str, Vec<&str>> = BTreeMap::new(); + for (new, old) in renames { + claims.entry(old.as_str()).or_default().push(new.as_str()); + } + let ambiguous: std::collections::BTreeSet<&str> = claims + .iter() + .filter(|(_, news)| news.len() > 1) + .map(|(old, _)| *old) + .collect(); + let resolved: std::collections::BTreeSet<&str> = renames + .iter() + .filter(|(new, old)| { + !ambiguous.contains(old.as_str()) + && candidate_types.contains_key(new.as_str()) + && !base_types.contains_key(new.as_str()) + && base_types.contains_key(old.as_str()) + }) + .map(|(new, _)| new.as_str()) + .collect(); + for (old, news) in &claims { + if ambiguous.contains(old) && base_types.contains_key(*old) { + changes.push(SchemaChange { + column: (*old).to_string(), + kind: "rename_ambiguous".to_string(), + detail: format!( + "declared renames for `{old}` are ambiguous: {}", + news.join(", ") + ), + safety: "review".to_string(), + }); + } + } + // Base columns renamed to a new candidate name — `old -> new`. + let renamed_away: BTreeMap<&str, &str> = renames + .iter() + .filter(|(new, _)| resolved.contains(new.as_str())) + .map(|(new, old)| (old.as_str(), new.as_str())) + .collect(); for (name, (base_type, base_nullable)) in &base_types { match candidate_types.get(name) { - None => changes.push(SchemaChange { - column: name.clone(), - kind: "removed".to_string(), - detail: format!("{base_type} removed"), - safety: "full_rebuild_required".to_string(), - }), + None => { + if let Some(new_name) = renamed_away.get(name.as_str()) { + let (new_type, _) = &candidate_types[*new_name]; + let type_note = if new_type != base_type { + format!(" (type {base_type} -> {new_type})") + } else { + String::new() + }; + changes.push(SchemaChange { + column: (*new_name).to_string(), + kind: "renamed".to_string(), + detail: format!("`{name}` renamed to `{new_name}`{type_note}"), + // Declaring a rename proves intent, not + // compatibility — consumers selecting the old name + // break, so it gates like a removal. + safety: "full_rebuild_required".to_string(), + }); + } else { + changes.push(SchemaChange { + column: name.clone(), + kind: "removed".to_string(), + detail: format!("{base_type} removed"), + safety: "full_rebuild_required".to_string(), + }); + } + } Some((candidate_type, candidate_nullable)) => { // Type and nullability are independent changes — a column // that moved on both reports both. @@ -518,8 +597,11 @@ pub(crate) fn compare_column_lists( } else { "nullable -> not null".to_string() }, + // Downstream direction: losing the NOT NULL + // guarantee breaks consumers that relied on it; + // gaining it can only reject producer data. safety: if *candidate_nullable { - "safe".to_string() + "error".to_string() } else { "review".to_string() }, @@ -529,7 +611,7 @@ pub(crate) fn compare_column_lists( } } for (name, (candidate_type, _)) in &candidate_types { - if !base_types.contains_key(name) { + if !base_types.contains_key(name) && !resolved.contains(name.as_str()) { changes.push(SchemaChange { column: name.clone(), kind: "added".to_string(), @@ -691,6 +773,7 @@ mod tests { strategy: DiffStrategy::Keyed, policy: DiffPolicy::default(), sample_fraction: None, + renames: Default::default(), } } @@ -777,11 +860,99 @@ mod tests { data_type: "varchar".to_string(), nullable: false, }]; - let changes = compare_column_lists(&candidate, &base); + let changes = compare_column_lists(&candidate, &base, &BTreeMap::new()); assert_eq!(changes.len(), 2); assert_eq!(changes[0].kind, "changed"); assert_eq!(changes[1].kind, "nullability"); assert_eq!(changes[1].detail, "not null -> nullable"); + // Losing a NOT NULL guarantee breaks downstream consumers. + assert_eq!(changes[1].safety, "error"); + } + + #[test] + fn declared_rename_reports_renamed_not_removed_and_added() { + let info = |name: &str| crate::adapter::ColumnInfo { + name: name.to_string(), + data_type: "bigint".to_string(), + nullable: true, + }; + let candidate = [info("id"), info("new_col")]; + let base = [info("id"), info("old_col")]; + let mut renames = BTreeMap::new(); + renames.insert("new_col".to_string(), "old_col".to_string()); + let changes = compare_column_lists(&candidate, &base, &renames); + assert_eq!(changes.len(), 1); + assert_eq!(changes[0].kind, "renamed"); + assert_eq!(changes[0].column, "new_col"); + // A declared rename proves intent, not compatibility — consumers + // selecting the old name break, so it is a breaking change. + assert_eq!(changes[0].safety, "full_rebuild_required"); + + // Without the declaration the same pair is a breaking removal. + let changes = compare_column_lists(&candidate, &base, &BTreeMap::new()); + assert!(changes.iter().any(|change| change.kind == "removed")); + assert!(changes.iter().any(|change| change.kind == "added")); + } + + #[test] + fn ambiguous_rename_resolves_nothing() { + // Two new columns claiming the same old name: resolve neither — the + // removal stays breaking and both additions report, plus an + // ambiguity diagnostic explains why. + let info = |name: &str| crate::adapter::ColumnInfo { + name: name.to_string(), + data_type: "bigint".to_string(), + nullable: true, + }; + let candidate = [info("id"), info("new_a"), info("new_b")]; + let base = [info("id"), info("old_col")]; + let mut renames = BTreeMap::new(); + renames.insert("new_a".to_string(), "old_col".to_string()); + renames.insert("new_b".to_string(), "old_col".to_string()); + let changes = compare_column_lists(&candidate, &base, &renames); + let removed = changes + .iter() + .find(|change| change.column == "old_col" && change.kind == "removed") + .expect("old_col removal"); + assert_eq!(removed.safety, "full_rebuild_required"); + for new in ["new_a", "new_b"] { + let added = changes + .iter() + .find(|change| change.column == new && change.kind == "added") + .unwrap_or_else(|| panic!("{new} must report added")); + assert_eq!(added.safety, "safe"); + } + assert!(changes + .iter() + .any(|change| change.kind == "rename_ambiguous" && change.column == "old_col")); + } + + #[test] + fn stale_rename_neither_hides_addition_nor_removal() { + let info = |name: &str| crate::adapter::ColumnInfo { + name: name.to_string(), + data_type: "bigint".to_string(), + nullable: true, + }; + // `old_col` was never a base column — the declaration explains + // nothing, and the genuinely-new `new_col` still reports added. + let candidate = [info("id"), info("new_col")]; + let base = [info("id")]; + let mut renames = BTreeMap::new(); + renames.insert("new_col".to_string(), "old_col".to_string()); + let changes = compare_column_lists(&candidate, &base, &renames); + assert_eq!(changes.len(), 1); + assert_eq!(changes[0].kind, "added"); + assert_eq!(changes[0].column, "new_col"); + + // `new_col` already existed on the base side — the declaration is + // ambiguous, so `old_col`'s disappearance stays a breaking removal. + let candidate = [info("id"), info("new_col")]; + let base = [info("id"), info("new_col"), info("old_col")]; + let changes = compare_column_lists(&candidate, &base, &renames); + assert_eq!(changes.len(), 1); + assert_eq!(changes[0].kind, "removed"); + assert_eq!(changes[0].column, "old_col"); } #[test] diff --git a/crates/phlo-transform-engine/src/lib.rs b/crates/phlo-transform-engine/src/lib.rs index e62676b..2b9fccf 100644 --- a/crates/phlo-transform-engine/src/lib.rs +++ b/crates/phlo-transform-engine/src/lib.rs @@ -11,6 +11,7 @@ pub mod artifacts; pub mod branch_diff; pub mod cancel; pub mod changed; +pub mod contracts; pub mod diff; pub mod environment; pub mod error; @@ -33,11 +34,15 @@ pub use artifacts::{ }; pub use branch_diff::{ branch_diff, materialized_for_environment, model_keys, retarget, seeds_for_environment, - BranchDiffReport, BranchDiffRequest, DatasetDiff, DatasetKind, DatasetStatus, ModelRowDiff, - ModelSchemaDiff, + BranchDiffReport, BranchDiffRequest, DatasetDiff, DatasetKind, DatasetStatus, + ModelContractDiff, ModelRowDiff, ModelSchemaDiff, }; pub use cancel::CancelHandle; pub use changed::changed_models; +pub use contracts::{ + breaking_contract_changes, contract_diff, effective_key, key_change, ContractChange, + ContractSafety, RecordedKey, +}; pub use diff::{ diff, DiffPolicy, DiffReport, DiffRequest, DiffStrategy, PolicyResult, RowSummary, SchemaChange, }; diff --git a/crates/phlo-transform-engine/src/plan.rs b/crates/phlo-transform-engine/src/plan.rs index 5d8ef2a..4d27e50 100644 --- a/crates/phlo-transform-engine/src/plan.rs +++ b/crates/phlo-transform-engine/src/plan.rs @@ -1090,10 +1090,7 @@ pub fn diff_reasons(model: &CompiledModel, current: &MaterializedRecord) -> Vec< )); } if current.version.contract_hash != desired.contract_hash { - reasons.push(PlanReason::simple( - ReasonKind::ContractChange, - "contract or assertions changed", - )); + reasons.push(contract_diff_reason(model, current)); } if current.version.dependency_hash != desired.dependency_hash { reasons.push(dependency_diff_reason(model, current)); @@ -1120,6 +1117,65 @@ pub fn diff_reasons(model: &CompiledModel, current: &MaterializedRecord) -> Vec< reasons } +/// Explain a contract-hash change with the structured contract diff when +/// both sides' contracts are available — names the breaking changes instead +/// of reporting an opaque hash change. +fn contract_diff_reason(model: &CompiledModel, current: &MaterializedRecord) -> PlanReason { + let Some(changes) = model + .contract + .as_ref() + .map(|desired| crate::contracts::contract_diff(current.contract.as_ref(), Some(desired))) + else { + // The model dropped its contract — report what the base recorded. + if let Some(previous) = ¤t.contract { + let changes = crate::contracts::contract_diff(Some(previous), None); + return contract_reason_from_changes(changes); + } + return PlanReason::simple(ReasonKind::ContractChange, "contract or assertions changed"); + }; + if changes.is_empty() { + // Assertions changed, or the recorded contract is missing — the hash + // moved but no column-level diff is derivable. + return PlanReason::simple(ReasonKind::ContractChange, "contract or assertions changed"); + } + contract_reason_from_changes(changes) +} + +fn contract_reason_from_changes(changes: Vec) -> PlanReason { + let breaking = changes + .iter() + .filter(|change| change.safety == crate::contracts::ContractSafety::Breaking) + .count(); + let summary = changes + .iter() + .take(3) + .map(|change| { + if change.column.is_empty() { + change.detail.clone() + } else { + format!("{}: {}", change.column, change.detail) + } + }) + .collect::>() + .join("; "); + let more = if changes.len() > 3 { + format!(" (+{} more)", changes.len() - 3) + } else { + String::new() + }; + PlanReason::simple( + ReasonKind::ContractChange, + format!( + "contract changed: {summary}{more}{}", + if breaking > 0 { + format!(" — {breaking} breaking") + } else { + String::new() + } + ), + ) +} + /// Explain a changed dependency hash in terms of which dependency's version /// input moved, using the detail recorded at materialisation time. fn dependency_diff_reason(model: &CompiledModel, current: &MaterializedRecord) -> PlanReason { diff --git a/crates/phlo-transform-engine/src/run.rs b/crates/phlo-transform-engine/src/run.rs index 43b4306..0617d02 100644 --- a/crates/phlo-transform-engine/src/run.rs +++ b/crates/phlo-transform-engine/src/run.rs @@ -1715,6 +1715,13 @@ impl Runner { .filter(|key| !key.is_empty()), adapter: Some(self.adapter.name().to_string()), output_identity: result.output_identity.clone(), + contract: model.contract.clone(), + // Always a definite value: `Some([])` records + // "materialised keyless", distinguishing it + // from a legacy row's NULL (unknown). + effective_key: Some( + crate::contracts::effective_key(model).unwrap_or_default(), + ), run_id: run_id.clone(), materialized_at: finished_at.clone(), })?; diff --git a/crates/phlo-transform-engine/src/state.rs b/crates/phlo-transform-engine/src/state.rs index 3df6298..c9332ff 100644 --- a/crates/phlo-transform-engine/src/state.rs +++ b/crates/phlo-transform-engine/src/state.rs @@ -10,6 +10,7 @@ use std::sync::Mutex; use rusqlite::Connection; use serde::Serialize; +use phlo_transform_core::semantic::ModelContract; use phlo_transform_core::{ModelVersion, VersionDetail}; use crate::error::EngineError; @@ -182,10 +183,71 @@ pub struct MaterializedRecord { /// cross-environment cache hit. #[serde(skip_serializing_if = "Option::is_none")] pub output_identity: Option, + /// The model's contract at materialisation time — lets promotion diff + /// the candidate's contract against what the base environment recorded + /// instead of comparing opaque contract hashes. + #[serde(skip_serializing_if = "Option::is_none")] + pub contract: Option, + /// The effective row-identity key at materialisation time — the union + /// of the `key` incremental strategy's columns and each unique + /// assertion's columns, one set per claim. `Some(vec![])` means the + /// record proves the model had no key; `None` means nothing about the + /// key was persisted (rows written before the column existed). + #[serde(skip_serializing_if = "Option::is_none")] + pub effective_key: Option>>, pub run_id: String, pub materialized_at: String, } +impl MaterializedRecord { + /// What this record proves about the row-identity key it was + /// materialised with. + /// + /// - a persisted `effective_key` (including an empty set — recorded + /// keyless) is known evidence; + /// - a legacy row with an incremental `key` strategy reconstructs its + /// columns: before `effective_key` existed that strategy was the only + /// key concept state recorded; + /// - anything else is `Unknown` — its historical key cannot be told + /// from "no key", and promotion evidence must not treat it as + /// keyless. + pub fn recorded_key(&self) -> crate::contracts::RecordedKey { + use crate::contracts::RecordedKey; + if let Some(key) = &self.effective_key { + let mut key = key.clone(); + for claim in &mut key { + claim.sort(); + } + key.sort(); + return RecordedKey::Known((!key.is_empty()).then_some(key)); + } + if self.incremental_strategy.as_deref() == Some("key") { + if let Some(claim) = self + .incremental_key + .as_deref() + .and_then(incremental_key_claim) + { + return RecordedKey::Known(Some(claim)); + } + } + RecordedKey::Unknown + } +} + +/// The canonical key claim a persisted `incremental_key` string encodes — +/// the comma-joined columns, sorted as `effective_key` claims are. `None` +/// for an empty column list. +pub(crate) fn incremental_key_claim(raw: &str) -> Option>> { + let mut columns: Vec = raw + .split(',') + .map(str::trim) + .filter(|column| !column.is_empty()) + .map(str::to_string) + .collect(); + columns.sort(); + (!columns.is_empty()).then_some(vec![columns]) +} + /// A seed load recorded against an environment. #[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)] pub struct SeedRecord { @@ -411,6 +473,8 @@ impl SqliteStateStore { version_detail TEXT, adapter TEXT, output_identity TEXT, + contract_json TEXT, + effective_key_json TEXT, PRIMARY KEY (model_id, environment) ); CREATE TABLE IF NOT EXISTS incremental_state ( @@ -460,6 +524,14 @@ impl SqliteStateStore { "ALTER TABLE model_versions ADD COLUMN output_identity TEXT", [], ); + let _ = connection.execute( + "ALTER TABLE model_versions ADD COLUMN contract_json TEXT", + [], + ); + let _ = connection.execute( + "ALTER TABLE model_versions ADD COLUMN effective_key_json TEXT", + [], + ); // Run-progress columns added for resumable runs. let _ = connection.execute("ALTER TABLE runs ADD COLUMN plan_json TEXT", []); // The Nessie commit a run validated — added for promotion provenance. @@ -473,6 +545,46 @@ impl SqliteStateStore { ] { let _ = connection.execute(column, []); } + // Backfill `effective_key_json` for rows written before the column + // existed: an incremental `key` strategy's columns were the era's + // effective key — the only key concept state then recorded. Rows + // without one stay NULL: their key is unknown, not absent. + let legacy_keys = { + let mut statement = connection + .prepare( + "SELECT model_id, environment, incremental_key FROM model_versions + WHERE effective_key_json IS NULL + AND incremental_strategy = 'key' + AND incremental_key IS NOT NULL", + ) + .map_err(|error| EngineError::State(error.to_string()))?; + let rows = statement + .query_map([], |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, String>(1)?, + row.get::<_, String>(2)?, + )) + }) + .map_err(|error| EngineError::State(error.to_string()))? + .collect::>>() + .map_err(|error| EngineError::State(error.to_string()))?; + rows + }; + for (model_id, environment, raw) in legacy_keys { + let Some(claim) = incremental_key_claim(&raw) else { + continue; + }; + let json = serde_json::to_string(&claim) + .map_err(|error| EngineError::State(error.to_string()))?; + connection + .execute( + "UPDATE model_versions SET effective_key_json = ?1 + WHERE model_id = ?2 AND environment = ?3", + rusqlite::params![json, model_id, environment], + ) + .map_err(|error| EngineError::State(error.to_string()))?; + } Ok(Self { connection: Mutex::new(connection), }) @@ -899,6 +1011,14 @@ impl StateStore for SqliteStateStore { .detail .as_ref() .map(|detail| serde_json::to_string(detail).unwrap_or_default()); + let contract = record + .contract + .as_ref() + .map(|contract| serde_json::to_string(contract).unwrap_or_default()); + let effective_key = record + .effective_key + .as_ref() + .map(|key| serde_json::to_string(key).unwrap_or_default()); // Ordered upsert: a record is only applied when it is not older than // what is already stored. Two concurrent runs can finish out of // order — the run that materialised later physically overwrote the @@ -910,8 +1030,8 @@ impl StateStore for SqliteStateStore { (model_id, environment, version_hash, sql_hash, config_hash, contract_hash, dependency_hash, source_state_hash, compiler_version, target_hash, target, run_id, materialized_at, incremental_strategy, incremental_key, version_detail, - adapter, output_identity) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18) + adapter, output_identity, contract_json, effective_key_json) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, ?19, ?20) ON CONFLICT (model_id, environment) DO UPDATE SET version_hash = excluded.version_hash, sql_hash = excluded.sql_hash, @@ -928,7 +1048,9 @@ impl StateStore for SqliteStateStore { incremental_key = excluded.incremental_key, version_detail = excluded.version_detail, adapter = excluded.adapter, - output_identity = excluded.output_identity + output_identity = excluded.output_identity, + contract_json = excluded.contract_json, + effective_key_json = excluded.effective_key_json WHERE excluded.materialized_at >= model_versions.materialized_at", rusqlite::params![ record.model_id, @@ -949,6 +1071,8 @@ impl StateStore for SqliteStateStore { detail, record.adapter, record.output_identity, + contract, + effective_key, ], ) .map_err(|error| EngineError::State(error.to_string()))?; @@ -1227,7 +1351,7 @@ impl StateStore for SqliteStateStore { const MATERIALIZED_COLUMNS: &str = "model_id, environment, version_hash, sql_hash, config_hash, \ contract_hash, dependency_hash, source_state_hash, compiler_version, target_hash, target, \ run_id, materialized_at, incremental_strategy, incremental_key, version_detail, adapter, \ - output_identity"; + output_identity, contract_json, effective_key_json"; fn materialized_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result { let environment: String = row.get(1)?; @@ -1256,6 +1380,12 @@ fn materialized_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result>(18)? + .and_then(|json| serde_json::from_str(&json).ok()), + effective_key: row + .get::<_, Option>(19)? + .and_then(|json| serde_json::from_str(&json).ok()), run_id: row.get(11)?, materialized_at: row.get(12)?, }) @@ -1351,6 +1481,8 @@ mod tests { incremental_key: None, adapter: Some("trino".to_string()), output_identity: Some(format!("snap:{hash}")), + contract: None, + effective_key: Some(Vec::new()), run_id: run_id.to_string(), materialized_at: at.to_string(), } @@ -1366,14 +1498,9 @@ mod tests { let writer_a = SqliteStateStore::open(&path).unwrap(); let writer_b = SqliteStateStore::open(&path).unwrap(); - writer_b - .record_materialized(&materialized( - "assay.results", - "v2", - "2026-01-01T00:00:02Z", - "run-b", - )) - .unwrap(); + let mut winning = materialized("assay.results", "v2", "2026-01-01T00:00:02Z", "run-b"); + winning.effective_key = Some(vec![vec!["sample_id".to_string()]]); + writer_b.record_materialized(&winning).unwrap(); writer_a .record_materialized(&materialized( "assay.results", @@ -1388,6 +1515,11 @@ mod tests { .expect("record"); assert_eq!(record.version.hash, "v2", "the stale write must lose"); assert_eq!(record.run_id, "run-b"); + // The persisted effective key survives the write and the read. + assert_eq!( + record.effective_key.as_deref(), + Some(&[vec!["sample_id".to_string()]][..]) + ); // An equal or later timestamp still applies — same-run re-records // and genuinely newer writes are unaffected. @@ -1406,6 +1538,54 @@ mod tests { assert_eq!(record.version.hash, "v3"); } + /// Reopening a database backfills `effective_key_json` for legacy rows + /// that recorded an incremental `key` strategy — the era's only key + /// concept — and leaves rows with no key evidence NULL (unknown, not + /// keyless). + #[test] + fn reopening_backfills_legacy_incremental_keys() { + use crate::contracts::RecordedKey; + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("state.db"); + + { + let store = SqliteStateStore::open(&path).unwrap(); + // A legacy incremental-key row: nothing in `effective_key`, + // the incremental fields carry the era's key. + let mut keyed = materialized("assay.keyed", "v1", "2026-01-01T00:00:01Z", "run-1"); + keyed.incremental_strategy = Some("key".to_string()); + keyed.incremental_key = Some("b, a".to_string()); + keyed.effective_key = None; + store.record_materialized(&keyed).unwrap(); + // A legacy row with no key evidence at all stays NULL. + let mut bare = materialized("assay.bare", "v1", "2026-01-01T00:00:01Z", "run-1"); + bare.effective_key = None; + store.record_materialized(&bare).unwrap(); + } + + let store = SqliteStateStore::open(&path).unwrap(); + let keyed = store + .materialized_version("assay.keyed", Some("prod")) + .unwrap() + .expect("keyed record"); + assert_eq!( + keyed.effective_key.as_deref(), + Some(&[vec!["a".to_string(), "b".to_string()]][..]), + "the backfill writes the canonical sorted claim" + ); + assert_eq!( + keyed.recorded_key().as_deref(), + Some(&[vec!["a".to_string(), "b".to_string()]][..]) + ); + + let bare = store + .materialized_version("assay.bare", Some("prod")) + .unwrap() + .expect("bare record"); + assert_eq!(bare.effective_key, None, "no evidence to backfill from"); + assert_eq!(bare.recorded_key(), RecordedKey::Unknown); + } + #[test] fn out_of_order_seed_writes_do_not_regress() { let dir = tempfile::tempdir().unwrap(); diff --git a/crates/phlo-transform-engine/src/state_postgres.rs b/crates/phlo-transform-engine/src/state_postgres.rs index 29a4c80..2e8a22f 100644 --- a/crates/phlo-transform-engine/src/state_postgres.rs +++ b/crates/phlo-transform-engine/src/state_postgres.rs @@ -26,8 +26,8 @@ use phlo_transform_core::{ModelVersion, VersionDetail}; use crate::error::EngineError; use crate::events::ExecutionStatus; use crate::state::{ - status_str, MaterializedRecord, ModelRunRecord, RunRecord, RunSummary, SeedRecord, - SeedRunRecord, StateStore, StoredPlan, StoredRun, TestRunRecord, + incremental_key_claim, status_str, MaterializedRecord, ModelRunRecord, RunRecord, RunSummary, + SeedRecord, SeedRunRecord, StateStore, StoredPlan, StoredRun, TestRunRecord, }; use crate::util::now_rfc3339; @@ -40,7 +40,7 @@ const MODEL_RUN_COLUMNS: &str = "run_id, model_id, materialization, status, star const MATERIALIZED_COLUMNS: &str = "model_id, environment, version_hash, sql_hash, config_hash, \ contract_hash, dependency_hash, source_state_hash, compiler_version, target_hash, target, \ run_id, materialized_at, incremental_strategy, incremental_key, version_detail, adapter, \ - output_identity"; + output_identity, contract_json, effective_key_json"; fn parse_status(value: &str) -> ExecutionStatus { match value { @@ -106,6 +106,8 @@ fn model_run_from_row(row: &Row) -> ModelRunRecord { fn materialized_from_row(row: &Row) -> MaterializedRecord { let environment: String = row.get(1); let detail: Option = row.get(15); + let contract: Option = row.get(18); + let effective_key: Option = row.get(19); MaterializedRecord { model_id: row.get(0), environment: if environment.is_empty() { @@ -129,6 +131,8 @@ fn materialized_from_row(row: &Row) -> MaterializedRecord { incremental_key: row.get(14), adapter: row.get(16), output_identity: row.get(17), + contract: contract.and_then(|json| serde_json::from_str(&json).ok()), + effective_key: effective_key.and_then(|json| serde_json::from_str(&json).ok()), run_id: row.get(11), materialized_at: row.get(12), } @@ -317,6 +321,8 @@ fn ensure_schema(client: &mut Client) -> Result<(), String> { version_detail TEXT, adapter TEXT, output_identity TEXT, + contract_json TEXT, + effective_key_json TEXT, PRIMARY KEY (model_id, environment) ); CREATE TABLE IF NOT EXISTS incremental_state ( @@ -357,6 +363,8 @@ fn ensure_schema(client: &mut Client) -> Result<(), String> { "ALTER TABLE model_versions ADD COLUMN version_detail TEXT", "ALTER TABLE model_versions ADD COLUMN adapter TEXT", "ALTER TABLE model_versions ADD COLUMN output_identity TEXT", + "ALTER TABLE model_versions ADD COLUMN contract_json TEXT", + "ALTER TABLE model_versions ADD COLUMN effective_key_json TEXT", "ALTER TABLE runs ADD COLUMN plan_json TEXT", "ALTER TABLE runs ADD COLUMN reference_hash TEXT", "ALTER TABLE runs ADD COLUMN seq BIGINT GENERATED ALWAYS AS IDENTITY", @@ -370,6 +378,33 @@ fn ensure_schema(client: &mut Client) -> Result<(), String> { ] { let _ = client.execute(column, &[]); } + // Backfill `effective_key_json` for rows written before the column + // existed: an incremental `key` strategy's columns were the era's + // effective key — the only key concept state then recorded. Rows + // without one stay NULL: their key is unknown, not absent. + let legacy_keys = client + .query( + "SELECT model_id, environment, incremental_key FROM model_versions + WHERE effective_key_json IS NULL + AND incremental_strategy = 'key' + AND incremental_key IS NOT NULL", + &[], + ) + .map_err(|error| error.to_string())?; + for row in legacy_keys { + let raw: String = row.get(2); + let Some(claim) = incremental_key_claim(&raw) else { + continue; + }; + let json = serde_json::to_string(&claim).map_err(|error| error.to_string())?; + client + .execute( + "UPDATE model_versions SET effective_key_json = $1 + WHERE model_id = $2 AND environment = $3", + &[&json, &row.get::<_, String>(0), &row.get::<_, String>(1)], + ) + .map_err(|error| error.to_string())?; + } Ok(()) } @@ -734,6 +769,14 @@ impl StateStore for PostgresStateStore { .detail .as_ref() .map(|detail| serde_json::to_string(detail).unwrap_or_default()); + let contract = record + .contract + .as_ref() + .map(|contract| serde_json::to_string(contract).unwrap_or_default()); + let effective_key = record + .effective_key + .as_ref() + .map(|key| serde_json::to_string(key).unwrap_or_default()); let environment = record.environment.clone().unwrap_or_default(); // Ordered upsert: only applied when the incoming record is not // older than what is stored — a later-materialised run physically @@ -746,8 +789,8 @@ impl StateStore for PostgresStateStore { (model_id, environment, version_hash, sql_hash, config_hash, contract_hash, dependency_hash, source_state_hash, compiler_version, target_hash, target, run_id, materialized_at, incremental_strategy, incremental_key, version_detail, - adapter, output_identity) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18) + adapter, output_identity, contract_json, effective_key_json) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20) ON CONFLICT (model_id, environment) DO UPDATE SET version_hash = EXCLUDED.version_hash, sql_hash = EXCLUDED.sql_hash, @@ -764,7 +807,9 @@ impl StateStore for PostgresStateStore { incremental_key = EXCLUDED.incremental_key, version_detail = EXCLUDED.version_detail, adapter = EXCLUDED.adapter, - output_identity = EXCLUDED.output_identity + output_identity = EXCLUDED.output_identity, + contract_json = EXCLUDED.contract_json, + effective_key_json = EXCLUDED.effective_key_json WHERE EXCLUDED.materialized_at >= model_versions.materialized_at", &[ &record.model_id, @@ -785,6 +830,8 @@ impl StateStore for PostgresStateStore { &detail, &record.adapter, &record.output_identity, + &contract, + &effective_key, ], ) .map_err(map_error)?; @@ -1128,6 +1175,8 @@ mod tests { incremental_key: None, adapter: Some("trino".to_string()), output_identity: Some("snap:1".to_string()), + contract: None, + effective_key: Some(vec![vec!["sample_id".to_string()]]), run_id: "run-pg-1".to_string(), materialized_at: "2026-01-01T00:00:01Z".to_string(), }) @@ -1139,6 +1188,10 @@ mod tests { assert_eq!(record.version.hash, "v1"); assert_eq!(record.adapter.as_deref(), Some("trino")); assert_eq!(record.output_identity.as_deref(), Some("snap:1")); + assert_eq!( + record.effective_key.as_deref(), + Some(&[vec!["sample_id".to_string()]][..]) + ); assert_eq!(store.materialized_by_hash("v1").expect("by hash").len(), 1); // Same-key writes are ordered by materialisation time: a later diff --git a/crates/phlo-transform-engine/tests/engine.rs b/crates/phlo-transform-engine/tests/engine.rs index da9a5f2..81d1e17 100644 --- a/crates/phlo-transform-engine/tests/engine.rs +++ b/crates/phlo-transform-engine/tests/engine.rs @@ -12,20 +12,21 @@ use std::time::Duration; use async_trait::async_trait; use phlo_transform_core::{ - compile, compile_with_options, resolve_selection, Compilation, DataType, DiffPolicySpec, - EmptySchemaProvider, EmptySourceStateProvider, IncrementalStrategy, Materialization, ModelId, - ModelOrigin, Nullability, Relation, RelationSchema, SchemaColumn, Selection, SelectorSet, - SemanticModel, SemanticProject, SemanticSeed, SemanticTest, SourceId, SourceStateProvider, - StaticSchemaProvider, StaticSourceStateProvider, TestId, + compile, compile_with_options, resolve_selection, ColumnContract, Compilation, DataType, + DiffPolicySpec, EmptySchemaProvider, EmptySourceStateProvider, IncrementalStrategy, + Materialization, ModelContract, ModelId, ModelOrigin, Nullability, Relation, RelationSchema, + SchemaColumn, Selection, SelectorSet, SemanticModel, SemanticProject, SemanticSeed, + SemanticTest, SourceId, SourceStateProvider, StaticSchemaProvider, StaticSourceStateProvider, + TestId, }; use phlo_transform_engine::{ branch_diff, changed_models, collect_source_states, ensure_environment, materialized_for_environment, Adapter, AdapterError, ArtifactWriter, BranchDiffRequest, - CancelHandle, CatalogRequest, ColumnInfo, DatasetStatus, EngineError, EngineEvent, - EnvironmentSpec, ExecutionStatus, FailureCategory, MaterializedRecord, Membership, ModelResult, - ModelRunRecord, Plan, PlanAction, PlanOptions, Planner, PromotionRecord, QueryResult, - ReasonKind, RetryPolicy, RunOptions, RunRecord, RunResult, RunSummary, Runner, SeedRecord, - SeedRunRecord, SqliteStateStore, StateStore, StoredPlan, StoredRun, TestRunRecord, + CancelHandle, CatalogRequest, ColumnInfo, ContractSafety, DatasetStatus, EngineError, + EngineEvent, EnvironmentSpec, ExecutionStatus, FailureCategory, MaterializedRecord, Membership, + ModelResult, ModelRunRecord, Plan, PlanAction, PlanOptions, Planner, PromotionRecord, + QueryResult, ReasonKind, RetryPolicy, RunOptions, RunRecord, RunResult, RunSummary, Runner, + SeedRecord, SeedRunRecord, SqliteStateStore, StateStore, StoredPlan, StoredRun, TestRunRecord, }; /// How a target should fail: the error to return, and how many attempts it @@ -1122,6 +1123,8 @@ async fn plan_prod_after( incremental_key: None, adapter: record_adapter.map(str::to_string), output_identity: output_identity.map(str::to_string), + contract: None, + effective_key: None, run_id: "run-dev".to_string(), materialized_at: "t".to_string(), }) @@ -1223,6 +1226,8 @@ async fn materialisation_by_another_adapter_rebuilds() { incremental_key: None, adapter: Some("trino".to_string()), output_identity: None, + contract: None, + effective_key: None, run_id: "run-prod".to_string(), materialized_at: "t".to_string(), }) @@ -1255,6 +1260,8 @@ async fn materialisation_by_an_unrecorded_adapter_rebuilds() { incremental_key: None, adapter: None, output_identity: Some("snap:dev".to_string()), + contract: None, + effective_key: None, run_id: "run-prod".to_string(), materialized_at: "t".to_string(), }) @@ -1502,6 +1509,8 @@ fn concurrent_sqlite_writers_complete() { incremental_key: None, adapter: Some("fake".to_string()), output_identity: None, + contract: None, + effective_key: None, run_id: format!("{prefix}-run"), materialized_at: "t".to_string(), }) @@ -4289,6 +4298,10 @@ fn materialized_at( incremental_key: None, adapter: Some("fake".to_string()), output_identity: None, + contract: None, + // A materialisation today records its key evidence even when the + // model has no key: an empty set, never an ambiguous NULL. + effective_key: Some(Vec::new()), run_id: "run-1".to_string(), materialized_at: at.to_string(), } @@ -4653,6 +4666,222 @@ async fn branch_diff_folds_default_environment_into_main() { assert_eq!(status("main.dropped"), Some(DatasetStatus::Removed)); } +fn contract(names: &[&str]) -> ModelContract { + ModelContract { + enforced: true, + columns: names + .iter() + .map(|name| ColumnContract { + name: (*name).to_string(), + data_type: None, + nullable: None, + }) + .collect(), + renames: BTreeMap::new(), + } +} + +/// A contract change reports structurally — the removed column is breaking +/// and its lineage impact names the downstream consumer. +#[tokio::test] +async fn branch_diff_reports_contract_changes_and_impacts() { + let mut base = model("main.base", "select 1 as id, 2 as legacy"); + base.contract = Some(contract(&["id"])); + let downstream = model("main.down", "select id, legacy from main.base"); + let compilation = compile(&SemanticProject::in_memory(vec![base, downstream])); + assert!(compilation.is_ok(), "{:?}", compilation.diagnostics); + + // The base environment recorded the contract while `legacy` was still + // part of it. + let state = SqliteStateStore::in_memory().unwrap(); + let relation = ref_relation("phlo_main", compilation.model_by_name("main.base").unwrap()); + let mut record = materialized("main.base", "main", "v1", &relation); + record.contract = Some(contract(&["id", "legacy"])); + state.record_materialized(&record).unwrap(); + + let report = branch_diff( + Arc::new(FakeAdapter::default()), + Some(&state), + &compilation, + &branch_diff_request(), + ) + .await + .unwrap(); + + let contract = report + .contract_changes + .iter() + .find(|diff| diff.model == "main.base") + .expect("contract diff for main.base"); + let removed = contract + .changes + .iter() + .find(|change| change.column == "legacy") + .expect("legacy change"); + assert_eq!(removed.kind, "removed"); + assert_eq!(removed.safety, ContractSafety::Breaking); + + let impact = report + .impacts + .iter() + .find(|impact| impact.model == "main.base" && impact.column.as_deref() == Some("legacy")) + .expect("impact for the breaking column"); + assert_eq!(impact.source, "contract"); + assert!( + impact + .downstream_models + .iter() + .any(|model| model == "main.down"), + "expected main.down downstream of the broken column: {:?}", + impact.downstream_models + ); +} + +/// A rename declared through `[model..renames]` classifies as a +/// breaking change — it proves intent, not compatibility — and produces +/// a lineage impact. +#[tokio::test] +async fn branch_diff_declared_contract_rename_is_breaking() { + let mut base = model("main.base", "select 1 as id, 2 as new_col"); + let mut desired = contract(&["id", "new_col"]); + desired + .renames + .insert("new_col".to_string(), "legacy".to_string()); + base.contract = Some(desired); + let compilation = compile(&SemanticProject::in_memory(vec![base])); + assert!(compilation.is_ok(), "{:?}", compilation.diagnostics); + + let state = SqliteStateStore::in_memory().unwrap(); + let relation = ref_relation("phlo_main", compilation.model_by_name("main.base").unwrap()); + let mut record = materialized("main.base", "main", "v1", &relation); + record.contract = Some(contract(&["id", "legacy"])); + state.record_materialized(&record).unwrap(); + + let report = branch_diff( + Arc::new(FakeAdapter::default()), + Some(&state), + &compilation, + &branch_diff_request(), + ) + .await + .unwrap(); + + let contract = report + .contract_changes + .iter() + .find(|diff| diff.model == "main.base") + .expect("contract diff"); + let renamed = contract + .changes + .iter() + .find(|change| change.column == "new_col") + .expect("new_col change"); + assert_eq!(renamed.kind, "renamed"); + // A declared rename proves intent, not compatibility: consumers + // selecting `legacy` break, so promotion must gate on it. + assert_eq!(renamed.safety, ContractSafety::Breaking); + assert!( + !report.impacts.is_empty(), + "a breaking rename produces lineage impact" + ); +} + +/// The historical key comes from the persisted `effective_key`, not the +/// incremental fields: a base materialised with only `unique(sample_id)` +/// remembers `[[sample_id]]`, so dropping the assertion reports a breaking +/// key removal and changing it reports a breaking key change. +#[tokio::test] +async fn branch_diff_reports_persisted_effective_key_changes() { + // Base has no key in the workspace; the recorded materialisation had a + // unique-only key — no incremental strategy at all. + let keyless = model("main.base", "select 1 as id"); + let compilation = compile(&SemanticProject::in_memory(vec![keyless])); + assert!(compilation.is_ok(), "{:?}", compilation.diagnostics); + + let state = SqliteStateStore::in_memory().unwrap(); + let relation = ref_relation("phlo_main", compilation.model_by_name("main.base").unwrap()); + let mut record = materialized("main.base", "main", "v1", &relation); + record.effective_key = Some(vec![vec!["sample_id".to_string()]]); + state.record_materialized(&record).unwrap(); + + let report = branch_diff( + Arc::new(FakeAdapter::default()), + Some(&state), + &compilation, + &branch_diff_request(), + ) + .await + .unwrap(); + let contract = report + .contract_changes + .iter() + .find(|diff| diff.model == "main.base") + .expect("key removal reports a contract diff"); + let removed = contract + .changes + .iter() + .find(|change| change.kind == "key_removed") + .expect("key_removed change"); + assert_eq!(removed.safety, ContractSafety::Breaking); + + // Changed: the workspace claims `unique(batch_id)` against the recorded + // `unique(sample_id)` — breaking, not a fresh declaration. + let mut keyed = model("main.base", "select 1 as id"); + keyed.directives.keys.push("batch_id".to_string()); + let compilation = compile(&SemanticProject::in_memory(vec![keyed])); + assert!(compilation.is_ok(), "{:?}", compilation.diagnostics); + + let report = branch_diff( + Arc::new(FakeAdapter::default()), + Some(&state), + &compilation, + &branch_diff_request(), + ) + .await + .unwrap(); + let contract = report + .contract_changes + .iter() + .find(|diff| diff.model == "main.base") + .expect("key change reports a contract diff"); + let changed = contract + .changes + .iter() + .find(|change| change.kind == "key_changed") + .expect("key_changed change"); + assert_eq!(changed.safety, ContractSafety::Breaking); +} + +/// The plan-level contract-change reason names the breaking columns rather +/// than reporting an opaque hash change. +#[test] +fn diff_reasons_describes_breaking_contract_changes() { + let mut base = model("main.base", "select 1 as id, 2 as legacy"); + base.contract = Some(contract(&["id"])); + let compilation = compile(&SemanticProject::in_memory(vec![base])); + assert!(compilation.is_ok(), "{:?}", compilation.diagnostics); + let compiled = compilation.model_by_name("main.base").unwrap(); + + let relation = Relation { + catalog: Some("phlo_main".to_string()), + schema: "main".to_string(), + table: "base".to_string(), + }; + let mut record = materialized("main.base", "main", "v9", &relation); + record.contract = Some(contract(&["id", "legacy"])); + + let reasons = phlo_transform_engine::diff_reasons(compiled, &record); + let reason = reasons + .iter() + .find(|reason| reason.kind == ReasonKind::ContractChange) + .expect("contract change reason"); + assert!( + reason.detail.contains("legacy") && reason.detail.contains("breaking"), + "expected the breaking column named: {}", + reason.detail + ); +} + #[test] fn materialized_for_environment_merges_default_env_by_latest() { let state = SqliteStateStore::in_memory().unwrap(); diff --git a/crates/phlo-transform-trino/tests/nessie_wap_e2e.rs b/crates/phlo-transform-trino/tests/nessie_wap_e2e.rs index f25a649..8e136db 100644 --- a/crates/phlo-transform-trino/tests/nessie_wap_e2e.rs +++ b/crates/phlo-transform-trino/tests/nessie_wap_e2e.rs @@ -248,6 +248,7 @@ async fn wap_candidate_on_nessie_branch_is_promoted() { strategy: DiffStrategy::Keyed, policy: DiffPolicy::default(), sample_fraction: None, + renames: Default::default(), }, ) .await diff --git a/crates/phlo-transform-trino/tests/trino_e2e.rs b/crates/phlo-transform-trino/tests/trino_e2e.rs index 82a0e78..2a56e3b 100644 --- a/crates/phlo-transform-trino/tests/trino_e2e.rs +++ b/crates/phlo-transform-trino/tests/trino_e2e.rs @@ -223,6 +223,7 @@ async fn computes_keyed_data_diff_against_trino() { strategy: DiffStrategy::Keyed, policy: DiffPolicy::default(), sample_fraction: None, + renames: Default::default(), }; let report = diff(adapter, &request).await.expect("diff runs"); @@ -284,6 +285,7 @@ async fn diff_supports_partitions_tolerances_schema_and_sampling() { strategy, policy, sample_fraction: None, + renames: Default::default(), }; // Tolerance: the tiny difference is within bounds. diff --git a/crates/phlo-transform-trino/tests/trino_incremental_e2e.rs b/crates/phlo-transform-trino/tests/trino_incremental_e2e.rs index 9be99c3..7a1b39b 100644 --- a/crates/phlo-transform-trino/tests/trino_incremental_e2e.rs +++ b/crates/phlo-transform-trino/tests/trino_incremental_e2e.rs @@ -233,6 +233,7 @@ async fn incremental_key_merges_on_trino_iceberg() { }, policy: DiffPolicy::default(), sample_fraction: None, + renames: Default::default(), }, ) .await diff --git a/docs/diff.md b/docs/diff.md index 006a993..b1e58ed 100644 --- a/docs/diff.md +++ b/docs/diff.md @@ -45,6 +45,11 @@ The report also carries: - **schema** — per-model added/removed columns, type changes and nullability changes, each with a safety classification; +- **contracts** — per-model contract changes: the workspace's declared + contract against what the base environment last recorded, classified + `safe`/`review`/`breaking` (see *Contract diff* below); +- **impacts** — for every breaking schema or contract change, the downstream + models and tests lineage says it would break; - **rows** — `count(*)` per side plus the delta; - **diffs** — with `--full`, a keyed value diff per changed model that declares keys, plus an aggregate diff for any changed model that declares @@ -122,9 +127,60 @@ relative_tolerance = 1e-6 `schema_changes` is populated from the candidate and base relation schemas: added, removed (`full_rebuild_required`), changed (`review` for numeric -widening, `error` otherwise) and nullability (`safe` when widening to -nullable, `review` when narrowing to not-null) columns appear alongside the -row-level result. +widening, `error` otherwise) and nullability (`error` when relaxing to +nullable — it removes a guarantee consumers may rely on — `review` when +tightening to not-null) columns appear alongside the row-level result. + +A column rename declared in `phlo.toml` turns an unexplained breaking +removal into an explicit `renamed` entry — still a breaking change for +consumers selecting the old name, so promotion gates on it — and lets the +diff compare the new column against its predecessor's type: + +```toml +[model."assay.results".renames] +result_value = "concentration" +``` + +## Contract diff + +`contract_changes` compares each model's declared contract (`contract_hash` +made legible) against the contract the base environment last materialised +with — contracts are persisted on materialisation records, so the report +describes what promotion would actually change rather than a hash that +moved: + +```text +contract_added contract declared where none was recorded safe +contract_removed contract dropped breaking +enforcement_changed enforcement toggled review (on) / breaking (off) +added contract column added safe +removed contract column removed breaking +renamed declared rename (new in desired, old in base) breaking +rename_ambiguous several new names claim one old name review +type_changed numeric widening `review`, else breaking +nullability_changed relaxed (lost NOT NULL guarantee) breaking + tightened review +key_changed effective key changed breaking +key_removed effective key dropped breaking +key_added effective key declared review +``` + +The effective key is the union of a `key` incremental strategy's columns +and unique-assertion columns — the same identity concept from either +source — compared against the key the base environment recorded at +materialisation time. + +A stale rename declaration — one whose new or old name is absent from the +contracts — resolves nothing: the underlying removal or addition reports +normally. Rename declarations must be one-to-one: when several new names +claim the same old name, none resolve — the old column reports `removed`, +every new column reports `added`, and `rename_ambiguous` explains why. The +`impacts` section resolves every `breaking` entry through the column +lineage graph into the downstream models and tests that consume it. + +At promotion time the same comparison runs live (workspace contract against +the target's recorded contract), so a contract edited after `diff` cannot +reach `promote` on the artifact's stale analysis. ## Policies diff --git a/docs/state.md b/docs/state.md index 388e7c4..7e50816 100644 --- a/docs/state.md +++ b/docs/state.md @@ -95,8 +95,16 @@ This is covered by `cache_reuse_across_environments_is_reported_as_cached`, The state store gains a `model_versions` table recording, per model and environment, the `ModelVersion` attached to the most recent successful materialisation (plus target, run id, timestamp, incremental strategy/key, -`version_detail`, the producing `adapter`, and the strong `output_identity` -the adapter observed on the relation right after writing). API: +`version_detail`, the producing `adapter`, the strong `output_identity` +the adapter observed on the relation right after writing, the declared +`contract`, and the `effective_key` — the row-identity key the model +materialised with, persisted so later comparisons need not reconstruct it). +`effective_key` is tri-state: a recorded set of claims, an empty set +(recorded "no key"), or NULL for rows that predate the column. Opening the +database backfills legacy rows that recorded an incremental `key` strategy; +rows with no surviving key evidence read as *unknown* — promotion treats an +unverifiable key as a breaking change, not as proof the base was keyless. +API: ```text record_materialized(record) @@ -224,7 +232,8 @@ version differs from the current compile. (`--environment` filters); - `state show ` — one run's model, seed and test records; - `state model ` — the recorded materialised version for the - effective environment: version components, target, adapter, run; + effective environment: version components, target, adapter, contract + summary, run; - `state promotions` — promotion history. All four honour `--json`. diff --git a/docs/wap.md b/docs/wap.md index 0b12af1..b1b5635 100644 --- a/docs/wap.md +++ b/docs/wap.md @@ -126,10 +126,18 @@ PASS conflicts — candidate merges cleanly - `tests` — no test in that run failed. - `blocked` — no model, seed or test was left blocked or cancelled. - `schema` — a fresh audited diff inspected this pair at these commits and - found no breaking schema changes, or they were waived with - `--allow-breaking-schema`. Without an audited artifact the gate fails - closed: "no evidence" never reads as "no changes". The waiver covers the - schema gate only — it does not waive provenance or run binding. + found no breaking schema or contract changes, or they were waived with + `--allow-breaking-schema`. Physical schema breaks come from the audited + diff; contract breaks (a dropped contract column, a removed contract, + enforcement turned off, an incompatible type change, a relaxed + nullability guarantee, a declared rename, a changed or dropped effective + key) are computed live at promotion time — the workspace's desired + contracts against the contracts the target environment last recorded — + so editing a contract after `diff` cannot bypass the gate on a stale + artifact. A state-store error fails promotion rather than reading as "no + contracts recorded". Without an audited artifact the gate fails closed: + "no evidence" never reads as "no changes". The waiver covers the schema + gate only — it does not waive provenance or run binding. - `data_diff` — only evaluated with `--require-diff`; the audited diff (`branch_diff.json` from a `--full` branch diff) must exist, pass its policies, cover this exact candidate→target pair **at the commits being diff --git a/fixtures/contract-renames/phlo.toml b/fixtures/contract-renames/phlo.toml new file mode 100644 index 0000000..3a64d84 --- /dev/null +++ b/fixtures/contract-renames/phlo.toml @@ -0,0 +1,12 @@ +[transform] +default_catalog = "memory" +default_schema = "default" + +[model."assay.results".contract] +enforced = true + +[model."assay.results".columns.result_value] +type = "DOUBLE" + +[model."assay.results".renames] +result_value = "concentration" diff --git a/fixtures/contract-renames/workflows/assay/transforms/results.sql b/fixtures/contract-renames/workflows/assay/transforms/results.sql new file mode 100644 index 0000000..ada6c3c --- /dev/null +++ b/fixtures/contract-renames/workflows/assay/transforms/results.sql @@ -0,0 +1 @@ +select sample_id, concentration as result_value from external.raw_results