From 46be39dfb7fbfb2b70e61819d6065b45810c41c9 Mon Sep 17 00:00:00 2001
From: Ruben Fiszel <ruben@windmill.dev>
Date: Sun, 5 Jul 2026 22:34:58 +0200
Subject: [PATCH] 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>
---
 backend/windmill-api-assets/src/lib.rs        | 118 ++++++++++++++++++
 backend/windmill-api/openapi.yaml             |  32 +++++
 docs/ducklake-materialization.md              |  34 +++++
 .../assets/AssetGraph/AssetGraphCanvas.svelte |  27 ++++
 .../assets/AssetGraph/boundedCascade.test.ts  |  71 ++++++++++-
 .../assets/AssetGraph/boundedCascade.ts       |  14 +++
 .../lib/components/assets/AssetGraph/types.ts |  17 +++
 7 files changed, 311 insertions(+), 2 deletions(-)

diff --git a/backend/windmill-api-assets/src/lib.rs b/backend/windmill-api-assets/src/lib.rs
index 2cebf9ed13..c0a93c039b 100644
--- a/backend/windmill-api-assets/src/lib.rs
+++ b/backend/windmill-api-assets/src/lib.rs
@@ -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,
     }))
 }
 
diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml
index afe8cbc0b6..1726234a8d 100644
--- a/backend/windmill-api/openapi.yaml
+++ b/backend/windmill-api/openapi.yaml
@@ -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:
diff --git a/docs/ducklake-materialization.md b/docs/ducklake-materialization.md
index 38c38a01f2..50e7c21de0 100644
--- a/docs/ducklake-materialization.md
+++ b/docs/ducklake-materialization.md
@@ -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
diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte
index 41f5ddc3f5..d2ae645a97 100644
--- a/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte
+++ b/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte
@@ -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 = ''
 				}
diff --git a/frontend/src/lib/components/assets/AssetGraph/boundedCascade.test.ts b/frontend/src/lib/components/assets/AssetGraph/boundedCascade.test.ts
index fa735682b7..0c3b9dea64 100644
--- a/frontend/src/lib/components/assets/AssetGraph/boundedCascade.test.ts
+++ b/frontend/src/lib/components/assets/AssetGraph/boundedCascade.test.ts
@@ -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', () => {
diff --git a/frontend/src/lib/components/assets/AssetGraph/boundedCascade.ts b/frontend/src/lib/components/assets/AssetGraph/boundedCascade.ts
index 44a26d9c71..8b401d8017 100644
--- a/frontend/src/lib/components/assets/AssetGraph/boundedCascade.ts
+++ b/frontend/src/lib/components/assets/AssetGraph/boundedCascade.ts
@@ -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 }
 }
diff --git a/frontend/src/lib/components/assets/AssetGraph/types.ts b/frontend/src/lib/components/assets/AssetGraph/types.ts
index 9d4db75861..5dea3319c7 100644
--- a/frontend/src/lib/components/assets/AssetGraph/types.ts
+++ b/frontend/src/lib/components/assets/AssetGraph/types.ts
@@ -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 =