mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-08-26 00:01:37 +00:00
fix(pipelines): order data_test relationships refs before the tested script in a cascade (#9934)
* fix(pipelines): order data_test relationships refs before the tested script in a cascade Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix(pipelines): key custom-test reads by (usage_kind, path) to avoid same-path flow collisions Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
GitHub
@@ -797,6 +797,25 @@ enum TriggerEdge {
|
||||
},
|
||||
}
|
||||
|
||||
// Ordering-only "must-run-after" edge: `runnable_path`'s data test reads
|
||||
// `asset` (a `// data_test relationships` ref, or a custom test whose body
|
||||
// reads a known pipeline asset), so the asset's in-pipeline producer must
|
||||
// materialize before `runnable_path` runs. NOT a data-consumption edge — the
|
||||
// tested script doesn't ingest the asset's rows, it only needs the table to
|
||||
// exist at test time. Rendered dashed (like macro edges) and fed into the
|
||||
// cascade topo-sort so a cold cascade orders the referenced dimension first.
|
||||
// Only emitted when the referenced asset has a producer in the graph; an
|
||||
// external table (no producer) adds no edge — the runtime error stands.
|
||||
#[derive(Serialize, Debug)]
|
||||
struct TestEdge {
|
||||
producer_kind: AssetUsageKind,
|
||||
producer_path: String,
|
||||
runnable_kind: AssetUsageKind,
|
||||
runnable_path: String,
|
||||
asset_kind: AssetKind,
|
||||
asset_path: String,
|
||||
}
|
||||
|
||||
#[derive(Serialize, Debug)]
|
||||
struct AssetGraphResponse {
|
||||
assets: Vec<GraphAssetNode>,
|
||||
@@ -805,6 +824,8 @@ struct AssetGraphResponse {
|
||||
triggers: Vec<TriggerEdge>,
|
||||
#[serde(skip_serializing_if = "Vec::is_empty", default)]
|
||||
macro_edges: Vec<MacroEdge>,
|
||||
#[serde(skip_serializing_if = "Vec::is_empty", default)]
|
||||
test_edges: Vec<TestEdge>,
|
||||
}
|
||||
|
||||
async fn asset_graph(
|
||||
@@ -1257,6 +1278,102 @@ async fn asset_graph(
|
||||
runnable_set.insert((AssetUsageKind::Script, e.consumer_path.clone()));
|
||||
}
|
||||
|
||||
// Data-test ordering edges. A `// data_test relationships <col> -> <asset>`
|
||||
// (and, best-effort, a custom `// data_test <script>` whose body reads a
|
||||
// pipeline asset) needs the referenced asset materialized before the tested
|
||||
// script runs — but that dependency is otherwise absent from the execution
|
||||
// DAG, so a cold cascade can run the tested script first and hard-fail
|
||||
// ("Catalog Error: Table … does not exist"). Resolve the referenced asset's
|
||||
// producer(s) from the write edges built above and add an ordering-only
|
||||
// edge producer → testing_script. No producer (external table) ⇒ no edge.
|
||||
let mut producers_by_asset: std::collections::HashMap<
|
||||
(AssetKind, String),
|
||||
Vec<(AssetUsageKind, String)>,
|
||||
> = Default::default();
|
||||
for e in &edges {
|
||||
if matches!(e.access_type.as_deref(), Some("w") | Some("rw")) {
|
||||
producers_by_asset
|
||||
.entry((e.asset_kind, e.asset_path.clone()))
|
||||
.or_default()
|
||||
.push((e.runnable_kind, e.runnable_path.clone()));
|
||||
}
|
||||
}
|
||||
// Assets a runnable reads, keyed by (usage_kind, path) — used to order a
|
||||
// custom-test *script*'s producers before whichever member declares
|
||||
// `// data_test <path>`. Keyed on the kind too because a flow can share a
|
||||
// script's path; `// data_test <path>` resolves a deployed script body, so
|
||||
// the lookup below pins `Script` and never pulls a same-path flow's reads.
|
||||
let mut reads_by_runnable: std::collections::HashMap<
|
||||
(AssetUsageKind, String),
|
||||
Vec<(AssetKind, String)>,
|
||||
> = Default::default();
|
||||
for e in &edges {
|
||||
if matches!(e.access_type.as_deref(), None | Some("r") | Some("rw")) {
|
||||
reads_by_runnable
|
||||
.entry((e.runnable_kind, e.runnable_path.clone()))
|
||||
.or_default()
|
||||
.push((e.asset_kind, e.asset_path.clone()));
|
||||
}
|
||||
}
|
||||
let mut test_edges: Vec<TestEdge> = Vec::new();
|
||||
// Dedup on (producer, tested_script, asset) so several tests referencing the
|
||||
// same asset, or a repeated producer, yield one edge.
|
||||
let mut seen_test_edges: std::collections::HashSet<(String, String, AssetKind, String)> =
|
||||
Default::default();
|
||||
for (member_path, ann) in &annotations_by_path {
|
||||
for dt in &ann.data_tests {
|
||||
use windmill_common::assets::DataTest;
|
||||
let referenced: Vec<(AssetKind, String)> = match dt {
|
||||
DataTest::Relationships { to_kind, to_path, .. } => {
|
||||
vec![(
|
||||
windmill_common::assets::asset_kind_from_parser(*to_kind),
|
||||
to_path.clone(),
|
||||
)]
|
||||
}
|
||||
// Best-effort: the custom test *script*'s parsed reads. Reading
|
||||
// the member's OWN output resolves to the member as producer and
|
||||
// is dropped by the self-edge guard below.
|
||||
DataTest::Custom { path } => reads_by_runnable
|
||||
.get(&(AssetUsageKind::Script, path.clone()))
|
||||
.cloned()
|
||||
.unwrap_or_default(),
|
||||
_ => vec![],
|
||||
};
|
||||
for (asset_kind, asset_path) in referenced {
|
||||
let Some(producers) = producers_by_asset.get(&(asset_kind, asset_path.clone()))
|
||||
else {
|
||||
continue;
|
||||
};
|
||||
for (producer_kind, producer_path) in producers {
|
||||
// Skip a self-edge: the tested script produces the asset it
|
||||
// tests (a materialize + relationships/custom on its own output).
|
||||
if *producer_kind == AssetUsageKind::Script && producer_path == member_path {
|
||||
continue;
|
||||
}
|
||||
if seen_test_edges.insert((
|
||||
producer_path.clone(),
|
||||
member_path.clone(),
|
||||
asset_kind,
|
||||
asset_path.clone(),
|
||||
)) {
|
||||
test_edges.push(TestEdge {
|
||||
producer_kind: *producer_kind,
|
||||
producer_path: producer_path.clone(),
|
||||
runnable_kind: AssetUsageKind::Script,
|
||||
runnable_path: member_path.clone(),
|
||||
asset_kind,
|
||||
asset_path: asset_path.clone(),
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
for e in &test_edges {
|
||||
runnable_set.insert((e.producer_kind, e.producer_path.clone()));
|
||||
runnable_set.insert((e.runnable_kind, e.runnable_path.clone()));
|
||||
}
|
||||
|
||||
// Fork data-environment state per ducklake asset. The parent's rows are read on the plain
|
||||
// pool: fork membership does not imply parent membership, and defer already exposes the
|
||||
// parent's data to fork jobs — surfacing its materialization status is strictly less.
|
||||
@@ -1421,6 +1538,7 @@ async fn asset_graph(
|
||||
edges,
|
||||
triggers,
|
||||
macro_edges,
|
||||
test_edges,
|
||||
}))
|
||||
}
|
||||
|
||||
|
||||
@@ -21595,6 +21595,38 @@ paths:
|
||||
type: string
|
||||
via_use:
|
||||
type: boolean
|
||||
test_edges:
|
||||
type: array
|
||||
description: >-
|
||||
Ordering-only "must-run-after" edges — a `// data_test relationships`
|
||||
(or custom test reading a pipeline asset) requires the referenced
|
||||
asset's producer to run before the tested script. Not a data-consumption
|
||||
edge; fed into the cascade topo-sort so cold runs order correctly.
|
||||
Omitted when empty.
|
||||
items:
|
||||
type: object
|
||||
required:
|
||||
[
|
||||
producer_kind,
|
||||
producer_path,
|
||||
runnable_kind,
|
||||
runnable_path,
|
||||
asset_kind,
|
||||
asset_path,
|
||||
]
|
||||
properties:
|
||||
producer_kind:
|
||||
$ref: "#/components/schemas/AssetUsageKind"
|
||||
producer_path:
|
||||
type: string
|
||||
runnable_kind:
|
||||
$ref: "#/components/schemas/AssetUsageKind"
|
||||
runnable_path:
|
||||
type: string
|
||||
asset_kind:
|
||||
$ref: "#/components/schemas/AssetKind"
|
||||
asset_path:
|
||||
type: string
|
||||
|
||||
/w/{workspace}/assets/macros:
|
||||
get:
|
||||
|
||||
@@ -412,6 +412,40 @@ seams: add a parsed variant, emit its check/reader SQL into the summary, read
|
||||
it back in the worker. Nothing about the closed set of *today's* keywords is
|
||||
load-bearing.
|
||||
|
||||
### Cascade ordering (`test_edges`)
|
||||
|
||||
A `relationships` test — and, best-effort, a custom test whose body reads a
|
||||
pipeline asset — needs the *referenced* asset to exist at test time, but it is
|
||||
**not** a data-consumption `// on`/read of that asset, so it contributes no
|
||||
lineage edge. Without an ordering constraint, a cold cascade can run the tested
|
||||
script before the referenced dimension is materialized and hard-fail
|
||||
(`Catalog Error: Table … does not exist`).
|
||||
|
||||
`asset_graph` (`windmill-api-assets/src/lib.rs`) closes this by resolving the
|
||||
referenced asset's in-pipeline **producer** (from the write edges it already
|
||||
builds) and emitting an ordering-only `test_edges` entry
|
||||
`producer → testing_script`. It is:
|
||||
|
||||
- **Only emitted when a producer exists in the graph.** An external table (no
|
||||
in-pipeline producer) adds no edge — the missing dependency is real and the
|
||||
runtime error is the correct signal. Self-edges (a script relationship-testing
|
||||
its own output) are dropped.
|
||||
- **Rendered distinctly** — amber dashed "test needs" link on the canvas, apart
|
||||
from lineage (blue/gray solid) and triggers (gray) — because the tested script
|
||||
does not ingest the asset's rows.
|
||||
- **Fed into the cascade topo-sort** via `buildLineageDag`
|
||||
(`AssetGraph/boundedCascade.ts`), routed *through* the referenced asset node
|
||||
(`asset → testing_script`) so the existing `producer → asset` write edge
|
||||
extends into `producer → asset → testing_script` and the two-hop
|
||||
`buildLineageDownstreamMap` invariant holds. Bounded and full-pipeline runs
|
||||
(`computeInducedSchedule`) then order the producer first.
|
||||
|
||||
Scope caveat: this orders *within a single client-driven cascade* (dev run /
|
||||
bounded run / full-pipeline run). It does not change the production reactive
|
||||
asset-dispatch of two independently-scheduled roots — there, `relationships`
|
||||
targets on a disjoint root should still be co-scheduled or bound with an
|
||||
explicit `// on <dim>` if a hard ordering is required.
|
||||
|
||||
### Scoping decisions (v1)
|
||||
|
||||
- **Partition scope.** When `// partitioned`, built-in checks are scoped to the
|
||||
|
||||
@@ -222,6 +222,7 @@
|
||||
| 'add-anchor'
|
||||
| 'data-test'
|
||||
| 'macro'
|
||||
| 'test-dependency'
|
||||
unsaved?: boolean
|
||||
// Edge from a missing-trigger placeholder — styled red dashed to
|
||||
// signal "this script declared `// on kafka` but no trigger row
|
||||
@@ -523,6 +524,22 @@
|
||||
})
|
||||
}
|
||||
|
||||
// Data-test ordering edges (producer → tested script): the referenced
|
||||
// asset must be materialized before the test runs, so the cascade orders
|
||||
// the producer first. Rendered as a dashed "must run after" link, distinct
|
||||
// from data flow — the tested script doesn't consume the asset's rows.
|
||||
for (const te of g.test_edges ?? []) {
|
||||
const producerId = `${te.producer_kind}:${te.producer_path}`
|
||||
const testedId = `${te.runnable_kind}:${te.runnable_path}`
|
||||
if (!runnableNodeIds.has(producerId) || !runnableNodeIds.has(testedId)) continue
|
||||
edges.push({
|
||||
id: `testdep:${producerId}->${testedId}`,
|
||||
source: producerId,
|
||||
target: testedId,
|
||||
kind: 'test-dependency'
|
||||
})
|
||||
}
|
||||
|
||||
// Non-asset triggers (schedule + native) are rendered as source nodes
|
||||
// above the pipeline script. Real (non-missing) nodes are
|
||||
// deduplicated per (kind, ref) tuple so a single schedule shared
|
||||
@@ -951,6 +968,16 @@
|
||||
label = e.via_use ? 'uses lib' : 'macros'
|
||||
labelStyle = 'fill: rgb(139 92 246); font-size: 10px; font-weight: 600;'
|
||||
break
|
||||
case 'test-dependency':
|
||||
// Producer → tested script: amber dashed ordering link. Not
|
||||
// data flow (blue/gray) nor execution trigger (gray "triggers")
|
||||
// — it only says "the test needs this asset to exist first".
|
||||
style = 'stroke: rgb(217 119 6); stroke-width: 1.25px;'
|
||||
strokeDasharray = '5 3'
|
||||
markerColor = 'rgb(217 119 6)'
|
||||
label = 'test needs'
|
||||
labelStyle = 'fill: rgb(217 119 6); font-size: 10px; font-weight: 600;'
|
||||
break
|
||||
default:
|
||||
style = ''
|
||||
}
|
||||
|
||||
@@ -16,15 +16,17 @@ import { computeInducedSchedule } from './graphTraversal'
|
||||
type W = [script: string, asset: string] // producer write edge (datatable)
|
||||
type R = [script: string, asset: string] // pure-read edge (datatable)
|
||||
type S = [script: string, asset: string] // `// on <asset>` subscription
|
||||
type T = [producer: string, tested: string, asset: string] // `// data_test` ordering edge
|
||||
|
||||
function graph(opts: {
|
||||
scripts?: string[]
|
||||
writes?: W[]
|
||||
reads?: R[]
|
||||
subs?: S[]
|
||||
tests?: T[]
|
||||
native?: Array<[kind: NativeTriggerKind, script: string]>
|
||||
}): AssetGraphResponse {
|
||||
const { scripts = [], writes = [], reads = [], subs = [], native = [] } = opts
|
||||
const { scripts = [], writes = [], reads = [], subs = [], tests = [], native = [] } = opts
|
||||
const triggers: AssetGraphTrigger[] = [
|
||||
...subs.map(
|
||||
([s, a]) =>
|
||||
@@ -60,7 +62,15 @@ function graph(opts: {
|
||||
access_type: 'r' as const
|
||||
}))
|
||||
],
|
||||
triggers
|
||||
triggers,
|
||||
test_edges: tests.map(([producer, tested, a]) => ({
|
||||
producer_kind: 'script' as const,
|
||||
producer_path: producer,
|
||||
runnable_kind: 'script' as const,
|
||||
runnable_path: tested,
|
||||
asset_kind: 'datatable' as const,
|
||||
asset_path: a
|
||||
}))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -95,6 +105,22 @@ describe('buildLineageDag', () => {
|
||||
expect([...(dag.down.get(sn('u')) ?? [])]).toEqual([asset('x')])
|
||||
expect(dag.up.get(sn('u'))).toBeUndefined() // asset is not upstream of its own writer
|
||||
})
|
||||
|
||||
it('routes a data_test ordering edge through the referenced asset', () => {
|
||||
// prod writes x; tested has a `// data_test` against x (prod → tested edge).
|
||||
// The DAG must place x (and thus prod) upstream of tested so a cascade
|
||||
// materializes x first.
|
||||
const g = graph({
|
||||
scripts: ['prod', 'tested'],
|
||||
writes: [['prod', 'x']],
|
||||
tests: [['prod', 'tested', 'x']]
|
||||
})
|
||||
const dag = buildLineageDag(g)
|
||||
// asset x → tested (routed through the asset, not a direct prod → tested hop)
|
||||
expect([...(dag.down.get(asset('x')) ?? [])]).toEqual([sn('tested')])
|
||||
// prod → x → tested makes prod an ancestor of tested.
|
||||
expect(ancestors(dag, sn('tested'))).toEqual(new Set([asset('x'), sn('prod')]))
|
||||
})
|
||||
})
|
||||
|
||||
describe('ancestors / descendants', () => {
|
||||
@@ -271,6 +297,47 @@ describe('buildLineageDownstreamMap (read-aware scheduling)', () => {
|
||||
expect(readAware.indegree.get('c')).toBe(1)
|
||||
expect(readAware.nodes).toEqual(['a', 'c'])
|
||||
})
|
||||
|
||||
it('orders a disjoint-root producer before a data_test that references it', () => {
|
||||
// Two disjoint roots (the HD-1 repro): `dim` produces the dimension
|
||||
// `dimc`; `fct` produces `fcto` and has a `// data_test relationships`
|
||||
// against `dimc`. Without the test edge they are unordered and a cold
|
||||
// cascade can run `fct` first ("table dimc does not exist"). The test
|
||||
// edge must place `dim` strictly before `fct`.
|
||||
const g = graph({
|
||||
scripts: ['dim', 'fct'],
|
||||
writes: [
|
||||
['dim', 'dimc'],
|
||||
['fct', 'fcto']
|
||||
],
|
||||
tests: [['dim', 'fct', 'dimc']]
|
||||
})
|
||||
const selected = new Set(['dim', 'fct'])
|
||||
const map = buildLineageDownstreamMap(g)
|
||||
expect([...(map.get('dim') ?? [])]).toEqual(['fct'])
|
||||
const schedule = computeInducedSchedule(g, selected, map)
|
||||
expect(schedule.roots).toEqual(['dim'])
|
||||
expect(schedule.indegree.get('fct')).toBe(1)
|
||||
expect(schedule.nodes).toEqual(['dim', 'fct'])
|
||||
expect(schedule.cyclic).toEqual([])
|
||||
})
|
||||
|
||||
it('adds no ordering when the referenced asset has no in-pipeline producer', () => {
|
||||
// `fct` tests against an external `ext` table nothing produces — the
|
||||
// backend emits no test edge, so the frontend sees none and `fct` stays
|
||||
// an independent root (the runtime error stands, as designed).
|
||||
const g = graph({
|
||||
scripts: ['dim', 'fct'],
|
||||
writes: [['fct', 'fcto']],
|
||||
tests: [] // no producer for `ext` ⇒ backend omitted the edge
|
||||
})
|
||||
const schedule = computeInducedSchedule(
|
||||
g,
|
||||
new Set(['dim', 'fct']),
|
||||
buildLineageDownstreamMap(g)
|
||||
)
|
||||
expect(schedule.roots.sort()).toEqual(['dim', 'fct'])
|
||||
})
|
||||
})
|
||||
|
||||
describe('assetUriToNodeId', () => {
|
||||
|
||||
@@ -70,10 +70,17 @@ export type LineageDag = {
|
||||
* - producer script → asset (write / rw edges)
|
||||
* - asset → reader script (pure-read edges — a data dependency)
|
||||
* - asset → subscriber script (`// on <asset>` triggers)
|
||||
* - asset → testing script (`// data_test` ordering edges)
|
||||
*
|
||||
* An `rw` edge is treated as production only (script → asset); emitting the
|
||||
* reverse asset → script too would make every upsert a 2-cycle through its own
|
||||
* asset.
|
||||
*
|
||||
* `test_edges` are modeled through the referenced asset (asset → testing
|
||||
* script), NOT as a direct producer → testing-script hop: the producer already
|
||||
* has a write edge to that asset, so this yields producer → asset → testing
|
||||
* script and keeps the two-hop (script → asset → script) invariant that
|
||||
* `buildLineageDownstreamMap` relies on.
|
||||
*/
|
||||
export function buildLineageDag(g: AssetGraphResponse): LineageDag {
|
||||
const down = new Map<string, Set<string>>()
|
||||
@@ -107,6 +114,13 @@ export function buildLineageDag(g: AssetGraphResponse): LineageDag {
|
||||
if (t.trigger_kind !== 'asset' || t.runnable_kind !== 'script') continue
|
||||
addEdge(assetKey(t), scriptNodeId(t.runnable_path))
|
||||
}
|
||||
// Data-test ordering edges: the referenced asset must exist before the
|
||||
// tested script runs. Routed through the asset node so the existing
|
||||
// producer → asset write edge extends into producer → asset → testing script.
|
||||
for (const t of g.test_edges ?? []) {
|
||||
if (t.runnable_kind !== 'script') continue
|
||||
addEdge(assetKey(t), scriptNodeId(t.runnable_path))
|
||||
}
|
||||
|
||||
return { down, up, nodes }
|
||||
}
|
||||
|
||||
@@ -147,12 +147,29 @@ export interface AssetGraphMacroEdge {
|
||||
unsaved?: boolean
|
||||
}
|
||||
|
||||
// Ordering-only "must-run-after" edge: `runnable_path`'s `// data_test`
|
||||
// (a `relationships` ref, or a custom test reading a pipeline asset) needs
|
||||
// `asset` materialized before the tested script runs — but the tested script
|
||||
// doesn't consume the asset's rows, so this is NOT a lineage edge. Resolved
|
||||
// server-side to the referenced asset's in-pipeline producer; fed into the
|
||||
// cascade topo-sort (buildLineageDag) so a cold cascade orders the referenced
|
||||
// dimension first, and rendered dashed on the canvas (like macro edges).
|
||||
export interface AssetGraphTestEdge {
|
||||
producer_kind: GraphUsageKind
|
||||
producer_path: string
|
||||
runnable_kind: GraphUsageKind
|
||||
runnable_path: string
|
||||
asset_kind: AssetKind
|
||||
asset_path: string
|
||||
}
|
||||
|
||||
export interface AssetGraphResponse {
|
||||
assets: AssetGraphAssetNode[]
|
||||
runnables: AssetGraphRunnableNode[]
|
||||
edges: AssetGraphEdge[]
|
||||
triggers: AssetGraphTrigger[]
|
||||
macro_edges?: AssetGraphMacroEdge[]
|
||||
test_edges?: AssetGraphTestEdge[]
|
||||
}
|
||||
|
||||
export type AssetGraphNodeData =
|
||||
|
||||
Reference in New Issue
Block a user