From d4b4374de8f8a7875b050c16d1236fcd0355812b Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sat, 4 Jul 2026 17:10:52 +0000 Subject: [PATCH] feat(pipelines): capture violating-row samples for data tests (#9919) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(pipelines): capture violating-row samples for data tests Co-Authored-By: Claude Fable 5 * fix(pipelines): byte-accurate sample cap and leaf-level payload sanitize Co-Authored-By: Claude Fable 5 * chore: bump ee-repo-ref to WAP guard probe adaptation Co-Authored-By: Claude Fable 5 * docs: WAP failures are counts-only — samples exist only on commit-then-test Co-Authored-By: Claude Fable 5 * docs: qualify where sample row data appears — job result and failed-job log line Co-Authored-By: Claude Fable 5 * docs: error handlers receive the full result incl. samples, like any failed job Co-Authored-By: Claude Fable 5 * chore: update ee-repo-ref to 80d309edebb899e36a3bdcdf4ea73c4db070534d This commit updates the EE repository reference after PR #646 was merged in windmill-ee-private. Previous ee-repo-ref: 16e916bf11f26381920560b55771fce693e668c6 New ee-repo-ref: 80d309edebb899e36a3bdcdf4ea73c4db070534d Automated by sync-ee-ref workflow. --------- Co-authored-by: Claude Fable 5 Co-authored-by: windmill-internal-app[bot] --- backend/ee-repo-ref.txt | 2 +- .../windmill-parser/src/sql_materialize.rs | 244 ++++++++++++------ .../windmill-duckdb-ffi-internal/src/lib.rs | 149 +++++++++++ .../windmill-worker/src/duckdb_executor.rs | 148 +++++++++-- docs/ducklake-materialization.md | 77 +++++- .../src/lib/components/DataTestsResult.svelte | 67 +++-- .../src/lib/components/DisplayResult.svelte | 61 +++-- 7 files changed, 604 insertions(+), 144 deletions(-) diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 2c16f06c50..e6344939d1 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -2fab310d4f50ed7c34857d69c9b854f4491bf217 +80d309edebb899e36a3bdcdf4ea73c4db070534d diff --git a/backend/parsers/windmill-parser/src/sql_materialize.rs b/backend/parsers/windmill-parser/src/sql_materialize.rs index d684cf431b..145e636904 100644 --- a/backend/parsers/windmill-parser/src/sql_materialize.rs +++ b/backend/parsers/windmill-parser/src/sql_materialize.rs @@ -837,16 +837,29 @@ pub fn materialize_result_sql( if checks.is_empty() { return format!("SELECT {base_cols};"); } - // Per-test breakdown. Each check's violating-count is computed once as a CTE - // column (`c0`, `c1`, …); the `data_tests` list-of-struct then references - // those columns — DuckDB rejects scalar subqueries *inside* a struct/list - // literal, hence the CTE. Names are single-quote-escaped. The result row - // carries the whole breakdown so the worker runs every test (no - // abort-on-first) and decides pass/fail itself. - let cte_cols = checks + // Per-test breakdown. Each check's one-row probe `(v, s)` becomes a CTE + // (`_wm_t0`, `_wm_t1`, …); `_wm_tr` cross-joins them (all one-row, so the + // join stays one row) and the `data_tests` list-of-struct references the + // flattened columns — DuckDB rejects scalar subqueries *inside* a + // struct/list literal, hence the CTE lift. Names are single-quote-escaped. + // The result row carries the whole breakdown so the worker runs every + // test (no abort-on-first) and decides pass/fail itself. + let probe_ctes = checks .iter() .enumerate() - .map(|(i, c)| format!("{} AS c{i}", c.violating)) + .map(|(i, c)| format!("_wm_t{i} AS ({})", c.probe)) + .collect::>() + .join(", "); + let tr_cols = checks + .iter() + .enumerate() + .map(|(i, _)| format!("_wm_t{i}.v AS c{i}, _wm_t{i}.s AS s{i}")) + .collect::>() + .join(", "); + let tr_from = checks + .iter() + .enumerate() + .map(|(i, _)| format!("_wm_t{i}")) .collect::>() .join(", "); let list_items = checks @@ -854,12 +867,12 @@ pub fn materialize_result_sql( .enumerate() .map(|(i, c)| { let name = c.name.replace('\'', "''"); - format!("{{'test': '{name}', 'violating': c{i}}}") + format!("{{'test': '{name}', 'violating': c{i}, 'sample': s{i}}}") }) .collect::>() .join(", "); format!( - "WITH _wm_tr AS (SELECT {cte_cols}) \ + "WITH {probe_ctes}, _wm_tr AS (SELECT {tr_cols} FROM {tr_from}) \ SELECT {base_cols}, [{list_items}] AS data_tests FROM _wm_tr;" ) } @@ -880,18 +893,19 @@ fn terminate(stmt: &str) -> String { // // A data test is the FIRST extensible annotation: the parser yields a // `DataTest` from a known vocabulary, and this module turns each into a -// *check* — a `(name, violating-row-count query)` pair — that runs against the -// freshly-materialized target after the write commits. The materialize summary -// query embeds every check's count in one `data_tests` column, so all tests -// run in a single pass (no abort-on-first) and the worker, not the SQL, -// decides pass/fail and reports the full per-test breakdown. +// *check* — a `(name, probe)` pair whose probe counts and samples the +// violating rows — that runs against the freshly-materialized target after +// the write commits. The materialize summary query embeds every check's +// outcome in one `data_tests` column, so all tests run in a single pass (no +// abort-on-first) and the worker, not the SQL, decides pass/fail and reports +// the full per-test breakdown. // -// The pattern is deliberately open: a verifier is just `(name, count query)`. -// Built-ins differ only in their count query; the `Custom` escape hatch -// supplies its own (a user SELECT returning the violating rows). A sibling -// annotation family (column-lineage) can emit its own checks through the same -// `push_check` shape rather than bolting on a parallel mechanism. See -// `docs/ducklake-materialization.md`. +// The pattern is deliberately open: a verifier is just `(name, violating-rows +// query)` handed to `push_check`. Built-ins differ only in their rows query; +// the `Custom` escape hatch supplies its own (a user SELECT returning the +// violating rows). A sibling annotation family (column-lineage) can emit its +// own checks through the same `push_check` shape rather than bolting on a +// parallel mechanism. See `docs/ducklake-materialization.md`. use crate::asset_parser::{AssetKind, DataTest}; @@ -926,19 +940,31 @@ pub enum DataTestResolved { Custom { path: String, body: String }, } -/// One compiled data-test check: a human-readable `name` and a scalar SQL -/// expression (`violating`) yielding the number of rows that violate it (0 = -/// pass). The materialize summary query embeds every check's count so the -/// worker gets the whole breakdown in one result — all tests run (no -/// abort-on-first) and the worker, not the SQL, decides pass/fail. +/// One compiled data-test check: a human-readable `name` and a one-row probe +/// query yielding `(v, s)` — the violating-row count (0 = pass) and a bounded +/// `to_json` sample of the violating rows as a VARCHAR (NULL when there are +/// none, or when the serialized sample exceeds the size cap). The materialize +/// summary query embeds every check's outcome so the worker gets the whole +/// breakdown in one result — all tests run (no abort-on-first) and the +/// worker, not the SQL, decides pass/fail. The sample is decoration only: +/// enforcement reads `v`, never `s`. #[derive(Debug, Clone, PartialEq, Eq)] pub struct DataTestCheck { pub name: String, - /// Scalar subquery yielding the violating-row count, e.g. - /// `(SELECT count(*) AS v FROM (…))`. - pub violating: String, + /// One-row subquery `SELECT … AS v, … AS s FROM (…)` counting and + /// sampling the check's violating rows in a single scan. + pub probe: String, } +/// Row cap on a data-test sample. Bounded so the sample stays a debugging aid +/// (the full count is still reported); no ORDER BY on the violating rows, so +/// which rows land in the sample is nondeterministic. +const SAMPLE_MAX_ROWS: usize = 20; +/// Byte cap on one serialized sample. Oversized samples are dropped entirely +/// (NULL), never truncated — truncated JSON would fail parsing downstream +/// after paying the bytes anyway. +const SAMPLE_MAX_BYTES: usize = 51_200; + /// The SQL a set of data tests compiles to: referenced-asset `ATTACH` /// statements (resolved by the executor's ATTACH-transform pass) and the /// per-test checks, both in declaration order. @@ -1002,11 +1028,40 @@ fn partition_scope(ctx: &DataTestCtx, prefix: &str, table_alias: Option<&str>) - format!("{prefix}{}", conds.join(" AND ")) } -// Record one check: its display `name` plus `count_query` (which yields a -// single-column violating-row count) wrapped as a scalar subquery. -fn push_check(out: &mut DataTestChecks, name: String, count_query: String) { - out.checks - .push(DataTestCheck { name, violating: format!("({count_query})") }); +// Record one check: its display `name` plus `rows_query`, the SELECT of its +// violating rows. The probe counts and samples those rows in one scan: +// `_wm_v` (the subquery alias) referenced as a column is the whole row as a +// struct; `to_json(...)::VARCHAR` keeps the sample a JSON *string* through +// the FFI — expanded rows would be visible to the executor's key-recursive +// `extract_i64(result, "rows"/"snapshot_id")` scans, which a user column of +// the same name could corrupt. `list()` over zero rows and an over-cap +// sample both degrade to NULL (`s` is optional by contract). +fn push_check(out: &mut DataTestChecks, name: String, rows_query: String) { + // `strlen` counts bytes (unlike `length`, characters) — the cap bounds + // payload size on the wire, so bytes are the right unit. + let probe = format!( + "SELECT v, CASE WHEN strlen(s_raw) <= {SAMPLE_MAX_BYTES} THEN s_raw END AS s \ + FROM (SELECT count(*) AS v, \ + to_json(list(_wm_v ORDER BY _wm_rn) FILTER (WHERE _wm_rn <= {SAMPLE_MAX_ROWS}))::VARCHAR AS s_raw \ + FROM (SELECT _wm_v, row_number() OVER () AS _wm_rn FROM ({rows_query}) _wm_v))" + ); + out.checks.push(DataTestCheck { name, probe }); +} + +// `SELECT *` for a sample rows-query, excluding the synthetic physical +// partition column on partitioned targets — it's Windmill's storage detail, +// not part of the producer's logical output (same rule as schema capture). +// `qualifier` scopes the star when the query aliases the target (`_wm_src`). +fn sample_star(ctx: &DataTestCtx, qualifier: Option<&str>) -> String { + let star = match qualifier { + Some(q) => format!("{q}.*"), + None => "*".to_string(), + }; + if ctx.partitioned { + format!("{star} EXCLUDE ({})", quote_ident(ctx.partition_col)) + } else { + star + } } /// Compile resolved data tests into ATTACH statements + per-test checks for @@ -1028,28 +1083,34 @@ pub fn build_data_test_checks( DataTestResolved::BuiltIn(DataTest::Unique { column }) => { let c = quote_ident(column); let scope = partition_scope(ctx, " AND ", None); + // The rows are the GROUP BY result — one `{value, count}` per + // duplicated key — so the count (number of duplicated values, + // not of rows) and the sample share one grain and can't + // contradict each other in the UI. let q = format!( - "SELECT count(*) AS v FROM (SELECT {c} FROM {t} WHERE {c} IS NOT NULL{scope} \ - GROUP BY {c} HAVING count(*) > 1)" + "SELECT {c} AS \"value\", count(*) AS \"count\" FROM {t} \ + WHERE {c} IS NOT NULL{scope} GROUP BY {c} HAVING count(*) > 1" ); push_check(&mut out, format!("unique({column})"), q); } DataTestResolved::BuiltIn(DataTest::NotNull { column }) => { let c = quote_ident(column); let scope = partition_scope(ctx, " AND ", None); - let q = format!("SELECT count(*) AS v FROM {t} WHERE {c} IS NULL{scope}"); + let star = sample_star(ctx, None); + let q = format!("SELECT {star} FROM {t} WHERE {c} IS NULL{scope}"); push_check(&mut out, format!("not_null({column})"), q); } DataTestResolved::BuiltIn(DataTest::AcceptedValues { column, values }) => { let c = quote_ident(column); let scope = partition_scope(ctx, " AND ", None); + let star = sample_star(ctx, None); let list = values .iter() .map(|v| quote_lit(v)) .collect::>() .join(", "); let q = format!( - "SELECT count(*) AS v FROM {t} WHERE {c} IS NOT NULL AND {c} NOT IN ({list}){scope}" + "SELECT {star} FROM {t} WHERE {c} IS NOT NULL AND {c} NOT IN ({list}){scope}" ); push_check(&mut out, format!("accepted_values({column})"), q); } @@ -1113,8 +1174,9 @@ pub fn build_data_test_checks( // segment so the dot stays a schema separator, not a literal. let rt = quote_qualified(ref_table); let scope = partition_scope(ctx, " AND ", Some("_wm_src")); + let star = sample_star(ctx, Some("_wm_src")); let q = format!( - "SELECT count(*) AS v FROM {t} _wm_src \ + "SELECT {star} FROM {t} _wm_src \ WHERE _wm_src.{c} IS NOT NULL{scope} \ AND NOT EXISTS (SELECT 1 FROM {alias}.{rt} _wm_ref \ WHERE _wm_ref.{rc} = _wm_src.{c})" @@ -1150,8 +1212,7 @@ pub fn build_data_test_checks( stmts.len() )); } - let q = format!("SELECT count(*) AS v FROM ({})", stmts[0]); - push_check(&mut out, format!("custom({path})"), q); + push_check(&mut out, format!("custom({path})"), stmts[0].to_string()); } } } @@ -1678,21 +1739,34 @@ mod tests { // short, asset-free names (the asset is shown once by the breakdown). assert_eq!(sql.checks[0].name, "unique(order_id)"); assert_eq!(sql.checks[1].name, "not_null(user_id)"); - // each `violating` is a scalar count subquery. + // each probe counts and samples the violating rows in one scan, with + // the size guard on the serialized sample. + for c in &sql.checks { + assert!(c + .probe + .starts_with("SELECT v, CASE WHEN strlen(s_raw) <= 51200 THEN s_raw END AS s")); + assert!(c.probe.contains("count(*) AS v")); + assert!(c.probe.contains("FILTER (WHERE _wm_rn <= 20)")); + } + // unique: groups non-null keys within the slice, having count>1; the + // sample is `{value, count}` pairs at the same grain as the count. assert!(sql.checks[0] - .violating - .starts_with("(SELECT count(*) AS v FROM")); - // unique: groups non-null keys within the slice, having count>1 - assert!(sql.checks[0] - .violating + .probe .contains("GROUP BY \"order_id\" HAVING count(*) > 1")); assert!(sql.checks[0] - .violating + .probe + .contains("SELECT \"order_id\" AS \"value\", count(*) AS \"count\"")); + assert!(sql.checks[0] + .probe .contains("\"order_id\" IS NOT NULL AND \"_wm_partition\" = '2026-06-19'")); - // not_null: null rows in the slice + // not_null: null rows in the slice; the sample excludes the synthetic + // partition column (storage detail, not producer output). assert!(sql.checks[1] - .violating + .probe .contains("WHERE \"user_id\" IS NULL AND \"_wm_partition\" = '2026-06-19'")); + assert!(sql.checks[1] + .probe + .contains("SELECT * EXCLUDE (\"_wm_partition\") FROM")); } #[test] @@ -1701,8 +1775,9 @@ mod tests { column: "id".into(), })]; let sql = build_data_test_checks(&tests, &ctx_unpartitioned()).unwrap(); - assert!(sql.checks[0].violating.contains("WHERE \"id\" IS NULL")); - assert!(!sql.checks[0].violating.contains("_wm_partition")); + assert!(sql.checks[0].probe.contains("WHERE \"id\" IS NULL")); + assert!(!sql.checks[0].probe.contains("_wm_partition")); + assert!(!sql.checks[0].probe.contains("EXCLUDE")); } #[test] @@ -1719,14 +1794,14 @@ mod tests { ]; let sql = build_data_test_checks(&tests, &ctx_scd2()).unwrap(); assert!(sql.checks[0] - .violating + .probe .contains("WHERE \"customer_id\" IS NOT NULL AND is_current")); assert!(sql.checks[1] - .violating + .probe .contains("WHERE \"tier\" IS NULL AND is_current")); - assert!(sql.checks[2].violating.contains("AND is_current")); + assert!(sql.checks[2].probe.contains("AND is_current")); // no partition scope leaks in (scd2 is unpartitioned in v1) - assert!(!sql.checks[0].violating.contains("_wm_partition")); + assert!(!sql.checks[0].probe.contains("_wm_partition")); } #[test] @@ -1736,10 +1811,8 @@ mod tests { values: vec!["paid".into(), "o'brien".into()], })]; let sql = build_data_test_checks(&tests, &ctx_unpartitioned()).unwrap(); - assert!(sql.checks[0] - .violating - .contains("NOT IN ('paid', 'o''brien')")); - assert!(sql.checks[0].violating.contains("\"status\" IS NOT NULL")); + assert!(sql.checks[0].probe.contains("NOT IN ('paid', 'o''brien')")); + assert!(sql.checks[0].probe.contains("\"status\" IS NOT NULL")); } #[test] @@ -1763,14 +1836,19 @@ mod tests { assert_eq!(sql.attaches.len(), 1, "same db attached once"); assert_eq!(sql.attaches[0], "ATTACH 'datatable://prod' AS _wm_ref_0;"); assert!(sql.checks[0] - .violating + .probe .contains("NOT EXISTS (SELECT 1 FROM _wm_ref_0.\"users\"")); assert!(sql.checks[1] - .violating + .probe .contains("NOT EXISTS (SELECT 1 FROM _wm_ref_0.\"buyers\"")); assert!(sql.checks[0] - .violating + .probe .contains("_wm_src.\"_wm_partition\" = '2026-06-19'")); + // sample rows come from the aliased target and drop the synthetic + // partition column. + assert!(sql.checks[0] + .probe + .contains("SELECT _wm_src.* EXCLUDE (\"_wm_partition\") FROM")); assert_eq!( sql.checks[0].name, "relationships(user_id -> prod/users.id)" @@ -1807,7 +1885,7 @@ mod tests { "same-lake ref must not ATTACH again" ); assert!(sql.checks[0] - .violating + .probe .contains("NOT EXISTS (SELECT 1 FROM _wm_target.\"users\"")); } @@ -1828,10 +1906,10 @@ mod tests { ); assert!( sql.checks[0] - .violating + .probe .contains("FROM _wm_ref_0.\"main\".\"dim_products\""), "schema-qualified target should be quoted per segment: {}", - sql.checks[0].violating + sql.checks[0].probe ); } @@ -1853,10 +1931,11 @@ mod tests { body: "SELECT * FROM _wm_target.orders WHERE amount < 0;".into(), }]; let sql = build_data_test_checks(&tests, &ctx_unpartitioned()).unwrap(); - // trailing ; stripped, wrapped as a count subquery - assert!(sql.checks[0].violating.contains( - "SELECT count(*) AS v FROM (SELECT * FROM _wm_target.orders WHERE amount < 0)" - )); + // trailing ; stripped, body embedded as the probe's rows query + assert!(sql.checks[0] + .probe + .contains("FROM (SELECT * FROM _wm_target.orders WHERE amount < 0) _wm_v")); + assert!(sql.checks[0].probe.contains("count(*) AS v")); assert_eq!(sql.checks[0].name, "custom(f/tests/amount)"); } @@ -1883,14 +1962,8 @@ mod tests { #[test] fn materialize_result_sql_embeds_data_tests_breakdown() { let checks = vec![ - DataTestCheck { - name: "unique(order_id)".into(), - violating: "(SELECT count(*) AS v FROM q0)".into(), - }, - DataTestCheck { - name: "custom(f/t)".into(), - violating: "(SELECT count(*) AS v FROM q1)".into(), - }, + DataTestCheck { name: "unique(order_id)".into(), probe: "SELECT v, s FROM q0".into() }, + DataTestCheck { name: "custom(f/t)".into(), probe: "SELECT v, s FROM q1".into() }, ]; let sql = materialize_result_sql( "_wm_target.orders", @@ -1900,10 +1973,17 @@ mod tests { false, &checks, ); - // counts computed once in a CTE, referenced by the list-of-struct. - assert!(sql.starts_with("WITH _wm_tr AS (SELECT (SELECT count(*) AS v FROM q0) AS c0,")); - assert!(sql.contains("[{'test': 'unique(order_id)', 'violating': c0}, ")); - assert!(sql.contains("{'test': 'custom(f/t)', 'violating': c1}] AS data_tests")); + // each probe runs once as a one-row CTE; _wm_tr cross-joins them and + // the list-of-struct references the flattened count/sample columns. + assert!(sql.starts_with( + "WITH _wm_t0 AS (SELECT v, s FROM q0), _wm_t1 AS (SELECT v, s FROM q1), \ + _wm_tr AS (SELECT _wm_t0.v AS c0, _wm_t0.s AS s0, _wm_t1.v AS c1, _wm_t1.s AS s1 \ + FROM _wm_t0, _wm_t1)" + )); + assert!(sql.contains("[{'test': 'unique(order_id)', 'violating': c0, 'sample': s0}, ")); + assert!( + sql.contains("{'test': 'custom(f/t)', 'violating': c1, 'sample': s1}] AS data_tests") + ); assert!(sql.contains("FROM _wm_tr;")); // no tests -> plain summary, no CTE / data_tests column. let plain = materialize_result_sql( diff --git a/backend/windmill-duckdb-ffi-internal/src/lib.rs b/backend/windmill-duckdb-ffi-internal/src/lib.rs index 054f892220..25347098f2 100644 --- a/backend/windmill-duckdb-ffi-internal/src/lib.rs +++ b/backend/windmill-duckdb-ffi-internal/src/lib.rs @@ -794,6 +794,155 @@ mod temporal_json_tests { assert_eq!(json_of(3), serde_json::json!("2026-07-01")); assert_eq!(json_of(4), serde_json::json!("10:30:00")); } + + // The data-test sample probe shape emitted by + // `windmill-parser::sql_materialize::build_data_test_checks`: one scan + // yielding the violating-row count plus a bounded `to_json` sample of the + // rows as a VARCHAR. These tests gate that design against the *bundled* + // engine (json extension availability, row-as-struct alias reference, + // NULL degrade on zero rows / oversized samples, exotic column types). + fn sample_probe_sql(rows_query: &str, max_len: usize) -> String { + format!( + "SELECT v, CASE WHEN strlen(s_raw) <= {max_len} THEN s_raw END AS s \ + FROM (SELECT count(*) AS v, \ + to_json(list(_wm_v ORDER BY _wm_rn) FILTER (WHERE _wm_rn <= 20))::VARCHAR AS s_raw \ + FROM (SELECT _wm_v, row_number() OVER () AS _wm_rn FROM ({rows_query}) _wm_v))" + ) + } + + fn run_sample_probe( + conn: &duckdb::Connection, + rows_query: &str, + max_len: usize, + ) -> (i64, Option) { + let mut stmt = conn + .prepare(&sample_probe_sql(rows_query, max_len)) + .unwrap(); + let mut rows = stmt.query([]).unwrap(); + let row = rows.next().unwrap().unwrap(); + (row.get(0).unwrap(), row.get(1).unwrap()) + } + + #[test] + fn data_test_sample_probe_counts_and_samples() { + let conn = duckdb::Connection::open_in_memory().unwrap(); + conn.execute_batch( + "CREATE TABLE t (id INT, name VARCHAR); \ + INSERT INTO t VALUES (1, 'a'), (2, NULL), (3, NULL);", + ) + .unwrap(); + let (v, s) = run_sample_probe(&conn, "SELECT * FROM t WHERE name IS NULL", 51200); + assert_eq!(v, 2); + let parsed: serde_json::Value = serde_json::from_str(&s.unwrap()).unwrap(); + assert_eq!( + parsed, + serde_json::json!([{"id": 2, "name": null}, {"id": 3, "name": null}]) + ); + } + + #[test] + fn data_test_sample_probe_zero_rows_yields_null_sample() { + let conn = duckdb::Connection::open_in_memory().unwrap(); + conn.execute_batch("CREATE TABLE t (id INT); INSERT INTO t VALUES (1);") + .unwrap(); + let (v, s) = run_sample_probe(&conn, "SELECT * FROM t WHERE id IS NULL", 51200); + assert_eq!(v, 0); + assert!(s.is_none()); + } + + #[test] + fn data_test_sample_probe_caps_at_20_rows_but_counts_all() { + let conn = duckdb::Connection::open_in_memory().unwrap(); + conn.execute_batch("CREATE TABLE t AS SELECT range AS id FROM range(50);") + .unwrap(); + let (v, s) = run_sample_probe(&conn, "SELECT * FROM t", 51200); + assert_eq!(v, 50); + let parsed: serde_json::Value = serde_json::from_str(&s.unwrap()).unwrap(); + assert_eq!(parsed.as_array().unwrap().len(), 20); + } + + #[test] + fn data_test_sample_probe_oversized_sample_degrades_to_null() { + let conn = duckdb::Connection::open_in_memory().unwrap(); + conn.execute_batch("CREATE TABLE t AS SELECT repeat('x', 1000) AS big FROM range(5);") + .unwrap(); + let (v, s) = run_sample_probe(&conn, "SELECT * FROM t", 100); + assert_eq!(v, 5); + assert!(s.is_none()); + } + + #[test] + fn data_test_sample_probe_codegen_row_query_shapes() { + // The exact rows-query shapes emitted by `build_data_test_checks`: + // unique's `{value, count}` grain and the star-EXCLUDE forms used on + // partitioned targets (plain and `_wm_src.`-qualified). + let conn = duckdb::Connection::open_in_memory().unwrap(); + conn.execute_batch( + "CREATE TABLE t (id INT, name VARCHAR, _wm_partition VARCHAR); \ + INSERT INTO t VALUES (1, 'a', 'p'), (1, 'b', 'p'), (2, NULL, 'p');", + ) + .unwrap(); + let (v, s) = run_sample_probe( + &conn, + "SELECT \"id\" AS \"value\", count(*) AS \"count\" FROM t \ + WHERE \"id\" IS NOT NULL GROUP BY \"id\" HAVING count(*) > 1", + 51200, + ); + assert_eq!(v, 1); + let parsed: serde_json::Value = serde_json::from_str(&s.unwrap()).unwrap(); + assert_eq!(parsed, serde_json::json!([{"value": 1, "count": 2}])); + + let (v, s) = run_sample_probe( + &conn, + "SELECT * EXCLUDE (\"_wm_partition\") FROM t WHERE name IS NULL", + 51200, + ); + assert_eq!(v, 1); + assert_eq!( + serde_json::from_str::(&s.unwrap()).unwrap(), + serde_json::json!([{"id": 2, "name": null}]) + ); + + let (v, s) = run_sample_probe( + &conn, + "SELECT _wm_src.* EXCLUDE (\"_wm_partition\") FROM t _wm_src WHERE _wm_src.name IS NULL", + 51200, + ); + assert_eq!(v, 1); + assert_eq!( + serde_json::from_str::(&s.unwrap()).unwrap(), + serde_json::json!([{"id": 2, "name": null}]) + ); + } + + #[test] + fn data_test_sample_probe_survives_exotic_types() { + let conn = duckdb::Connection::open_in_memory().unwrap(); + conn.execute_batch( + "CREATE TABLE t AS SELECT \ + INTERVAL 3 DAY AS iv, \ + 12345678901234567890123456789::HUGEINT AS hi, \ + '\\xDE\\xAD'::BLOB AS bl, \ + [1, 2, 3] AS li, \ + {'a': 1, 'b': 'x'} AS st, \ + TIMESTAMPTZ '2026-07-01 23:13:42+00' AS tstz, \ + DECIMAL '12.34' AS dec;", + ) + .unwrap(); + let (v, s) = run_sample_probe(&conn, "SELECT * FROM t", 51200); + assert_eq!(v, 1); + let parsed: serde_json::Value = serde_json::from_str(&s.unwrap()).unwrap(); + let row = &parsed.as_array().unwrap()[0]; + // Exact renderings are DuckDB's to_json choices — the gate is only + // that every type serializes without error and parses back as JSON. + assert!(row.get("iv").is_some()); + assert!(row.get("hi").is_some()); + assert!(row.get("bl").is_some()); + assert_eq!(row["li"], serde_json::json!([1, 2, 3])); + assert_eq!(row["st"], serde_json::json!({"a": 1, "b": "x"})); + assert!(row.get("tstz").is_some()); + assert!(row.get("dec").is_some()); + } } fn json_value_to_duckdb_value( diff --git a/backend/windmill-worker/src/duckdb_executor.rs b/backend/windmill-worker/src/duckdb_executor.rs index a90c80acfc..31db95b91a 100644 --- a/backend/windmill-worker/src/duckdb_executor.rs +++ b/backend/windmill-worker/src/duckdb_executor.rs @@ -11,7 +11,7 @@ use serde_json::{json, Value}; use uuid::Uuid; use windmill_common::error::{to_anyhow, Error, Result}; use windmill_common::utils::sanitize_string_from_password; -use windmill_common::worker::{get_memory, Connection, SqlResultCollectionStrategy}; +use windmill_common::worker::{get_memory, to_raw_value, Connection, SqlResultCollectionStrategy}; use windmill_common::workspaces::{ get_datatable_resource_from_db_unchecked, get_ducklake_from_db_unchecked, DucklakeCatalogResourceType, @@ -773,10 +773,14 @@ fn extract_i64(result: &RawValue, field: &str) -> Option { } // One data test's outcome as carried by the materialize summary's `data_tests` -// column: its display name and how many rows violated it (0 = pass). +// column: its display name, how many rows violated it (0 = pass), and an +// optional bounded sample of the violating rows. The sample is decoration +// only — enforcement reads `violating`, never `sample` — so a NULL, dropped +// (over the size cap) or unparseable sample must never affect pass/fail. struct DataTestOutcome { name: String, violating: i64, + sample: Option, } // Pull the per-test breakdown out of the materialize summary result. The @@ -793,7 +797,21 @@ fn extract_data_tests(result: &RawValue) -> Vec { .get("violating") .and_then(|x| x.as_i64().or_else(|| x.as_f64().map(|f| f as i64))) .unwrap_or(0); - out.push(DataTestOutcome { name: name.clone(), violating }); + // The probe serializes the sample as a JSON *string* + // (a VARCHAR through the FFI), deliberately — parse it + // only here, so sampled user columns named `rows` / + // `snapshot_id` stay invisible to the key-recursive + // `extract_i64` scans over the summary result. Accept + // a native array too (FFI JSON-column quirk); anything + // else (NULL, size-capped, garbage) degrades to None. + let sample = o.get("sample").and_then(|s| match s { + arr @ Value::Array(_) => Some(arr.clone()), + Value::String(txt) => serde_json::from_str::(txt) + .ok() + .filter(|v| v.is_array()), + _ => None, + }); + out.push(DataTestOutcome { name: name.clone(), violating, sample }); } } } @@ -1304,7 +1322,42 @@ pub async fn do_duckdb( Some(&breakdown), ) .await; - return Err(Error::ExecutionErr(breakdown)); + // Structured failure payload: the queue wraps it as + // `{"error": {...}}` (`WrappedError`) and result_processor + // derives the run description from the top-level `message`, + // so keep `message` at the top and add no `error` nesting of + // our own. Samples ride only on failed tests, only in this + // structured result — the message text stays counts-only. + let has_samples = tests.iter().any(|t| t.violating > 0 && t.sample.is_some()); + let message = if has_samples { + // Wording must not match the UI's breakdown-line parsing + // (no ✓/✗, no `— N violating`), which older-result + // rendering still relies on. + format!( + "{breakdown}\n\nSamples of the violating rows are \ + attached to this run's result (error.data_tests)." + ) + } else { + breakdown + }; + let data_tests = tests + .iter() + .map(|t| { + let mut o = json!({ "test": t.name, "violating": t.violating }); + if t.violating > 0 { + if let Some(s) = &t.sample { + o["sample"] = s.clone(); + } + } + o + }) + .collect::>(); + return Err(Error::ExecutionRawError(to_raw_value(&json!({ + "message": message, + "name": "ExecutionErr", + "step_id": job.flow_step_id, + "data_tests": data_tests, + })))); } record_mat( conn, @@ -1347,14 +1400,46 @@ pub async fn do_duckdb( match result { Ok(result) => Ok(result), Err(e) => { - // Passwords might appear in the error message - let mut err_str = e.to_string(); - for pwd in hidden_passwords.lock().unwrap().iter() { - if let Some(sanitized) = sanitize_string_from_password(&err_str, &pwd.clone()) { - err_str = sanitized; + // Passwords might appear in the error message — and, for the + // structured data-test failure, in sampled row data read from an + // attached database — so every outgoing error is sanitized here. + let sanitize = |mut s: String| { + for pwd in hidden_passwords.lock().unwrap().iter() { + if let Some(sanitized) = sanitize_string_from_password(&s, &pwd.clone()) { + s = sanitized; + } } + s + }; + match e { + // The structured payload must keep its variant: flattening it + // to a string (the arm below) would strip the data-test + // samples result_processor places verbatim into the failed + // job's result. Sanitize its string *leaves*, not the + // serialized text: a secret containing `"` or `\` is + // JSON-escaped in the text, so a plain-substring pass would + // miss it — and samples carry raw row data from attached + // databases. + Error::ExecutionRawError(raw) => { + fn walk(v: &mut Value, f: &dyn Fn(String) -> String) { + match v { + Value::String(s) => *s = f(std::mem::take(s)), + Value::Array(a) => a.iter_mut().for_each(|x| walk(x, f)), + Value::Object(o) => o.values_mut().for_each(|x| walk(x, f)), + _ => {} + } + } + Err(match serde_json::from_str::(raw.get()) { + Ok(mut v) => { + walk(&mut v, &sanitize); + Error::ExecutionRawError(to_raw_value(&v)) + } + // Not valid JSON (shouldn't happen) — redact as text. + Err(_) => Error::ExecutionErr(sanitize(raw.get().to_string())), + }) + } + e => Err(Error::ExecutionErr(sanitize(e.to_string()))), } - Err(Error::ExecutionErr(err_str)) } } } @@ -2908,15 +2993,22 @@ mod tests { // `data_tests` array (how the FFI serialises the list-of-struct). let r = raw( r#"[{"rows":3,"snapshot_id":17,"materialized":"ducklake://a/b", - "data_tests":[{"test":"unique(order_id)","violating":0}, - {"test":"accepted_values(status)","violating":2}]}]"#, + "data_tests":[{"test":"unique(order_id)","violating":0,"sample":null}, + {"test":"accepted_values(status)","violating":2, + "sample":"[{\"id\":1,\"status\":\"bad\"}]"}]}]"#, ); let out = extract_data_tests(&r); assert_eq!(out.len(), 2); assert_eq!(out[0].name, "unique(order_id)"); assert_eq!(out[0].violating, 0); + assert!(out[0].sample.is_none()); assert_eq!(out[1].name, "accepted_values(status)"); assert_eq!(out[1].violating, 2); + // The probe emits the sample as a JSON string; it parses to rows here. + assert_eq!( + out[1].sample, + Some(serde_json::json!([{"id": 1, "status": "bad"}])) + ); } #[test] @@ -2927,10 +3019,30 @@ mod tests { assert_eq!(out.len(), 1); assert_eq!(out[0].name, "not_null(x)"); assert_eq!(out[0].violating, 1); + assert!(out[0].sample.is_none()); // Absent column (no tests) -> empty, no panic. assert!(extract_data_tests(&raw(r#"[{"rows":3}]"#)).is_empty()); } + #[test] + fn extract_data_tests_sample_degrades_to_none_never_flips_outcome() { + // sample is optional by contract: native array accepted; non-array + // JSON, garbage text, and absence all degrade to None without + // touching `violating`. + let r = raw(r#"[{"data_tests":[ + {"test":"a","violating":1,"sample":[{"id":9}]}, + {"test":"b","violating":2,"sample":"{\"not\":\"an array\"}"}, + {"test":"c","violating":3,"sample":"not json at all"}, + {"test":"d","violating":4}]}]"#); + let out = extract_data_tests(&r); + assert_eq!(out.len(), 4); + assert_eq!(out[0].sample, Some(serde_json::json!([{"id": 9}]))); + for (i, t) in out.iter().enumerate().skip(1) { + assert!(t.sample.is_none(), "test {} should have no sample", t.name); + assert_eq!(t.violating, i as i64 + 1); + } + } + #[test] fn extract_schema_parses_nested_and_string_encoded() { // Real shape: the summary row carries a nested `output_schema` @@ -2959,9 +3071,15 @@ mod tests { #[test] fn format_data_test_breakdown_lists_all_with_marks() { let tests = vec![ - DataTestOutcome { name: "unique(order_id)".into(), violating: 1 }, - DataTestOutcome { name: "not_null(user_id)".into(), violating: 0 }, - DataTestOutcome { name: "accepted_values(status)".into(), violating: 2 }, + DataTestOutcome { name: "unique(order_id)".into(), violating: 1, sample: None }, + DataTestOutcome { name: "not_null(user_id)".into(), violating: 0, sample: None }, + DataTestOutcome { + name: "accepted_values(status)".into(), + violating: 2, + // The breakdown is counts-only by design — samples never + // appear in the error text. + sample: Some(serde_json::json!([{"status": "bad"}])), + }, ]; let msg = format_data_test_breakdown("analytics/orders", &tests); assert_eq!( diff --git a/docs/ducklake-materialization.md b/docs/ducklake-materialization.md index 8329958baa..8f99b4c2c9 100644 --- a/docs/ducklake-materialization.md +++ b/docs/ducklake-materialization.md @@ -327,18 +327,73 @@ The reusable shape, in three layers, each a clean extension seam: match arm + its sub-parser; the `Custom` arm is the open fallback. A sibling family reuses this head-keyword dispatch rather than adding a parallel list. 2. **Compile** (`sql_materialize.rs::build_data_test_checks`). Each test becomes - a **check**: `(name, violating-row-count query)`. Built-ins differ only in - their count query; `Custom` supplies its own (the user's SELECT of violating - rows). Referenced assets (relationships) emit an `ATTACH` resolved by the - same transform pass as the user's own. + a **check**: `(name, probe)`, where the probe is a one-row query counting + *and sampling* the test's violating rows in a single scan. Built-ins differ + only in their violating-rows query; `Custom` supplies its own (the user's + SELECT of violating rows). Referenced assets (relationships) emit an + `ATTACH` resolved by the same transform pass as the user's own. 3. **Execute** (`duckdb_executor.rs`). The materialize summary query embeds - every check's count in one `data_tests` list-of-struct column (computed in a - CTE, since DuckDB rejects subqueries inside struct literals), so **all tests - run in a single pass** against the freshly-materialized slice — no - abort-on-first. The worker reads the breakdown from the result and decides - pass/fail: any violation **fails the run** (record `Failed`, propagate up the - cascade) with an error listing *every* test (✓/✗ + counts); a clean run - returns the per-test summary so the UI can render a checklist. + every check's outcome in one `data_tests` list-of-struct column (probes are + lifted into one-row CTEs and cross-joined, since DuckDB rejects subqueries + inside struct literals), so **all tests run in a single pass** against the + freshly-materialized slice — no abort-on-first. The worker reads the + breakdown from the result and decides pass/fail: any violation **fails the + run** (record `Failed`, propagate up the cascade) with an error listing + *every* test (✓/✗ + counts); a clean run returns the per-test summary so + the UI can render a checklist. + +### Violating-row samples + +Each check's probe also captures a bounded **sample of its violating rows** +(≤20 rows, ≤50KB serialized per test — over-cap samples are *dropped*, never +truncated, since truncated JSON fails parsing after paying the bytes). The +sample is **optional by contract** at every layer: enforcement reads only the +count; NULL / dropped / unparseable samples degrade to "no sample" and can +never flip pass/fail. Where they surface: + +- **Failed run**: the worker returns a structured error payload + (`Error::ExecutionRawError`), so the failed job's result carries + `error.data_tests: [{test, violating, sample?}]` with samples on failed + tests only. The error *message* stays counts-only, plus a pointer at the + payload — so run descriptions and `materialized_partition.error` carry no + row data. The full payload is the job *result*, and follows every failed + job's result wherever that already goes: the worker's failed-job log line + (`add_completed_job_error`) and, on EE, the args of configured + global/workspace error handlers — the same pre-existing paths that carry + e.g. a failed process's own `result.json`. A handler that formats only + `error.message` (the common alert shape) stays counts-only; one that wants + the offending rows can read `error.data_tests`. The duckdb executor's + password-sanitize catch-all must special-case `ExecutionRawError` — + flattening it to a string would silently strip the payload. +- **UI**: `DataTestsResult.svelte` renders a per-failed-test expandable + `AutoDataTable`; `DisplayResult.svelte` prefers the structured payload and + falls back to text-parsing the breakdown for results predating it. +- **Enterprise (write-audit-publish)**: samples only exist on the + commit-then-test path. Under the EE in-transaction guard a violation aborts + pre-commit and the slice rolls back, so the violating rows cease to exist — + WAP failures are counts-only by construction (the guard's in-SQL ✓/✗ + message), and the guard projects only the probe's count column + (`pipeline_advanced_ee.rs::data_test_guard_sql`). + +Sample mechanics worth knowing: + +- The sample rides as a **JSON string** (`to_json(...)::VARCHAR`) through the + summary row, deliberately: expanded rows would be visible to the executor's + key-recursive `extract_i64(result, "rows"/"snapshot_id")` scans, which a + user column of the same name could corrupt. It is parsed only inside + `extract_data_tests`. +- No `ORDER BY` on the violating rows — *which* rows land in the sample is + nondeterministic (labelled "sample" in the UI for that reason). +- `unique` samples `{value, count}` pairs at its count's grain (number of + duplicated *values*), so count and sample can't contradict each other. +- On partitioned targets the synthetic `_wm_partition` column is `EXCLUDE`d + from samples (same rule as schema capture). +- A custom body joining with `SELECT *` can yield duplicate column names — + duplicate JSON keys keep the last value; harmless but visible. +- The probe shape is gated by in-memory tests against the bundled engine in + `windmill-duckdb-ffi-internal` (json extension availability, zero-row NULL + degrade, exotic types) — a sample-expr runtime error would fail the summary + read after the write committed, so keep those green when touching the shape. A new annotation family that produces post-materialize checks (or, for column-lineage, post-materialize *metadata reads*) plugs into the same three diff --git a/frontend/src/lib/components/DataTestsResult.svelte b/frontend/src/lib/components/DataTestsResult.svelte index cba28f3a82..385af187df 100644 --- a/frontend/src/lib/components/DataTestsResult.svelte +++ b/frontend/src/lib/components/DataTestsResult.svelte @@ -1,15 +1,22 @@
    {#each tests as t (t.test)} -
  • - {#if t.violating > 0} - - failed: - {t.test} - — {t.violating} violating row{t.violating === 1 ? '' : 's'} - {:else} - - passed: - {t.test} +
  • +
    + {#if t.violating > 0} + + failed: + {t.test} + — {t.violating} violating row{t.violating === 1 ? '' : 's'} + {#if t.sample && t.sample.length > 0} + + {/if} + {:else} + + passed: + {t.test} + {/if} +
    + {#if t.violating > 0 && t.sample && t.sample.length > 0 && expanded.has(t.test)} +
    +
    + sample of the violating rows ({t.sample.length} of {t.violating}, unordered) +
    + +
    {/if}
    • {/each} diff --git a/frontend/src/lib/components/DisplayResult.svelte b/frontend/src/lib/components/DisplayResult.svelte index cbd81086a9..efb1dd2da1 100644 --- a/frontend/src/lib/components/DisplayResult.svelte +++ b/frontend/src/lib/components/DisplayResult.svelte @@ -574,24 +574,53 @@ // formats are produced by this repo's worker (see duckdb_executor.rs); the // derivation is inert (undefined) for every other DisplayResult use. let dataTests = $derived.by(() => { + // Both structured shapes carry `[{ test, violating, sample? }]`; the + // sample (bounded violating-row rows) may arrive as a JSON string (the + // worker keeps it string-typed through the summary row) and is optional + // by contract — anything malformed degrades to no sample, never to a + // dropped checklist. + const normalize = ( + dt: any + ): Array<{ test: string; violating: number; sample?: Record[] }> | undefined => { + if (typeof dt === 'string') { + try { + dt = JSON.parse(dt) + } catch { + return undefined + } + } + if ( + !Array.isArray(dt) || + dt.length === 0 || + !dt.every((x) => x && typeof x.test === 'string' && typeof x.violating === 'number') + ) { + return undefined + } + return dt.map((x) => { + let sample = x.sample + if (typeof sample === 'string') { + try { + sample = JSON.parse(sample) + } catch { + sample = undefined + } + } + if (!Array.isArray(sample) || !sample.every((r) => r && typeof r === 'object')) { + sample = undefined + } + return { test: x.test, violating: x.violating, sample } + }) + } // Success: structured column on the summary row. const row = Array.isArray(result) ? (result as any)?.[0] : (result as any) - let dt = row?.data_tests - if (typeof dt === 'string') { - try { - dt = JSON.parse(dt) - } catch { - dt = undefined - } - } - if ( - Array.isArray(dt) && - dt.length > 0 && - dt.every((x) => x && typeof x.test === 'string' && typeof x.violating === 'number') - ) { - return dt as Array<{ test: string; violating: number }> - } - // Failure: parse the worker's breakdown out of the error message. + const fromRow = normalize(row?.data_tests) + if (fromRow) return fromRow + // Failure: the worker attaches the same structured breakdown (plus + // per-failed-test samples) to the error payload. + const fromError = normalize((result as any)?.error?.data_tests) + if (fromError) return fromError + // Failure fallback for results predating the structured error payload: + // parse the worker's breakdown out of the error message. const msg = (result as any)?.error?.message if (typeof msg === 'string' && msg.includes('data tests failed on')) { const out: Array<{ test: string; violating: number }> = []