From 42e11c6570b62ffaa86598438fa8ddf462c4035f Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sat, 4 Jul 2026 08:40:00 +0000 Subject: [PATCH] =?UTF-8?q?feat(pipelines):=20schema=20contracts=20?= =?UTF-8?q?=E2=80=94=20save-time=20consumer=20checks=20vs=20captured=20sch?= =?UTF-8?q?emas=20(#9917)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(pipelines): schema contracts — save-time consumer checks vs captured schemas Co-Authored-By: Claude Fable 5 * refactor: move schemaContractContext above schemaCanEvolve doc comment Co-Authored-By: Claude Fable 5 * fix: emit scd2/on_schema_change in CLI local graph, address review notes Co-Authored-By: Claude Fable 5 * fix: gate editor _current ignore-suppression on scd2, matching backend Co-Authored-By: Claude Fable 5 --------- Co-authored-by: Claude Fable 5 --- ...368a6862e35350ae43383fb2335e2c82a9419.json | 29 + ...eb5a0665c90c4926f1afb689d6cd16bd5b722.json | 29 + ...3ef4ab1ccfdc4ed33904e83e5130ac0bf5896.json | 41 ++ .../windmill-parser/src/asset_parser.rs | 67 ++ .../tests/fixtures/pipeline_annotations.json | 37 ++ .../tests/pipeline_annotations_parity.rs | 15 + backend/windmill-api-assets/src/lib.rs | 17 + backend/windmill-api-scripts/src/scripts.rs | 53 ++ backend/windmill-api/openapi.yaml | 77 +++ backend/windmill-common/src/assets.rs | 2 +- backend/windmill-common/src/lib.rs | 1 + .../windmill-common/src/schema_contracts.rs | 614 ++++++++++++++++++ cli/src/commands/pipeline/localGraph.ts | 33 +- docs/ducklake-materialization.md | 31 + docs/pipelines-vs-dbt.md | 44 +- frontend/src/lib/components/Editor.svelte | 91 ++- .../src/lib/components/ScriptBuilder.svelte | 6 + .../src/lib/components/ScriptEditor.svelte | 43 +- .../AssetGraph/AssetGraphDetailsPane.svelte | 12 +- .../AssetGraph/PipelineGraphEditor.svelte | 6 + .../parsePipelineAnnotations.parity.test.ts | 5 + .../AssetGraph/parsePipelineAnnotations.ts | 13 +- .../assets/AssetGraph/schemaContracts.test.ts | 276 ++++++++ .../assets/AssetGraph/schemaContracts.ts | 440 +++++++++++++ .../lib/components/assets/AssetGraph/types.ts | 11 +- .../(logged)/pipeline/[folder]/+page.svelte | 8 + 26 files changed, 1972 insertions(+), 29 deletions(-) create mode 100644 backend/.sqlx/query-36da136cf9b9554702e3b3ec364368a6862e35350ae43383fb2335e2c82a9419.json create mode 100644 backend/.sqlx/query-431a5eb2eb5a0b57e096cc78f68eb5a0665c90c4926f1afb689d6cd16bd5b722.json create mode 100644 backend/.sqlx/query-c6c186eb3a108e1f695bf5730c23ef4ab1ccfdc4ed33904e83e5130ac0bf5896.json create mode 100644 backend/windmill-common/src/schema_contracts.rs create mode 100644 frontend/src/lib/components/assets/AssetGraph/schemaContracts.test.ts create mode 100644 frontend/src/lib/components/assets/AssetGraph/schemaContracts.ts diff --git a/backend/.sqlx/query-36da136cf9b9554702e3b3ec364368a6862e35350ae43383fb2335e2c82a9419.json b/backend/.sqlx/query-36da136cf9b9554702e3b3ec364368a6862e35350ae43383fb2335e2c82a9419.json new file mode 100644 index 0000000000..d05abd4f3a --- /dev/null +++ b/backend/.sqlx/query-36da136cf9b9554702e3b3ec364368a6862e35350ae43383fb2335e2c82a9419.json @@ -0,0 +1,29 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT DISTINCT path AS \"asset_path!\", usage_path AS \"producer_path!\"\n FROM asset\n WHERE workspace_id = $1 AND kind = 'ducklake' AND path = ANY($2)\n AND usage_kind = 'script' AND usage_access_type IN ('w', 'rw')", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "asset_path!", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "producer_path!", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Text", + "TextArray" + ] + }, + "nullable": [ + false, + false + ] + }, + "hash": "36da136cf9b9554702e3b3ec364368a6862e35350ae43383fb2335e2c82a9419" +} diff --git a/backend/.sqlx/query-431a5eb2eb5a0b57e096cc78f68eb5a0665c90c4926f1afb689d6cd16bd5b722.json b/backend/.sqlx/query-431a5eb2eb5a0b57e096cc78f68eb5a0665c90c4926f1afb689d6cd16bd5b722.json new file mode 100644 index 0000000000..b554ef8b55 --- /dev/null +++ b/backend/.sqlx/query-431a5eb2eb5a0b57e096cc78f68eb5a0665c90c4926f1afb689d6cd16bd5b722.json @@ -0,0 +1,29 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT DISTINCT ON (path) path, content\n FROM script\n WHERE workspace_id = $1 AND path = ANY($2)\n AND archived = false AND deleted = false\n ORDER BY path, created_at DESC", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "path", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "content", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Text", + "TextArray" + ] + }, + "nullable": [ + false, + false + ] + }, + "hash": "431a5eb2eb5a0b57e096cc78f68eb5a0665c90c4926f1afb689d6cd16bd5b722" +} diff --git a/backend/.sqlx/query-c6c186eb3a108e1f695bf5730c23ef4ab1ccfdc4ed33904e83e5130ac0bf5896.json b/backend/.sqlx/query-c6c186eb3a108e1f695bf5730c23ef4ab1ccfdc4ed33904e83e5130ac0bf5896.json new file mode 100644 index 0000000000..7cfc1f109f --- /dev/null +++ b/backend/.sqlx/query-c6c186eb3a108e1f695bf5730c23ef4ab1ccfdc4ed33904e83e5130ac0bf5896.json @@ -0,0 +1,41 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT DISTINCT ON (asset_path)\n asset_path, version, columns AS \"columns: Json>\", captured_at\n FROM materialized_asset_schema\n WHERE workspace_id = $1 AND asset_kind = 'ducklake' AND asset_path = ANY($2)\n ORDER BY asset_path, version DESC", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "asset_path", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "version", + "type_info": "Int8" + }, + { + "ordinal": 2, + "name": "columns: Json>", + "type_info": "Jsonb" + }, + { + "ordinal": 3, + "name": "captured_at", + "type_info": "Timestamptz" + } + ], + "parameters": { + "Left": [ + "Text", + "TextArray" + ] + }, + "nullable": [ + false, + false, + false, + false + ] + }, + "hash": "c6c186eb3a108e1f695bf5730c23ef4ab1ccfdc4ed33904e83e5130ac0bf5896" +} diff --git a/backend/parsers/windmill-parser/src/asset_parser.rs b/backend/parsers/windmill-parser/src/asset_parser.rs index 299e3a70f9..3e6d433405 100644 --- a/backend/parsers/windmill-parser/src/asset_parser.rs +++ b/backend/parsers/windmill-parser/src/asset_parser.rs @@ -252,6 +252,8 @@ pub struct RetrySpec { // history (`valid_from`/`valid_to`/`is_current`). The leading keyword `scd2` is a // recognized alias for `history`. `deletes=close` (scd2 only) also closes a key // that disappears from the snapshot; default leaves absent keys current. +// `on_schema_change=ignore` suppresses downstream schema-contract warnings for +// the produced asset (save-time metadata only; default `warn`). #[derive(Serialize, Debug, PartialEq, Clone)] pub struct MaterializeSpec { pub target_kind: AssetKind, @@ -276,6 +278,32 @@ pub struct MaterializeSpec { // `hard_deletes=close`). Default (false) leaves absent keys current. #[serde(skip_serializing_if = "std::ops::Not::not", default)] pub close_deleted: bool, + // `on_schema_change=ignore` opts this producer's asset out of downstream + // schema-contract warnings (gap #2b): consumers referencing columns the + // captured schema no longer has warn by default (`warn`); `ignore` declares + // the schema deliberately unstable and suppresses those warnings. Save-time + // metadata only — the materialize write strategy is unaffected. + #[serde(skip_serializing_if = "OnSchemaChange::is_warn", default)] + pub on_schema_change: OnSchemaChange, +} + +// dbt's `on_schema_change` narrowed to the save-time contract check: `warn` +// (default) surfaces consumer warnings, `ignore` suppresses them. dbt's `fail` +// is deliberately not offered — saves are never hard-blocked (a deliberate +// upstream reshape must not fail every consumer save); a CI/CLI gate can layer +// it on later without touching the grammar. +#[derive(Serialize, Debug, PartialEq, Eq, Clone, Copy, Default)] +#[serde(rename_all = "lowercase")] +pub enum OnSchemaChange { + #[default] + Warn, + Ignore, +} + +impl OnSchemaChange { + pub fn is_warn(&self) -> bool { + matches!(self, OnSchemaChange::Warn) + } } // `// data_test …` — a data-quality assertion run against the @@ -917,6 +945,14 @@ fn parse_materialize_spec(s: &str) -> Option { // `deletes=close` (scd2 only) opts into hard-delete-close; any other value // (or absence) keeps the soft-delete default. let close_deleted = opts.get("deletes").map(|v| v == "close").unwrap_or(false); + // `on_schema_change=ignore` suppresses downstream contract warnings; any + // other value (or absence) keeps the `warn` default, fail-safe like + // `deletes=` above. + let on_schema_change = if opts.get("on_schema_change").map(String::as_str) == Some("ignore") { + OnSchemaChange::Ignore + } else { + OnSchemaChange::Warn + }; Some(MaterializeSpec { target_kind, target_path: path.to_string(), @@ -926,6 +962,7 @@ fn parse_materialize_spec(s: &str) -> Option { scd2, track, close_deleted, + on_schema_change, }) } @@ -1623,6 +1660,36 @@ mod pipeline_annotation_tests { assert!(!out.materialize.expect("materialize").close_deleted); } + #[test] + fn materialize_on_schema_change_opt() { + let out = parse_pipeline_annotations( + "// materialize ducklake://a/orders on_schema_change=ignore", + ); + let m = out.materialize.expect("materialize"); + assert_eq!(m.on_schema_change, OnSchemaChange::Ignore); + // default is warn + let out = parse_pipeline_annotations("// materialize ducklake://a/orders"); + assert_eq!( + out.materialize.expect("materialize").on_schema_change, + OnSchemaChange::Warn + ); + // unknown value keeps the warn default (fail-safe, like `deletes=`); + // `fail` is deliberately unrecognized in v1 + let out = + parse_pipeline_annotations("// materialize ducklake://a/orders on_schema_change=fail"); + assert_eq!( + out.materialize.expect("materialize").on_schema_change, + OnSchemaChange::Warn + ); + // composes with other opts + let out = parse_pipeline_annotations( + "// materialize ducklake://a/dim key=id history on_schema_change=ignore", + ); + let m = out.materialize.expect("materialize"); + assert!(m.scd2); + assert_eq!(m.on_schema_change, OnSchemaChange::Ignore); + } + #[test] fn materialize_key_without_history_is_plain_merge() { let out = parse_pipeline_annotations("// materialize ducklake://a/dim key=id"); diff --git a/backend/parsers/windmill-parser/tests/fixtures/pipeline_annotations.json b/backend/parsers/windmill-parser/tests/fixtures/pipeline_annotations.json index f7977239c1..5ee2945d18 100644 --- a/backend/parsers/windmill-parser/tests/fixtures/pipeline_annotations.json +++ b/backend/parsers/windmill-parser/tests/fixtures/pipeline_annotations.json @@ -310,6 +310,43 @@ } } }, + { + "name": "materialize on_schema_change=ignore opt", + "code": "// pipeline\n// materialize ducklake://analytics/orders on_schema_change=ignore\nSELECT 1;", + "expected": { + "in_pipeline": true, + "asset_triggers": [], + "native_triggers": [], + "partition": null, + "freshness": null, + "tag": null, + "retry": null, + "materialize": { + "target_kind": "ducklake", + "target_path": "analytics/orders", + "on_schema_change": "ignore" + } + } + }, + { + "name": "materialize on_schema_change unknown value keeps warn default (fail unrecognized in v1)", + "code": "// materialize ducklake://analytics/orders key=id on_schema_change=fail\nSELECT 1;", + "expected": { + "in_pipeline": false, + "asset_triggers": [], + "native_triggers": [], + "partition": null, + "freshness": null, + "tag": null, + "retry": null, + "materialize": { + "target_kind": "ducklake", + "target_path": "analytics/orders", + "unique_key": "id", + "on_schema_change": "warn" + } + } + }, { "name": "materialize key without history is plain merge (SCD1, not scd2)", "code": "// materialize ducklake://analytics/dim key=id\nSELECT 1;", diff --git a/backend/parsers/windmill-parser/tests/pipeline_annotations_parity.rs b/backend/parsers/windmill-parser/tests/pipeline_annotations_parity.rs index 2df4acd2e4..b51b33faf8 100644 --- a/backend/parsers/windmill-parser/tests/pipeline_annotations_parity.rs +++ b/backend/parsers/windmill-parser/tests/pipeline_annotations_parity.rs @@ -71,6 +71,13 @@ struct ExpectedMaterialize { track: Vec, #[serde(default)] close_deleted: bool, + // "warn" | "ignore"; absent === "warn" (the default). + #[serde(default = "default_on_schema_change")] + on_schema_change: String, +} + +fn default_on_schema_change() -> String { + "warn".to_string() } #[derive(Deserialize)] @@ -209,6 +216,14 @@ fn pipeline_annotation_fixtures_match() { m.close_deleted, e.close_deleted, "{ctx}: materialize close_deleted" ); + let osc = match m.on_schema_change { + windmill_parser::asset_parser::OnSchemaChange::Warn => "warn", + windmill_parser::asset_parser::OnSchemaChange::Ignore => "ignore", + }; + assert_eq!( + osc, e.on_schema_change, + "{ctx}: materialize on_schema_change" + ); } (got, want) => panic!( "{ctx}: materialize mismatch — got {:?}, want present={}", diff --git a/backend/windmill-api-assets/src/lib.rs b/backend/windmill-api-assets/src/lib.rs index 2eb24cee6c..587307afbb 100644 --- a/backend/windmill-api-assets/src/lib.rs +++ b/backend/windmill-api-assets/src/lib.rs @@ -664,6 +664,13 @@ struct GraphRunnableNode { // `merge` / any partitioned write INSERTs into a fixed-schema table. #[serde(skip_serializing_if = "Option::is_none", default)] materialize_strategy: Option, + // `on_schema_change=ignore` on the managed materialize — the producer's + // opt-out from downstream schema-contract warnings. Threaded to the editor + // so its client-side contract mirror suppresses the same warnings the + // server check does. Only serialized when set to `ignore` (default `warn` + // is absent). Lockstep with TS `AssetGraphRunnableNode.materialize_on_schema_change`. + #[serde(skip_serializing_if = "Option::is_none", default)] + materialize_on_schema_change: Option, // Macros this script provides to the workspace registry (deployed // `// macros` library). Drives the library node state + details-pane // signature list. Lockstep with TS `AssetGraphRunnableNode.macros`. @@ -1289,8 +1296,12 @@ async fn asset_graph( } }), materialize_strategy: ann.and_then(|a| a.materialize.as_ref()).and_then(|m| { + // Precedence mirrors the runtime strategy derivation: + // scd2 (`history`) > append > merge (`key=`) > replace. if m.manual { None + } else if m.scd2 { + Some("scd2".to_string()) } else if m.append { Some("append".to_string()) } else if m.unique_key.is_some() { @@ -1299,6 +1310,12 @@ async fn asset_graph( Some("replace".to_string()) } }), + materialize_on_schema_change: ann + .and_then(|a| a.materialize.as_ref()) + .filter(|m| { + m.on_schema_change == windmill_common::assets::OnSchemaChange::Ignore + }) + .map(|_| "ignore".to_string()), macros: (usage_kind == AssetUsageKind::Script) .then(|| macros_by_provider.get(&path)) .flatten() diff --git a/backend/windmill-api-scripts/src/scripts.rs b/backend/windmill-api-scripts/src/scripts.rs index 49a4091f07..e38d1925f0 100644 --- a/backend/windmill-api-scripts/src/scripts.rs +++ b/backend/windmill-api-scripts/src/scripts.rs @@ -141,6 +141,8 @@ pub fn workspaced_service() -> Router { // CI test results .route("/ci_test_results/{kind}/{*path}", get(get_ci_test_results)) .route("/ci_test_results_batch", post(get_ci_test_results_batch)) + // Save-time schema-contract check (pipelines gap #2b) + .route("/check_schema_contracts", post(check_schema_contracts)) } #[derive(Serialize, FromRow)] @@ -1305,6 +1307,14 @@ async fn create_script_internal<'c>( ns.path ); } + // `manual` materialize never captures a schema (no wrap codegen), so + // there is no contract for `on_schema_change` to mute downstream. + if m.manual && m.on_schema_change == windmill_parser::asset_parser::OnSchemaChange::Ignore { + tracing::warn!( + "script {}: `on_schema_change=ignore` on a `manual` materialize is inert — manual mode captures no schema, so consumers have no contract to check", + ns.path + ); + } } // `// macros` — this script is a workspace macro library: its body is // CREATE [OR REPLACE] MACRO statements plus plain setup, registered into @@ -3966,3 +3976,46 @@ async fn get_ci_test_results_batch( Ok(Json(result_map)) } + +#[derive(Deserialize)] +struct CheckSchemaContractsRequest { + language: ScriptLang, + content: String, +} + +#[derive(Serialize)] +struct CheckSchemaContractsResponse { + warnings: Vec, +} + +// Save-time schema-contract check (pipelines gap #2b): validate the given +// script content's asset references (body column reads, `// column` lineage, +// `// data_test relationships`) against the latest captured producer schemas +// and return WARNINGS — never errors, and deploy never blocks on this. The +// frontend calls it right after a successful deploy (post-commit, so a +// self-produced target resolves to the fresh content) and the editor mirrors +// the same diff client-side; this endpoint is the authoritative check. Parsing +// uses the same server path as deploy (`effective_script_assets` + +// `parse_pipeline_annotations`) so the verdict matches what deployed. +async fn check_schema_contracts( + authed: ApiAuthed, + Extension(user_db): Extension, + Path(w_id): Path, + Json(req): Json, +) -> JsonResult { + let assets = crate::asset_inference::effective_script_assets(&req.language, &req.content, None) + .unwrap_or_default(); + let ann = parse_pipeline_annotations(&req.content); + let mut tx = user_db.begin(&authed).await?; + let warnings = windmill_common::schema_contracts::check_schema_contracts( + &mut tx, + &w_id, + &assets, + &ann.column_lineage, + &ann.data_tests, + ann.materialize.as_ref(), + ) + .await?; + tx.commit().await?; + Ok(Json(CheckSchemaContractsResponse { warnings })) +} diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index fbad5fffd4..1f836cfb2b 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -9187,6 +9187,50 @@ paths: items: $ref: "#/components/schemas/CiTestResult" + /w/{workspace}/scripts/check_schema_contracts: + post: + summary: check a script's asset references against captured producer schemas + description: | + Save-time schema-contract check for data pipelines: validates the given + script content's asset references (body column reads, `// column` lineage, + `// data_test relationships`) against the latest captured producer schemas + and returns warnings. Warnings never block a save/deploy; an asset whose + producer declares `on_schema_change=ignore` is suppressed to a single + informational entry. + operationId: checkSchemaContracts + tags: + - script + parameters: + - $ref: "#/components/parameters/WorkspaceId" + requestBody: + required: true + content: + application/json: + schema: + type: object + required: + - language + - content + properties: + language: + $ref: "#/components/schemas/ScriptLang" + content: + type: string + responses: + "200": + description: contract warnings (empty when all references match) + content: + application/json: + schema: + type: object + required: + - warnings + properties: + warnings: + type: array + items: + $ref: "#/components/schemas/ContractWarning" + /w/{workspace}/scripts/raw_temp/store: post: summary: store raw script content temporarily for CLI lock generation @@ -29903,6 +29947,39 @@ components: kind: $ref: "#/components/schemas/AssetKind" required: [path, kind] + ContractWarning: + description: | + One save-time schema-contract warning: a consumer reference that does + not match the referenced asset's latest captured schema. + `schema_version`/`captured_at` identify the capture the check ran + against (as-of the producer's last run, not its latest save). + type: object + required: [kind, asset_path, message] + properties: + kind: + type: string + enum: + - missing_column + - missing_lineage_source + - missing_relationship_column + - relationship_type_mismatch + - suppressed + asset_path: + type: string + column: + type: string + expected_type: + type: string + found_type: + type: string + schema_version: + type: integer + format: int64 + captured_at: + type: string + format: date-time + message: + type: string Volume: type: object required: diff --git a/backend/windmill-common/src/assets.rs b/backend/windmill-common/src/assets.rs index 722ad29747..63309a1400 100644 --- a/backend/windmill-common/src/assets.rs +++ b/backend/windmill-common/src/assets.rs @@ -6,7 +6,7 @@ use crate::{error, scripts::ScriptHash}; pub use windmill_parser::asset_parser::{ merge_column_lineage, parse_pipeline_annotations, ColumnLineage, ColumnRef, DataTest, - PartitionKind, PipelineAnnotations, RetrySpec, TriggerSpec, PARTITION_TOKEN, + OnSchemaChange, PartitionKind, PipelineAnnotations, RetrySpec, TriggerSpec, PARTITION_TOKEN, }; pub use windmill_types::assets::*; diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index baa3e4b500..acf97734d8 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -65,6 +65,7 @@ pub mod materialization; pub mod min_version; pub mod notify_events; pub mod runtime_assets; +pub mod schema_contracts; pub mod workspace_dependencies; #[cfg(feature = "private")] diff --git a/backend/windmill-common/src/schema_contracts.rs b/backend/windmill-common/src/schema_contracts.rs new file mode 100644 index 0000000000..b605aae33b --- /dev/null +++ b/backend/windmill-common/src/schema_contracts.rs @@ -0,0 +1,614 @@ +//! Save-time schema-contract check (pipelines gap #2b): validate a consumer +//! script's asset references against the latest *captured* producer schema +//! (`materialized_asset_schema`, written post-materialize by #2a) and return +//! WARNINGS — never errors. A deliberate upstream reshape must not fail every +//! consumer save; blocking (`on_schema_change=fail`) is deliberately not +//! offered in v1. +//! +//! Ducklake-only: `// materialize` targets are ducklake-only in v1, so nothing +//! else has a captured schema to check against; an asset with no captured +//! schema produces no warnings (first deploy, datatable, external tables). +//! +//! The comparison itself (`diff_contract`) is pure so it can be unit-tested +//! and mirrored 1:1 by the editor-side TS check +//! (frontend/src/lib/components/assets/AssetGraph/schemaContracts.ts); the +//! async wrapper owns the DB reads (schemas + producer resolution) and runs on +//! the caller's RLS-scoped transaction. + +use std::collections::{HashMap, HashSet}; + +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use sqlx::types::Json; +use sqlx::{Postgres, Transaction}; +use windmill_parser::asset_parser::{ + ColumnLineage, DataTest, MaterializeSpec, OnSchemaChange, PARTITION_TOKEN, +}; +use windmill_types::assets::{AssetKind, AssetWithAltAccessType}; + +use crate::error::Result; +use crate::materialization::SchemaColumn; + +/// Columns the materialize engine adds/manages; never part of the captured +/// schema, so consumer reads of them must not warn (`_wm_partition` is +/// filtered out of the DESCRIBE capture on purpose). +const RESERVED_COLUMNS: &[&str] = &["_wm_partition"]; + +#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum ContractWarningKind { + /// A column the body reads/writes is absent from the captured schema. + MissingColumn, + /// A `// column … <- .` source column is absent. + MissingLineageSource, + /// A `// data_test relationships … -> .` ref column is absent. + MissingRelationshipColumn, + /// Relationship join columns have different captured types (may still + /// coerce at run time — phrased as "differs", not "will fail"). + RelationshipTypeMismatch, + /// Warnings for this asset were suppressed by the producer's + /// `on_schema_change=ignore` (one informational entry per asset). + Suppressed, +} + +/// One save-time contract warning. `schema_version`/`captured_at` identify the +/// capture the check ran against, so a stale-capture warning is +/// self-explaining (the schema is as-of the producer's last run, not its +/// latest save). +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)] +pub struct ContractWarning { + pub kind: ContractWarningKind, + /// Normalized ducklake asset path (`/`). + pub asset_path: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub column: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub expected_type: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub found_type: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub schema_version: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub captured_at: Option>, + pub message: String, +} + +/// The latest captured schema of one asset, as loaded by the wrapper. +#[derive(Debug, Clone)] +pub struct CapturedSchema { + pub columns: Vec, + pub version: i64, + pub captured_at: DateTime, +} + +impl CapturedSchema { + /// Case-insensitive column lookup — DuckDB matches unquoted identifiers + /// case-insensitively, and the body parser preserves source casing while + /// DESCRIBE returns stored casing. + fn find(&self, name: &str) -> Option<&SchemaColumn> { + self.columns + .iter() + .find(|c| c.name.eq_ignore_ascii_case(name)) + } +} + +fn is_reserved(name: &str) -> bool { + RESERVED_COLUMNS + .iter() + .any(|r| r.eq_ignore_ascii_case(name)) +} + +/// Strip the `{partition}` token a declared URI may carry (`// on +/// ducklake://lake/t/{partition}` or pasted refs) so lookups hit the captured +/// path. Body-inferred paths never carry it, but annotation refs can. +pub fn normalize_asset_path(path: &str) -> String { + path.replace(&format!("/{}", PARTITION_TOKEN), "") + .replace(PARTITION_TOKEN, "") + .trim_end_matches('/') + .to_string() +} + +fn column_list(schema: &CapturedSchema) -> String { + schema + .columns + .iter() + .map(|c| c.name.as_str()) + .collect::>() + .join(", ") +} + +/// Pure comparison: consumer refs vs captured schemas. `schemas` is keyed by +/// normalized ducklake path (the `_current` → base-table fallback is resolved +/// by the wrapper before this runs); `ignored` holds paths whose producer +/// declared `on_schema_change=ignore`. +pub fn diff_contract( + assets: &[AssetWithAltAccessType], + column_lineage: &[ColumnLineage], + data_tests: &[DataTest], + materialize: Option<&MaterializeSpec>, + schemas: &HashMap, + ignored: &HashSet, +) -> Vec { + let mut warnings: Vec = vec![]; + + // W1 — body-read/written columns missing from the captured schema. Assets + // whose column set the parser could not derive (`columns: None`, e.g. + // wildcard SELECT or non-SQL access) are skipped fail-safe; a literal "*" + // key is skipped defensively for the same reason. + for a in assets { + if a.kind != AssetKind::Ducklake { + continue; + } + let Some(columns) = a.columns.as_ref() else { + continue; + }; + let path = normalize_asset_path(&a.path); + let Some(schema) = schemas.get(&path) else { + continue; + }; + for col in columns.keys() { + if col == "*" || is_reserved(col) { + continue; + } + if schema.find(col).is_none() { + warnings.push(ContractWarning { + kind: ContractWarningKind::MissingColumn, + asset_path: path.clone(), + column: Some(col.clone()), + expected_type: None, + found_type: None, + schema_version: Some(schema.version), + captured_at: Some(schema.captured_at), + message: format!( + "column `{col}` of ducklake://{path} is not in its captured schema \ + (v{}, columns: {})", + schema.version, + column_list(schema) + ), + }); + } + } + } + + // W2 — `// column` lineage source refs. Only annotation-declared lineage + // reaches this fn (AST-inferred lineage is redundant with W1 and can + // mis-attribute aliases). + for cl in column_lineage { + for input in &cl.inputs { + if input.from_kind != windmill_parser::asset_parser::AssetKind::Ducklake { + continue; + } + let path = normalize_asset_path(&input.from_path); + let Some(schema) = schemas.get(&path) else { + continue; + }; + if is_reserved(&input.from_column) { + continue; + } + if schema.find(&input.from_column).is_none() { + warnings.push(ContractWarning { + kind: ContractWarningKind::MissingLineageSource, + asset_path: path.clone(), + column: Some(input.from_column.clone()), + expected_type: None, + found_type: None, + schema_version: Some(schema.version), + captured_at: Some(schema.captured_at), + message: format!( + "`// column {}` reads `{}` from ducklake://{path}, which is not in \ + its captured schema (v{})", + cl.column, input.from_column, schema.version + ), + }); + } + } + } + + // W3 — relationships data-test refs: the referenced column must exist; + // when both sides have captured types, flag a difference. Types come from + // DuckDB DESCRIBE on both sides so verbatim spellings are comparable; the + // runtime probe's IN-subquery still coerces, so a difference is "differs", + // never "will fail". + let own_schema = materialize + .filter(|m| m.target_kind == windmill_parser::asset_parser::AssetKind::Ducklake) + .and_then(|m| schemas.get(&normalize_asset_path(&m.target_path))); + for dt in data_tests { + let DataTest::Relationships { column, to_kind, to_path, to_column } = dt else { + continue; + }; + if *to_kind != windmill_parser::asset_parser::AssetKind::Ducklake { + continue; + } + let path = normalize_asset_path(to_path); + let Some(schema) = schemas.get(&path) else { + continue; + }; + match schema.find(to_column) { + None => { + warnings.push(ContractWarning { + kind: ContractWarningKind::MissingRelationshipColumn, + asset_path: path.clone(), + column: Some(to_column.clone()), + expected_type: None, + found_type: None, + schema_version: Some(schema.version), + captured_at: Some(schema.captured_at), + message: format!( + "`// data_test relationships {column}` references \ + ducklake://{path}.{to_column}, which is not in its captured schema \ + (v{})", + schema.version + ), + }); + } + Some(ref_col) => { + if let Some(own_col) = own_schema.and_then(|s| s.find(column)) { + if !own_col.data_type.eq_ignore_ascii_case(&ref_col.data_type) { + warnings.push(ContractWarning { + kind: ContractWarningKind::RelationshipTypeMismatch, + asset_path: path.clone(), + column: Some(to_column.clone()), + expected_type: Some(own_col.data_type.clone()), + found_type: Some(ref_col.data_type.clone()), + schema_version: Some(schema.version), + captured_at: Some(schema.captured_at), + message: format!( + "`// data_test relationships {column}` joins `{}` ({}) to \ + ducklake://{path}.{to_column} ({}) — captured types differ", + column, own_col.data_type, ref_col.data_type + ), + }); + } + } + } + } + } + + // W4 — producer opted the asset out (`on_schema_change=ignore`): drop its + // warnings, leaving one informational entry per suppressed asset so the + // response still records that a mismatch exists but was muted upstream. + if !ignored.is_empty() { + let mut suppressed_assets: Vec = vec![]; + warnings.retain(|w| { + if ignored.contains(&w.asset_path) { + if !suppressed_assets.contains(&w.asset_path) { + suppressed_assets.push(w.asset_path.clone()); + } + false + } else { + true + } + }); + for path in suppressed_assets { + warnings.push(ContractWarning { + kind: ContractWarningKind::Suppressed, + asset_path: path.clone(), + column: None, + expected_type: None, + found_type: None, + schema_version: None, + captured_at: None, + message: format!( + "schema mismatches on ducklake://{path} suppressed by its producer's \ + `on_schema_change=ignore`" + ), + }); + } + } + + warnings +} + +/// Load captured schemas + producer modes and run the contract check for one +/// consumer script's parsed refs. +/// +/// Runs on the caller's RLS-scoped transaction (`user_db`), consistent with +/// the `listAssetSchemas` read path: a producer script the caller cannot read +/// simply stays unresolved and keeps the default `warn` behavior. Draft-only +/// producers have no `asset` write edges yet and likewise default to `warn`. +pub async fn check_schema_contracts( + tx: &mut Transaction<'_, Postgres>, + workspace_id: &str, + assets: &[AssetWithAltAccessType], + column_lineage: &[ColumnLineage], + data_tests: &[DataTest], + materialize: Option<&MaterializeSpec>, +) -> Result> { + // Referenced ducklake paths (normalized) across every ref family the diff + // inspects — plus the consumer's own materialize target (for W3 types). + let mut paths: HashSet = HashSet::new(); + for a in assets { + if a.kind == AssetKind::Ducklake && a.columns.is_some() { + paths.insert(normalize_asset_path(&a.path)); + } + } + for cl in column_lineage { + for input in &cl.inputs { + if input.from_kind == windmill_parser::asset_parser::AssetKind::Ducklake { + paths.insert(normalize_asset_path(&input.from_path)); + } + } + } + for dt in data_tests { + if let DataTest::Relationships { to_kind, to_path, .. } = dt { + if *to_kind == windmill_parser::asset_parser::AssetKind::Ducklake { + paths.insert(normalize_asset_path(to_path)); + } + } + } + if let Some(m) = materialize { + if m.target_kind == windmill_parser::asset_parser::AssetKind::Ducklake { + paths.insert(normalize_asset_path(&m.target_path)); + } + } + if paths.is_empty() { + return Ok(vec![]); + } + + // A managed scd2 producer (re)creates a `_current` view with the base + // table's columns; only the base table's schema is captured. Include the + // base path in the lookup so `_current` readers can fall back to it (the + // fallback itself is gated on the producer's spec below). + let mut lookup_paths: HashSet = paths.clone(); + for p in &paths { + if let Some(base) = p.strip_suffix("_current") { + if !base.is_empty() { + lookup_paths.insert(base.to_string()); + } + } + } + let lookup_vec: Vec = lookup_paths.into_iter().collect(); + + let schema_rows = sqlx::query!( + r#"SELECT DISTINCT ON (asset_path) + asset_path, version, columns AS "columns: Json>", captured_at + FROM materialized_asset_schema + WHERE workspace_id = $1 AND asset_kind = 'ducklake' AND asset_path = ANY($2) + ORDER BY asset_path, version DESC"#, + workspace_id, + &lookup_vec, + ) + .fetch_all(&mut **tx) + .await?; + let mut schemas: HashMap = schema_rows + .into_iter() + .map(|r| { + ( + r.asset_path, + CapturedSchema { + columns: r.columns.0, + version: r.version, + captured_at: r.captured_at, + }, + ) + }) + .collect(); + + // Producer resolution: write edges on the referenced assets → latest + // non-archived producer content → parsed `// materialize` spec. Drives + // both the `on_schema_change=ignore` suppression and the `_current` + // fallback. Flow writers are excluded — they cannot carry the annotation. + let paths_vec: Vec = paths.iter().cloned().collect(); + let producer_edges = sqlx::query!( + r#"SELECT DISTINCT path AS "asset_path!", usage_path AS "producer_path!" + FROM asset + WHERE workspace_id = $1 AND kind = 'ducklake' AND path = ANY($2) + AND usage_kind = 'script' AND usage_access_type IN ('w', 'rw')"#, + workspace_id, + &paths_vec, + ) + .fetch_all(&mut **tx) + .await?; + + let mut ignored: HashSet = HashSet::new(); + if !producer_edges.is_empty() { + let producer_paths: Vec = producer_edges + .iter() + .map(|e| e.producer_path.clone()) + .collect::>() + .into_iter() + .collect(); + // Same latest-content pattern as the asset-graph endpoint; NOT + // `get_latest_script_hash`, whose `lock IS NOT NULL` filter transiently + // excludes a just-deployed producer pending its dependency job. + let producer_rows = sqlx::query!( + r#"SELECT DISTINCT ON (path) path, content + FROM script + WHERE workspace_id = $1 AND path = ANY($2) + AND archived = false AND deleted = false + ORDER BY path, created_at DESC"#, + workspace_id, + &producer_paths, + ) + .fetch_all(&mut **tx) + .await?; + let producer_specs: HashMap> = producer_rows + .into_iter() + .map(|r| { + ( + r.path, + windmill_parser::asset_parser::parse_pipeline_annotations(&r.content) + .materialize, + ) + }) + .collect(); + + for edge in &producer_edges { + let Some(Some(spec)) = producer_specs.get(&edge.producer_path) else { + continue; + }; + if spec.target_kind != windmill_parser::asset_parser::AssetKind::Ducklake { + continue; + } + let target = normalize_asset_path(&spec.target_path); + // `on_schema_change=ignore` — any producer declaring it wins. + if spec.on_schema_change == OnSchemaChange::Ignore + && (target == edge.asset_path + || (spec.scd2 && format!("{target}_current") == edge.asset_path)) + { + ignored.insert(edge.asset_path.clone()); + } + // `_current` fallback: a managed scd2 producer's view has exactly + // the base table's columns. + if spec.scd2 + && !spec.manual + && format!("{target}_current") == edge.asset_path + && !schemas.contains_key(&edge.asset_path) + { + if let Some(base) = schemas.get(&target).cloned() { + schemas.insert(edge.asset_path.clone(), base); + } + } + } + } + + Ok(diff_contract( + assets, + column_lineage, + data_tests, + materialize, + &schemas, + &ignored, + )) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::BTreeMap; + use windmill_parser::asset_parser::{parse_pipeline_annotations, ColumnRef}; + use windmill_types::assets::AssetUsageAccessType; + + fn schema(cols: &[(&str, &str)]) -> CapturedSchema { + CapturedSchema { + columns: cols + .iter() + .map(|(n, t)| SchemaColumn { name: n.to_string(), data_type: t.to_string() }) + .collect(), + version: 2, + captured_at: DateTime::::MIN_UTC, + } + } + + fn read_asset(path: &str, cols: &[&str]) -> AssetWithAltAccessType { + AssetWithAltAccessType { + path: path.to_string(), + kind: AssetKind::Ducklake, + access_type: Some(AssetUsageAccessType::R), + alt_access_type: None, + columns: Some( + cols.iter() + .map(|c| (c.to_string(), AssetUsageAccessType::R)) + .collect::>(), + ), + } + } + + #[test] + fn missing_read_column_warns_case_insensitively() { + let schemas = HashMap::from([( + "lake/orders".to_string(), + schema(&[("Order_ID", "BIGINT"), ("amount_usd", "DOUBLE")]), + )]); + let assets = vec![read_asset("lake/orders", &["order_id", "amount"])]; + let w = diff_contract(&assets, &[], &[], None, &schemas, &HashSet::new()); + // order_id matches case-insensitively; amount is gone + assert_eq!(w.len(), 1); + assert_eq!(w[0].kind, ContractWarningKind::MissingColumn); + assert_eq!(w[0].column.as_deref(), Some("amount")); + assert_eq!(w[0].schema_version, Some(2)); + } + + #[test] + fn unknown_columns_wildcard_and_reserved_are_skipped() { + let schemas = HashMap::from([("lake/orders".to_string(), schema(&[("id", "BIGINT")]))]); + // columns: None (wildcard SELECT) — skipped entirely + let mut a = read_asset("lake/orders", &[]); + a.columns = None; + assert!(diff_contract(&[a], &[], &[], None, &schemas, &HashSet::new()).is_empty()); + // literal "*" and the reserved partition column are skipped + let a = read_asset("lake/orders", &["*", "_wm_partition", "id"]); + assert!(diff_contract(&[a], &[], &[], None, &schemas, &HashSet::new()).is_empty()); + } + + #[test] + fn asset_without_captured_schema_is_silent() { + let assets = vec![read_asset("lake/unknown", &["whatever"])]; + assert!( + diff_contract(&assets, &[], &[], None, &HashMap::new(), &HashSet::new()).is_empty() + ); + } + + #[test] + fn partition_token_is_normalized() { + let schemas = HashMap::from([("lake/orders".to_string(), schema(&[("id", "BIGINT")]))]); + let assets = vec![read_asset("lake/orders/{partition}", &["gone"])]; + let w = diff_contract(&assets, &[], &[], None, &schemas, &HashSet::new()); + assert_eq!(w.len(), 1); + assert_eq!(w[0].asset_path, "lake/orders"); + } + + #[test] + fn lineage_ref_missing_column_warns() { + let schemas = HashMap::from([("lake/orders".to_string(), schema(&[("id", "BIGINT")]))]); + let lineage = vec![ColumnLineage { + column: "total".to_string(), + inputs: vec![ColumnRef { + from_kind: windmill_parser::asset_parser::AssetKind::Ducklake, + from_path: "lake/orders".to_string(), + from_column: "amount".to_string(), + }], + }]; + let w = diff_contract(&[], &lineage, &[], None, &schemas, &HashSet::new()); + assert_eq!(w.len(), 1); + assert_eq!(w[0].kind, ContractWarningKind::MissingLineageSource); + } + + #[test] + fn relationships_missing_and_type_mismatch() { + let schemas = HashMap::from([ + ("lake/customers".to_string(), schema(&[("id", "VARCHAR")])), + ( + "lake/orders".to_string(), + schema(&[("customer_id", "BIGINT")]), + ), + ]); + let ann = parse_pipeline_annotations( + "// materialize ducklake://lake/orders\n\ + // data_test relationships customer_id -> ducklake://lake/customers.id\n\ + // data_test relationships customer_id -> ducklake://lake/customers.uuid\n\ + SELECT 1;", + ); + let w = diff_contract( + &[], + &[], + &ann.data_tests, + ann.materialize.as_ref(), + &schemas, + &HashSet::new(), + ); + assert_eq!(w.len(), 2); + assert!(w + .iter() + .any(|w| w.kind == ContractWarningKind::RelationshipTypeMismatch + && w.expected_type.as_deref() == Some("BIGINT") + && w.found_type.as_deref() == Some("VARCHAR"))); + assert!(w + .iter() + .any(|w| w.kind == ContractWarningKind::MissingRelationshipColumn + && w.column.as_deref() == Some("uuid"))); + } + + #[test] + fn ignored_asset_suppresses_to_single_note() { + let schemas = HashMap::from([("lake/orders".to_string(), schema(&[("id", "BIGINT")]))]); + let assets = vec![read_asset("lake/orders", &["a", "b"])]; + let ignored = HashSet::from(["lake/orders".to_string()]); + let w = diff_contract(&assets, &[], &[], None, &schemas, &ignored); + assert_eq!(w.len(), 1); + assert_eq!(w[0].kind, ContractWarningKind::Suppressed); + // and nothing at all when there was nothing to suppress + let assets = vec![read_asset("lake/orders", &["id"])]; + assert!(diff_contract(&assets, &[], &[], None, &schemas, &ignored).is_empty()); + } +} diff --git a/cli/src/commands/pipeline/localGraph.ts b/cli/src/commands/pipeline/localGraph.ts index 9d40a0e4b7..98413876a5 100644 --- a/cli/src/commands/pipeline/localGraph.ts +++ b/cli/src/commands/pipeline/localGraph.ts @@ -44,9 +44,14 @@ export type GraphRunnable = { column_lineage?: unknown[]; // `// materialize ` target + strategy — the script's declared output, // so the UI anchors column lineage / the materialize badge to it (the producer - // write-edge is emitted separately). + // write-edge is emitted separately). `scd2` also identifies the producer of a + // `_current` view for the editor's schema-contract fallback. materialize_target?: { kind: string; path: string }; - materialize_strategy?: "replace" | "append" | "merge"; + materialize_strategy?: "replace" | "append" | "merge" | "scd2"; + // `on_schema_change=ignore` — producer's opt-out from downstream + // schema-contract warnings; only present when set (default `warn` is absent), + // mirroring the deployed graph node. + materialize_on_schema_change?: string; }; export type GraphEdge = { runnable_kind: string; @@ -110,6 +115,9 @@ type ParseAssetsRaw = { manual?: boolean; append?: boolean; unique_key?: string; + scd2?: boolean; + // "ignore" when set; the default `warn` is skipped in serialization. + on_schema_change?: string; }; // `// tag ` — the worker tag the deployed pipeline routes to. tag?: string; @@ -428,16 +436,18 @@ export async function buildLocalPipelineGraph(args: { // deployed pipeline would (both `pipeline run --local` and `/pipeline_dev`). pipelineScripts.push(out.tag ? { ...s, tag: out.tag } : s); const mat = out.materialize; - // Managed-materialize write strategy, derived like the deployed graph - // (append → append; key=→ merge; else replace). Manual mode has no - // managed strategy. + // Managed-materialize write strategy, derived like the deployed graph — + // precedence mirrors the runtime: scd2 (`history`) > append > merge + // (key=) > replace. Manual mode has no managed strategy. const materialize_strategy = mat && !mat.manual - ? mat.append - ? "append" - : mat.unique_key - ? "merge" - : "replace" + ? mat.scd2 + ? "scd2" + : mat.append + ? "append" + : mat.unique_key + ? "merge" + : "replace" : undefined; runnables.push({ path: s.path, @@ -453,6 +463,9 @@ export async function buildLocalPipelineGraph(args: { : {}), ...(mat ? { materialize_target: { kind: mat.target_kind, path: mat.target_path } } : {}), ...(materialize_strategy ? { materialize_strategy } : {}), + ...(mat?.on_schema_change === "ignore" + ? { materialize_on_schema_change: "ignore" } + : {}), }); for (const a of out.assets ?? []) { diff --git a/docs/ducklake-materialization.md b/docs/ducklake-materialization.md index 4481d7555b..83c25f3eec 100644 --- a/docs/ducklake-materialization.md +++ b/docs/ducklake-materialization.md @@ -49,6 +49,8 @@ stays separate because it is cross-cutting (cascade + scheduling + materialize). // materialize ducklake://analytics/orders_daily append → managed, append // materialize ducklake://analytics/dim_customer key=id history → managed, SCD type 2 history // materialize manual ducklake://analytics/orders_daily → track-only escape hatch +// materialize ducklake://analytics/raw_events on_schema_change=ignore +// → managed, downstream contract warnings muted ``` - **managed (default)** — the script is *setup + one trailing `SELECT`*; Windmill @@ -366,6 +368,35 @@ load-bearing. - **Managed only.** `// materialize manual` + `// data_test` is rejected with a clear error (we can't know the manual script's target alias / partition col). +## Schema contracts (save-time, gap #2b) + +The captured schema (#2a) is read back as a *contract*: at save/deploy time, +every consumer's asset references — body-read/written columns, `// column` +lineage sources, `// data_test relationships` refs — are diffed against the +latest `materialized_asset_schema` version of each referenced ducklake asset, +and mismatches surface as **warnings** (deploy never blocks; dbt's +`on_schema_change=fail` is deliberately absent in v1). The diff lives in +`windmill_common::schema_contracts` (endpoint: +`POST /w/{ws}/scripts/check_schema_contracts`, called by the UI right after a +save) and is mirrored 1:1 by the editor (`schemaContracts.ts`): live Monaco +warning squiggles from the WASM buffer parse, plus column-name completion for +annotation refs fed by the same captured schemas. + +Comparison rules worth knowing: column names are case-insensitive (DuckDB +unquoted-identifier semantics); `_wm_partition` is whitelisted (it's excluded +from capture); `{partition}` tokens are stripped before lookup; a +`_current` reference falls back to the scd2 base table's capture; a +relationships join across two captured assets also flags a captured-type +*difference* (the runtime probe coerces, so it's "differs", not "will fail"). +An asset with no capture (never materialized, `manual` mode, any +`datatable://`) produces no warnings. + +The producer opts a deliberately unstable schema out with +`// materialize … on_schema_change=ignore` — consumers then get a single +informational "suppressed" note instead of per-column warnings. On `manual` +materialize the option is inert (nothing is captured) and deploy logs a +warning saying so. + ## Scoping decision: DuckLake vs DataTable **Make DuckLake the materialization/versioning substrate; keep DataTable as the diff --git a/docs/pipelines-vs-dbt.md b/docs/pipelines-vs-dbt.md index c073c88b8c..4ad8e65fb9 100644 --- a/docs/pipelines-vs-dbt.md +++ b/docs/pipelines-vs-dbt.md @@ -51,7 +51,7 @@ Asset-centric, polyglot, annotation-driven, event-aware: | Column lineage | No | **Shipped** (`// column`); docs site still TODO | | Snapshots / SCD2 | Yes (`key=… history`) | Managed strategy | | Selective execution grammar | No | UI/CLI surface | -| Schema contracts | No, but design metadata model | TODO with design work | +| Schema contracts | No, but design metadata model | **Shipped** (capture #2a + save-time check #2b) | | Packages / community / macros | No | **Shipped** (`// macros` workspace macro libraries) | | Semantic layer / metrics | No | Large additive scope | @@ -140,21 +140,43 @@ Today: `requestRunCascadeSignal` in the canvas, `// tag` annotation parsed. Graph + tags + last-run state has all the inputs. UI/CLI surface, not abstraction work. -### 6. Schema contracts +### 6. Schema contracts — **shipped** dbt: `contract: enforced` + `columns: [{name, data_type}]`. Compile-time check that model output matches the declaration. -Today: `// on datatable://users/active` is a string. Rename a column -upstream → downstream breaks at runtime, silently. +**Shipped**, capture-then-validate (no declaration to maintain): the schema a +managed `// materialize` run captures post-DESCRIBE into the versioned +`materialized_asset_schema` sidecar (#2a) *is* the contract, and every +consumer save/deploy is validated against the latest capture. Three surfaces, +same diff (`windmill_common::schema_contracts`, mirrored 1:1 in +`schemaContracts.ts`): -This is the item where the current asset abstraction is thinnest. -To do contracts well: capture output schemas after a run (substrate-specific -DESCRIBE), persist them as asset metadata, validate consumer references at -save time. The asset-as-typed-node model accommodates it — but **where** -schemas live (asset row, sidecar?), **when** they're captured (post-run? -edit-time?), and **how** versioning works are non-trivial design choices. -Worth doing intentionally now while the asset surface is still young. +- **Save-time (authoritative)**: `POST /w/{ws}/scripts/check_schema_contracts` + runs on the deployed content right after a save and returns **warnings — + never errors**. A deliberate upstream reshape must not fail every consumer + save; that's why dbt's `on_schema_change=fail` is deliberately not offered + in v1 (a CI/CLI gate can layer it on later without touching the grammar). +- **Editor flycheck**: the WASM parse that already runs on the open buffer + feeds the same diff, surfacing mismatches as live Monaco warning squiggles + anchored to the offending read / `// column` / `// data_test` line. +- **Autocompletion**: annotation refs (`// column out <- + ducklake://lake/orders.`, `// data_test relationships … ->`) complete + column names from the captured schema — the broken ref never gets typed. + +What's checked (ducklake-only — the only substrate with capture in v1; +datatable refs have no captured schema and stay silent): body-read/written +columns missing from the capture, `// column` lineage sources, and +`// data_test relationships` refs — including a captured-type *difference* on +the join columns when both sides are captured (phrased "differs", since the +runtime probe's IN-subquery still coerces). Column names compare +case-insensitively (DuckDB unquoted-identifier semantics); the managed +`_wm_partition` column is whitelisted; a `_current` scd2 view falls back +to its base table's capture (identical columns by construction). + +The producer-side escape hatch is `// materialize … on_schema_change=ignore`: +it declares the schema deliberately unstable and collapses downstream +warnings for that asset into a single informational note. Default is `warn`. ### 7. Packages / community / macros — **shipped** (macro libraries) diff --git a/frontend/src/lib/components/Editor.svelte b/frontend/src/lib/components/Editor.svelte index ae413d2484..78a75734c9 100644 --- a/frontend/src/lib/components/Editor.svelte +++ b/frontend/src/lib/components/Editor.svelte @@ -91,6 +91,11 @@ listWorkspaceMacrosCached, macroDefinitionSql } from '$lib/components/assets/workspaceMacros' + import { + fetchLatestSchema, + normalizeAssetPath, + type ContractMarker + } from '$lib/components/assets/AssetGraph/schemaContracts' import * as htmllang from '$lib/svelteMonarch' import { conf, language } from '$lib/vueMonarch' @@ -162,6 +167,11 @@ preparedAssetsSqlQueries?: InferAssetsSqlQueryDetails[] | undefined // To execute preview scripts with the right worker group customTag?: string + // Live schema-contract diagnostics (pipelines gap #2b): owner-scoped + // warning markers computed by the caller (ScriptEditor's contract + // mirror) from the buffer's asset refs vs captured producer schemas. + // Warning severity only — contracts never block; empty clears. + schemaContractMarkers?: ContractMarker[] } let { @@ -194,7 +204,8 @@ enablePreprocessorSnippet = false, rawAppRunnableKey = undefined, preparedAssetsSqlQueries, - customTag + customTag, + schemaContractMarkers = [] }: Props = $props() $effect.pre(() => { @@ -676,6 +687,53 @@ }) } + let schemaContractCompletor: IDisposable | undefined = undefined + + // Column-name completion for pipeline annotation refs (`// column out <- + // ducklake://lake/orders.|`, `// data_test relationships col -> + // ducklake://lake/customers.|`): suggests the referenced asset's *captured* + // columns (with types) so a broken ref never gets typed — the prevention + // side of the schema-contract check. Only fires on annotation comment lines + // with a ducklake URI right before the cursor's `.`; schemas come from the + // short-TTL contract cache, so per-keystroke cost is a map lookup. + function addSchemaContractCompletions() { + schemaContractCompletor?.dispose() + schemaContractCompletor = languages.registerCompletionItemProvider(lang, { + triggerCharacters: ['.'], + provideCompletionItems: async function (model, position) { + // Read the store per request, not at registration — the provider + // outlives a workspace switch. + const workspace = $workspaceStore + if (!workspace) return { suggestions: [] } + const before = model.getLineContent(position.lineNumber).slice(0, position.column - 1) + if (!/^\s*(\/\/|--|#)\s*(column|data_test|on|materialize)\b/.test(before)) { + return { suggestions: [] } + } + const uri = before.match(/ducklake:\/\/([\w/.{}-]+?)\.$/) + if (!uri) return { suggestions: [] } + const schema = await fetchLatestSchema(workspace, normalizeAssetPath(uri[1])) + if (!schema) return { suggestions: [] } + const word = model.getWordUntilPosition(position) + const range = { + startLineNumber: position.lineNumber, + endLineNumber: position.lineNumber, + startColumn: word.startColumn, + endColumn: word.endColumn + } + return { + suggestions: schema.columns.map((c) => ({ + label: c.name, + kind: languages.CompletionItemKind.Field, + detail: `${c.type} · captured schema v${schema.version}`, + insertText: c.name, + range, + sortText: 'a' + c.name + })) + } + } + }) + } + let sqlSchemaCompletor: IDisposable | undefined = undefined async function updateSchema(newSchemaRes: string | undefined) { @@ -1885,6 +1943,7 @@ sqlTypeCompletor && sqlTypeCompletor.dispose() resultCollectionCompletor && resultCollectionCompletor.dispose() workspaceMacroCompletor && workspaceMacroCompletor.dispose() + schemaContractCompletor && schemaContractCompletor.dispose() preprocessorCompletor && preprocessorCompletor.dispose() timeoutModel && clearTimeout(timeoutModel) changeChainStart = undefined @@ -1968,6 +2027,36 @@ : workspaceMacroCompletor?.dispose() }) + // Pipeline annotation grammar is language-agnostic (`//` / `--` / `#` + // comment headers), so contract-ref completions register for every script + // language that can be a pipeline member. The provider line-gates itself, + // so it is inert outside annotation lines. + $effect(() => { + initialized && ['duckdb', 'python3', 'bun', 'deno', 'nativets'].includes(scriptLang ?? '') + ? untrack(() => addSchemaContractCompletions()) + : schemaContractCompletor?.dispose() + }) + + // Schema-contract markers arrive as a prop because the mirror can finish + // computing before Monaco initializes on mount — reacting to `initialized` + // re-applies the pending set once the model exists. The ever-set latch + // keeps unrelated editors from calling setModelMarkers with [] forever. + let contractMarkersEverSet = false + $effect(() => { + const ms = schemaContractMarkers + if (!initialized || (ms.length === 0 && !contractMarkersEverSet)) return + contractMarkersEverSet = true + untrack(() => { + const model = editor?.getModel() + if (!model) return + meditor.setModelMarkers( + model, + 'schema-contracts', + ms.map((m) => ({ ...m, severity: MarkerSeverity.Warning })) + ) + }) + }) + $effect(() => { initialized && canHavePreprocessor(lang) && enablePreprocessorSnippet ? untrack(() => addPreprocessorCompletions(lang)) diff --git a/frontend/src/lib/components/ScriptBuilder.svelte b/frontend/src/lib/components/ScriptBuilder.svelte index fbfa4f1c1b..7d639b2c75 100644 --- a/frontend/src/lib/components/ScriptBuilder.svelte +++ b/frontend/src/lib/components/ScriptBuilder.svelte @@ -41,6 +41,7 @@ } from '$lib/utils' import Path from './Path.svelte' import { invalidateWorkspacePaths } from './PathNameAutocomplete.svelte' + import { notifyContractWarnings } from './assets/AssetGraph/schemaContracts' import ScriptEditor from './ScriptEditor.svelte' import { Alert, Button, Drawer, SecondsInput, Tab, TabContent, Tabs } from './common' import LanguageIcon from './common/languageIcons/LanguageIcon.svelte' @@ -654,6 +655,11 @@ // cache so it shows up immediately instead of after the 60s TTL. invalidateWorkspacePaths($workspaceStore!) + // Authoritative save-time schema-contract check (pipelines gap #2b): + // warn-only, post-commit so a self-produced target resolves to the + // content just deployed. Fire-and-forget — must never gate the deploy. + notifyContractWarnings($workspaceStore!, script.language, script.content) + if (!initialPath) { await CaptureService.moveCapturesAndConfigs({ workspace: $workspaceStore!, diff --git a/frontend/src/lib/components/ScriptEditor.svelte b/frontend/src/lib/components/ScriptEditor.svelte index f1944c73cc..50b5c8da95 100644 --- a/frontend/src/lib/components/ScriptEditor.svelte +++ b/frontend/src/lib/components/ScriptEditor.svelte @@ -105,6 +105,11 @@ import { canHavePreprocessor } from '$lib/script_helpers' import { assetEq, type AssetWithAltAccessType } from './assets/lib' import type { ColumnLineage } from './assets/AssetGraph/parsePipelineAnnotations' + import { + computeContractMarkers, + type ContractMarker, + type SchemaContractGraphContext + } from './assets/AssetGraph/schemaContracts' import { editor as meditor } from 'monaco-editor' import type { ReviewChangesOpts } from './copilot/chat/monaco-adapter' import GitRepoViewer from './GitRepoViewer.svelte' @@ -202,6 +207,11 @@ // regular /scripts/edit route keeps its current open-by-default UX; // the session preview opts in to save vertical real estate. initialTestPanelCollapsed?: boolean + // Producer-side facts for the live schema-contract diagnostics + // (`on_schema_change=ignore` suppression + scd2 `_current` fallback), + // built by the pipeline page from the resolved graph. Absent outside the + // pipeline editor — the check still runs, just without suppression. + schemaContractContext?: SchemaContractGraphContext } let { @@ -242,7 +252,8 @@ previewLayout = 'right', onTestStateChange, onTestJob, - initialTestPanelCollapsed = false + initialTestPanelCollapsed = false, + schemaContractContext = undefined }: Props = $props() $effect(() => { @@ -614,6 +625,35 @@ } ) + // Live schema-contract diagnostics (pipelines gap #2b): diff the buffer's + // asset refs against the captured producer schemas and surface mismatches + // as Monaco warning squiggles — the as-you-type mirror of the authoritative + // save-time check. The result is a prop on Editor (not an imperative call) + // because this can resolve before Monaco initializes on mount. Sequenced so + // a slow schema fetch can't overwrite the markers of a newer keystroke. + let contractMarkers: ContractMarker[] = $state([]) + let contractCheckSeq = 0 + watch([() => inferAssetsRes.current, () => schemaContractContext], () => { + const res = inferAssetsRes.current + const workspace = $workspaceStore + const seq = ++contractCheckSeq + if (!workspace || !res || res.status === 'error') { + contractMarkers = [] + return + } + const bufferCode = code + computeContractMarkers( + workspace, + bufferCode, + (res.assets ?? []) as AssetWithAltAccessType[], + schemaContractContext + ) + .then((markers) => { + if (seq === contractCheckSeq) contractMarkers = markers + }) + .catch((e) => console.error('schema-contract diagnostics failed', e)) + }) + watch([() => code, () => lang], () => { if (lang !== 'ansible') return inferAnsibleExecutionMode(code).then((v) => { @@ -2602,6 +2642,7 @@ bind:code={editorCode} bind:websocketAlive bind:this={editor} + schemaContractMarkers={contractMarkers} {yContent} awareness={wsProvider?.awareness} on:change={(e) => { diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte index 1abe21f76d..c9b921be2e 100644 --- a/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte @@ -1,5 +1,5 @@