diff --git a/backend/.sqlx/query-5e50ba0ae27b09a3ea1530c223e5039aa631bb5b0993ad09bc9a1381f6715f19.json b/backend/.sqlx/query-5e50ba0ae27b09a3ea1530c223e5039aa631bb5b0993ad09bc9a1381f6715f19.json new file mode 100644 index 0000000000..17d372249b --- /dev/null +++ b/backend/.sqlx/query-5e50ba0ae27b09a3ea1530c223e5039aa631bb5b0993ad09bc9a1381f6715f19.json @@ -0,0 +1,47 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO materialized_partition\n (workspace_id, asset_kind, asset_path, partition, status,\n snapshot_id, row_count, job_id, materialized_at, error)\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, now(), $9)\n ON CONFLICT (workspace_id, asset_kind, asset_path, partition)\n DO UPDATE SET status = EXCLUDED.status,\n snapshot_id = EXCLUDED.snapshot_id,\n row_count = EXCLUDED.row_count,\n job_id = EXCLUDED.job_id,\n materialized_at = now(),\n error = EXCLUDED.error", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + { + "Custom": { + "name": "asset_kind", + "kind": { + "Enum": [ + "s3object", + "resource", + "variable", + "ducklake", + "datatable", + "volume" + ] + } + } + }, + "Varchar", + "Text", + { + "Custom": { + "name": "materialization_status", + "kind": { + "Enum": [ + "running", + "materialized", + "failed" + ] + } + } + }, + "Int8", + "Int8", + "Uuid", + "Text" + ] + }, + "nullable": [] + }, + "hash": "5e50ba0ae27b09a3ea1530c223e5039aa631bb5b0993ad09bc9a1381f6715f19" +} diff --git a/backend/.sqlx/query-c28e066bf24263f9e3b48a2ecabdf037178ba246092dbc1d2b3816e7a9269dc3.json b/backend/.sqlx/query-c28e066bf24263f9e3b48a2ecabdf037178ba246092dbc1d2b3816e7a9269dc3.json new file mode 100644 index 0000000000..7981dbc983 --- /dev/null +++ b/backend/.sqlx/query-c28e066bf24263f9e3b48a2ecabdf037178ba246092dbc1d2b3816e7a9269dc3.json @@ -0,0 +1,111 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT asset_kind AS \"asset_kind: AssetKind\", asset_path, partition,\n status AS \"status: MaterializationStatus\", snapshot_id,\n row_count, job_id, materialized_at, error\n FROM materialized_partition\n WHERE workspace_id = $1 AND asset_kind = $2 AND asset_path = $3\n ORDER BY partition DESC", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "asset_kind: AssetKind", + "type_info": { + "Custom": { + "name": "asset_kind", + "kind": { + "Enum": [ + "s3object", + "resource", + "variable", + "ducklake", + "datatable", + "volume" + ] + } + } + } + }, + { + "ordinal": 1, + "name": "asset_path", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "partition", + "type_info": "Text" + }, + { + "ordinal": 3, + "name": "status: MaterializationStatus", + "type_info": { + "Custom": { + "name": "materialization_status", + "kind": { + "Enum": [ + "running", + "materialized", + "failed" + ] + } + } + } + }, + { + "ordinal": 4, + "name": "snapshot_id", + "type_info": "Int8" + }, + { + "ordinal": 5, + "name": "row_count", + "type_info": "Int8" + }, + { + "ordinal": 6, + "name": "job_id", + "type_info": "Uuid" + }, + { + "ordinal": 7, + "name": "materialized_at", + "type_info": "Timestamptz" + }, + { + "ordinal": 8, + "name": "error", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Text", + { + "Custom": { + "name": "asset_kind", + "kind": { + "Enum": [ + "s3object", + "resource", + "variable", + "ducklake", + "datatable", + "volume" + ] + } + } + }, + "Text" + ] + }, + "nullable": [ + false, + false, + false, + false, + true, + true, + true, + false, + true + ] + }, + "hash": "c28e066bf24263f9e3b48a2ecabdf037178ba246092dbc1d2b3816e7a9269dc3" +} diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 367c17e912..79c61dd687 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -14354,6 +14354,7 @@ dependencies = [ "windmill-parser", "windmill-parser-py", "windmill-parser-py-asset", + "windmill-parser-sql", "windmill-parser-sql-asset", "windmill-parser-ts", "windmill-parser-ts-asset", diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 151e85295b..8bf4c6e5dd 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -ba677ea142011462ad4dfe77e8375a6dd274cdef +23b5f55a943dd4d4f72a5406398b68f22782a8b8 \ No newline at end of file diff --git a/backend/migrations/20260619170118_add_materialized_partition.down.sql b/backend/migrations/20260619170118_add_materialized_partition.down.sql new file mode 100644 index 0000000000..4932338956 --- /dev/null +++ b/backend/migrations/20260619170118_add_materialized_partition.down.sql @@ -0,0 +1,3 @@ +DROP INDEX IF EXISTS idx_materialized_partition_asset_status; +DROP TABLE IF EXISTS materialized_partition; +DROP TYPE IF EXISTS MATERIALIZATION_STATUS; diff --git a/backend/migrations/20260619170118_add_materialized_partition.up.sql b/backend/migrations/20260619170118_add_materialized_partition.up.sql new file mode 100644 index 0000000000..da5f806e4f --- /dev/null +++ b/backend/migrations/20260619170118_add_materialized_partition.up.sql @@ -0,0 +1,29 @@ +-- Per-partition materialization state for managed `// materialize` assets. +-- One row per (asset, partition): the latest materialization of that slice. +-- Drives: the partition-status grid (CE observability), run-stale/gap +-- detection, and the EE backfill worklist (missing/failed partitions). The +-- `partition` column uses '' as the sentinel for an unpartitioned (whole-table) +-- materialization, since partition is part of the primary key and cannot be +-- NULL. +CREATE TYPE MATERIALIZATION_STATUS AS ENUM ('running', 'materialized', 'failed'); + +CREATE TABLE IF NOT EXISTS materialized_partition ( + workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id) ON DELETE CASCADE ON UPDATE CASCADE, + asset_kind ASSET_KIND NOT NULL, + asset_path VARCHAR(255) NOT NULL, + partition TEXT NOT NULL DEFAULT '', + status MATERIALIZATION_STATUS NOT NULL, + -- DuckLake snapshot id produced by the write; NULL while running / on + -- failure. The pin that makes downstream reads reproducible. + snapshot_id BIGINT, + row_count BIGINT, + job_id UUID, + materialized_at TIMESTAMPTZ NOT NULL DEFAULT now(), + error TEXT, + PRIMARY KEY (workspace_id, asset_kind, asset_path, partition) +); + +-- Backfill enumeration / grid "show only gaps": filter an asset's partitions +-- by status without scanning the whole table. +CREATE INDEX IF NOT EXISTS idx_materialized_partition_asset_status + ON materialized_partition (workspace_id, asset_kind, asset_path, status); diff --git a/backend/parsers/windmill-parser/src/asset_parser.rs b/backend/parsers/windmill-parser/src/asset_parser.rs index ce0be81f00..61e30a343f 100644 --- a/backend/parsers/windmill-parser/src/asset_parser.rs +++ b/backend/parsers/windmill-parser/src/asset_parser.rs @@ -107,6 +107,11 @@ pub struct ParseAssetsOutput { // The delay is a raw duration string parsed at deploy (parser-light). #[serde(skip_serializing_if = "Option::is_none", default)] pub retry: Option, + // `// materialize [manual] [append] [key=]` — + // managed-materialization target + its strategy. At most one per script. + // Drives the worker's write-strategy + snapshot capture. + #[serde(skip_serializing_if = "Option::is_none", default)] + pub materialize: Option, } #[derive(Serialize, Debug, PartialEq, Clone)] @@ -209,6 +214,27 @@ pub struct RetrySpec { pub delay: Option, } +// `// materialize [manual] [append] [key=]` — declares that this +// script produces a *managed* materialization of `` (a `ducklake://` +// table). By default the runtime generates the write DDL around the script's +// single trailing `SELECT` and owns idempotency, partition-state and snapshot +// capture. `manual` is the escape hatch: the script writes its own DDL and the +// runtime only records state (track-only). The reconciliation strategy options +// (`append`, `key=`) apply to managed mode: none → DELETE-by-partition + +// INSERT (replace); `key=` → MERGE (dedup within slice); `append` → +// INSERT-only. `append` wins if both are given (deploy-time warning). +#[derive(Serialize, Debug, PartialEq, Clone)] +pub struct MaterializeSpec { + pub target_kind: AssetKind, + pub target_path: String, + #[serde(skip_serializing_if = "std::ops::Not::not", default)] + pub manual: bool, + #[serde(skip_serializing_if = "std::ops::Not::not", default)] + pub append: bool, + #[serde(skip_serializing_if = "Option::is_none", default)] + pub unique_key: Option, +} + // `// trigger any` (default) vs `// trigger all`. `Any` = OR: any trigger // firing runs the script (current behaviour). `All` = AND: the script // runs only once every partition-bearing input has materialized at the @@ -239,6 +265,7 @@ pub struct PipelineAnnotations { pub debounce_default: Option, pub tag: Option, pub retry: Option, + pub materialize: Option, } impl ParseAssetsOutput { @@ -262,6 +289,7 @@ impl ParseAssetsOutput { debounce_default: pipeline.debounce_default, tag: pipeline.tag, retry: pipeline.retry, + materialize: pipeline.materialize, } } } @@ -571,6 +599,15 @@ pub fn parse_pipeline_annotations(code: &str) -> PipelineAnnotations { continue; } + if let Some(after_kw) = consume_keyword(rest, "materialize") { + if out.materialize.is_none() { + if let Some(spec) = parse_materialize_spec(after_kw.trim()) { + out.materialize = Some(spec); + } + } + continue; + } + if let Some(after_kw) = consume_keyword(rest, "on") { let spec_text = after_kw.trim(); if spec_text.is_empty() { @@ -618,6 +655,35 @@ fn parse_retry_spec(s: &str) -> Option { Some(RetrySpec { count, delay }) } +// Parse a `// materialize [manual] [append] [key=]` right-hand +// side. An optional leading `manual` token (whitespace-delimited) opts out of +// managed mode (track-only). The next whitespace token is the target asset URI +// (default-syntax shorthands enabled, so `ducklake` → `ducklake://main`); the +// remainder are strategy options — bare `append` and `key=` (merge key), +// which apply to managed mode only. A missing/empty target yields `None` (the +// annotation is dropped, fail-safe). +fn parse_materialize_spec(s: &str) -> Option { + let (manual, rest) = match s.strip_prefix("manual") { + Some(after) if after.is_empty() || after.starts_with(char::is_whitespace) => { + (true, after.trim_start()) + } + _ => (false, s), + }; + let mut it = rest.trim().splitn(2, char::is_whitespace); + let asset_tok = it.next()?; + let opts_str = it.next().unwrap_or(""); + let (target_kind, path) = parse_asset_syntax(asset_tok.trim(), true)?; + if path.is_empty() { + return None; + } + let append = opts_str.split_whitespace().any(|t| t == "append"); + let unique_key = parse_kv_opts(opts_str) + .get("key") + .filter(|k| !k.is_empty()) + .cloned(); + Some(MaterializeSpec { target_kind, target_path: path.to_string(), manual, append, unique_key }) +} + // Parse a `// partitioned [opts]` right-hand side. Recognized kinds: // `daily`, `hourly`, `weekly`, `monthly` (with optional tz/format/start), // and `dynamic key=""` (plus optional format). @@ -1044,6 +1110,67 @@ mod pipeline_annotation_tests { assert!(out.retry.is_none()); } + #[test] + fn materialize_managed_default() { + let out = parse_pipeline_annotations("// materialize ducklake://analytics/orders_daily"); + let m = out.materialize.expect("materialize"); + assert_eq!(m.target_kind, AssetKind::Ducklake); + assert_eq!(m.target_path, "analytics/orders_daily"); + // managed by default; replace strategy (no append / key) + assert!(!m.manual); + assert!(!m.append); + assert_eq!(m.unique_key, None); + } + + #[test] + fn materialize_manual_escape_hatch() { + let out = + parse_pipeline_annotations("// materialize manual ducklake://analytics/orders_daily"); + let m = out.materialize.expect("materialize"); + assert!(m.manual); + assert_eq!(m.target_path, "analytics/orders_daily"); + } + + #[test] + fn materialize_merge_and_append_options() { + let out = + parse_pipeline_annotations("// materialize ducklake://a/orders_daily key=order_id"); + let m = out.materialize.expect("materialize"); + assert_eq!(m.unique_key.as_deref(), Some("order_id")); + assert!(!m.append); + + let out = parse_pipeline_annotations("// materialize ducklake://a/events append"); + let m = out.materialize.expect("materialize"); + assert!(m.append); + assert_eq!(m.unique_key, None); + } + + #[test] + fn materialize_default_syntax_shorthand() { + let out = parse_pipeline_annotations("// materialize ducklake"); + let m = out.materialize.expect("materialize"); + assert_eq!(m.target_kind, AssetKind::Ducklake); + assert_eq!(m.target_path, "main"); + assert!(!m.manual); + } + + #[test] + fn materialize_manual_only_is_dropped() { + // `manual` with no target is not a valid materialization. + let out = parse_pipeline_annotations("// materialize manual"); + assert!(out.materialize.is_none()); + } + + #[test] + fn materialize_first_wins() { + let out = parse_pipeline_annotations( + "// materialize ducklake://a/x\n# materialize manual ducklake://b/y", + ); + let m = out.materialize.expect("materialize"); + assert_eq!(m.target_path, "a/x"); + assert!(!m.manual); + } + #[test] fn combined() { let code = concat!( @@ -1053,7 +1180,8 @@ mod pipeline_annotation_tests { "// partitioned daily tz=\"UTC\"\n", "// freshness 2h\n", "// tag heavy\n", - "// retry 3 5s\n" + "// retry 3 5s\n", + "// materialize ducklake://analytics/orders_daily key=order_id\n" ); let out = parse_pipeline_annotations(code); assert!(out.in_pipeline); @@ -1064,6 +1192,10 @@ mod pipeline_annotation_tests { let r = out.retry.expect("retry"); assert_eq!(r.count, 3); assert_eq!(r.delay.as_deref(), Some("5s")); + let m = out.materialize.expect("materialize"); + assert!(!m.manual); + assert_eq!(m.target_path, "analytics/orders_daily"); + assert_eq!(m.unique_key.as_deref(), Some("order_id")); } #[test] diff --git a/backend/parsers/windmill-parser/src/lib.rs b/backend/parsers/windmill-parser/src/lib.rs index 5a7e90bf7e..19bc5602cd 100644 --- a/backend/parsers/windmill-parser/src/lib.rs +++ b/backend/parsers/windmill-parser/src/lib.rs @@ -13,6 +13,7 @@ use serde::Serialize; use serde_json::Value; pub mod asset_parser; +pub mod sql_materialize; /// S3 output format for SQL queries (moved here to avoid pulling sqlx into WASM via windmill-types) #[derive(Clone, Copy, Debug)] diff --git a/backend/parsers/windmill-parser/src/sql_materialize.rs b/backend/parsers/windmill-parser/src/sql_materialize.rs new file mode 100644 index 0000000000..c1b0cc97aa --- /dev/null +++ b/backend/parsers/windmill-parser/src/sql_materialize.rs @@ -0,0 +1,817 @@ +//! Eligibility classifier + materialization SQL codegen for managed `// materialize`. +//! +//! Managed `// materialize` (the default) promises the script is "setup +//! statements, then one trailing SELECT" — Windmill generates the write DDL +//! around that SELECT (the `// materialize manual` escape hatch opts out and +//! writes its own DDL). This module is the single source of truth for *which +//! block is that SELECT* and *what DDL gets generated*, so save-time validation +//! (deploy path) and run-time codegen (DuckDB executor) can never disagree. +//! +//! Everything here is pure and string-level: no SQL is executed, no type +//! inference is done. The classifier is leading-keyword based and deliberately +//! conservative — anything it can't positively recognize as a read-only output +//! or a known-safe setup statement is rejected, so a script is only accepted +//! for managed mode when its shape is unambiguous. + +/// One top-level statement's role in a wrap-mode script. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum BlockClass { + /// Read-only relation the wrap writes from: `SELECT` / `WITH …SELECT` / + /// `FROM` (DuckDB from-first) / `VALUES` / `TABLE x` / `(UN)PIVOT`. + Output, + /// Known-safe preamble: `ATTACH` / `INSTALL` / `LOAD` / `SET` / `PRAGMA` / + /// `USE` / `CREATE TEMP …`. Runs verbatim before the generated write. + Setup, + /// Anything that writes or whose effect we can't vouch for: non-temp + /// `CREATE` / `INSERT` / `UPDATE` / `DELETE` / `MERGE` / `DROP` / `COPY` / + /// `ALTER` / `TRUNCATE`, or an unrecognized leading keyword. Disqualifies + /// managed mode (the user should use `// materialize manual`). + Disallowed, +} + +/// A script accepted for wrapping: zero+ setup blocks then one terminal SELECT. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct WrapPlan { + /// Setup statements in source order, verbatim, **without** trailing `;`. + pub setup: Vec, + /// The single terminal output statement, verbatim, **without** trailing `;`. + pub output: String, +} + +/// Why a script is not eligible for managed `// materialize`. Carries enough to +/// render the targeted save-time messages. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum WrapError { + /// No statements at all (empty / comments only). + Empty, + /// No terminal SELECT — nothing to wrap. + NoOutput, + /// More than one top-level SELECT. `count` is how many were found. + MultipleOutputs { count: usize }, + /// A SELECT exists but isn't the last statement (something runs after it). + OutputNotLast, + /// A write/unknown statement appears among the setup blocks. `snippet` is a + /// short prefix of the offending statement for the error message. + DisallowedBlock { snippet: String }, +} + +impl WrapError { + /// Human-facing, actionable message (matches the spec's rejection text). + pub fn message(&self) -> String { + let base = + "managed `// materialize` requires the script to be setup statements then a single trailing SELECT"; + let manual = "use `// materialize manual` to write the DDL yourself"; + match self { + WrapError::Empty => format!("{base}: the script is empty."), + WrapError::NoOutput => format!("{base}: found no SELECT — {manual}."), + WrapError::MultipleOutputs { count } => format!( + "{base}: found {count} SELECT statements; combine them with a CTE, or {manual}." + ), + WrapError::OutputNotLast => format!( + "{base}: found statements after the SELECT — move them above it, or {manual}." + ), + WrapError::DisallowedBlock { snippet } => { + format!("{base}: `{snippet}` writes or is unrecognized — {manual}.") + } + } + } +} + +/// Split SQL into top-level, `;`-separated statements, skipping line comments +/// (`-- …`), block comments (`/* … */`), single-quoted strings (`'…'` with +/// `''` escape) and double-quoted identifiers (`"…"`). Semicolons inside any of +/// those are not separators. Returns each statement trimmed, comments stripped, +/// empties dropped. Self-contained so the parser crate stays dependency-free; +/// it must stay behaviourally aligned with the executor's block splitter (both +/// route wrap through `classify_wrap`, so the split they see is this one). +pub fn split_statements(sql: &str) -> Vec { + let mut out = Vec::new(); + let mut cur = String::new(); + let bytes = sql.as_bytes(); + let mut i = 0; + let n = bytes.len(); + while i < n { + let c = bytes[i] as char; + // line comment — `--` (SQL) or `//`. The `//` form is not SQL, but it + // is how Windmill pipeline annotations (`// materialize`, `// pipeline`, + // …) are written, and they sit above the SQL in the same script; strip + // them so they don't pollute the first statement block's classification + // or the generated setup SQL. + if (c == '-' && i + 1 < n && bytes[i + 1] == b'-') + || (c == '/' && i + 1 < n && bytes[i + 1] == b'/') + { + while i < n && bytes[i] != b'\n' { + i += 1; + } + continue; + } + // block comment + if c == '/' && i + 1 < n && bytes[i + 1] == b'*' { + i += 2; + while i + 1 < n && !(bytes[i] == b'*' && bytes[i + 1] == b'/') { + i += 1; + } + i += 2; + continue; + } + // single-quoted string + if c == '\'' { + cur.push(c); + i += 1; + while i < n { + cur.push(bytes[i] as char); + if bytes[i] == b'\'' { + // doubled '' is an escaped quote, stay in string + if i + 1 < n && bytes[i + 1] == b'\'' { + cur.push('\''); + i += 2; + continue; + } + i += 1; + break; + } + i += 1; + } + continue; + } + // double-quoted identifier + if c == '"' { + cur.push(c); + i += 1; + while i < n { + cur.push(bytes[i] as char); + if bytes[i] == b'"' { + i += 1; + break; + } + i += 1; + } + continue; + } + if c == ';' { + let t = cur.trim(); + if !t.is_empty() { + out.push(t.to_string()); + } + cur.clear(); + i += 1; + continue; + } + cur.push(c); + i += 1; + } + let t = cur.trim(); + if !t.is_empty() { + out.push(t.to_string()); + } + out +} + +/// Lowercased top-level keyword tokens of a single statement (parens collapsed +/// away: tokens *inside* balanced `(...)` are skipped, so a CTE body's verbs +/// don't leak up). Strings/identifiers are already gone from the split, but we +/// re-guard quotes defensively. Used to disambiguate `WITH …` and `CREATE …`. +fn top_level_keywords(stmt: &str) -> Vec { + let mut toks = Vec::new(); + let mut cur = String::new(); + let mut depth: i32 = 0; + let bytes = stmt.as_bytes(); + let mut i = 0; + let n = bytes.len(); + let flush = |cur: &mut String, toks: &mut Vec| { + if !cur.is_empty() { + toks.push(cur.to_lowercase()); + cur.clear(); + } + }; + while i < n { + let c = bytes[i] as char; + if c == '\'' || c == '"' { + let q = bytes[i]; + i += 1; + while i < n && bytes[i] != q { + i += 1; + } + i += 1; + continue; + } + if c == '(' { + flush(&mut cur, &mut toks); + depth += 1; + i += 1; + continue; + } + if c == ')' { + if depth > 0 { + depth -= 1; + } + i += 1; + continue; + } + if depth > 0 { + i += 1; + continue; + } + if c.is_alphanumeric() || c == '_' { + cur.push(c); + } else { + flush(&mut cur, &mut toks); + } + i += 1; + } + flush(&mut cur, &mut toks); + toks +} + +const OUTPUT_KW: &[&str] = &["select", "from", "values", "table", "pivot", "unpivot"]; +const SETUP_KW: &[&str] = &["attach", "install", "load", "set", "pragma", "use"]; +const WRITE_VERBS: &[&str] = &["insert", "update", "delete", "merge"]; + +/// Classify a single statement by its leading keyword (with `WITH`/`CREATE` +/// disambiguation). See [`BlockClass`]. +pub fn classify_block(stmt: &str) -> BlockClass { + let kws = top_level_keywords(stmt); + let Some(first) = kws.first().map(String::as_str) else { + return BlockClass::Disallowed; + }; + + // CREATE TEMP … is setup (staging); any other CREATE is a write. + if first == "create" { + let temp = kws + .iter() + .skip(1) + .take(3) + .any(|k| k == "temp" || k == "temporary"); + return if temp { + BlockClass::Setup + } else { + BlockClass::Disallowed + }; + } + + // WITH … : the main statement's verb decides. CTE bodies are parenthesized, + // so their verbs are not in `kws`; the first top-level write verb or SELECT + // after the CTE list is the real one. + if first == "with" { + for k in kws.iter().skip(1) { + if k == "select" { + return BlockClass::Output; + } + if WRITE_VERBS.contains(&k.as_str()) { + return BlockClass::Disallowed; + } + } + // `WITH x AS (...) SELECT` where SELECT got collapsed is impossible + // (SELECT here is top-level), so a WITH with no top-level verb is a + // malformed/unknown statement — reject conservatively. + return BlockClass::Disallowed; + } + + if OUTPUT_KW.contains(&first) { + return BlockClass::Output; + } + if SETUP_KW.contains(&first) { + return BlockClass::Setup; + } + BlockClass::Disallowed +} + +/// Validate a script for managed `// materialize` and, on success, return the +/// setup/output split. Enforces the four conditions from the spec: +/// 1. exactly one Output block, 2. it is last, 3. all preceding blocks are +/// Setup, 4. nothing after it. +pub fn classify_wrap(sql: &str) -> Result { + let stmts = split_statements(sql); + if stmts.is_empty() { + return Err(WrapError::Empty); + } + let classes: Vec = stmts.iter().map(|s| classify_block(s)).collect(); + + let output_idxs: Vec = classes + .iter() + .enumerate() + .filter(|(_, c)| **c == BlockClass::Output) + .map(|(i, _)| i) + .collect(); + + match output_idxs.len() { + 0 => return Err(WrapError::NoOutput), + 1 => {} + count => return Err(WrapError::MultipleOutputs { count }), + } + let out_idx = output_idxs[0]; + if out_idx != stmts.len() - 1 { + return Err(WrapError::OutputNotLast); + } + // Everything before the output must be Setup (no Disallowed preamble). + for (i, c) in classes.iter().enumerate().take(out_idx) { + if *c != BlockClass::Setup { + return Err(WrapError::DisallowedBlock { snippet: snippet(&stmts[i]) }); + } + } + Ok(WrapPlan { setup: stmts[..out_idx].to_vec(), output: stmts[out_idx].clone() }) +} + +fn snippet(stmt: &str) -> String { + let one_line: String = stmt.split_whitespace().collect::>().join(" "); + if one_line.chars().count() > 40 { + let truncated: String = one_line.chars().take(40).collect(); + format!("{truncated}…") + } else { + one_line + } +} + +// --------------------------------------------------------------------------- +// Codegen +// --------------------------------------------------------------------------- + +/// How a (partition of a) materialized table is reconciled on each run. +/// Derived at deploy from `unique_key`/`append`: `append` → `Append`, else +/// `unique_key` → `Merge`, else `Replace`. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum MaterializeStrategy { + /// DELETE the current partition, then INSERT — partition becomes exactly + /// what the SELECT returned. Full-refresh of the slice. + Replace, + /// Upsert within the slice on `unique_key` (delete-by-key + insert); rows + /// absent from the SELECT are left in place. + Merge { unique_key: String }, + /// INSERT only — immutable event-log semantics. + Append, +} + +/// Inputs to materialization codegen, all resolved at run time by the worker. +/// Pure: produces SQL text; executes nothing. +#[derive(Debug, Clone)] +pub struct MaterializeCodegen<'a> { + /// Fully-qualified target, e.g. `_wm_target.orders_daily`. Always qualified + /// so a user `USE …;` in setup can't redirect the write. + pub target_qualified: &'a str, + /// The user's output SELECT (verbatim, no trailing `;`) — embedded as a + /// subquery so its own shape is irrelevant to the generated wrapper. + pub select_sql: &'a str, + /// Physical partition column added to the managed table. + pub partition_col: &'a str, + /// SQL expression for the current partition value — a literal like + /// `'2026-06-19'` or a bind placeholder. The caller is responsible for + /// safe quoting/binding. + pub partition_value_sql: &'a str, + /// Whether `// partitioned` applies. When false the table is unpartitioned + /// and the partition column / `SET PARTITIONED BY` are omitted. + pub partitioned: bool, + pub strategy: MaterializeStrategy, +} + +impl<'a> MaterializeCodegen<'a> { + /// The ordered statements that perform the materialization, to be run after + /// the setup blocks and inside the caller's execution. The first-run + /// bootstrap is idempotent (`IF NOT EXISTS`), so this is safe to run every + /// time. The DELETE/INSERT body is wrapped in one transaction so a partial + /// failure leaves the prior snapshot intact. Every strategy reduces to + /// DELETE+INSERT (no `MERGE INTO`) — see the `Merge` arm for why. + pub fn statements(&self) -> Vec { + let t = self.target_qualified; + let sel = self.select_sql; + let pcol = self.partition_col; + let pval = self.partition_value_sql; + let mut out = Vec::new(); + + // Whole-table replace: rebuild the table to match the SELECT's *current* + // schema each run with one atomic `CREATE OR REPLACE` (which DuckLake + // still snapshots). This is the only path that survives a changed SELECT + // or a pre-existing table with a different schema — the persist-and- + // mutate paths below fix the schema at first create. + if !self.partitioned && matches!(self.strategy, MaterializeStrategy::Replace) { + out.push(format!( + "CREATE OR REPLACE TABLE {t} AS SELECT * FROM ({sel});" + )); + return out; + } + + // Persist-and-mutate (partitioned, or merge/append): bootstrap the table + // if absent, then write into it. The schema is fixed at first create — + // a later SELECT-schema change needs a manual rebuild (schema evolution + // is a follow-up). + if self.partitioned { + out.push(format!( + "CREATE TABLE IF NOT EXISTS {t} AS \ + SELECT *, CAST(NULL AS VARCHAR) AS {pcol} FROM ({sel}) WHERE false;" + )); + out.push(format!("ALTER TABLE {t} SET PARTITIONED BY ({pcol});")); + } else { + out.push(format!( + "CREATE TABLE IF NOT EXISTS {t} AS SELECT * FROM ({sel}) WHERE false;" + )); + } + + out.push("BEGIN TRANSACTION;".to_string()); + // The rows to write, with the partition column appended when partitioned. + let source = if self.partitioned { + format!("SELECT *, {pval} AS {pcol} FROM ({sel})") + } else { + format!("SELECT * FROM ({sel})") + }; + match &self.strategy { + MaterializeStrategy::Replace => { + // Only reached when partitioned (whole-table replace returned above). + out.push(format!("DELETE FROM {t} WHERE {pcol} = {pval};")); + out.push(format!("INSERT INTO {t} {source};")); + } + MaterializeStrategy::Append => { + out.push(format!("INSERT INTO {t} {source};")); + } + MaterializeStrategy::Merge { unique_key } => { + // Upsert within the slice via delete-by-key + insert (dbt's + // `delete+insert`): rows whose key is in the incoming SELECT are + // replaced, others are left in place. This deliberately avoids + // `MERGE INTO` — DuckLake's MERGE fails writing the first rows of + // a fresh partition (HTTP 404 on the new parquet), and a failed + // write leaves the table needing a DROP. DELETE+INSERT is the + // same write shape as `replace`, which is reliable. The DELETE is + // scoped to the current partition when partitioned so it stays + // slice-local (a key present in another partition is untouched). + let scope = if self.partitioned { + format!("{pcol} = {pval} AND ") + } else { + String::new() + }; + out.push(format!( + "DELETE FROM {t} WHERE {scope}{unique_key} IN (SELECT {unique_key} FROM ({sel}));" + )); + out.push(format!("INSERT INTO {t} {source};")); + } + } + out.push("COMMIT;".to_string()); + out + } +} + +/// The read that captures the DuckLake snapshot id produced by the write, for +/// the given attach alias (e.g. `_wm_target`). The worker runs this last and +/// records the result into `materialized_partition`. +pub fn snapshot_capture_sql(alias: &str) -> String { + format!("SELECT max(snapshot_id) AS snapshot_id FROM ducklake_snapshots('{alias}');") +} + +/// Reserved attach alias for the materialization target, fully-qualified in all +/// generated SQL so a user `USE …;` in the setup blocks can't redirect the +/// write. The worker resolves the real `ATTACH 'ducklake:…' AS _wm_target (…)` +/// from the target ducklake's config and passes it in as `target_attach`. +pub const TARGET_ALIAS: &str = "_wm_target"; + +/// Assemble the full ordered statement list the DuckDB executor runs for a +/// managed `// materialize` script. This is the single entry point the worker +/// calls; it composes the already-tested pieces (classifier split → target +/// ATTACH → strategy codegen → snapshot capture) so their ordering lives in one +/// tested place rather than inline in the executor. +/// +/// `target_attach` is the real `ATTACH 'ducklake:…' AS _wm_target (…);` string +/// the worker built from config (it depends on resolved credentials, so it +/// can't be generated here). `target_table` is the table within that catalog +/// (e.g. `orders_daily`), referenced as `_wm_target.`. `asset_path` is +/// the full `/
` for the result summary. The trailing statement is +/// a one-row summary read (asset / rows / snapshot_id) that is both the job's +/// result (a useful preview) and what the worker records. +pub fn build_wrap_blocks( + plan: &WrapPlan, + target_attach: &str, + target_table: &str, + asset_path: &str, + partition_col: &str, + partition_value_sql: &str, + partitioned: bool, + strategy: MaterializeStrategy, +) -> Vec { + let target_qualified = format!("{TARGET_ALIAS}.{target_table}"); + let cg = MaterializeCodegen { + target_qualified: &target_qualified, + select_sql: &plan.output, + partition_col, + partition_value_sql, + partitioned, + strategy, + }; + let mut blocks: Vec = Vec::new(); + // Setup blocks come from the splitter with their `;` stripped — re-terminate + // each so that when the executor re-joins and re-splits the assembled query, + // adjacent statements (e.g. the user ATTACH and the synthetic target ATTACH) + // don't merge into one malformed statement. + blocks.extend(plan.setup.iter().map(|s| terminate(s))); + blocks.push(target_attach.to_string()); + blocks.extend(cg.statements()); + blocks.push(materialize_result_sql( + &target_qualified, + asset_path, + partition_col, + partition_value_sql, + partitioned, + )); + blocks +} + +/// The trailing one-row summary the materialize run returns: the asset it +/// produced, the row count of the materialized slice (the partition when +/// partitioned, else the whole table), and the DuckLake snapshot it created. +/// This is both a useful preview result and the row the worker records. +pub fn materialize_result_sql( + target_qualified: &str, + asset_path: &str, + partition_col: &str, + partition_value_sql: &str, + partitioned: bool, +) -> String { + let (count_expr, partition_sel) = if partitioned { + // Row count is the slice this run wrote (the partition); `partition` + // lets the UI label the count and scope the preview to it. + ( + format!( + "(SELECT count(*) FROM {target_qualified} WHERE {partition_col} = {partition_value_sql})" + ), + format!("{partition_value_sql} AS partition, "), + ) + } else { + ( + format!("(SELECT count(*) FROM {target_qualified})"), + String::new(), + ) + }; + format!( + "SELECT 'ducklake://{asset_path}' AS materialized, \ + {partition_sel}{count_expr} AS rows, \ + (SELECT max(snapshot_id) FROM ducklake_snapshots('{TARGET_ALIAS}')) AS snapshot_id;" + ) +} + +// Ensure a statement ends with a single `;`. +fn terminate(stmt: &str) -> String { + let t = stmt.trim_end(); + if t.ends_with(';') { + t.to_string() + } else { + format!("{t};") + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn ok(sql: &str) -> WrapPlan { + classify_wrap(sql).expect("expected wrap-eligible") + } + fn err(sql: &str) -> WrapError { + classify_wrap(sql).expect_err("expected wrap-ineligible") + } + + #[test] + fn split_respects_strings_comments_idents() { + let sql = "SET x=1; -- a; comment\nSELECT ';' AS a, \"weird;col\" /* ; */ FROM t;"; + let s = split_statements(sql); + assert_eq!(s.len(), 2); + assert_eq!(s[0], "SET x=1"); + assert!(s[1].starts_with("SELECT")); + assert!(s[1].contains("\"weird;col\"")); + } + + #[test] + fn split_handles_escaped_quote() { + let s = split_statements("SELECT 'it''s; fine' AS a;"); + assert_eq!(s.len(), 1); + assert!(s[0].contains("it''s; fine")); + } + + #[test] + fn pipeline_annotations_are_stripped() { + // The real shape: `//` annotation lines above the SQL must not pollute + // the first block's classification (regression — they were being read + // as a leading `pipeline` keyword and rejected). + let p = ok("// pipeline\n// materialize ducklake://main/t\n// partitioned daily\nATTACH 'ducklake://main' AS dl;\nSELECT 1 AS id"); + assert_eq!(p.setup.len(), 1); + // The annotation lines are gone — the setup block starts at the real + // SQL (the `//` inside `ducklake://main` is legitimately retained). + assert!(p.setup[0].starts_with("ATTACH")); + assert!(p.output.starts_with("SELECT")); + } + + #[test] + fn bare_select_is_eligible() { + let p = ok("SELECT a, b FROM t WHERE c = '{partition}'"); + assert!(p.setup.is_empty()); + assert!(p.output.starts_with("SELECT")); + } + + #[test] + fn setup_then_select_is_eligible() { + let p = ok( + "ATTACH 'ducklake://main' AS dl;\n SET memory_limit='4GB';\n SELECT * FROM dl.orders", + ); + assert_eq!(p.setup.len(), 2); + assert!(p.output.starts_with("SELECT")); + } + + #[test] + fn create_temp_staging_is_setup() { + let p = ok("CREATE TEMP TABLE s AS SELECT 1; SELECT * FROM s"); + assert_eq!(p.setup.len(), 1); + assert_eq!( + classify_block("CREATE TEMP TABLE s AS SELECT 1"), + BlockClass::Setup + ); + assert_eq!( + classify_block("CREATE OR REPLACE TEMPORARY VIEW v AS SELECT 1"), + BlockClass::Setup + ); + } + + #[test] + fn with_cte_select_is_output_write_is_disallowed() { + assert_eq!( + classify_block("WITH x AS (SELECT 1) SELECT * FROM x"), + BlockClass::Output + ); + // CTE whose main statement inserts is a write, even though it starts WITH. + assert_eq!( + classify_block("WITH x AS (SELECT 1) INSERT INTO t SELECT * FROM x"), + BlockClass::Disallowed + ); + } + + #[test] + fn from_first_and_values_are_output() { + assert_eq!(classify_block("FROM t SELECT a"), BlockClass::Output); + assert_eq!(classify_block("VALUES (1),(2)"), BlockClass::Output); + assert_eq!(classify_block("TABLE t"), BlockClass::Output); + } + + #[test] + fn trailing_write_rejected() { + assert_eq!( + err("SELECT * FROM t; INSERT INTO u VALUES (1)"), + WrapError::OutputNotLast + ); + } + + #[test] + fn write_in_preamble_rejected() { + match err("INSERT INTO t VALUES (1); SELECT * FROM t") { + WrapError::DisallowedBlock { snippet } => assert!(snippet.starts_with("INSERT")), + e => panic!("wrong error: {e:?}"), + } + } + + #[test] + fn multiple_selects_rejected() { + assert_eq!( + err("SELECT 1; SELECT 2"), + WrapError::MultipleOutputs { count: 2 } + ); + } + + #[test] + fn no_select_and_empty_rejected() { + assert_eq!(err("CREATE TABLE t (a INT)"), WrapError::NoOutput); + assert_eq!(err(" -- just a comment\n"), WrapError::Empty); + } + + #[test] + fn use_cannot_redirect_is_classified_setup() { + // `USE` is allowed setup; generated SQL is fully qualified regardless. + assert_eq!(classify_block("USE dl"), BlockClass::Setup); + } + + #[test] + fn codegen_replace_partitioned() { + let cg = MaterializeCodegen { + target_qualified: "_wm_target.orders_daily", + select_sql: "SELECT a FROM dl.orders", + partition_col: "_wm_partition", + partition_value_sql: "'2026-06-19'", + partitioned: true, + strategy: MaterializeStrategy::Replace, + }; + let st = cg.statements(); + assert!(st[0].contains("CREATE TABLE IF NOT EXISTS _wm_target.orders_daily")); + assert!(st[0].contains("CAST(NULL AS VARCHAR) AS _wm_partition")); + assert!(st.iter().any( + |s| s == "ALTER TABLE _wm_target.orders_daily SET PARTITIONED BY (_wm_partition);" + )); + assert!(st.iter().any(|s| s.starts_with( + "DELETE FROM _wm_target.orders_daily WHERE _wm_partition = '2026-06-19'" + ))); + assert!(st.iter().any(|s| s.contains( + "INSERT INTO _wm_target.orders_daily SELECT *, '2026-06-19' AS _wm_partition" + ))); + assert_eq!(st.first().map(|_| &st[st.len() - 1]).unwrap(), "COMMIT;"); + } + + #[test] + fn codegen_merge_is_delete_by_key_plus_insert() { + let cg = MaterializeCodegen { + target_qualified: "_wm_target.orders_daily", + select_sql: "SELECT order_id, amount FROM dl.orders", + partition_col: "_wm_partition", + partition_value_sql: "'2026-06-19'", + partitioned: true, + strategy: MaterializeStrategy::Merge { unique_key: "order_id".to_string() }, + }; + let st = cg.statements(); + // upsert = delete-by-key (partition-scoped) + insert — NO `MERGE INTO` + // (DuckLake's MERGE fails on fresh partitions). + assert!(!st.iter().any(|s| s.contains("MERGE INTO"))); + let del = st + .iter() + .find(|s| s.starts_with("DELETE FROM")) + .expect("delete stmt"); + assert!(del.contains( + "WHERE _wm_partition = '2026-06-19' AND order_id IN (SELECT order_id FROM (SELECT order_id, amount FROM dl.orders))" + )); + assert!(st + .iter() + .any(|s| s.starts_with("INSERT INTO _wm_target.orders_daily SELECT *, '2026-06-19'"))); + } + + #[test] + fn codegen_append_inserts_only() { + let cg = MaterializeCodegen { + target_qualified: "_wm_target.events", + select_sql: "SELECT * FROM dl.raw", + partition_col: "_wm_partition", + partition_value_sql: "'2026-06-19'", + partitioned: true, + strategy: MaterializeStrategy::Append, + }; + let st = cg.statements(); + assert!(st + .iter() + .any(|s| s.starts_with("INSERT INTO _wm_target.events"))); + assert!(!st.iter().any(|s| s.starts_with("DELETE"))); + assert!(!st.iter().any(|s| s.starts_with("MERGE"))); + } + + #[test] + fn codegen_whole_table_replace_is_create_or_replace() { + // Unpartitioned replace must use CREATE OR REPLACE so a changed SELECT + // schema (or a pre-existing table with a different schema) doesn't break + // — and nothing else (no bootstrap / DELETE / INSERT / txn). + let cg = MaterializeCodegen { + target_qualified: "_wm_target.customer_dim", + select_sql: "SELECT a, b, c FROM dl.src", + partition_col: "_wm_partition", + partition_value_sql: "''", + partitioned: false, + strategy: MaterializeStrategy::Replace, + }; + let st = cg.statements(); + assert_eq!( + st, + vec![ + "CREATE OR REPLACE TABLE _wm_target.customer_dim AS SELECT * FROM (SELECT a, b, c FROM dl.src);" + .to_string() + ] + ); + } + + #[test] + fn snapshot_capture_targets_alias() { + assert_eq!( + snapshot_capture_sql("_wm_target"), + "SELECT max(snapshot_id) AS snapshot_id FROM ducklake_snapshots('_wm_target');" + ); + } + + #[test] + fn build_wrap_blocks_orders_setup_attach_codegen_snapshot() { + let plan = ok("ATTACH 'ducklake://main' AS dl;\n SELECT a FROM dl.orders WHERE d = '{p}'"); + let blocks = build_wrap_blocks( + &plan, + "ATTACH 'ducklake:postgres:…' AS _wm_target (DATA_PATH 's3://b/p');", + "orders_daily", + "main/orders_daily", + "_wm_partition", + "'2026-06-19'", + true, + MaterializeStrategy::Replace, + ); + // setup block first, then the target ATTACH, then codegen, then result. + assert!(blocks[0].starts_with("ATTACH 'ducklake://main' AS dl")); + // every setup block must be `;`-terminated so re-splitting can't merge it + // with the synthetic target ATTACH that follows. + assert!(blocks[0].ends_with(';')); + assert_eq!( + blocks[1], + "ATTACH 'ducklake:postgres:…' AS _wm_target (DATA_PATH 's3://b/p');" + ); + assert!(blocks.iter().any(|b| b.contains("_wm_target.orders_daily"))); + assert!(blocks.iter().any(|b| b.starts_with( + "DELETE FROM _wm_target.orders_daily WHERE _wm_partition = '2026-06-19'" + ))); + // the trailing block is the one-row summary (asset / rows / snapshot_id), + // partition-scoped for the row count + let last = blocks.last().unwrap(); + assert!(last.contains("'ducklake://main/orders_daily' AS materialized")); + assert!(last.contains("'2026-06-19' AS partition")); + assert!(last.contains("WHERE _wm_partition = '2026-06-19') AS rows")); + assert!(last.contains("ducklake_snapshots('_wm_target')")); + } +} diff --git a/backend/parsers/windmill-parser/tests/fixtures/pipeline_annotations.json b/backend/parsers/windmill-parser/tests/fixtures/pipeline_annotations.json index 158246022c..083ee7a8ba 100644 --- a/backend/parsers/windmill-parser/tests/fixtures/pipeline_annotations.json +++ b/backend/parsers/windmill-parser/tests/fixtures/pipeline_annotations.json @@ -234,5 +234,72 @@ "tag": null, "retry": null } + }, + { + "name": "materialize managed (default) with merge key", + "code": "// pipeline\n// materialize ducklake://analytics/orders_daily key=order_id\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_daily", + "unique_key": "order_id" + } + } + }, + { + "name": "materialize manual escape hatch, first value wins", + "code": "// materialize manual ducklake://analytics/orders_daily\n// materialize ducklake://other/x\nexport function main() {}", + "expected": { + "in_pipeline": false, + "asset_triggers": [], + "native_triggers": [], + "partition": null, + "freshness": null, + "tag": null, + "retry": null, + "materialize": { + "target_kind": "ducklake", + "target_path": "analytics/orders_daily", + "manual": true + } + } + }, + { + "name": "materialize default-syntax shorthand with append", + "code": "// materialize ducklake append\nexport function main() {}", + "expected": { + "in_pipeline": false, + "asset_triggers": [], + "native_triggers": [], + "partition": null, + "freshness": null, + "tag": null, + "retry": null, + "materialize": { + "target_kind": "ducklake", + "target_path": "main", + "append": true + } + } + }, + { + "name": "materialize manual with no target is dropped", + "code": "// materialize manual\nexport function main() {}", + "expected": { + "in_pipeline": false, + "asset_triggers": [], + "native_triggers": [], + "partition": null, + "freshness": null, + "tag": null, + "retry": null + } } ] diff --git a/backend/parsers/windmill-parser/tests/pipeline_annotations_parity.rs b/backend/parsers/windmill-parser/tests/pipeline_annotations_parity.rs index 3efc9f0c9d..5e863fc06e 100644 --- a/backend/parsers/windmill-parser/tests/pipeline_annotations_parity.rs +++ b/backend/parsers/windmill-parser/tests/pipeline_annotations_parity.rs @@ -34,6 +34,22 @@ struct Expected { freshness: Option, tag: Option, retry: Option, + // Default-on-absent so the pre-existing fixtures (which omit it) keep + // deserializing; only fixtures exercising materialization set it. + #[serde(default)] + materialize: Option, +} + +#[derive(Deserialize)] +struct ExpectedMaterialize { + target_kind: String, + target_path: String, + #[serde(default)] + manual: bool, + #[serde(default)] + append: bool, + #[serde(default)] + unique_key: Option, } #[derive(Deserialize)] @@ -153,5 +169,25 @@ fn pipeline_annotation_fixtures_match() { want.is_some() ), } + + match (&got.materialize, &f.expected.materialize) { + (None, None) => {} + (Some(m), Some(e)) => { + assert_eq!( + kind_str(m.target_kind), + e.target_kind, + "{ctx}: materialize kind" + ); + assert_eq!(m.target_path, e.target_path, "{ctx}: materialize path"); + assert_eq!(m.manual, e.manual, "{ctx}: materialize manual"); + assert_eq!(m.append, e.append, "{ctx}: materialize append"); + assert_eq!(m.unique_key, e.unique_key, "{ctx}: materialize key"); + } + (got, want) => panic!( + "{ctx}: materialize mismatch — got {:?}, want present={}", + got, + want.is_some() + ), + } } } diff --git a/backend/windmill-api-assets/src/lib.rs b/backend/windmill-api-assets/src/lib.rs index be685b2fe6..a982b02d35 100644 --- a/backend/windmill-api-assets/src/lib.rs +++ b/backend/windmill-api-assets/src/lib.rs @@ -22,6 +22,63 @@ pub fn workspaced_service() -> Router { .route("/list_favorites", get(list_favorites)) .route("/graph", get(asset_graph)) .route("/pipelines", get(list_pipeline_folders)) + .route("/partitions", get(list_partitions)) + .route("/record_materialization", post(record_materialization)) +} + +#[derive(Deserialize)] +struct PartitionsQuery { + // The materialized asset path (`/
`). + path: String, +} + +// Per-partition materialization status for a ducklake asset — drives the +// partition-status grid and the backfill worklist. Materialization targets are +// ducklake-only in v1, so the kind is fixed. +async fn list_partitions( + authed: ApiAuthed, + Path(w_id): Path, + Extension(user_db): Extension, + Query(q): Query, +) -> JsonResult> { + let mut tx = user_db.begin(&authed).await?; + let rows = windmill_common::materialization::list_materialized_partitions( + &mut *tx, + &w_id, + AssetKind::Ducklake, + &q.path, + ) + .await?; + tx.commit().await?; + Ok(Json(rows)) +} + +// Record a materialization outcome from a polyglot (Python/TS) `wmill.ducklake` +// helper running as a pipeline step. The DuckDB `// materialize` engine records +// this itself; the SDK helpers post here instead so SDK-materialized slices show +// up in the grid identically. RLS-scoped to the caller's workspace. +async fn record_materialization( + authed: ApiAuthed, + Path(w_id): Path, + Extension(user_db): Extension, + Json(req): Json, +) -> JsonResult<()> { + let mut tx = user_db.begin(&authed).await?; + windmill_common::materialization::record_materialization( + &mut *tx, + &w_id, + req.asset_kind, + &req.asset_path, + &req.partition, + req.status, + req.snapshot_id, + req.row_count, + req.job_id, + req.error.as_deref(), + ) + .await?; + tx.commit().await?; + Ok(Json(())) } #[derive(Deserialize)] diff --git a/backend/windmill-api-scripts/Cargo.toml b/backend/windmill-api-scripts/Cargo.toml index 2dee051c93..ecf0ff0b61 100644 --- a/backend/windmill-api-scripts/Cargo.toml +++ b/backend/windmill-api-scripts/Cargo.toml @@ -26,6 +26,7 @@ windmill-parser-ts.workspace = true windmill-parser.workspace = true windmill-parser-ts-asset.workspace = true windmill-parser-sql-asset.workspace = true +windmill-parser-sql.workspace = true windmill-parser-yaml.workspace = true axum.workspace = true diff --git a/backend/windmill-api-scripts/src/scripts.rs b/backend/windmill-api-scripts/src/scripts.rs index 8115e2ef78..ea54b83c31 100644 --- a/backend/windmill-api-scripts/src/scripts.rs +++ b/backend/windmill-api-scripts/src/scripts.rs @@ -1251,6 +1251,72 @@ async fn create_script_internal<'c>( windmill_common::pipeline_advanced::freshness_enforcement_todo() ); } + // `// materialize` materializes a `ducklake:///
` target from a + // DuckDB script. These two constraints hold for *both* modes: a non-DuckLake + // target would otherwise deploy, register a producer in the asset graph, then + // silently no-op at run time (`build_materialized_query` returns `Ok(None)`), + // and a non-DuckDB script never reaches the executor that records state. The + // managed-only checks (single trailing SELECT, no SQL args) come after — a + // `manual` script owns its DDL and skips them. + if let Some(m) = pipeline_annotations.materialize.as_ref() { + if ns.language != ScriptLang::DuckDb { + return Err(Error::BadRequest(format!( + "`// materialize` is only supported for DuckDB scripts, not {}. Use the \ + wmll.ducklake helpers to materialize from other languages.", + ns.language.as_str() + ))); + } + if m.target_kind != windmill_parser::asset_parser::AssetKind::Ducklake { + return Err(Error::BadRequest( + "`// materialize` only supports a DuckLake target \ + (`ducklake:///
`); other asset kinds aren't materializable." + .to_string(), + )); + } + if !m.target_path.contains('/') { + return Err(Error::BadRequest(format!( + "`// materialize` needs a table in the target: \ + `ducklake://{0}/
` (got `ducklake://{0}`).", + m.target_path + ))); + } + if !m.manual { + if let Err(e) = windmill_parser::sql_materialize::classify_wrap(&ns.content) { + return Err(Error::BadRequest(e.message())); + } + // Managed materialize strips line comments when it wraps the SELECT, + // so a `-- $name (TYPE)` declaration is lost while its `$name` + // reference survives in the embedded SELECT — it would run unbound. + // Managed materialize takes no SQL args (the partition is supplied by + // the engine, not bound). Reject declared args with a clear error. + if let Ok(sig) = windmill_parser_sql::parse_duckdb_sig(&ns.content) { + if !sig.args.is_empty() { + let names = sig + .args + .iter() + .map(|a| format!("${}", a.name)) + .collect::>() + .join(", "); + return Err(Error::BadRequest(format!( + "managed `// materialize` cannot take SQL arguments ({names}): wrapping your \ + SELECT drops the `-- $arg` declarations, so they would run unbound. The \ + partition is supplied by the engine — reference its value with the \ + `{{partition}}` token, or use `// materialize manual` to write the DDL (and \ + bind args) yourself." + ))); + } + } + } + // `key=` (merge) and `append` are mutually exclusive reconciliation + // strategies; append (INSERT-only) wins. Surface the conflict rather + // than silently dropping the dedup the author may have intended. + if m.unique_key.is_some() && m.append { + tracing::warn!( + "script {}: both `key=` and `append` set on // materialize; append wins (INSERT-only, no dedup)", + ns.path + ); + } + } let in_pipeline = pipeline_annotations.in_pipeline; // `// trigger all` → AND join barrier (else OR, the default). let pipeline_join_all = !pipeline_annotations.join_mode.is_any(); @@ -1290,6 +1356,26 @@ async fn create_script_internal<'c>( &ns.content, ns.assets.take(), ); + // Register the `// materialize` target as a write asset so the deployed + // asset graph shows this script as the producer of the managed table — the + // body's `SELECT` doesn't express the write (the runtime generates it), so + // server-side inference wouldn't otherwise link it. + let effective_assets = if let Some(m) = pipeline_annotations.materialize.as_ref() { + let kind = windmill_common::assets::asset_kind_from_parser(m.target_kind); + let mut a = effective_assets.unwrap_or_default(); + if !a.iter().any(|x| x.kind == kind && x.path == m.target_path) { + a.push(windmill_common::assets::AssetWithAltAccessType { + path: m.target_path.clone(), + kind, + access_type: Some(windmill_common::assets::AssetUsageAccessType::W), + alt_access_type: None, + columns: None, + }); + } + Some(a) + } else { + effective_assets + }; let auto_kind = if in_pipeline { Some("pipeline".to_string()) } else if ci_test_refs.is_some() { diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 1e0e3cef92..8675c7724a 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -22576,6 +22576,13 @@ components: type: string parent_hash: type: string + auto_parent: + type: boolean + description: >- + When true, the backend resolves the parent to the current deployed + head for this path within the transaction (ignoring parent_hash), + instead of failing with a "lineage must be linear" error when the + supplied parent_hash is stale. summary: type: string description: diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index c501bbed6e..7c2a475361 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -61,6 +61,7 @@ pub mod indexer; pub mod instance_config; pub mod job_metrics; pub mod log_context; +pub mod materialization; pub mod min_version; pub mod notify_events; pub mod runtime_assets; diff --git a/backend/windmill-common/src/materialization.rs b/backend/windmill-common/src/materialization.rs new file mode 100644 index 0000000000..ccc1f7782e --- /dev/null +++ b/backend/windmill-common/src/materialization.rs @@ -0,0 +1,135 @@ +//! CE materialization state — the per-partition status recorded by the managed +//! `// materialize` write (in windmill-worker), read by the partition-status +//! grid and by the EE backfill worklist. +//! +//! The write engine and this state are CE; only automatic partition +//! *resolution* (`partition_ee`) and *backfill* orchestration +//! (`pipeline_advanced_ee`) are enterprise. This module is the shared seam: +//! the EE backfill enumerates the partitions in a range, diffs them against +//! these rows to find the missing/failed set, and pushes one CE materialization +//! job per gap (with an explicit `partition` arg — which runs idempotently and +//! upserts the row here). Nothing about that orchestration lives in this file; +//! it only needs the rows to exist, which is why recording is CE. + +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use sqlx::PgExecutor; +use uuid::Uuid; + +use crate::assets::AssetKind; +use crate::error::Result; + +/// Sentinel `partition` value for an unpartitioned (whole-table) +/// materialization — partition is part of the primary key and cannot be NULL. +pub const UNPARTITIONED: &str = ""; + +/// Mirrors the `MATERIALIZATION_STATUS` pg enum (see migration +/// `20260619170118_add_materialized_partition`). +#[derive(sqlx::Type, Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[sqlx(type_name = "MATERIALIZATION_STATUS", rename_all = "lowercase")] +#[serde(rename_all = "lowercase")] +pub enum MaterializationStatus { + Running, + Materialized, + Failed, +} + +/// The materialization outcome an agent worker (`Connection::Http`, no direct +/// DB) sends to the API to be recorded. Mirrors the `record_materialization` +/// args; the API handler unpacks it and calls that function with its own DB. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RecordMaterializationRequest { + pub asset_kind: AssetKind, + pub asset_path: String, + pub partition: String, + pub status: MaterializationStatus, + pub snapshot_id: Option, + pub row_count: Option, + pub job_id: Option, + pub error: Option, +} + +/// Upsert the latest materialization state for one (asset, partition) slice. +/// The worker records the terminal outcome once the write finishes: +/// `Materialized` (with the DuckLake `snapshot_id` + `row_count`) or `Failed` +/// (with `error`). `Running` mirrors the pg enum but has no writer in this flow. +/// Idempotent: re-running the same partition overwrites the row — exactly the +/// backfill / failure-recovery contract. +#[allow(clippy::too_many_arguments)] +pub async fn record_materialization<'e>( + executor: impl PgExecutor<'e>, + workspace_id: &str, + asset_kind: AssetKind, + asset_path: &str, + partition: &str, + status: MaterializationStatus, + snapshot_id: Option, + row_count: Option, + job_id: Option, + error: Option<&str>, +) -> Result<()> { + sqlx::query!( + "INSERT INTO materialized_partition + (workspace_id, asset_kind, asset_path, partition, status, + snapshot_id, row_count, job_id, materialized_at, error) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, now(), $9) + ON CONFLICT (workspace_id, asset_kind, asset_path, partition) + DO UPDATE SET status = EXCLUDED.status, + snapshot_id = EXCLUDED.snapshot_id, + row_count = EXCLUDED.row_count, + job_id = EXCLUDED.job_id, + materialized_at = now(), + error = EXCLUDED.error", + workspace_id, + asset_kind as AssetKind, + asset_path, + partition, + status as MaterializationStatus, + snapshot_id, + row_count, + job_id, + error, + ) + .execute(executor) + .await?; + Ok(()) +} + +/// One materialized-partition row, for the status grid / backfill diff. +#[derive(sqlx::FromRow, Debug, Clone, Serialize)] +pub struct MaterializedPartition { + pub asset_kind: AssetKind, + pub asset_path: String, + pub partition: String, + pub status: MaterializationStatus, + pub snapshot_id: Option, + pub row_count: Option, + pub job_id: Option, + pub materialized_at: DateTime, + pub error: Option, +} + +/// All recorded partitions for one asset, newest first — the grid's data and +/// the backfill worklist's "what already exists" set. +pub async fn list_materialized_partitions<'e>( + executor: impl PgExecutor<'e>, + workspace_id: &str, + asset_kind: AssetKind, + asset_path: &str, +) -> Result> { + let rows = sqlx::query_as!( + MaterializedPartition, + r#"SELECT asset_kind AS "asset_kind: AssetKind", asset_path, partition, + status AS "status: MaterializationStatus", snapshot_id, + row_count, job_id, materialized_at, error + FROM materialized_partition + WHERE workspace_id = $1 AND asset_kind = $2 AND asset_path = $3 + ORDER BY partition DESC"#, + workspace_id, + asset_kind as AssetKind, + asset_path, + ) + .fetch_all(executor) + .await?; + Ok(rows) +} diff --git a/backend/windmill-worker/src/agent_workers.rs b/backend/windmill-worker/src/agent_workers.rs index 9bf0ea1834..5320c8ef6c 100644 --- a/backend/windmill-worker/src/agent_workers.rs +++ b/backend/windmill-worker/src/agent_workers.rs @@ -78,4 +78,22 @@ pub async fn get_datatable_resource_from_agent_http( .await } +/// Record a materialization outcome from an agent worker (no direct DB) via the +/// API, so `materialized_partition` state lands the same as on a Sql worker. +// Only called from the duckdb executor, which is itself `#[cfg(feature = "duckdb")]`. +#[cfg(feature = "duckdb")] +pub async fn record_materialization_from_agent_http( + client: &HttpClient, + w_id: &str, + req: &windmill_common::materialization::RecordMaterializationRequest, +) -> anyhow::Result<()> { + client + .post( + &format!("/api/w/{}/agent_workers/record_materialization", w_id), + None, + req, + ) + .await +} + pub const UPDATE_PING_URL: &str = "/api/agent_workers/update_ping"; diff --git a/backend/windmill-worker/src/duckdb_executor.rs b/backend/windmill-worker/src/duckdb_executor.rs index 1a474ee9a0..3b6a07f702 100644 --- a/backend/windmill-worker/src/duckdb_executor.rs +++ b/backend/windmill-worker/src/duckdb_executor.rs @@ -32,6 +32,178 @@ use crate::sql_utils::remove_comments; use windmill_common::client::AuthedClient; use windmill_object_store::DEFAULT_STORAGE; +// What a `// materialize` run records into `materialized_partition` once it +// finishes. `asset_path` is the full `/
` (the asset identity); +// `partition` is "" for an unpartitioned (whole-table) materialization. +struct MaterializeExec { + asset_kind: windmill_common::assets::AssetKind, + asset_path: String, + partition: String, +} + +// If `query` declares `// materialize `, return what to record plus, +// for the default managed mode, the rewritten managed-write SQL (in `manual` +// mode the script writes its own DDL, so the rewrite is `None`). The rewritten +// SQL contains a synthetic `ATTACH 'ducklake://' AS _wm_target` that the +// normal ATTACH-transform pass resolves to real credentials — the same path as +// the user's own ATTACH. Returns `None` when there is no materialize annotation +// or the target isn't a ducklake (only ducklake is materialized in v1). +fn build_materialized_query( + query: &str, + partition_value: Option<&str>, +) -> Result, MaterializeExec)>> { + use windmill_parser::asset_parser::{parse_pipeline_annotations, AssetKind as PAssetKind}; + use windmill_parser::sql_materialize::{ + build_wrap_blocks, classify_wrap, MaterializeStrategy, TARGET_ALIAS, + }; + + let ann = parse_pipeline_annotations(query); + let Some(m) = ann.materialize else { + return Ok(None); + }; + if m.target_kind != PAssetKind::Ducklake { + return Ok(None); + } + let partitioned = ann.partition.is_some(); + let partition = partition_value.unwrap_or("").to_string(); + // Partition *resolution* is enterprise; in its absence a partitioned + // materialize only runs with an explicit `partition` arg. Fail loudly rather + // than silently materialize the wrong (empty) slice. + if partitioned && partition.is_empty() { + return Err(Error::ExecutionErr( + "materialize: a `// partitioned` script ran with no resolved partition — pass an \ + explicit `partition` arg, or enable enterprise partition resolution" + .to_string(), + )); + } + // Convention: `ducklake:///
` — is the configured + // ducklake (resolved like a user ATTACH),
is the rest. + let (ducklake_name, table) = m + .target_path + .split_once('/') + .unwrap_or((m.target_path.as_str(), "")); + let meta = MaterializeExec { + asset_kind: windmill_common::assets::AssetKind::Ducklake, + asset_path: m.target_path.clone(), + partition: partition.clone(), + }; + + if m.manual { + // Escape hatch: the script owns its DDL; we only record state. + return Ok(Some((None, meta))); + } + if table.is_empty() { + return Err(Error::ExecutionErr(format!( + "materialize: target `ducklake://{}` has no table (use ducklake:///
)", + m.target_path + ))); + } + let mut plan = classify_wrap(query).map_err(|e| Error::ExecutionErr(e.message()))?; + // Resolve the `{partition}` token (same token `// on` asset URIs use) to the + // current partition value everywhere in the managed script, so a partitioned + // materialize can filter its source by the active slice, e.g. + // `WHERE day = {partition}`. The token is always replaced by a *complete* + // escaped SQL literal (`'…'` with `'` doubled) whether or not the author + // quoted it — so a run caller can't pass metacharacters that break out of + // the literal and alter statement boundaries. The pre-quoted form + // `'{partition}'` is matched first so it doesn't become `''…''`. Only + // meaningful when partitioned. + if partitioned { + let lit = format!("'{}'", partition.replace('\'', "''")); + let tok = windmill_common::assets::PARTITION_TOKEN; + let quoted_tok = format!("'{tok}'"); + plan.output = plan.output.replace("ed_tok, &lit).replace(tok, &lit); + for s in plan.setup.iter_mut() { + *s = s.replace("ed_tok, &lit).replace(tok, &lit); + } + } + let strategy = if m.append { + MaterializeStrategy::Append + } else if let Some(uk) = m.unique_key { + MaterializeStrategy::Merge { unique_key: uk } + } else { + MaterializeStrategy::Replace + }; + // Inline the partition as an escaped SQL literal (DuckLake has no bind for + // the partition column in our generated DDL). + let pval = format!("'{}'", partition.replace('\'', "''")); + let synthetic_attach = format!("ATTACH 'ducklake://{ducklake_name}' AS {TARGET_ALIAS};"); + let blocks = build_wrap_blocks( + &plan, + &synthetic_attach, + table, + &m.target_path, + "_wm_partition", + &pval, + partitioned, + strategy, + ); + Ok(Some((Some(blocks.join("\n")), meta))) +} + +// Pull a named i64 field (`snapshot_id` / `rows`) out of the trailing summary +// read — which in wrap mode is the job result. Shape-tolerant (object / array / +// nested), returns None if absent (literal mode, or capture failed). +fn extract_i64(result: &RawValue, field: &str) -> Option { + fn find(v: &Value, field: &str) -> Option { + match v { + Value::Number(n) => n.as_i64(), + Value::Object(m) => m.get(field).and_then(|x| find(x, field)), + Value::Array(a) => a.iter().find_map(|x| find(x, field)), + _ => None, + } + } + find(&serde_json::from_str::(result.get()).ok()?, field) +} + +// Best-effort record of a materialization outcome. On a Sql connection it writes +// the row directly; on an agent worker (Http, no direct DB) it posts to the API +// so state lands the same way. Never fails the job — a lost row degrades the +// grid, not the run. +async fn record_mat( + conn: &Connection, + w_id: &str, + job_id: Uuid, + meta: &MaterializeExec, + status: windmill_common::materialization::MaterializationStatus, + snapshot_id: Option, + row_count: Option, + error: Option<&str>, +) { + let req = windmill_common::materialization::RecordMaterializationRequest { + asset_kind: meta.asset_kind, + asset_path: meta.asset_path.clone(), + partition: meta.partition.clone(), + status, + snapshot_id, + row_count, + job_id: Some(job_id), + error: error.map(|e| e.to_string()), + }; + let res: anyhow::Result<()> = match conn { + Connection::Sql(db) => windmill_common::materialization::record_materialization( + db, + w_id, + req.asset_kind, + &req.asset_path, + &req.partition, + req.status, + req.snapshot_id, + req.row_count, + req.job_id, + req.error.as_deref(), + ) + .await + .map_err(|e| anyhow::anyhow!("{e:#}")), + Connection::Http(client) => { + crate::agent_workers::record_materialization_from_agent_http(client, w_id, &req).await + } + }; + if let Err(e) = res { + tracing::warn!("failed to record materialization state: {e:#}"); + } +} + pub async fn do_duckdb( job: &MiniPulledJob, client: &AuthedClient, @@ -68,6 +240,30 @@ pub async fn do_duckdb( let mut hidden_passwords = hidden_passwords.clone(); let mut bigquery_credentials = None; + // Materialization (`// materialize`): rewrite a wrap script into managed + // DDL (its synthetic target ATTACH is resolved by the transform pass + // below, like the user's own ATTACH); a literal script is left as-is. + // `materialize` also carries what to record once the run finishes. + let partition_value: Option = job + .args + .as_ref() + .and_then(|a| a.0.get(windmill_common::partition::PARTITION_ARG)) + .and_then(|rv| serde_json::from_str::(rv.get()).ok()) + .filter(|s| !s.is_empty()); + let materialize = if query.contains("materialize") { + build_materialized_query(query, partition_value.as_deref())? + } else { + None + }; + let materialized_query; + let query: &str = match &materialize { + Some((Some(rewritten), _)) => { + materialized_query = rewritten.clone(); + &materialized_query + } + _ => query, + }; + let sig = parse_duckdb_sig(query)?.args; let mut job_args = build_args_values(job, client, conn).await?; @@ -199,6 +395,19 @@ pub async fn do_duckdb( let (result, column_order) = match result { Ok(r) => r, Err(e) => { + if let Some((_, meta)) = &materialize { + record_mat( + conn, + &job.workspace_id, + job.id, + meta, + windmill_common::materialization::MaterializationStatus::Failed, + None, + None, + Some(&e.to_string()), + ) + .await; + } if let Some(s3_proxy_err) = S3_PROXY_LAST_ERRORS_CACHE.get(&client.token) { return Err(Error::ExecutionErr(format!( "{}\n\nS3 Related Error: {}", @@ -210,6 +419,24 @@ pub async fn do_duckdb( } }; + if let Some((_, meta)) = &materialize { + // In wrap mode the job result is the summary read (snapshot_id + + // rows); in literal mode there is none, so both stay None. + let snapshot_id = extract_i64(&result, "snapshot_id"); + let row_count = extract_i64(&result, "rows"); + record_mat( + conn, + &job.workspace_id, + job.id, + meta, + windmill_common::materialization::MaterializationStatus::Materialized, + snapshot_id, + row_count, + None, + ) + .await; + } + drop(bigquery_credentials); *column_order_ref = column_order; diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 1ca77310ff..45ae8c3c25 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -4324,20 +4324,25 @@ async fn resolve_partition_for_job( job: &MiniPulledJob, code: &str, conn: &Connection, -) -> error::Result> { +) -> error::Result<(Option, bool)> { use windmill_common::partition::{resolve_partition, PARTITION_ARG}; use windmill_parser::asset_parser::PartitionKind; - // Only deployed scripts participate in asset pipelines. Cheap - // substring guard so the overwhelming majority of script jobs (no - // `// partitioned` line) skip the full annotation scan on the hot - // path; a false positive only costs one extra parse, never wrong. - if !matches!(job.kind, JobKind::Script) || !code.contains("partitioned") { - return Ok(None); + // Only deployed scripts participate in asset pipelines. Cheap substring + // guard so the overwhelming majority of script jobs skip the annotation + // scan; when one might be present we parse *once* here and reuse the result + // for both `in_pipeline` (→ WM_PIPELINE env, read by the wmll.ducklake SDK to + // record state) and `partition` resolution — no second parse downstream. The + // bool is whether the script is a `// pipeline` member. + if !matches!(job.kind, JobKind::Script) + || !(code.contains("pipeline") || code.contains("partitioned")) + { + return Ok((None, false)); } - let Some(spec) = windmill_parser::asset_parser::parse_pipeline_annotations(code).partition - else { - return Ok(None); + let ann = windmill_parser::asset_parser::parse_pipeline_annotations(code); + let in_pipeline = ann.in_pipeline; + let Some(spec) = ann.partition else { + return Ok((None, in_pipeline)); }; // Already resolved upstream — explicit run arg, backfill, or @@ -4349,7 +4354,7 @@ async fn resolve_partition_for_job( .is_some_and(|s| !s.is_empty()) }); if already_set { - return Ok(None); + return Ok((None, in_pipeline)); } // `dynamic` extracts from the triggering payload (the `trigger` object @@ -4382,7 +4387,7 @@ async fn resolve_partition_for_job( job_id = %job.id, "partitioned script resolved to no partition (before start anchor); running without one" ); - return Ok(None); + return Ok((None, in_pipeline)); }; // Persist back so dispatch_asset_triggers (which reads the producer's @@ -4404,7 +4409,7 @@ async fn resolve_partition_for_job( windmill_common::worker::to_raw_value(&value), ); updated.args = Some(Json(map)); - Ok(Some(updated)) + Ok((Some(updated), in_pipeline)) } #[tracing::instrument(level = "trace", skip_all)] @@ -4566,7 +4571,8 @@ async fn handle_code_execution_job( // `// partitioned` (if any) and shadow `job` with a clone whose args // carry the resolved `partition` for the rest of execution. let _job_with_partition; - let job = match resolve_partition_for_job(job, code, conn).await? { + let (resolved_job, in_pipeline) = resolve_partition_for_job(job, code, conn).await?; + let job = match resolved_job { Some(j) => { _job_with_partition = j; &_job_with_partition @@ -4619,6 +4625,7 @@ async fn handle_code_execution_job( lock, &modules, false, + in_pipeline, ) .await } @@ -4685,6 +4692,9 @@ pub async fn run_language_executor( lock: &Option, modules: &Option>, run_inline: bool, + // Whether the script is a `// pipeline` member (parsed once upstream) — sets + // WM_PIPELINE so the wmll.ducklake SDK helpers record materialization state. + in_pipeline: bool, ) -> error::Result> { // Defense-in-depth (GHSA-wxjq-w5pj-jqhx): the entrypoint override is // interpolated verbatim into a code position of the generated language @@ -5047,6 +5057,11 @@ mount {{ #[allow(unused_mut)] let mut envs = build_envs(envs.as_ref())?; + // Signal pipeline context to the script so the wmll.ducklake SDK helpers + // record materialization state (the grid/backfill) and skip it otherwise. + if in_pipeline { + envs.insert("WM_PIPELINE".to_string(), "true".to_string()); + } let Some(language) = language else { return Err(Error::ExecutionErr( @@ -5832,6 +5847,7 @@ pub fn init_worker_internal_server_inline_utils( &None, &None, true, + false, ) .await }) @@ -5913,6 +5929,7 @@ pub fn init_worker_internal_server_inline_utils( &content_info.lockfile, &content_info.modules, true, + false, ) .await }) diff --git a/cli/src/guidance/skills.gen.ts b/cli/src/guidance/skills.gen.ts index 1e6202b443..422e2bc03a 100644 --- a/cli/src/guidance/skills.gen.ts +++ b/cli/src/guidance/skills.gen.ts @@ -1102,6 +1102,29 @@ datatable(name: string = "main"): DatatableSqlTemplateFunction * let sql = wmill.ducklake("my_lake:analytics") */ ducklake(name: string = "main"): SqlTemplateFunction + +/** + * Idempotently materialize \`selectSql\` into a ducklake table for one + * partition (or the whole table when \`partition\` is omitted) — the client-side + * equivalent of the \`// materialize\` engine. + * With \`uniqueKey\` it upserts the slice (delete-by-key + insert); otherwise it + * replaces it (whole table → \`CREATE OR REPLACE\`; partition → delete + insert). + * Safe to re-run for the same partition (backfill / failure-recovery). + * + * Returns a lazy statement — call \`.execute()\` to run it: + * \`await wmill.upsertPartition({ table, selectSql, partition }).execute()\`. + */ +upsertPartition(opts: DucklakeMaterializeOptions): SqlStatement + +/** + * INSERT-only materialization (no dedup/replace) for append-only tables. + * Re-running the same partition duplicates rows — use only for immutable + * event-log sources. + * + * Returns a lazy statement — call \`.execute()\` to run it: + * \`await wmill.appendPartition({ table, selectSql, partition }).execute()\`. + */ +appendPartition(opts: Omit,): SqlStatement `, "write-script-bunnative": `--- name: write-script-bunnative @@ -1833,6 +1856,29 @@ datatable(name: string = "main"): DatatableSqlTemplateFunction * let sql = wmill.ducklake("my_lake:analytics") */ ducklake(name: string = "main"): SqlTemplateFunction + +/** + * Idempotently materialize \`selectSql\` into a ducklake table for one + * partition (or the whole table when \`partition\` is omitted) — the client-side + * equivalent of the \`// materialize\` engine. + * With \`uniqueKey\` it upserts the slice (delete-by-key + insert); otherwise it + * replaces it (whole table → \`CREATE OR REPLACE\`; partition → delete + insert). + * Safe to re-run for the same partition (backfill / failure-recovery). + * + * Returns a lazy statement — call \`.execute()\` to run it: + * \`await wmill.upsertPartition({ table, selectSql, partition }).execute()\`. + */ +upsertPartition(opts: DucklakeMaterializeOptions): SqlStatement + +/** + * INSERT-only materialization (no dedup/replace) for append-only tables. + * Re-running the same partition duplicates rows — use only for immutable + * event-log sources. + * + * Returns a lazy statement — call \`.execute()\` to run it: + * \`await wmill.appendPartition({ table, selectSql, partition }).execute()\`. + */ +appendPartition(opts: Omit,): SqlStatement `, "write-script-csharp": `--- name: write-script-csharp @@ -2656,6 +2702,29 @@ datatable(name: string = "main"): DatatableSqlTemplateFunction * let sql = wmill.ducklake("my_lake:analytics") */ ducklake(name: string = "main"): SqlTemplateFunction + +/** + * Idempotently materialize \`selectSql\` into a ducklake table for one + * partition (or the whole table when \`partition\` is omitted) — the client-side + * equivalent of the \`// materialize\` engine. + * With \`uniqueKey\` it upserts the slice (delete-by-key + insert); otherwise it + * replaces it (whole table → \`CREATE OR REPLACE\`; partition → delete + insert). + * Safe to re-run for the same partition (backfill / failure-recovery). + * + * Returns a lazy statement — call \`.execute()\` to run it: + * \`await wmill.upsertPartition({ table, selectSql, partition }).execute()\`. + */ +upsertPartition(opts: DucklakeMaterializeOptions): SqlStatement + +/** + * INSERT-only materialization (no dedup/replace) for append-only tables. + * Re-running the same partition duplicates rows — use only for immutable + * event-log sources. + * + * Returns a lazy statement — call \`.execute()\` to run it: + * \`await wmill.appendPartition({ table, selectSql, partition }).execute()\`. + */ +appendPartition(opts: Omit,): SqlStatement `, "write-script-duckdb": `--- name: write-script-duckdb @@ -4306,6 +4375,27 @@ def stream_result(stream) -> None # SqlQuery instance for fetching results def query(sql: str, *args) -> SqlQuery +# Idempotently materialize the rows of \`select_sql\` into ducklake +# \`table\` for one \`partition\` (or the whole table when \`partition\` is +# None). Client-side equivalent of the \`// materialize\` engine: with +# \`unique_key\` it upserts within the slice (delete-by-key + insert); +# without it, it replaces (whole table → CREATE OR REPLACE; partition → +# delete the partition + insert). Re-running the same slice is safe — the +# backfill / failure-recovery contract. +# +# The partition value is bound as a DuckDB arg (never string-interpolated) +# so it cannot inject SQL. \`select_sql\` is trusted (your own query). +def upsert_partition(table: str, select_sql: str, partition: str = None, unique_key: str = None, partition_col: str = '_wm_partition', schema: str = None) + +# INSERT-only materialization (no dedup / no replace) for an immutable +# event-log table — for one \`partition\`, or the whole table when +# \`partition\` is None. NOTE: unlike \`upsert_partition\`, re-running the same +# slice duplicates rows — use only for append-only sources. +def append_partition(table: str, select_sql: str, partition: str = None, partition_col: str = '_wm_partition', schema: str = None) + +# Read a materialized ducklake table, optionally a single partition. +def read(table: str, partition: str = None, partition_col: str = '_wm_partition', schema: str = None) + # Execute query and fetch results. # # Args: diff --git a/docs/ducklake-materialization.md b/docs/ducklake-materialization.md new file mode 100644 index 0000000000..3b5c95c770 --- /dev/null +++ b/docs/ducklake-materialization.md @@ -0,0 +1,250 @@ +# DuckLake-native materialization + +Design sketch for "managed, versioned, incremental" assets built on the +DuckLake substrate. This is a companion to [`pipelines-vs-dbt.md`](./pipelines-vs-dbt.md) +and extends its **Path C (hybrid, partition-first)** recommendation. The new +contribution here is leveraging DuckLake's snapshot/time-travel layer, which +the earlier doc's incremental deep-dive did not use. The annotation grammar is +reconciled with that doc — `// partitioned` + `// unique_key` + `// append` +stay canonical; nothing here forks a competing vocabulary. + +## The core reframe + +dbt had to *build* a materialization engine (compile SQL → `CREATE TABLE AS` / +incremental `MERGE` / SCD2 snapshots) because the warehouse gives it nothing +but raw SQL over a mutable table. DuckLake hands us, at the storage layer, the +four things that engine exists to provide: + +| Capability | Source | +|---|---| +| ACID multi-statement transactions | DuckLake | +| A snapshot per commit + time-travel (`AT (VERSION => n)` / `AT (TIMESTAMP => ...)`) | DuckLake | +| Physical partitioning + pruning (`ALTER TABLE … SET PARTITIONED BY (…)`) | DuckLake | +| Schema evolution tracked in the catalog DB | DuckLake | + +Windmill already attaches DuckLake fully — see `transform_attach_ducklake` +(`backend/windmill-worker/src/duckdb_executor.rs:661`), which rewrites a user's +`ATTACH 'ducklake://name' AS dl` into the real +`ATTACH 'ducklake:postgres:…' AS dl (DATA_PATH 's3://…', OVERRIDE_DATA_PATH +TRUE, AUTOMATIC_MIGRATION TRUE)` at `duckdb_executor.rs:730`. But today every +write is a destructive overwrite and **all four capabilities above are thrown +away** — snapshots are never surfaced, partitioning is purely an orchestration +concept disconnected from the physical layout. + +So the materialization engine we need is a thin layer — a write-strategy +wrapper + snapshot capture — not a dbt rebuild. This is exactly the +"buy A's 80% without B's dialect-rewriting tax" tradeoff `pipelines-vs-dbt.md` +argued for; DuckLake is what makes the remaining 20% (versioning, reproducible +reads, materialization history) nearly free instead of a second project. + +## Annotation grammar (final) + +One self-documenting line; managed-by-default. Strategy options live *on* the +`materialize` line (they have no meaning without it), while `// partitioned` +stays separate because it is cross-cutting (cascade + scheduling + materialize). + +``` +// materialize ducklake://analytics/orders_daily → managed, replace (default) +// materialize ducklake://analytics/orders_daily key=order_id → managed, merge +// materialize ducklake://analytics/orders_daily append → managed, append +// materialize manual ducklake://analytics/orders_daily → track-only escape hatch +``` + +- **managed (default)** — the script is *setup + one trailing `SELECT`*; Windmill + generates the write DDL, captures the DuckLake snapshot, and records state. + DuckDB-only; validated at deploy (a non-SELECT script is rejected with a clear + error pointing to the `wmll.ducklake` helpers). +- **`manual`** — escape hatch: the script writes its own DDL; Windmill only + records state (no snapshot capture, no idempotency guarantee). Rare; explicit. +- **`key=`** → MERGE (dedup within slice); **`append`** → INSERT-only; + neither → DELETE-by-partition + INSERT (replace). `append` wins over `key` if + both are given (deploy warning). +- **`// partitioned `** — unit of work + state + backfill (separate; + cross-cutting). Polyglot / multi-statement writes use the `wmll.ducklake` + helpers instead of `// materialize`. + +There is no `wrap` keyword — `materialize` *is* "manage the write," so it was +redundant; the only reason for it was to carve out the weak track-only mode, +which is now the explicit `manual` opt-out. + +DuckLake snapshots are **orthogonal to all of the above** — they apply to every +strategy automatically because every write is a DuckLake commit. The user never +annotates for versioning; they get it. + +## Executor codegen + +### The seam + +`run_duckdb` already splits the script into statement blocks and rewrites +custom `ATTACH` blocks in a single pass before execution +(`duckdb_executor.rs:114-160`): + +```rust +let query_block_list = parse_sql_blocks(&query, true); +// each block: remove_comments → if ducklake/datatable ATTACH, expand; else passthrough +``` + +All blocks run in order on one DuckDB connection. Materialize has two modes: + +1. **Managed (default).** The user writes *setup + one trailing `SELECT`*. Windmill + replaces that SELECT with generated statements, wrapped in an *explicit + DuckLake transaction it controls* — never textual `BEGIN/COMMIT` injected + around the user's other statements (fragile across their own `ATTACH`s and + multi-statement SQL). For a partitioned `replace` the SELECT block expands to: + + ```sql + -- generated for: // partitioned daily ; target = dl.orders_daily ; partition = '2026-06-19' + CREATE TABLE IF NOT EXISTS dl.orders_daily AS + SELECT *, CAST(NULL AS VARCHAR) AS _wm_partition FROM () WHERE false; -- first-run bootstrap + ALTER TABLE dl.orders_daily SET PARTITIONED BY (_wm_partition); + BEGIN TRANSACTION; + DELETE FROM dl.orders_daily WHERE _wm_partition = '2026-06-19'; + INSERT INTO dl.orders_daily SELECT *, '2026-06-19' AS _wm_partition FROM (); + COMMIT; + ``` + + The strategy variants are all DELETE+INSERT-shaped — no `MERGE INTO`, which + DuckLake can't reliably run on a fresh partition (it 404s writing the first + rows): + - **whole-table replace** (no `// partitioned`) → a single `CREATE OR REPLACE + TABLE … AS ` (handles schema changes, still snapshots). + - **`key=`** → `DELETE FROM … WHERE [ AND] IN (SELECT + FROM ())` then `INSERT` (upsert within the slice). + - **`append`** → the `DELETE` is dropped (insert-only). + +2. **`manual`.** The user writes their own DDL inside their own `BEGIN … COMMIT`; + Windmill injects nothing into the body and only records state (no snapshot + capture, no idempotency guarantee). + +The `_wm_partition` column is the physical link the orchestration layer lacks: on +first materialize Windmill runs `ALTER TABLE … SET PARTITIONED BY (_wm_partition)` +so DuckLake prunes on read and DELETE-by-partition rewrites only that partition's +Parquet files. + +> Storage: DuckLake writes go to `s3://_default_/` through the windmill S3 proxy +> (`/api/w/{ws}/s3_proxy`, gated behind the `parquet` + `private` features). The +> proxy must sign the SigV4 canonical URI with **single** percent-encoding — the +> SigV4 default (`Double`) 401s Hive-partition keys like `_wm_partition=2026-06-19` +> (the `=` double-encodes to `%253D` vs the client's `%3D`). + +### Run summary capture + +After the generated blocks, Windmill appends one read block — it is both the job's +result (a useful preview rendered as the materialized table) and the row it records: + +```sql +SELECT 'ducklake:///
' AS materialized, + '' AS partition, -- only when partitioned + (SELECT count(*) FROM [WHERE _wm_partition = '']) AS rows, + (SELECT max(snapshot_id) FROM ducklake_snapshots('')) AS snapshot_id; +``` + +The `snapshot_id` and `rows` are persisted as `materialized_partition` metadata. +One extra round-trip per materialization, no new infra. + +## Metadata schema + +Extends the `materialized_partitions` table proposed in `pipelines-vs-dbt.md` +§"First implementation slice" with the DuckLake snapshot id: + +``` +materialized_partition ( + workspace_id TEXT, + asset_kind TEXT, -- 'ducklake' + asset_path TEXT, -- 'analytics/orders_daily' + partition TEXT, -- '2026-06-19' (NULL for unpartitioned) + snapshot_id BIGINT, -- DuckLake snapshot produced by this materialize + row_count BIGINT, + job_id UUID, + materialized_at TIMESTAMPTZ, + PRIMARY KEY (workspace_id, asset_kind, asset_path, partition) +) +``` + +This one table drives four things at once: + +- **Observability** — "last materialized: snapshot 42, 1.2M rows, 09:14" per + asset node (closes the Dagster-catalog gap from the v1-readiness review). +- **Run-stale / gap detection** — which partitions exist, which are missing. +- **Backfill** — the missing/failed set *is* the backfill worklist. +- **Snapshot pinning** — see below. + +## Reproducibility — the beyond-dbt part + +Because every materialization records the snapshot it produced, a downstream +consumer can read the *exact* upstream snapshot its run saw: + +```sql +FROM dl.orders_daily AT (VERSION => $WM_UPSTREAM_SNAPSHOT) +``` + +The cascade already threads a `trigger` blob (producer path, partition) to each +subscriber; add the producer's captured `snapshot_id` to it, and a consumer's +read is pinned to the upstream state at dispatch time. That makes the *whole +pipeline* reproducible and time-travelable — something dbt has no native answer +for (dbt models are always "whatever's in the warehouse now"). It also gives +rollback (re-point an asset to snapshot N) and "what did this table look like at +the failing run" debugging, for free off the same captured ids. + +This is the differentiator worth leaning on. It is not catch-up to dbt; it is a +capability dbt structurally cannot offer, and DuckLake gives it to us at the +cost of recording one integer per run. + +It also means **we do not build SCD2 snapshots** (gap #4 in `pipelines-vs-dbt.md`): +DuckLake time-travel is a strictly better answer for most of what dbt's +`{% snapshot %}` is used for. One fewer engine to write. + +## Scoping decision: DuckLake vs DataTable + +**Make DuckLake the materialization/versioning substrate; keep DataTable as the +live operational table with no versioning.** DataTable is plain Postgres +(`transform_attach_datatable`, `duckdb_executor.rs:742`) — no native snapshots +or time-travel — so giving *it* the versioned/incremental story means building +MVCC-on-top ourselves (history tables, SCD2), precisely the complexity this +DuckLake approach exists to avoid. Clean split: + +- `ducklake://` → analytics, versioned, reproducible, backfillable. +- `datatable://` → mutable app/operational state; partition idempotency via + DELETE+INSERT still works, but no snapshot/time-travel layer. + +Don't try to give both the full treatment for v1. + +## v1 slice (smallest viable) + +1. **Partition runtime context** — resolve `(value, start, end)` and surface as + `WM_PARTITION*` bind/env (Path C step 1; partly built per the + pipeline-partition-runtime work). +2. **Physical partition wiring** — `_wm_partition` column + `SET PARTITIONED BY` + on first materialize for `ducklake://` targets. +3. **Strategy templates** — DELETE+INSERT default (`CREATE OR REPLACE` for the + whole table); delete-by-key + insert when `key=`; INSERT-only when + `append`. Managed `// materialize` wraps a single-SELECT DuckDB script behind + these templates; `// materialize manual` opts out. +4. **Snapshot + metadata capture** — append `ducklake_snapshots` read, persist + `materialized_partition` rows. +5. **Surface it** — last-materialized/snapshot/row-count on the asset node; + missing-partition set feeds the backfill UI. +6. *v1.x* — snapshot pinning across the cascade (`$WM_UPSTREAM_SNAPSHOT`), + rollback, time-travel read helper. + +Steps 1–5 are a thin annotation+template layer plus one metadata table and one +extra read per run. They deliver managed/incremental/versioned assets, +idempotent partitioned materialization, the backfill substrate, and +materialization observability together — and stay recognizably Windmill-shaped. + +## Open decisions + +These ride on top of the six in `pipelines-vs-dbt.md` §"Decisions either path +forces"; DuckLake-specific: + +1. **Bootstrap of `SET PARTITIONED BY`.** First-materialize detection — table + absent vs. present-but-unpartitioned. Idempotent re-apply. +2. **Snapshot retention / compaction.** DuckLake snapshots accumulate; when do + we expire old ones, and does pinning hold a snapshot alive past retention? +3. **Pin scope.** Pin only direct producers, or the full transitive upstream + set per run? Storage and "stale pin" semantics differ. +4. **Managed multi-statement.** *Resolved:* managed `// materialize` accepts + setup statements (ATTACH/SET/…) followed by exactly one trailing SELECT, and + rejects anything else at deploy with a clear error pointing to + `// materialize manual`. The classifier (`sql_materialize.rs`) is the single + source of truth. diff --git a/docs/pipelines-vs-dbt.md b/docs/pipelines-vs-dbt.md index 9ec6dd11d3..4a957b2a42 100644 --- a/docs/pipelines-vs-dbt.md +++ b/docs/pipelines-vs-dbt.md @@ -280,6 +280,11 @@ introspection per substrate is its own project. Ships A's 80% case first without committing to B's dialect-rewriting tax. Wrapping becomes opt-in convenience for users who want dbt-style ergonomics. +> See [`ducklake-materialization.md`](./ducklake-materialization.md) for the +> DuckLake-native realization of this path: how snapshots make the assets +> versioned/reproducible for free, the executor codegen seam, and the +> materialization-metadata schema. + ### Decisions either path forces 1. **Partition window provenance.** Scheduler tick? Trigger event time diff --git a/frontend/src/lib/components/DisplayResult.svelte b/frontend/src/lib/components/DisplayResult.svelte index 707cda66e1..4c4577d02f 100644 --- a/frontend/src/lib/components/DisplayResult.svelte +++ b/frontend/src/lib/components/DisplayResult.svelte @@ -7,7 +7,7 @@ import { base } from '$lib/base' import { downloadViaClient, shouldDownloadViaClient } from '$lib/utils/downloadFile' import { appendViewToken } from '$lib/viewToken' - import { Button, Drawer, DrawerContent } from './common' + import { Badge, Button, Drawer, DrawerContent } from './common' import { ClipboardCopy, Download, @@ -16,8 +16,10 @@ Braces, Highlighter, ArrowDownFromLine, + Database, Loader2 } from 'lucide-svelte' + import DucklakeResultPreview from './assets/AssetGraph/DucklakeResultPreview.svelte' import Portal from '$lib/components/Portal.svelte' import DisplayResultControlBar from './DisplayResultControlBar.svelte' @@ -68,6 +70,7 @@ | 'filename' | 's3object' | 's3object-list' + | 'materialized' | 'plain' | 'markdown' | 'map' @@ -195,6 +198,28 @@ return keys.includes('s3') && typeof result.s3 === 'string' } + // The materialize-run summary — `[{ materialized: 'ducklake://…', rows, + // snapshot_id }]` or the bare object. The shape is narrow (a `ducklake://` + // value plus a `snapshot_id` key) so an ordinary user result isn't hijacked. + function parseMaterializedResult( + res: any + ): + | { materialized: string; partition?: string; rows?: number; snapshot_id?: number | null } + | undefined { + const obj = Array.isArray(res) && res.length === 1 ? res[0] : res + if ( + obj && + typeof obj === 'object' && + typeof obj.materialized === 'string' && + obj.materialized.startsWith('ducklake://') && + 'snapshot_id' in obj + ) { + return obj + } + return undefined + } + + let showMaterializedPreview = $state(true) let is_render_all = $state(false) let download_as_csv = $state(false) function inferResultKind(result: any) { @@ -221,6 +246,13 @@ } try { let keys = result && typeof result === 'object' ? Object.keys(result) : [] + + if (parseMaterializedResult(result)) { + largeObject = false + is_render_all = false + return 'materialized' + } + is_render_all = keys.length == 1 && keys.includes('render_all') && Array.isArray(result['render_all']) @@ -832,6 +864,43 @@ > + {:else if !forceJson && resultKind === 'materialized'} + {@const m = parseMaterializedResult(result)} + {#if m} +
+
+ + {m.materialized} + {#if m.partition} + partition {m.partition} + {/if} + {#if typeof m.rows === 'number'} + + {m.rows} + {m.rows === 1 ? 'row' : 'rows'}{m.partition ? ' in partition' : ''} + + {/if} + {#if m.snapshot_id != null} + snapshot {m.snapshot_id} + {/if} +
+ + {#if showMaterializedPreview} +
+ +
+ {/if} +
+ {/if} {:else if !forceJson && resultKind === 's3object'} {@const s3object = parseS3Object(result) as typeof result}
| undefined, + l: string | undefined, + c: string + ) { + try { + if (l !== 'duckdb' || !s?.properties) return + const part = parsePipelineAnnotations(c).partition + if (!part) return + // Date-based partition kinds render a date / datetime picker; a dynamic + // key is a free-form string. + const format = + part.kind === 'hourly' + ? 'date-time' + : part.kind === 'daily' || part.kind === 'weekly' || part.kind === 'monthly' + ? 'date' + : undefined + if (!s.properties['partition']) { + s.properties['partition'] = { + type: 'string', + ...(format ? { format } : {}), + // ISO output so partition keys sort lexicographically (the date + // picker defaults to dd-MM-yyyy otherwise). + ...(format === 'date' ? { dateFormat: 'yyyy-MM-dd' } : {}), + description: + part.kind === 'dynamic' + ? 'Partition key value to materialize.' + : `Partition (${part.kind}) to materialize.` + } + if (Array.isArray(s.order) && !s.order.includes('partition')) { + s.order = ['partition', ...s.order] + } + } + // Pre-fill the *test* arg with the current slice for date kinds — a + // convenience default, kept on the args (not baked into the schema, + // where it would persist to the deployed script and go stale). + if (format && a && (a['partition'] == null || a['partition'] === '')) { + const now = new Date() + a['partition'] = + format === 'date' ? now.toISOString().slice(0, 10) : now.toISOString().slice(0, 16) + } + } catch (e) {} + } + async function inferModuleSchema() { if (activeModuleTab === null) return try { await inferArgs(effectiveLang, editorCode, testPanelSchema) + injectPartitionArg(testPanelSchema, testPanelArgs, effectiveLang, editorCode) moduleTestState[activeModuleTab] = { args: testPanelArgs, schema: testPanelSchema } } catch (e) { // Module code may be in-progress; silently ignore diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte index 39a6a2ae72..6073ee392c 100644 --- a/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte @@ -29,6 +29,7 @@ import SummaryPathDisplay from '$lib/components/SummaryPathDisplay.svelte' import S3FilePreview from '$lib/components/S3FilePreview.svelte' import DataTablePreview from './DataTablePreview.svelte' + import PartitionStatusGrid from './PartitionStatusGrid.svelte' import AssetRunsPanel from './AssetRunsPanel.svelte' import { Pane, Splitpanes } from 'svelte-splitpanes' import { fade } from 'svelte/transition' @@ -571,6 +572,12 @@ // deployed script, or a draft promoted from unsaved edits // to a deployed script) chains off it. parent_hash: script.hash ? String(script.hash) : undefined, + // Let the backend resolve the parent to the current head for + // this path (atomically, under an advisory lock) instead of + // rejecting a stale parent_hash with a "lineage must be + // linear" error — the pane is opened from a graph snapshot + // that can fall behind the deployed head between renders. + auto_parent: true, is_template: false, tag: script.tag, kind: script.kind as Script['kind'] | undefined, @@ -976,6 +983,8 @@ class="h-full" refreshKey={previewRefreshKey} /> + {:else if selection.asset_kind === 'ducklake'} + {:else}
No inline preview yet for {selection.asset_kind}. Use the producer/consumer arrows diff --git a/frontend/src/lib/components/assets/AssetGraph/BackfillRangeDialog.svelte b/frontend/src/lib/components/assets/AssetGraph/BackfillRangeDialog.svelte new file mode 100644 index 0000000000..1826ff6b41 --- /dev/null +++ b/frontend/src/lib/components/assets/AssetGraph/BackfillRangeDialog.svelte @@ -0,0 +1,79 @@ + + + open ?? false, (v) => (open = v)} title={`Backfill ${assetPath}`}> +
+ {#if !$enterpriseLicense} +

+ Partition backfill is an enterprise feature. Materializing a single partition is available + in the open-source edition; reprocessing a historical range requires an enterprise license. +

+ {:else} +

+ Re-runs the materialization for each partition in the range. Re-running a partition is + idempotent, so this is safe to repeat. +

+
+
+ From + +
+
+ To + +
+
+ {#if error} +

{error}

+ {/if} + {/if} +
+ + {#snippet actions()} + + + {/snippet} + diff --git a/frontend/src/lib/components/assets/AssetGraph/DucklakeResultPreview.svelte b/frontend/src/lib/components/assets/AssetGraph/DucklakeResultPreview.svelte new file mode 100644 index 0000000000..403a100aa2 --- /dev/null +++ b/frontend/src/lib/components/assets/AssetGraph/DucklakeResultPreview.svelte @@ -0,0 +1,136 @@ + + +
+ {#if partition} +
+ (scope = e.detail)}> + {#snippet children({ item })} + + + {/snippet} + +
+ {/if} + {#if colDefs.loading && !colDefs.current} +
+ +
+ {:else if !tableColDefs} +
+ + Couldn't load a preview of this table. +
+ {:else if dbTableOps} + + {#key [refreshKey, scope]} +
+ +
+ {/key} + {/if} +
diff --git a/frontend/src/lib/components/assets/AssetGraph/PartitionStatusGrid.svelte b/frontend/src/lib/components/assets/AssetGraph/PartitionStatusGrid.svelte new file mode 100644 index 0000000000..863130833d --- /dev/null +++ b/frontend/src/lib/components/assets/AssetGraph/PartitionStatusGrid.svelte @@ -0,0 +1,134 @@ + + +
+
+ Materialized partitions +
+ +
+
+ +
+ {#if partitions.loading} +
+ Loading partitions… +
+ {:else if partitions.error} +

Failed to load: {partitions.error.message}

+ {:else if !partitions.current?.length} +

+ No partitions materialized yet. They appear here after a // materialize run. +

+ {:else} +
+ + + + + + + + + + + {#each partitions.current as p (p.partition)} + + + + + + + + {#if p.error} + + {/if} + {/each} + +
PartitionStatusSnapshotRowsMaterialized
{p.partition || '(whole table)'} + + {p.status} + + {p.snapshot_id ?? '—'}{p.row_count ?? '—'}{new Date(p.materialized_at).toLocaleString()}
{p.error}
+ {/if} + + + + diff --git a/frontend/src/lib/components/assets/AssetGraph/PipelineInsertMenu.svelte b/frontend/src/lib/components/assets/AssetGraph/PipelineInsertMenu.svelte index 847f682049..e3c6c04378 100644 --- a/frontend/src/lib/components/assets/AssetGraph/PipelineInsertMenu.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/PipelineInsertMenu.svelte @@ -231,7 +231,7 @@
pathEl, { timeout: 50 }) })} @@ -240,7 +240,7 @@ {#each visibleOutputKinds.length ? visibleOutputKinds : PIPELINE_OUTPUT_KINDS as k} {@const isSelected = selected.outputId === k.id}