mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-08-18 16:02:10 +00:00
feat(pipelines): schema contracts — save-time consumer checks vs captured schemas (#9917)
* feat(pipelines): schema contracts — save-time consumer checks vs captured schemas Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * refactor: move schemaContractContext above schemaCanEvolve doc comment Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix: emit scd2/on_schema_change in CLI local graph, address review notes Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix: gate editor _current ignore-suppression on scd2, matching backend Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
GitHub
+29
@@ -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"
|
||||
}
|
||||
+29
@@ -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"
|
||||
}
|
||||
+41
@@ -0,0 +1,41 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "SELECT DISTINCT ON (asset_path)\n asset_path, version, columns AS \"columns: Json<Vec<SchemaColumn>>\", 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<Vec<SchemaColumn>>",
|
||||
"type_info": "Jsonb"
|
||||
},
|
||||
{
|
||||
"ordinal": 3,
|
||||
"name": "captured_at",
|
||||
"type_info": "Timestamptz"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Text",
|
||||
"TextArray"
|
||||
]
|
||||
},
|
||||
"nullable": [
|
||||
false,
|
||||
false,
|
||||
false,
|
||||
false
|
||||
]
|
||||
},
|
||||
"hash": "c6c186eb3a108e1f695bf5730c23ef4ab1ccfdc4ed33904e83e5130ac0bf5896"
|
||||
}
|
||||
@@ -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 <kind> …` — a data-quality assertion run against the
|
||||
@@ -917,6 +945,14 @@ fn parse_materialize_spec(s: &str) -> Option<MaterializeSpec> {
|
||||
// `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<MaterializeSpec> {
|
||||
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");
|
||||
|
||||
@@ -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;",
|
||||
|
||||
@@ -71,6 +71,13 @@ struct ExpectedMaterialize {
|
||||
track: Vec<String>,
|
||||
#[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={}",
|
||||
|
||||
@@ -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<String>,
|
||||
// `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<String>,
|
||||
// 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()
|
||||
|
||||
@@ -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<windmill_common::schema_contracts::ContractWarning>,
|
||||
}
|
||||
|
||||
// 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<UserDB>,
|
||||
Path(w_id): Path<String>,
|
||||
Json(req): Json<CheckSchemaContractsRequest>,
|
||||
) -> JsonResult<CheckSchemaContractsResponse> {
|
||||
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 }))
|
||||
}
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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::*;
|
||||
|
||||
|
||||
@@ -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")]
|
||||
|
||||
@@ -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 … <- <asset>.<col>` source column is absent.
|
||||
MissingLineageSource,
|
||||
/// A `// data_test relationships … -> <asset>.<col>` 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 (`<lake>/<table>`).
|
||||
pub asset_path: String,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub column: Option<String>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub expected_type: Option<String>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub found_type: Option<String>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub schema_version: Option<i64>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub captured_at: Option<DateTime<Utc>>,
|
||||
pub message: String,
|
||||
}
|
||||
|
||||
/// The latest captured schema of one asset, as loaded by the wrapper.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct CapturedSchema {
|
||||
pub columns: Vec<SchemaColumn>,
|
||||
pub version: i64,
|
||||
pub captured_at: DateTime<Utc>,
|
||||
}
|
||||
|
||||
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::<Vec<_>>()
|
||||
.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<String, CapturedSchema>,
|
||||
ignored: &HashSet<String>,
|
||||
) -> Vec<ContractWarning> {
|
||||
let mut warnings: Vec<ContractWarning> = 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<String> = 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<Vec<ContractWarning>> {
|
||||
// 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<String> = 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 `<dim>_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<String> = 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<String> = lookup_paths.into_iter().collect();
|
||||
|
||||
let schema_rows = sqlx::query!(
|
||||
r#"SELECT DISTINCT ON (asset_path)
|
||||
asset_path, version, columns AS "columns: Json<Vec<SchemaColumn>>", 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<String, CapturedSchema> = 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<String> = 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<String> = HashSet::new();
|
||||
if !producer_edges.is_empty() {
|
||||
let producer_paths: Vec<String> = producer_edges
|
||||
.iter()
|
||||
.map(|e| e.producer_path.clone())
|
||||
.collect::<HashSet<_>>()
|
||||
.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<String, Option<MaterializeSpec>> = 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::<Utc>::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::<BTreeMap<_, _>>(),
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
#[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());
|
||||
}
|
||||
}
|
||||
@@ -44,9 +44,14 @@ export type GraphRunnable = {
|
||||
column_lineage?: unknown[];
|
||||
// `// materialize <asset>` 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
|
||||
// `<dim>_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 <worker-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=<col> → 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=<col>) > 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 ?? []) {
|
||||
|
||||
@@ -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
|
||||
`<dim>_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
|
||||
|
||||
+33
-11
@@ -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 `<dim>_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)
|
||||
|
||||
|
||||
@@ -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))
|
||||
|
||||
@@ -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!,
|
||||
|
||||
@@ -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) => {
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
<script lang="ts">
|
||||
import { ScriptService, type Script } from '$lib/gen'
|
||||
import { ScriptService, type Script, type ScriptLang } from '$lib/gen'
|
||||
import { resource } from 'runed'
|
||||
import { base } from '$lib/base'
|
||||
import Button from '$lib/components/common/button/Button.svelte'
|
||||
@@ -38,6 +38,7 @@
|
||||
import S3FilePreview from '$lib/components/S3FilePreview.svelte'
|
||||
import DataTablePreview from './DataTablePreview.svelte'
|
||||
import DucklakeAssetPanel from './DucklakeAssetPanel.svelte'
|
||||
import { notifyContractWarnings, type SchemaContractGraphContext } from './schemaContracts'
|
||||
import AssetRunsPanel from './AssetRunsPanel.svelte'
|
||||
import { Pane, Splitpanes } from 'svelte-splitpanes'
|
||||
import { fade } from 'svelte/transition'
|
||||
@@ -146,6 +147,10 @@
|
||||
// `replace` producer). Forwarded to the Schema tab: version history when
|
||||
// true, a single fixed-schema view when false. Defaults to true (unknown).
|
||||
schemaCanEvolve?: boolean
|
||||
// Producer-side facts for the editor's live schema-contract diagnostics
|
||||
// (ignore suppression + scd2 `_current` fallback), built by the page from
|
||||
// the resolved graph and forwarded to ScriptEditor.
|
||||
schemaContractContext?: SchemaContractGraphContext
|
||||
// Bumped by the parent after dispatching a run so the runs panel
|
||||
// re-fetches the listing immediately (rather than waiting on its
|
||||
// background poll tick).
|
||||
@@ -254,6 +259,7 @@
|
||||
selectionProducers = [],
|
||||
selectionColumnGraph,
|
||||
schemaCanEvolve = true,
|
||||
schemaContractContext = undefined,
|
||||
runsRefreshKey,
|
||||
runsPendingJobId,
|
||||
onRunCompleted,
|
||||
@@ -664,6 +670,9 @@
|
||||
// same parent").
|
||||
if (typeof newHash === 'string' && newHash) script.hash = newHash
|
||||
sendUserToast(`Saved ${script.path}`)
|
||||
// Authoritative save-time schema-contract check (pipelines gap #2b):
|
||||
// warn-only, post-commit. Fire-and-forget — must never gate the save.
|
||||
notifyContractWarnings(workspace, script.language as ScriptLang, script.content)
|
||||
if (isDraft) {
|
||||
onDraftSaved?.(script.path)
|
||||
} else {
|
||||
@@ -1155,6 +1164,7 @@
|
||||
showCaptures={false}
|
||||
noSyncFromGithub
|
||||
requireValidAssets
|
||||
{schemaContractContext}
|
||||
lang={script.language}
|
||||
path={script.path}
|
||||
tag={script.tag}
|
||||
|
||||
@@ -22,6 +22,7 @@
|
||||
import type { RunnableRunState, PipelineEvent } from './activeRunnables.svelte'
|
||||
import type { PipelineOutputKind } from './pipelineTemplates'
|
||||
import type { PipelineEditorState } from './pipelineEditorState.svelte'
|
||||
import type { SchemaContractGraphContext } from './schemaContracts'
|
||||
|
||||
type RunProducer = { kind: 'script' | 'flow'; path: string; unsaved?: boolean; cascade?: boolean }
|
||||
|
||||
@@ -73,6 +74,7 @@
|
||||
selectionProducers = [],
|
||||
selectionColumnGraph,
|
||||
schemaCanEvolve = true,
|
||||
schemaContractContext = undefined,
|
||||
downstreamSubscribers = 0,
|
||||
onStartBoundedRunForOpen,
|
||||
canBoundedRunOpenScript = false,
|
||||
@@ -170,6 +172,9 @@
|
||||
/** Transitive column-lineage trace for a selected ducklake asset (route page). */
|
||||
selectionColumnGraph?: ColumnLineageGraph
|
||||
schemaCanEvolve?: boolean
|
||||
/** Producer-side facts for the editor's live schema-contract diagnostics
|
||||
* (ignore suppression + scd2 `_current` fallback), from the route page. */
|
||||
schemaContractContext?: SchemaContractGraphContext
|
||||
downstreamSubscribers?: number
|
||||
onStartBoundedRunForOpen?: (path: string) => void
|
||||
canBoundedRunOpenScript?: boolean
|
||||
@@ -486,6 +491,7 @@
|
||||
selectionProducers={activeDraft ? [] : selectionProducers}
|
||||
{selectionColumnGraph}
|
||||
{schemaCanEvolve}
|
||||
{schemaContractContext}
|
||||
{runsRefreshKey}
|
||||
{runsPendingJobId}
|
||||
{activeRunnable}
|
||||
|
||||
@@ -72,6 +72,8 @@ type Fixture = {
|
||||
scd2?: boolean
|
||||
track?: string[]
|
||||
close_deleted?: boolean
|
||||
// "warn" | "ignore"; absent === "warn" (the default)
|
||||
on_schema_change?: string
|
||||
} | null
|
||||
// Snake_case form matching the Rust `DataTest` serde output, so the one
|
||||
// corpus drives both sides. The TS parser emits this shape verbatim
|
||||
@@ -180,6 +182,9 @@ describe('parsePipelineAnnotations matches the shared Rust fixture corpus', () =
|
||||
expect(got.materialize?.closeDeleted ?? false, 'materialize close_deleted').toBe(
|
||||
f.expected.materialize.close_deleted ?? false
|
||||
)
|
||||
expect(got.materialize?.onSchemaChange ?? 'warn', 'materialize on_schema_change').toBe(
|
||||
f.expected.materialize.on_schema_change ?? 'warn'
|
||||
)
|
||||
}
|
||||
|
||||
expect(got.dataTests, 'data tests').toEqual(f.expected.data_tests ?? [])
|
||||
|
||||
@@ -117,6 +117,10 @@ export type MaterializeSpec = {
|
||||
track?: string[]
|
||||
// SCD2 hard-delete-close (`deletes=close`): close absent keys; absent === false
|
||||
closeDeleted?: boolean
|
||||
// `on_schema_change=ignore` opts the produced asset out of downstream
|
||||
// schema-contract warnings (save-time metadata only). Default `warn`;
|
||||
// `fail` is deliberately unrecognized in v1 (saves never hard-block).
|
||||
onSchemaChange?: 'warn' | 'ignore'
|
||||
}
|
||||
|
||||
// `// data_test <kind> …` — a data-quality assertion run against the
|
||||
@@ -264,7 +268,8 @@ function parseAssetSyntaxDefault(s: string): PipelineTriggerAsset | undefined {
|
||||
// managed mode; a leading `scd2` word is an alias for the `history` flag; the
|
||||
// next token is the target asset URI (default-syntax shorthands enabled); the
|
||||
// remainder are strategy options (`append` flag, `key=<col>`, `history` flag,
|
||||
// `track=<c1,c2,…>`, `deletes=close`). Missing/empty target → undefined (dropped).
|
||||
// `track=<c1,c2,…>`, `deletes=close`, `on_schema_change=ignore`). Missing/empty
|
||||
// target → undefined (dropped).
|
||||
function parseMaterializeSpec(s: string): MaterializeSpec | undefined {
|
||||
// One optional leading mode keyword: `manual` (track-only) or `scd2` (an alias
|
||||
// for the `history` flag below).
|
||||
@@ -304,6 +309,9 @@ function parseMaterializeSpec(s: string): MaterializeSpec | undefined {
|
||||
// `deletes=close` (scd2 only) opts into hard-delete-close; any other value
|
||||
// (or absence) keeps the soft-delete default.
|
||||
const closeDeleted = opts.get('deletes') === 'close'
|
||||
// `on_schema_change=ignore` suppresses downstream contract warnings; any
|
||||
// other value (or absence) keeps the `warn` default, fail-safe like `deletes=`.
|
||||
const onSchemaChange = opts.get('on_schema_change') === 'ignore' ? 'ignore' : 'warn'
|
||||
return {
|
||||
targetKind: asset.kind,
|
||||
targetPath: asset.path,
|
||||
@@ -312,7 +320,8 @@ function parseMaterializeSpec(s: string): MaterializeSpec | undefined {
|
||||
uniqueKey,
|
||||
scd2,
|
||||
track,
|
||||
closeDeleted
|
||||
closeDeleted,
|
||||
onSchemaChange
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,276 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import {
|
||||
buildSchemaContractContext,
|
||||
diffSchemaContracts,
|
||||
mapWarningsToMarkers,
|
||||
normalizeAssetPath,
|
||||
referencedDucklakePaths,
|
||||
type CapturedSchemaLite
|
||||
} from './schemaContracts'
|
||||
import { parsePipelineAnnotations } from './parsePipelineAnnotations'
|
||||
import type { AssetWithAltAccessType } from '../lib'
|
||||
|
||||
// Mirrors backend/windmill-common/src/schema_contracts.rs unit tests — the two
|
||||
// diffs must apply the same rules or the editor previews a different verdict
|
||||
// than the save-time check returns.
|
||||
|
||||
function schema(cols: [string, string][], version = 2): CapturedSchemaLite {
|
||||
return {
|
||||
columns: cols.map(([name, type]) => ({ name, type })),
|
||||
version,
|
||||
capturedAt: '2026-01-01T00:00:00Z'
|
||||
}
|
||||
}
|
||||
|
||||
function readAsset(path: string, cols: string[]): AssetWithAltAccessType {
|
||||
return {
|
||||
path,
|
||||
kind: 'ducklake',
|
||||
access_type: 'r',
|
||||
columns: Object.fromEntries(cols.map((c) => [c, 'r' as const]))
|
||||
}
|
||||
}
|
||||
|
||||
const NO_ANN = { columnLineage: [], dataTests: [] }
|
||||
|
||||
describe('diffSchemaContracts', () => {
|
||||
it('warns on a missing read column, matching case-insensitively', () => {
|
||||
const schemas = new Map([
|
||||
[
|
||||
'lake/orders',
|
||||
schema([
|
||||
['Order_ID', 'BIGINT'],
|
||||
['amount_usd', 'DOUBLE']
|
||||
])
|
||||
]
|
||||
])
|
||||
const w = diffSchemaContracts({
|
||||
...NO_ANN,
|
||||
assets: [readAsset('lake/orders', ['order_id', 'amount'])],
|
||||
schemas,
|
||||
ignored: new Set()
|
||||
})
|
||||
expect(w).toHaveLength(1)
|
||||
expect(w[0].kind).toBe('missing_column')
|
||||
expect(w[0].column).toBe('amount')
|
||||
expect(w[0].schema_version).toBe(2)
|
||||
})
|
||||
|
||||
it('skips unknown-column assets, "*" and reserved columns', () => {
|
||||
const schemas = new Map([['lake/orders', schema([['id', 'BIGINT']])]])
|
||||
const noColumns: AssetWithAltAccessType = {
|
||||
path: 'lake/orders',
|
||||
kind: 'ducklake',
|
||||
access_type: 'r'
|
||||
}
|
||||
expect(
|
||||
diffSchemaContracts({ ...NO_ANN, assets: [noColumns], schemas, ignored: new Set() })
|
||||
).toEqual([])
|
||||
expect(
|
||||
diffSchemaContracts({
|
||||
...NO_ANN,
|
||||
assets: [readAsset('lake/orders', ['*', '_wm_partition', 'id'])],
|
||||
schemas,
|
||||
ignored: new Set()
|
||||
})
|
||||
).toEqual([])
|
||||
})
|
||||
|
||||
it('is silent for assets without a captured schema', () => {
|
||||
expect(
|
||||
diffSchemaContracts({
|
||||
...NO_ANN,
|
||||
assets: [readAsset('lake/unknown', ['whatever'])],
|
||||
schemas: new Map(),
|
||||
ignored: new Set()
|
||||
})
|
||||
).toEqual([])
|
||||
})
|
||||
|
||||
it('normalizes the {partition} token before lookup', () => {
|
||||
const schemas = new Map([['lake/orders', schema([['id', 'BIGINT']])]])
|
||||
const w = diffSchemaContracts({
|
||||
...NO_ANN,
|
||||
assets: [readAsset('lake/orders/{partition}', ['gone'])],
|
||||
schemas,
|
||||
ignored: new Set()
|
||||
})
|
||||
expect(w).toHaveLength(1)
|
||||
expect(w[0].asset_path).toBe('lake/orders')
|
||||
})
|
||||
|
||||
it('warns on broken // column lineage refs', () => {
|
||||
const ann = parsePipelineAnnotations(
|
||||
'// column total <- ducklake://lake/orders.amount\nSELECT 1;'
|
||||
)
|
||||
const schemas = new Map([['lake/orders', schema([['id', 'BIGINT']])]])
|
||||
const w = diffSchemaContracts({
|
||||
assets: [],
|
||||
columnLineage: ann.columnLineage,
|
||||
dataTests: [],
|
||||
schemas,
|
||||
ignored: new Set()
|
||||
})
|
||||
expect(w).toHaveLength(1)
|
||||
expect(w[0].kind).toBe('missing_lineage_source')
|
||||
})
|
||||
|
||||
it('flags missing relationship columns and captured-type differences', () => {
|
||||
const ann = parsePipelineAnnotations(
|
||||
'// 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;'
|
||||
)
|
||||
const schemas = new Map([
|
||||
['lake/customers', schema([['id', 'VARCHAR']])],
|
||||
['lake/orders', schema([['customer_id', 'BIGINT']])]
|
||||
])
|
||||
const w = diffSchemaContracts({
|
||||
assets: [],
|
||||
columnLineage: ann.columnLineage,
|
||||
dataTests: ann.dataTests,
|
||||
materialize: ann.materialize,
|
||||
schemas,
|
||||
ignored: new Set()
|
||||
})
|
||||
expect(w).toHaveLength(2)
|
||||
expect(
|
||||
w.some(
|
||||
(x) =>
|
||||
x.kind === 'relationship_type_mismatch' &&
|
||||
x.expected_type === 'BIGINT' &&
|
||||
x.found_type === 'VARCHAR'
|
||||
)
|
||||
).toBe(true)
|
||||
expect(w.some((x) => x.kind === 'missing_relationship_column' && x.column === 'uuid')).toBe(
|
||||
true
|
||||
)
|
||||
})
|
||||
|
||||
it('suppresses ignored assets down to one informational note', () => {
|
||||
const schemas = new Map([['lake/orders', schema([['id', 'BIGINT']])]])
|
||||
const ignored = new Set(['lake/orders'])
|
||||
const w = diffSchemaContracts({
|
||||
...NO_ANN,
|
||||
assets: [readAsset('lake/orders', ['a', 'b'])],
|
||||
schemas,
|
||||
ignored
|
||||
})
|
||||
expect(w).toHaveLength(1)
|
||||
expect(w[0].kind).toBe('suppressed')
|
||||
expect(
|
||||
diffSchemaContracts({
|
||||
...NO_ANN,
|
||||
assets: [readAsset('lake/orders', ['id'])],
|
||||
schemas,
|
||||
ignored
|
||||
})
|
||||
).toEqual([])
|
||||
})
|
||||
})
|
||||
|
||||
describe('referencedDucklakePaths', () => {
|
||||
it('collects paths from reads, lineage, relationships and materialize', () => {
|
||||
const ann = parsePipelineAnnotations(
|
||||
'// materialize ducklake://lake/out\n' +
|
||||
'// column total <- ducklake://lake/a.amount\n' +
|
||||
'// data_test relationships k -> ducklake://lake/b.id\n' +
|
||||
'SELECT 1;'
|
||||
)
|
||||
const refs = referencedDucklakePaths({
|
||||
assets: [readAsset('lake/c/{partition}', ['x'])],
|
||||
columnLineage: ann.columnLineage,
|
||||
dataTests: ann.dataTests,
|
||||
materialize: ann.materialize
|
||||
})
|
||||
expect(refs.sort()).toEqual(['lake/a', 'lake/b', 'lake/c', 'lake/out'])
|
||||
})
|
||||
})
|
||||
|
||||
describe('buildSchemaContractContext', () => {
|
||||
it('derives ignored assets and scd2 _current bases from graph runnables', () => {
|
||||
const ctx = buildSchemaContractContext([
|
||||
{
|
||||
materialize_target: { kind: 'ducklake', path: 'lake/dim' },
|
||||
materialize_strategy: 'scd2',
|
||||
materialize_on_schema_change: 'ignore'
|
||||
},
|
||||
{
|
||||
materialize_target: { kind: 'ducklake', path: 'lake/orders' },
|
||||
materialize_strategy: 'replace'
|
||||
},
|
||||
// non-ducklake and absent targets are ignored
|
||||
{ materialize_target: { kind: 's3object', path: 'x/y' }, materialize_strategy: 'scd2' },
|
||||
{}
|
||||
])
|
||||
expect(ctx.ignoredAssets).toEqual(['lake/dim', 'lake/dim_current'])
|
||||
expect(ctx.scd2CurrentBases).toEqual({ 'lake/dim_current': 'lake/dim' })
|
||||
})
|
||||
|
||||
it('ignores _current only for scd2 producers (backend spec.scd2 gate)', () => {
|
||||
const ctx = buildSchemaContractContext([
|
||||
{
|
||||
materialize_target: { kind: 'ducklake', path: 'lake/t' },
|
||||
materialize_strategy: 'replace',
|
||||
materialize_on_schema_change: 'ignore'
|
||||
}
|
||||
])
|
||||
// a non-scd2 producer's `<base>_current` is an unrelated asset — it must
|
||||
// keep warning, exactly like the server-side check
|
||||
expect(ctx.ignoredAssets).toEqual(['lake/t'])
|
||||
expect(ctx.scd2CurrentBases).toEqual({})
|
||||
})
|
||||
})
|
||||
|
||||
describe('mapWarningsToMarkers', () => {
|
||||
it('anchors annotation warnings to their lines and body reads to the identifier', () => {
|
||||
const code =
|
||||
'-- pipeline\n' +
|
||||
'-- on ducklake://lake/orders\n' +
|
||||
'-- column total <- ducklake://lake/orders.amount\n' +
|
||||
'-- data_test relationships k -> ducklake://lake/customers.uuid\n' +
|
||||
'SELECT amount FROM dl.orders;'
|
||||
const markers = mapWarningsToMarkers(code, [
|
||||
{
|
||||
kind: 'missing_lineage_source',
|
||||
asset_path: 'lake/orders',
|
||||
column: 'amount',
|
||||
message: 'm1'
|
||||
},
|
||||
{
|
||||
kind: 'missing_relationship_column',
|
||||
asset_path: 'lake/customers',
|
||||
column: 'uuid',
|
||||
message: 'm2'
|
||||
},
|
||||
{ kind: 'missing_column', asset_path: 'lake/orders', column: 'amount', message: 'm3' },
|
||||
{ kind: 'suppressed', asset_path: 'lake/orders', message: 'hidden' }
|
||||
])
|
||||
expect(markers).toHaveLength(3)
|
||||
expect(markers[0].startLineNumber).toBe(3)
|
||||
expect(markers[1].startLineNumber).toBe(4)
|
||||
// body-read warning anchors to the first occurrence of the identifier,
|
||||
// which is the annotation line mentioning `amount` (line 3)
|
||||
expect(markers[2].startLineNumber).toBe(3)
|
||||
// token range is tight around the identifier, not the whole line
|
||||
const line3 = '-- column total <- ducklake://lake/orders.amount'
|
||||
expect(markers[2].startColumn).toBe(line3.indexOf('amount') + 1)
|
||||
})
|
||||
|
||||
it('falls back to the line mentioning the asset path', () => {
|
||||
const code = '# pipeline\n# on ducklake://lake/orders\nprint(1)'
|
||||
const markers = mapWarningsToMarkers(code, [
|
||||
{ kind: 'missing_column', asset_path: 'lake/orders', column: 'zzz', message: 'm' }
|
||||
])
|
||||
expect(markers[0].startLineNumber).toBe(2)
|
||||
})
|
||||
})
|
||||
|
||||
describe('normalizeAssetPath', () => {
|
||||
it('strips the partition token and trailing slashes', () => {
|
||||
expect(normalizeAssetPath('lake/orders/{partition}')).toBe('lake/orders')
|
||||
expect(normalizeAssetPath('lake/orders_{partition}')).toBe('lake/orders_')
|
||||
expect(normalizeAssetPath('lake/orders/')).toBe('lake/orders')
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,440 @@
|
||||
// Client-side mirror of the save-time schema-contract check (pipelines gap
|
||||
// #2b, backend/windmill-common/src/schema_contracts.rs). The backend endpoint
|
||||
// (`checkSchemaContracts`) is the authoritative check run on save; this mirror
|
||||
// drives the *live* editor surface (Monaco warning markers + completions) from
|
||||
// the WASM parse that already runs on the open buffer, so the two must apply
|
||||
// the same rules: ducklake-only, case-insensitive column names, `columns`
|
||||
// absent ⇒ skip, `_wm_partition` whitelisted, `{partition}` token stripped,
|
||||
// annotation-declared lineage only, asset without captured schema ⇒ silent.
|
||||
|
||||
import { AssetService, ScriptService, type ContractWarning, type ScriptLang } from '$lib/gen'
|
||||
import { sendUserToast } from '$lib/toast'
|
||||
import type { AssetWithAltAccessType } from '../lib'
|
||||
import {
|
||||
parsePipelineAnnotations,
|
||||
type ColumnLineage,
|
||||
type DataTest,
|
||||
type MaterializeSpec
|
||||
} from './parsePipelineAnnotations'
|
||||
|
||||
// Columns the materialize engine manages; excluded from the captured schema on
|
||||
// purpose, so reads of them must not warn.
|
||||
const RESERVED_COLUMNS = ['_wm_partition']
|
||||
|
||||
const PARTITION_TOKEN = '{partition}'
|
||||
|
||||
export type CapturedSchemaLite = {
|
||||
columns: { name: string; type: string }[]
|
||||
version: number
|
||||
capturedAt: string
|
||||
}
|
||||
|
||||
// Strip the `{partition}` token a declared URI may carry so lookups hit the
|
||||
// captured path (mirrors `normalize_asset_path`).
|
||||
export function normalizeAssetPath(path: string): string {
|
||||
return path
|
||||
.replaceAll('/' + PARTITION_TOKEN, '')
|
||||
.replaceAll(PARTITION_TOKEN, '')
|
||||
.replace(/\/+$/, '')
|
||||
}
|
||||
|
||||
function isReserved(name: string): boolean {
|
||||
return RESERVED_COLUMNS.some((r) => r.toLowerCase() === name.toLowerCase())
|
||||
}
|
||||
|
||||
function findColumn(
|
||||
schema: CapturedSchemaLite,
|
||||
name: string
|
||||
): { name: string; type: string } | undefined {
|
||||
const lower = name.toLowerCase()
|
||||
return schema.columns.find((c) => c.name.toLowerCase() === lower)
|
||||
}
|
||||
|
||||
export type SchemaContractInputs = {
|
||||
// Per-asset column reads/writes from the WASM parse (entries without a
|
||||
// `columns` map are skipped — wildcard/unknown access).
|
||||
assets: AssetWithAltAccessType[]
|
||||
// Annotation-declared `// column` lineage ONLY (not merged AST-inferred
|
||||
// lineage — redundant with body reads and alias-attribution can misfire).
|
||||
columnLineage: ColumnLineage[]
|
||||
dataTests: DataTest[]
|
||||
materialize?: MaterializeSpec
|
||||
// Latest captured schema per normalized ducklake path (after any
|
||||
// `_current` → base-table fallback the caller resolved).
|
||||
schemas: Map<string, CapturedSchemaLite>
|
||||
// Normalized paths whose producer declares `on_schema_change=ignore`.
|
||||
ignored: Set<string>
|
||||
}
|
||||
|
||||
// Mirrors backend `diff_contract` — same warning kinds and suppression
|
||||
// semantics, minus the human message wording (the editor renders its own).
|
||||
export function diffSchemaContracts(input: SchemaContractInputs): ContractWarning[] {
|
||||
const { assets, columnLineage, dataTests, materialize, schemas, ignored } = input
|
||||
const warnings: ContractWarning[] = []
|
||||
|
||||
// W1 — body-read/written columns missing from the captured schema.
|
||||
for (const a of assets) {
|
||||
if (a.kind !== 'ducklake' || a.columns == undefined) continue
|
||||
const path = normalizeAssetPath(a.path)
|
||||
const schema = schemas.get(path)
|
||||
if (!schema) continue
|
||||
for (const col of Object.keys(a.columns)) {
|
||||
if (col === '*' || isReserved(col)) continue
|
||||
if (!findColumn(schema, col)) {
|
||||
warnings.push({
|
||||
kind: 'missing_column',
|
||||
asset_path: path,
|
||||
column: col,
|
||||
schema_version: schema.version,
|
||||
captured_at: schema.capturedAt,
|
||||
message: `column \`${col}\` of ducklake://${path} is not in its captured schema (v${schema.version}, columns: ${schema.columns.map((c) => c.name).join(', ')})`
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// W2 — `// column` lineage source refs.
|
||||
for (const cl of columnLineage) {
|
||||
for (const input of cl.inputs) {
|
||||
if (input.from_kind !== 'ducklake' || isReserved(input.from_column)) continue
|
||||
const path = normalizeAssetPath(input.from_path)
|
||||
const schema = schemas.get(path)
|
||||
if (!schema) continue
|
||||
if (!findColumn(schema, input.from_column)) {
|
||||
warnings.push({
|
||||
kind: 'missing_lineage_source',
|
||||
asset_path: path,
|
||||
column: input.from_column,
|
||||
schema_version: schema.version,
|
||||
captured_at: schema.capturedAt,
|
||||
message: `\`// column ${cl.column}\` reads \`${input.from_column}\` from ducklake://${path}, which is not in its captured schema (v${schema.version})`
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// W3 — relationships refs: missing column, and captured-type difference
|
||||
// when the consumer's own materialize target has a capture. Types still
|
||||
// coerce at run time, so a difference is "differs", never "will fail".
|
||||
const ownSchema =
|
||||
materialize?.targetKind === 'ducklake'
|
||||
? schemas.get(normalizeAssetPath(materialize.targetPath))
|
||||
: undefined
|
||||
for (const dt of dataTests) {
|
||||
if (dt.type !== 'relationships' || dt.to_kind !== 'ducklake') continue
|
||||
const path = normalizeAssetPath(dt.to_path)
|
||||
const schema = schemas.get(path)
|
||||
if (!schema) continue
|
||||
const refCol = findColumn(schema, dt.to_column)
|
||||
if (!refCol) {
|
||||
warnings.push({
|
||||
kind: 'missing_relationship_column',
|
||||
asset_path: path,
|
||||
column: dt.to_column,
|
||||
schema_version: schema.version,
|
||||
captured_at: schema.capturedAt,
|
||||
message: `\`// data_test relationships ${dt.column}\` references ducklake://${path}.${dt.to_column}, which is not in its captured schema (v${schema.version})`
|
||||
})
|
||||
} else {
|
||||
const ownCol = ownSchema && findColumn(ownSchema, dt.column)
|
||||
if (ownCol && ownCol.type.toLowerCase() !== refCol.type.toLowerCase()) {
|
||||
warnings.push({
|
||||
kind: 'relationship_type_mismatch',
|
||||
asset_path: path,
|
||||
column: dt.to_column,
|
||||
expected_type: ownCol.type,
|
||||
found_type: refCol.type,
|
||||
schema_version: schema.version,
|
||||
captured_at: schema.capturedAt,
|
||||
message: `\`// data_test relationships ${dt.column}\` joins \`${dt.column}\` (${ownCol.type}) to ducklake://${path}.${dt.to_column} (${refCol.type}) — captured types differ`
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// W4 — producer `on_schema_change=ignore`: drop the asset's warnings,
|
||||
// leaving one informational entry per suppressed asset.
|
||||
if (ignored.size > 0) {
|
||||
const suppressed: string[] = []
|
||||
const kept = warnings.filter((w) => {
|
||||
if (ignored.has(w.asset_path)) {
|
||||
if (!suppressed.includes(w.asset_path)) suppressed.push(w.asset_path)
|
||||
return false
|
||||
}
|
||||
return true
|
||||
})
|
||||
warnings.length = 0
|
||||
warnings.push(...kept)
|
||||
for (const path of suppressed) {
|
||||
warnings.push({
|
||||
kind: 'suppressed',
|
||||
asset_path: path,
|
||||
message: `schema mismatches on ducklake://${path} suppressed by its producer's \`on_schema_change=ignore\``
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
return warnings
|
||||
}
|
||||
|
||||
// The ducklake paths a buffer references in ways the contract check inspects —
|
||||
// what the editor needs captured schemas for.
|
||||
export function referencedDucklakePaths(
|
||||
input: Pick<SchemaContractInputs, 'assets' | 'columnLineage' | 'dataTests' | 'materialize'>
|
||||
): string[] {
|
||||
const paths = new Set<string>()
|
||||
for (const a of input.assets) {
|
||||
if (a.kind === 'ducklake' && a.columns != undefined) paths.add(normalizeAssetPath(a.path))
|
||||
}
|
||||
for (const cl of input.columnLineage) {
|
||||
for (const i of cl.inputs) {
|
||||
if (i.from_kind === 'ducklake') paths.add(normalizeAssetPath(i.from_path))
|
||||
}
|
||||
}
|
||||
for (const dt of input.dataTests) {
|
||||
if (dt.type === 'relationships' && dt.to_kind === 'ducklake')
|
||||
paths.add(normalizeAssetPath(dt.to_path))
|
||||
}
|
||||
if (input.materialize?.targetKind === 'ducklake')
|
||||
paths.add(normalizeAssetPath(input.materialize.targetPath))
|
||||
return [...paths]
|
||||
}
|
||||
|
||||
// --- Captured-schema cache ------------------------------------------------
|
||||
|
||||
// Short-TTL cache so per-keystroke recomputes and completion requests don't
|
||||
// re-fetch. Captured schemas only change when a producer materializes, so a
|
||||
// briefly stale hit is fine — the authoritative save-time check re-reads.
|
||||
const SCHEMA_TTL_MS = 30_000
|
||||
const schemaCache = new Map<string, { at: number; value: CapturedSchemaLite | undefined }>()
|
||||
|
||||
export async function fetchLatestSchema(
|
||||
workspace: string,
|
||||
path: string
|
||||
): Promise<CapturedSchemaLite | undefined> {
|
||||
const key = `${workspace}:${path}`
|
||||
const hit = schemaCache.get(key)
|
||||
if (hit && Date.now() - hit.at < SCHEMA_TTL_MS) return hit.value
|
||||
let value: CapturedSchemaLite | undefined = undefined
|
||||
try {
|
||||
const versions = await AssetService.listAssetSchemas({ workspace, path })
|
||||
const latest = versions[0]
|
||||
if (latest) {
|
||||
value = {
|
||||
columns: latest.columns,
|
||||
version: latest.version,
|
||||
capturedAt: latest.captured_at
|
||||
}
|
||||
}
|
||||
} catch (e) {
|
||||
console.error('failed to fetch captured asset schema', path, e)
|
||||
}
|
||||
schemaCache.set(key, { at: Date.now(), value })
|
||||
return value
|
||||
}
|
||||
|
||||
// Resolve the schema map for a set of referenced paths, applying the scd2
|
||||
// `<dim>_current` → base-table fallback when the graph identifies the view's
|
||||
// producer as a managed scd2 materializer (the view is `SELECT * … WHERE
|
||||
// is_current`, so columns are identical).
|
||||
export async function fetchSchemasForPaths(
|
||||
workspace: string,
|
||||
paths: string[],
|
||||
scd2CurrentBase?: (path: string) => string | undefined
|
||||
): Promise<Map<string, CapturedSchemaLite>> {
|
||||
const out = new Map<string, CapturedSchemaLite>()
|
||||
await Promise.all(
|
||||
paths.map(async (p) => {
|
||||
let schema = await fetchLatestSchema(workspace, p)
|
||||
if (!schema && p.endsWith('_current')) {
|
||||
const base = scd2CurrentBase?.(p)
|
||||
if (base) schema = await fetchLatestSchema(workspace, base)
|
||||
}
|
||||
if (schema) out.set(p, schema)
|
||||
})
|
||||
)
|
||||
return out
|
||||
}
|
||||
|
||||
// --- Pipeline-graph context ---------------------------------------------------
|
||||
|
||||
// Producer-side facts the contract mirror needs but cannot derive from the
|
||||
// open buffer: which assets are muted (`on_schema_change=ignore`) and which
|
||||
// `<dim>_current` views map to an scd2 base table. Built by the pipeline page
|
||||
// from the resolved graph; absent outside the pipeline editor (standalone
|
||||
// script editor), where suppression simply doesn't apply client-side — the
|
||||
// save-time server check remains authoritative either way.
|
||||
export type SchemaContractGraphContext = {
|
||||
// Normalized asset paths whose producer declares `on_schema_change=ignore`.
|
||||
ignoredAssets: string[]
|
||||
// `<base>_current` → base for managed scd2 producers in the graph.
|
||||
scd2CurrentBases: Record<string, string>
|
||||
}
|
||||
|
||||
export function buildSchemaContractContext(
|
||||
runnables: Pick<
|
||||
import('./types').AssetGraphRunnableNode,
|
||||
'materialize_target' | 'materialize_strategy' | 'materialize_on_schema_change'
|
||||
>[]
|
||||
): SchemaContractGraphContext {
|
||||
const ignoredAssets: string[] = []
|
||||
const scd2CurrentBases: Record<string, string> = {}
|
||||
for (const r of runnables) {
|
||||
const t = r.materialize_target
|
||||
if (!t || t.kind !== 'ducklake') continue
|
||||
const base = normalizeAssetPath(t.path)
|
||||
if (r.materialize_on_schema_change === 'ignore') {
|
||||
ignoredAssets.push(base)
|
||||
// The `_current` companion is the producer's own view only for scd2 —
|
||||
// mirroring the backend's `spec.scd2` gate; for any other strategy a
|
||||
// `<base>_current` ref is an unrelated asset that must keep warning.
|
||||
if (r.materialize_strategy === 'scd2') {
|
||||
ignoredAssets.push(`${base}_current`)
|
||||
}
|
||||
}
|
||||
if (r.materialize_strategy === 'scd2') {
|
||||
scd2CurrentBases[`${base}_current`] = base
|
||||
}
|
||||
}
|
||||
return { ignoredAssets, scd2CurrentBases }
|
||||
}
|
||||
|
||||
// --- Save-time surface ------------------------------------------------------
|
||||
|
||||
// Run the authoritative backend check for just-deployed content and toast the
|
||||
// result. Never throws — a failed check must not taint a successful deploy.
|
||||
export async function notifyContractWarnings(
|
||||
workspace: string,
|
||||
language: ScriptLang,
|
||||
content: string
|
||||
): Promise<void> {
|
||||
// Every checkable ref carries the `ducklake` token (URIs and the bare
|
||||
// default-syntax shorthand alike) — skip the round-trip for the vast
|
||||
// majority of saves that can't produce a warning.
|
||||
if (!content.includes('ducklake')) return
|
||||
try {
|
||||
const { warnings } = await ScriptService.checkSchemaContracts({
|
||||
workspace,
|
||||
requestBody: { language, content }
|
||||
})
|
||||
const real = warnings.filter((w) => w.kind !== 'suppressed')
|
||||
if (real.length === 0) return
|
||||
sendUserToast(
|
||||
`Schema contract: ${real.length} warning${real.length > 1 ? 's' : ''}`,
|
||||
'warning',
|
||||
[],
|
||||
real.map((w) => `• ${w.message}`).join('\n'),
|
||||
10000
|
||||
)
|
||||
} catch (e) {
|
||||
console.error('schema-contract check failed', e)
|
||||
}
|
||||
}
|
||||
|
||||
// End-to-end live-editor check: parse the buffer's annotations, resolve the
|
||||
// captured schemas for everything it references, diff, and anchor the result
|
||||
// to source positions. Cheap per keystroke — the annotation parse is a line
|
||||
// scan and schema fetches hit the short-TTL cache.
|
||||
export async function computeContractMarkers(
|
||||
workspace: string,
|
||||
code: string,
|
||||
assets: AssetWithAltAccessType[],
|
||||
context?: SchemaContractGraphContext
|
||||
): Promise<ContractMarker[]> {
|
||||
const ann = parsePipelineAnnotations(code)
|
||||
const inputs = {
|
||||
assets,
|
||||
// Annotation-declared lineage only — body-inferred lineage is redundant
|
||||
// with the body-read check and its alias attribution can misfire.
|
||||
columnLineage: ann.columnLineage,
|
||||
dataTests: ann.dataTests,
|
||||
materialize: ann.materialize
|
||||
}
|
||||
const refs = referencedDucklakePaths(inputs)
|
||||
if (refs.length === 0) return []
|
||||
const schemas = await fetchSchemasForPaths(
|
||||
workspace,
|
||||
refs,
|
||||
context ? (p) => context.scd2CurrentBases[p] : undefined
|
||||
)
|
||||
if (schemas.size === 0) return []
|
||||
const warnings = diffSchemaContracts({
|
||||
...inputs,
|
||||
schemas,
|
||||
ignored: new Set(context?.ignoredAssets ?? [])
|
||||
})
|
||||
return mapWarningsToMarkers(code, warnings)
|
||||
}
|
||||
|
||||
// --- Editor marker mapping ---------------------------------------------------
|
||||
|
||||
export type ContractMarker = {
|
||||
message: string
|
||||
startLineNumber: number
|
||||
startColumn: number
|
||||
endLineNumber: number
|
||||
endColumn: number
|
||||
}
|
||||
|
||||
// Best-effort source anchoring: annotation-family warnings anchor to their
|
||||
// annotation line; body-read warnings anchor to the first occurrence of the
|
||||
// column identifier; fallback is the `// on`/first line mentioning the asset.
|
||||
export function mapWarningsToMarkers(code: string, warnings: ContractWarning[]): ContractMarker[] {
|
||||
const lines = code.split('\n')
|
||||
|
||||
function lineMatching(pred: (line: string) => boolean): number | undefined {
|
||||
const idx = lines.findIndex(pred)
|
||||
return idx >= 0 ? idx + 1 : undefined
|
||||
}
|
||||
|
||||
function tokenRange(
|
||||
lineNumber: number,
|
||||
token: string
|
||||
): { startColumn: number; endColumn: number } {
|
||||
const line = lines[lineNumber - 1] ?? ''
|
||||
const idx = line.toLowerCase().indexOf(token.toLowerCase())
|
||||
if (idx < 0) return { startColumn: 1, endColumn: line.length + 1 }
|
||||
return { startColumn: idx + 1, endColumn: idx + 1 + token.length }
|
||||
}
|
||||
|
||||
return warnings
|
||||
.filter((w) => w.kind !== 'suppressed')
|
||||
.map((w) => {
|
||||
let lineNumber: number | undefined
|
||||
let token: string | undefined = w.column ?? undefined
|
||||
switch (w.kind) {
|
||||
case 'missing_lineage_source':
|
||||
lineNumber = lineMatching(
|
||||
(l) => /^\s*(\/\/|--|#)\s*column\s/.test(l) && !!w.column && l.includes(w.column)
|
||||
)
|
||||
break
|
||||
case 'missing_relationship_column':
|
||||
case 'relationship_type_mismatch':
|
||||
lineNumber = lineMatching(
|
||||
(l) =>
|
||||
/^\s*(\/\/|--|#)\s*data_test\s+relationships\s/.test(l) && l.includes(w.asset_path)
|
||||
)
|
||||
break
|
||||
case 'missing_column': {
|
||||
// first body occurrence of the column identifier
|
||||
const re = new RegExp(`\\b${w.column?.replace(/[.*+?^${}()|[\]\\]/g, '\\$&')}\\b`, 'i')
|
||||
lineNumber = w.column ? lineMatching((l) => re.test(l)) : undefined
|
||||
break
|
||||
}
|
||||
}
|
||||
if (lineNumber == undefined) {
|
||||
// fallback: the `// on …` (or any) line mentioning the asset path
|
||||
lineNumber = lineMatching((l) => l.includes(w.asset_path)) ?? 1
|
||||
token = w.asset_path
|
||||
}
|
||||
const range = token
|
||||
? tokenRange(lineNumber, token)
|
||||
: { startColumn: 1, endColumn: (lines[lineNumber - 1]?.length ?? 0) + 1 }
|
||||
return {
|
||||
message: w.message,
|
||||
startLineNumber: lineNumber,
|
||||
endLineNumber: lineNumber,
|
||||
...range
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -49,8 +49,15 @@ export interface AssetGraphRunnableNode {
|
||||
// Managed `// materialize` write strategy. Absent for non-materializing or
|
||||
// `manual` scripts. Used (with `partition_kind`) to decide whether a
|
||||
// produced asset's schema can evolve: only whole-table `replace` can, since
|
||||
// `append`/`merge`/partitioned writes INSERT into a fixed-schema table.
|
||||
materialize_strategy?: 'replace' | 'append' | 'merge'
|
||||
// `append`/`merge`/`scd2`/partitioned writes INSERT into a fixed-schema
|
||||
// table. `scd2` also identifies the producer of a `<dim>_current` companion
|
||||
// view for the schema-contract `_current` → base-table fallback.
|
||||
materialize_strategy?: 'replace' | 'append' | 'merge' | 'scd2'
|
||||
// `on_schema_change=ignore` on the managed materialize — the producer's
|
||||
// opt-out from downstream schema-contract warnings. Only present when set
|
||||
// to `ignore` (default `warn` is absent). Threaded into the editor's
|
||||
// contract mirror so it suppresses the same warnings the server check does.
|
||||
materialize_on_schema_change?: string
|
||||
// Macros this script provides to the workspace registry (deployed
|
||||
// `// macros` library). Non-empty marks the node as a macro library;
|
||||
// drives the "defines N macros" badge and the details-pane signature
|
||||
|
||||
@@ -36,6 +36,7 @@
|
||||
type ColumnLineageGraph
|
||||
} from '$lib/components/assets/AssetGraph/columnLineageGraph'
|
||||
import { resolveGraph } from '$lib/components/assets/AssetGraph/resolveGraph'
|
||||
import { buildSchemaContractContext } from '$lib/components/assets/AssetGraph/schemaContracts'
|
||||
import {
|
||||
computeDownstreamClosure,
|
||||
computeInducedSchedule,
|
||||
@@ -1675,6 +1676,12 @@
|
||||
: EMPTY_COLUMN_GRAPH
|
||||
)
|
||||
|
||||
// Producer-side facts for the editor's live schema-contract diagnostics:
|
||||
// which assets are muted (`on_schema_change=ignore`) and which `_current`
|
||||
// views map to an scd2 base table. Derived from the same resolved graph the
|
||||
// canvas renders so the mirror suppresses exactly what the server check does.
|
||||
let schemaContractContext = $derived(buildSchemaContractContext(graphWithDraft.runnables))
|
||||
|
||||
// Whether the selected ducklake asset's captured schema can *evolve* (drives
|
||||
// the asset panel's Schema tab: version history vs. a single fixed schema).
|
||||
// Only a whole-table `replace` producer (CREATE OR REPLACE) can change
|
||||
@@ -2210,6 +2217,7 @@
|
||||
{selectionProducers}
|
||||
selectionColumnGraph={pe.activeDraft ? EMPTY_COLUMN_GRAPH : columnGraph}
|
||||
{schemaCanEvolve}
|
||||
{schemaContractContext}
|
||||
downstreamSubscribers={editedScriptDownstreamCount}
|
||||
onStartBoundedRunForOpen={startBoundedRun}
|
||||
canBoundedRunOpenScript={!!openScriptPath &&
|
||||
|
||||
Reference in New Issue
Block a user