mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-09-12 08:05:44 +00:00
feat(pipelines): capture violating-row samples for data tests (#9919)
* feat(pipelines): capture violating-row samples for data tests Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(pipelines): byte-accurate sample cap and leaf-level payload sanitize Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * chore: bump ee-repo-ref to WAP guard probe adaptation Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * docs: WAP failures are counts-only — samples exist only on commit-then-test Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * docs: qualify where sample row data appears — job result and failed-job log line Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * docs: error handlers receive the full result incl. samples, like any failed job Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * 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 <noreply@anthropic.com> Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>
This commit is contained in:
Ruben Fiszel
co-authored by
Claude Fable 5
windmill-internal-app[bot]
Claude Fable 5
windmill-internal-app[bot]
parent
5ad2de91a2
commit
d4b4374de8
@@ -1 +1 @@
|
||||
2fab310d4f50ed7c34857d69c9b854f4491bf217
|
||||
80d309edebb899e36a3bdcdf4ea73c4db070534d
|
||||
|
||||
@@ -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::<Vec<_>>()
|
||||
.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::<Vec<_>>()
|
||||
.join(", ");
|
||||
let tr_from = checks
|
||||
.iter()
|
||||
.enumerate()
|
||||
.map(|(i, _)| format!("_wm_t{i}"))
|
||||
.collect::<Vec<_>>()
|
||||
.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::<Vec<_>>()
|
||||
.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::<Vec<_>>()
|
||||
.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(
|
||||
|
||||
@@ -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<String>) {
|
||||
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::<serde_json::Value>(&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::<serde_json::Value>(&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(
|
||||
|
||||
@@ -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<i64> {
|
||||
}
|
||||
|
||||
// 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<Value>,
|
||||
}
|
||||
|
||||
// Pull the per-test breakdown out of the materialize summary result. The
|
||||
@@ -793,7 +797,21 @@ fn extract_data_tests(result: &RawValue) -> Vec<DataTestOutcome> {
|
||||
.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::<Value>(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::<Vec<_>>();
|
||||
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::<Value>(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!(
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -1,15 +1,22 @@
|
||||
<script lang="ts">
|
||||
// Renders the per-test breakdown a managed `// materialize` run attaches to
|
||||
// its result as `data_tests: [{ test, violating }]`. Shown above the raw
|
||||
// result so "which tests ran, and which passed/failed" is clear at a glance.
|
||||
// (On a failed run the job result is the error message — which already lists
|
||||
// every test — so this success-path checklist and that error text together
|
||||
// cover both outcomes.)
|
||||
import { CheckCircle2, FlaskConical, XCircle } from 'lucide-svelte'
|
||||
// its result as `data_tests: [{ test, violating, sample? }]`. Shown above
|
||||
// the raw result so "which tests ran, and which passed/failed" is clear at
|
||||
// a glance. A failed test may carry `sample` — a bounded, unordered sample
|
||||
// of its violating rows — rendered as an expandable table. The sample is
|
||||
// optional by contract: its absence only means no rows to show.
|
||||
import { CheckCircle2, ChevronDown, ChevronRight, FlaskConical, XCircle } from 'lucide-svelte'
|
||||
import { Button } from '$lib/components/common'
|
||||
import AutoDataTable from '$lib/components/table/AutoDataTable.svelte'
|
||||
import { SvelteSet } from 'svelte/reactivity'
|
||||
|
||||
let { tests }: { tests: Array<{ test: string; violating: number }> } = $props()
|
||||
let {
|
||||
tests
|
||||
}: { tests: Array<{ test: string; violating: number; sample?: Record<string, any>[] }> } =
|
||||
$props()
|
||||
|
||||
let failed = $derived(tests.filter((t) => t.violating > 0).length)
|
||||
let expanded = new SvelteSet<string>()
|
||||
</script>
|
||||
|
||||
<div
|
||||
@@ -32,18 +39,40 @@
|
||||
</div>
|
||||
<ul class="divide-y divide-border">
|
||||
{#each tests as t (t.test)}
|
||||
<li class="flex items-center gap-1.5 px-2 py-1 font-mono">
|
||||
{#if t.violating > 0}
|
||||
<XCircle size={13} class="shrink-0 text-red-600 dark:text-red-400" />
|
||||
<span class="sr-only">failed:</span>
|
||||
<span class="text-primary">{t.test}</span>
|
||||
<span class="text-red-600 dark:text-red-400"
|
||||
>— {t.violating} violating row{t.violating === 1 ? '' : 's'}</span
|
||||
>
|
||||
{:else}
|
||||
<CheckCircle2 size={13} class="shrink-0 text-emerald-600 dark:text-emerald-400" />
|
||||
<span class="sr-only">passed:</span>
|
||||
<span class="text-secondary">{t.test}</span>
|
||||
<li class="px-2 py-1">
|
||||
<div class="flex items-center gap-1.5 font-mono">
|
||||
{#if t.violating > 0}
|
||||
<XCircle size={13} class="shrink-0 text-red-600 dark:text-red-400" />
|
||||
<span class="sr-only">failed:</span>
|
||||
<span class="text-primary">{t.test}</span>
|
||||
<span class="text-red-600 dark:text-red-400"
|
||||
>— {t.violating} violating row{t.violating === 1 ? '' : 's'}</span
|
||||
>
|
||||
{#if t.sample && t.sample.length > 0}
|
||||
<Button
|
||||
variant="subtle"
|
||||
unifiedSize="sm"
|
||||
startIcon={{ icon: expanded.has(t.test) ? ChevronDown : ChevronRight }}
|
||||
onclick={() => {
|
||||
if (!expanded.delete(t.test)) expanded.add(t.test)
|
||||
}}
|
||||
>
|
||||
{expanded.has(t.test) ? 'hide sample' : `view sample (${t.sample.length})`}
|
||||
</Button>
|
||||
{/if}
|
||||
{:else}
|
||||
<CheckCircle2 size={13} class="shrink-0 text-emerald-600 dark:text-emerald-400" />
|
||||
<span class="sr-only">passed:</span>
|
||||
<span class="text-secondary">{t.test}</span>
|
||||
{/if}
|
||||
</div>
|
||||
{#if t.violating > 0 && t.sample && t.sample.length > 0 && expanded.has(t.test)}
|
||||
<div class="mt-1 mb-1">
|
||||
<div class="text-2xs text-tertiary mb-1">
|
||||
sample of the violating rows ({t.sample.length} of {t.violating}, unordered)
|
||||
</div>
|
||||
<AutoDataTable objects={t.sample} />
|
||||
</div>
|
||||
{/if}
|
||||
</li>
|
||||
{/each}
|
||||
|
||||
@@ -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<string, any>[] }> | 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 }> = []
|
||||
|
||||
Reference in New Issue
Block a user