feat: assets-only view of a pipeline, with producer and trigger chips

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Diego Imbert
2026-09-30 13:04:06 +02:00
co-authored by Claude Opus 5.5
parent 4cdbd11977
commit 77c5df6289
10 changed files with 422 additions and 55 deletions
@@ -3,6 +3,7 @@
import {
SvelteFlow,
Controls,
Panel,
MiniMap,
ConnectionLineType,
type Node,
@@ -17,7 +18,11 @@
import AssetGraphEdge from './AssetGraphEdge.svelte'
import PanToNode from './PanToNode.svelte'
import InitialFitView from './InitialFitView.svelte'
import { layoutAssetGraph } from './assetGraphLayout'
import { layoutAssetGraph, PIPELINE_NODE_EXTRA_ROW } from './assetGraphLayout'
import { assetsOnlyView, type AssetUpstream } from './assetsOnlyView'
import { TRIGGER_NODE_STYLE } from './TriggerNode.svelte'
import { describeCron } from '$lib/utils/describeCron'
import Toggle from '$lib/components/Toggle.svelte'
import { computeMutedReadKeys, dbtAssociations } from './resolveGraph'
import { buildDownstreamMap } from './graphTraversal'
import { buildLineageDownstreamMap } from './boundedCascade'
@@ -190,6 +195,9 @@
* canvas is embedded inline inside a scrollable container, so a wheel
* gesture over it scrolls the container instead of being captured. */
scrollZoom?: boolean
/** Offer the "Assets only" switch, which folds scripts and triggers into the
* assets they produce. */
assetsOnlyToggle?: boolean
}
let {
graph,
@@ -222,9 +230,12 @@
highlightActiveRun = false,
recomputedAssetIds,
assetRunStatus,
scrollZoom = true
scrollZoom = true,
assetsOnlyToggle = false
}: Props = $props()
let assetsOnly = $state(false)
// `${kind}:${path}` ids for the hovered / pinned runs (both script and flow
// variants, since the run row's kind isn't known here).
const runIds = (paths: string[] | undefined): Set<string> =>
@@ -252,6 +263,9 @@
| 'macro'
| 'test-dependency'
| 'dbt-ref'
// Assets-only view: an asset → an asset computed from it, through the
// script that is folded away.
| 'asset-flow'
unsaved?: boolean
// Muted read edge: a ducklake/s3 input read every run whose (default)
// auto cascade trigger is suppressed by `// mute` / `// mute all`.
@@ -885,6 +899,84 @@
let model = $derived(build(graph))
/** The label of the chip naming what runs an asset's producer. */
function triggerChip(t: (AssetUpstream & { multiple: false })['trigger']): {
label: string
missing?: boolean
draft?: boolean
} {
if (!t) return { label: 'Manual' }
if (t.kind === 'asset') return { label: 'On asset change' }
const style = TRIGGER_NODE_STYLE[t.kind as TriggerNodeKind]
const label =
t.kind === 'schedule' && t.data?.schedule
? (describeCron(t.data.schedule) ?? `Schedule (${t.data.schedule})`)
: (style?.label ?? t.kind)
return {
label,
missing: t.data?.missing && t.kind !== 'webhook' && t.kind !== 'data_upload',
draft: t.data?.draft
}
}
// Scripts and triggers fold into the assets they produce: each asset carries
// its producer's run state, language and trigger instead of drawing them.
function assetsOnlyModel(m: typeof model): typeof model {
const v = assetsOnlyView(m.nodes, m.edges)
const runnables = new Map(m.nodes.filter((n) => n.type === 'runnable').map((n) => [n.id, n]))
const nodes = m.nodes
.filter((n) => n.id === ADD_NODE_ID || v.nodeIds.has(n.id))
.map((n) => {
if (n.type !== 'asset') return n
const u = v.upstream.get(n.id)
const r = u && !u.multiple ? runnables.get(u.runnableId)?.data : undefined
return {
...n,
data: {
...n.data,
// A dbt model is built by its project, whose writes the canvas
// does not draw; its dbt chip already names it.
upstream: !u
? n.data.dbt && n.data.dbt.resource_type !== 'source'
? undefined
: { none: true }
: u.multiple || !r
? { multiple: true }
: {
path: r.path,
summary: r.summary,
language: r.language,
unsaved: r.unsaved,
runState: r.runState,
trigger: triggerChip(u.trigger),
onOpen: () =>
onselect?.({
kind: 'runnable',
runnable_kind: r.runnable_kind,
path: r.path
})
}
}
}
})
const edges: BuiltEdge[] = v.edges.map((e) => ({ ...e, kind: e.kind as BuiltEdge['kind'] }))
if (nodes.some((n) => n.id === ADD_NODE_ID)) {
const hasIncoming = new Set(edges.map((e) => e.target))
for (const n of nodes) {
if (n.id !== ADD_NODE_ID && !hasIncoming.has(n.id)) {
edges.push({
id: `add-anchor:${n.id}`,
source: ADD_NODE_ID,
target: n.id,
kind: 'add-anchor'
})
}
}
}
return { ...m, nodes, edges }
}
let view = $derived(assetsOnly ? assetsOnlyModel(model) : model)
// dbt association, surfaced by emphasis instead of edges. Hovering a model's
// dbt badge lights up the project node that materializes it; hovering the
// project node lights up every model it owns. Clicking the badge selects the
@@ -926,14 +1018,14 @@
// input; the rendered edges are untouched (both arrows still drawn).
let writeEdgePairs = $derived(
new Set(
model.edges.filter((e) => e.kind === 'lineage-write').map((e) => `${e.source}\n${e.target}`)
view.edges.filter((e) => e.kind === 'lineage-write').map((e) => `${e.source}\n${e.target}`)
)
)
let layoutInput = $derived({
nodes: model.nodes
nodes: view.nodes
.map((n) => ({ id: n.id, data: n.data }))
.sort((a, b) => (a.id < b.id ? -1 : a.id > b.id ? 1 : 0)),
edges: model.edges
edges: view.edges
.filter(
(e) =>
!(
@@ -954,7 +1046,9 @@
: 1
)
})
let layoutPositions = $derived(layoutAssetGraph(layoutInput, ADD_NODE_ID))
let layoutPositions = $derived(
layoutAssetGraph(layoutInput, ADD_NODE_ID, assetsOnly ? PIPELINE_NODE_EXTRA_ROW : 0)
)
let positionedNodes = $derived.by(() => {
// Compute bbox width from layout; shift every x so the graph is
@@ -969,7 +1063,7 @@
}
const bboxWidth = isFinite(minX) ? maxX - minX : 0
const xCenter = paneWidth / 2 - bboxWidth / 2
return model.nodes.map<Node>((n) => {
return view.nodes.map<Node>((n) => {
const p = layoutPositions.get(n.id) ?? { x: 0, y: 0 }
// Activity-panel emphasis (purely visual rings, kept off `selected`
// which swaps the details pane). Hover wins over pin so the cursor
@@ -1040,7 +1134,7 @@
let assetEmphasis = $derived.by<Map<string, 'input' | 'output'>>(() => {
const m = new Map<string, 'input' | 'output'>()
if (emphasizedRunIdSet.size === 0) return m
for (const e of model.edges) {
for (const e of view.edges) {
if (e.kind === 'lineage-write' && emphasizedRunIdSet.has(e.source)) {
m.set(e.target, 'output')
} else if (
@@ -1091,7 +1185,7 @@
}
let flowEdges = $derived.by(() =>
model.edges
view.edges
// Anchor edges are layout-only.
.filter((e) => e.kind !== 'add-anchor')
.map<Edge>((e) => {
@@ -1123,6 +1217,10 @@
animated = flowAnimated
markerColor = 'rgb(59 130 246)'
break
case 'asset-flow':
style = 'stroke: rgb(59 130 246); stroke-width: 1.25px;'
markerColor = 'rgb(59 130 246)'
break
case 'lineage-read':
style = 'stroke: rgb(156 163 175); stroke-width: 1px;'
animated = flowAnimated
@@ -1346,9 +1444,17 @@
--background-color={false}
>
<div class="absolute inset-0 !bg-surface-secondary h-full"></div>
<InitialFitView {nodes} fitKey={viewportFitKey} />
<InitialFitView {nodes} fitKey={`${viewportFitKey}:${assetsOnly}`} />
<PanToNode targetId={panToNodeId} {nodes} />
<Controls position="top-right" orientation="horizontal" showLock={false} class="!mr-10" />
{#if assetsOnlyToggle}
<Panel
position="top-left"
class="!m-3 px-2 py-1 rounded-md bg-surface border border-gray-200 dark:border-gray-700 shadow-sm"
>
<Toggle bind:checked={assetsOnly} size="xs" options={{ right: 'Assets only' }} />
</Panel>
{/if}
{#if showMinimap}
<!-- Node hues mirror the canvas: blue asset cards, amber triggers,
bordered neutral script cards. Visible strokes + rounded corners +
@@ -25,6 +25,9 @@
import type { PipelineOutputKind } from './pipelineTemplates'
import type { DbtAssetProvenance } from './types'
import DbtIcon from '$lib/components/icons/DbtIcon.svelte'
import LanguageIcon from '$lib/components/common/languageIcons/LanguageIcon.svelte'
import RunStateChip from './RunStateChip.svelte'
import type { RunnableRunState } from './activeRunnables.svelte'
import { useOperatingWorkspace } from '$lib/components/operatingWorkspace.svelte'
const operatingWorkspace = useOperatingWorkspace()
@@ -33,6 +36,19 @@
// `content` / `language` so the page-level run handler can dispatch to
// `runScriptPreview` instead of `runScriptByPath` (which 404s for
// non-deployed scripts).
export type AssetUpstreamChips =
| { none: true }
| { multiple: true }
| {
path: string
summary?: string
language?: ScriptLang
unsaved?: boolean
runState?: RunnableRunState
trigger: { label: string; missing?: boolean; draft?: boolean }
onOpen: () => void
}
export type AssetProducer = {
kind: 'script' | 'flow'
path: string
@@ -102,6 +118,11 @@
// instead of only settling once the job ends.
runStatus?: 'running' | 'materialized' | 'failed'
runRowCount?: number | null
/** Set in the assets-only view, where the producing script and its
* trigger are folded into the asset: `none` for an asset nothing in the
* pipeline writes, `multiple` when the producer or its trigger is
* ambiguous. */
upstream?: AssetUpstreamChips
}
// SvelteFlow injects this on the node component when the user clicks
// the node. Combined with our own `hovered` state to drive the
@@ -157,6 +178,8 @@
}
let showAdd = $derived(data.onAddScript != undefined)
let upstream = $derived(data.upstream)
let producer = $derived(upstream && 'path' in upstream ? upstream : undefined)
// dbt badge. `materialized` is dbt's own word rather than the Windmill
// strategy because `view` and `ephemeral` have no strategy, and showing the
@@ -224,6 +247,31 @@
)
</script>
{#snippet upstreamRow()}
{#if upstream}
<span
class={twMerge(
'truncate rounded-md border px-1.5 py-0.5 text-3xs leading-none',
producer?.trigger.missing
? 'border-red-300 dark:border-red-600 text-red-700 dark:text-red-300'
: 'border-gray-300 dark:border-gray-600 text-secondary',
producer?.trigger.draft && 'border-dashed'
)}
title={producer?.trigger.missing
? `No ${producer.trigger.label} trigger targets ${producer.path} yet`
: undefined}
>
{#if 'none' in upstream}
External source
{:else if producer}
{producer.trigger.label}{producer.trigger.draft ? ' · draft' : ''}
{:else}
Multiple upstream nodes
{/if}
</span>
{/if}
{/snippet}
<!-- onmouseenter/leave on the wrapper (not the inner card) so the run
button — which floats outside the card — keeps the hover state alive
when the cursor moves between the card and the button. -->
@@ -246,6 +294,7 @@
tooltip={data.error ? `${data.path}: ${data.error}` : data.path}
{selected}
tone={data.error ? 'error' : undefined}
footer={upstream ? upstreamRow : undefined}
>
{#snippet icon()}
<!-- Data identity carries the accent (luminance blue), pairing with
@@ -358,6 +407,29 @@
<History size={12} />
</span>
{/if}
{#if producer}
{#if producer.runState}
<RunStateChip runState={producer.runState} class="mr-1" />
{/if}
<button
type="button"
class="shrink-0 h-5 min-w-5 px-1 grid place-items-center rounded-md border bg-surface text-secondary hover:bg-surface-hover {producer.unsaved
? 'border-dashed border-gray-400 dark:border-gray-500'
: 'border-gray-300 dark:border-gray-600'}"
title={`Open ${producer.summary ? `${producer.summary} (${producer.path})` : producer.path}${producer.unsaved ? ' · draft' : ''}`}
aria-label="Open the script that produces this asset"
onclick={(e) => {
e.stopPropagation()
producer?.onOpen()
}}
>
{#if producer.language}
<LanguageIcon lang={producer.language} width={12} height={12} />
{:else}
<Code2 size={12} />
{/if}
</button>
{/if}
{/snippet}
</PipelineNodeCard>
{#if showGuardBadge}
@@ -25,6 +25,8 @@
if (!nodesInitialized.current) return
if (!nodes.some((n) => n.type !== 'add')) return
fittedFor = fitKey
void fitView({ padding: 0.1, maxZoom: 1 })
// A re-fit over nodes that keep their ids finds them already measured, one
// render before their new positions reach the flow.
requestAnimationFrame(() => void fitView({ padding: 0.1, maxZoom: 1 }))
})
</script>
@@ -505,6 +505,7 @@
{panToNodeId}
showMinimap={!stacked}
viewportFitKey={viewportFitKey ?? folder}
assetsOnlyToggle
/>
{#if boundBar}{@render boundBar()}{/if}
{#if mode === 'edit'}
@@ -6,7 +6,7 @@ and the controls hung around the card.
-->
<script lang="ts">
import { NODE } from '$lib/components/graph/util'
import { PIPELINE_NODE_HEIGHT } from './assetGraphLayout'
import { PIPELINE_NODE_EXTRA_ROW, PIPELINE_NODE_HEIGHT } from './assetGraphLayout'
import { twMerge } from 'tailwind-merge'
import type { Snippet } from 'svelte'
@@ -29,6 +29,8 @@ and the controls hung around the card.
/** Between the icon and the text: a live status the eye should find first. */
leading?: Snippet
trailing?: Snippet
/** A second row under the title, as wide as the card. */
footer?: Snippet
}
let {
@@ -42,12 +44,13 @@ and the controls hung around the card.
tone,
onclick,
leading,
trailing
trailing,
footer
}: Props = $props()
const cardClass = $derived(
twMerge(
'flex items-center w-full text-left font-normal rounded-md drop-shadow-sm overflow-hidden border transition-colors',
'flex flex-col justify-center w-full text-left font-normal rounded-md drop-shadow-sm overflow-hidden border transition-colors',
surface === 'primary' ? 'bg-surface' : 'bg-surface-secondary',
'border-gray-400 dark:border-gray-600 hover:border-gray-500 dark:hover:border-gray-500',
draft && 'border-dashed border-gray-400 dark:border-gray-500',
@@ -66,6 +69,7 @@ and the controls hung around the card.
'bg-amber-50 dark:bg-amber-900/30 border-amber-400 dark:border-amber-600 animate-pulse'
)
)
const minHeight = $derived(PIPELINE_NODE_HEIGHT + (footer ? PIPELINE_NODE_EXTRA_ROW : 0))
const toneText = $derived(
tone === 'danger'
? 'text-red-700 dark:text-red-400'
@@ -78,6 +82,13 @@ and the controls hung around the card.
</script>
{#snippet body()}
<span class="flex items-center w-full">{@render row()}</span>
{#if footer}
<span class="flex items-center gap-1 w-full min-w-0 pl-3 pr-2 pb-1.5">{@render footer()}</span>
{/if}
{/snippet}
{#snippet row()}
<span class="shrink-0 ml-3 mr-2.5 flex items-center">{@render icon()}</span>
{@render leading?.()}
<span class="flex flex-col min-w-0 flex-1 pr-1 py-1 leading-tight">
@@ -96,7 +107,7 @@ and the controls hung around the card.
type="button"
{onclick}
class={cardClass}
style="width: {NODE.width}px; min-height: {PIPELINE_NODE_HEIGHT}px;"
style="width: {NODE.width}px; min-height: {minHeight}px;"
title={tooltip}
>
{@render body()}
@@ -104,7 +115,7 @@ and the controls hung around the card.
{:else}
<div
class={cardClass}
style="width: {NODE.width}px; min-height: {PIPELINE_NODE_HEIGHT}px;"
style="width: {NODE.width}px; min-height: {minHeight}px;"
title={tooltip}
>
{@render body()}
@@ -0,0 +1,37 @@
<script lang="ts">
import { twMerge } from 'tailwind-merge'
import { CheckCircle2, Loader2, XCircle } from 'lucide-svelte'
import type { RunnableRunState } from './activeRunnables.svelte'
let { runState: rs, class: className }: { runState: RunnableRunState; class?: string } = $props()
</script>
<div
class={twMerge(
'shrink-0 flex items-center gap-0.5 px-1 py-0.5 rounded-sm',
rs.status === 'running'
? 'bg-blue-100 dark:bg-blue-900/40 text-blue-700 dark:text-blue-300'
: rs.status === 'success'
? 'bg-emerald-100 dark:bg-emerald-900/40 text-emerald-700 dark:text-emerald-300'
: 'bg-red-100 dark:bg-red-900/40 text-red-700 dark:text-red-300',
className
)}
title={`${
rs.status === 'running'
? 'Running now'
: rs.status === 'success'
? 'Last run succeeded'
: 'Last run failed'
}${rs.runs > 0 ? ` — ran ${rs.runs}× this session` : ''}`}
>
{#if rs.status === 'running'}
<Loader2 size={10} class="animate-spin" />
{:else if rs.status === 'success'}
<CheckCircle2 size={10} />
{:else}
<XCircle size={10} />
{/if}
{#if rs.runs > 0}
<span class="text-3xs leading-none tabular-nums">×{rs.runs}</span>
{/if}
</div>
@@ -1,7 +1,6 @@
<script lang="ts">
import { Handle, Position } from '@xyflow/svelte'
import {
CheckCircle2,
ChevronDown,
Code2,
EllipsisVertical,
@@ -15,7 +14,6 @@
Target,
Timer,
Trash2,
XCircle,
Zap
} from 'lucide-svelte'
import DbtIcon from '$lib/components/icons/DbtIcon.svelte'
@@ -26,6 +24,7 @@
import { parseDurationSecs } from './parsePipelineAnnotations'
import LanguageIcon from '$lib/components/common/languageIcons/LanguageIcon.svelte'
import PipelineNodeCard from './PipelineNodeCard.svelte'
import RunStateChip from './RunStateChip.svelte'
import type { ScriptLang } from '$lib/gen'
import DropdownV2 from '$lib/components/DropdownV2.svelte'
import Popover from '$lib/components/meltComponents/Popover.svelte'
@@ -315,35 +314,7 @@
</div>
{/if}
{#if data.runState}
{@const rs = data.runState}
<div
class={twMerge(
'shrink-0 flex items-center gap-0.5 px-1 py-0.5 mr-1 rounded-sm',
rs.status === 'running'
? 'bg-blue-100 dark:bg-blue-900/40 text-blue-700 dark:text-blue-300'
: rs.status === 'success'
? 'bg-emerald-100 dark:bg-emerald-900/40 text-emerald-700 dark:text-emerald-300'
: 'bg-red-100 dark:bg-red-900/40 text-red-700 dark:text-red-300'
)}
title={`${
rs.status === 'running'
? 'Running now'
: rs.status === 'success'
? 'Last run succeeded'
: 'Last run failed'
}${rs.runs > 0 ? ` — ran ${rs.runs}× this session` : ''}`}
>
{#if rs.status === 'running'}
<Loader2 size={10} class="animate-spin" />
{:else if rs.status === 'success'}
<CheckCircle2 size={10} />
{:else}
<XCircle size={10} />
{/if}
{#if rs.runs > 0}
<span class="text-3xs leading-none tabular-nums">×{rs.runs}</span>
{/if}
</div>
<RunStateChip runState={data.runState} class="mr-1" />
{/if}
{/snippet}
</PipelineNodeCard>
@@ -18,7 +18,8 @@ const SIBLING_GAP = NODE.gap.horizontal
// into each other's columns — but only ~2×, so they don't drift far apart.
const COMPONENT_GAP = SIBLING_GAP * 2
const LAYER_H = NODE_HEIGHT + LAYER_GAP
/** Extra height of an assets-only node: the row of upstream chips under the title. */
export const PIPELINE_NODE_EXTRA_ROW = 24
interface GraphInput {
nodes: Array<{ id: string; data: AssetGraphNodeData }>
@@ -60,8 +61,10 @@ interface Band {
// handled by dropping feedback edges (see the Kahn step below).
function layoutComponent(
nodes: GraphInput['nodes'],
edges: GraphInput['edges']
edges: GraphInput['edges'],
rowH: number
): Map<string, Positioned> {
const layerH = rowH + LAYER_GAP
const out = new Map<string, Positioned>()
const ids = new Set(nodes.map((n) => n.id))
@@ -155,14 +158,14 @@ function layoutComponent(
if (l > hi) hi = l
for (const k of treeChildren.get(cur)!) stack.push(k)
}
return { top: lo * LAYER_H, bottom: hi * LAYER_H + NODE_HEIGHT }
return { top: lo * layerH, bottom: hi * layerH + rowH }
}
// Recursive placement: node centered over its band, children packed
// side-by-side and centered within it.
function placeTree(id: string, left: number) {
const w = W.get(id)!
out.set(id, { x: left + w / 2, y: layer.get(id)! * LAYER_H })
out.set(id, { x: left + w / 2, y: layer.get(id)! * layerH })
const kids = treeChildren.get(id)!
if (kids.length === 0) return
const kidsW = kids.reduce((acc, k) => acc + W.get(k)!, 0) + SIBLING_GAP * (kids.length - 1)
@@ -246,7 +249,13 @@ function layoutComponent(
//
// Falls back to a stable grid if the component layout throws (defensive —
// cycles are already absorbed by feedback-edge dropping in layoutComponent).
export function layoutAssetGraph(graph: GraphInput, anchorId?: string): Map<string, Positioned> {
export function layoutAssetGraph(
graph: GraphInput,
anchorId?: string,
extraRowHeight = 0
): Map<string, Positioned> {
const rowH = NODE_HEIGHT + extraRowHeight
const layerH = rowH + LAYER_GAP
const byId = new Map<string, Positioned>()
if (graph.nodes.length === 0) return byId
@@ -302,7 +311,7 @@ export function layoutAssetGraph(graph: GraphInput, anchorId?: string): Map<stri
// since positions are node centers) plus the gutter.
let xOffset = 0
for (let c = 0; c < nComp; c++) {
const positions = layoutComponent(compNodes[c], compEdges[c])
const positions = layoutComponent(compNodes[c], compEdges[c], rowH)
let maxX = 0
for (const p of positions.values()) if (p.x > maxX) maxX = p.x
for (const [id, p] of positions) {
@@ -327,7 +336,7 @@ export function layoutAssetGraph(graph: GraphInput, anchorId?: string): Map<stri
if (p.x > maxX) maxX = p.x
if (p.y < minY) minY = p.y
}
byId.set(anchorId, { x: (minX + maxX) / 2, y: minY - LAYER_H })
byId.set(anchorId, { x: (minX + maxX) / 2, y: minY - layerH })
let nMinY = Infinity
for (const p of byId.values()) if (p.y < nMinY) nMinY = p.y
for (const p of byId.values()) p.y -= nMinY
@@ -339,7 +348,7 @@ export function layoutAssetGraph(graph: GraphInput, anchorId?: string): Map<stri
graph.nodes.forEach((n, i) => {
byId.set(n.id, {
x: (i % cols) * (NODE_WIDTH + SIBLING_GAP),
y: Math.floor(i / cols) * LAYER_H
y: Math.floor(i / cols) * layerH
})
})
return byId
@@ -0,0 +1,56 @@
import { describe, expect, it } from 'vitest'
import { assetsOnlyView } from './assetsOnlyView'
const node = (id: string, type: string, data: any = {}) => ({ id, type, data })
const edge = (source: string, target: string, kind: string) => ({
id: `${kind}:${source}->${target}`,
source,
target,
kind
})
describe('assetsOnlyView', () => {
const nodes = [
node('asset:ducklake:raw', 'asset'),
node('asset:ducklake:clean', 'asset'),
node('asset:ducklake:report', 'asset'),
node('script:f/p/clean', 'runnable'),
node('script:f/p/report', 'runnable'),
node('script:f/p/other', 'runnable'),
node('trigger:schedule:f/p/daily', 'trigger', { kind: 'schedule', schedule: '0 0 4 * * *' })
]
const edges = [
edge('trigger:schedule:f/p/daily', 'script:f/p/clean', 'trigger-native'),
edge('asset:ducklake:raw', 'script:f/p/clean', 'lineage-read'),
edge('script:f/p/clean', 'asset:ducklake:clean', 'lineage-write'),
// read-modify-write: no self edge
edge('asset:ducklake:clean', 'script:f/p/clean', 'lineage-read'),
edge('asset:ducklake:clean', 'script:f/p/report', 'trigger-asset'),
edge('asset:ducklake:clean', 'script:f/p/report', 'lineage-read'),
edge('script:f/p/report', 'asset:ducklake:report', 'lineage-write'),
edge('script:f/p/other', 'asset:ducklake:report', 'lineage-write')
]
const v = assetsOnlyView(nodes, edges)
it('joins each script input to its outputs, once', () => {
expect(v.edges.map((e) => `${e.source} -> ${e.target}`).sort()).toEqual([
'asset:ducklake:clean -> asset:ducklake:report',
'asset:ducklake:raw -> asset:ducklake:clean'
])
expect([...v.nodeIds].sort()).toEqual([
'asset:ducklake:clean',
'asset:ducklake:raw',
'asset:ducklake:report'
])
})
it('names the single producer and its trigger, or marks the asset ambiguous', () => {
expect(v.upstream.get('asset:ducklake:raw')).toBeUndefined()
expect(v.upstream.get('asset:ducklake:clean')).toMatchObject({
multiple: false,
runnableId: 'script:f/p/clean',
trigger: { kind: 'schedule', data: { schedule: '0 0 4 * * *' } }
})
expect(v.upstream.get('asset:ducklake:report')).toEqual({ multiple: true })
})
})
@@ -0,0 +1,102 @@
/** A node or edge of the canvas model, reduced to what the assets-only view reads. */
export type ViewNode = { id: string; type: string; data: any }
export type ViewEdge = {
id: string
source: string
target: string
kind: string
unsaved?: boolean
}
/** What an asset shows of the script and trigger that produce it, in the assets-only view. */
export type AssetUpstream =
| { multiple: true }
| {
multiple: false
/** Canvas id of the producing runnable (`script:<path>`). */
runnableId: string
/** The producer's trigger; undefined when nothing triggers it (run by hand). */
trigger?: { nodeId?: string; kind: string; data?: any }
}
/**
* The asset → asset view of a pipeline graph: runnables and triggers disappear,
* and every read (or asset trigger) of a script is joined to each of its writes,
* so an asset points at the assets computed from it. Each produced asset keeps a
* reference to its single producer and that producer's single trigger, or is
* marked `multiple` when either is ambiguous.
*/
export function assetsOnlyView<N extends ViewNode, E extends ViewEdge>(
nodes: N[],
edges: E[]
): {
nodeIds: Set<string>
edges: Array<ViewEdge>
upstream: Map<string, AssetUpstream>
} {
const nodeById = new Map(nodes.map((n) => [n.id, n]))
const inputsOf = new Map<string, E[]>()
const triggersOf = new Map<string, E[]>()
const producersOf = new Map<string, E[]>()
const push = (m: Map<string, E[]>, k: string, e: E) => m.set(k, [...(m.get(k) ?? []), e])
for (const e of edges) {
if (e.kind === 'lineage-write') push(producersOf, e.target, e)
else if (e.kind === 'lineage-read' || e.kind === 'trigger-asset') {
push(inputsOf, e.target, e)
if (e.kind === 'trigger-asset') push(triggersOf, e.target, e)
} else if (e.kind === 'trigger-native') push(triggersOf, e.target, e)
}
const out = new Map<string, ViewEdge>()
for (const [assetId, writes] of producersOf) {
for (const w of writes) {
for (const r of inputsOf.get(w.source) ?? []) {
// A read-modify-write of the same table is not a dependency on itself.
if (r.source === assetId) continue
const id = `flow:${r.source}->${assetId}`
const prev = out.get(id)
const unsaved = !!(w.unsaved || r.unsaved)
if (prev) prev.unsaved = prev.unsaved && unsaved
else out.set(id, { id, source: r.source, target: assetId, kind: 'asset-flow', unsaved })
}
}
}
for (const e of edges) if (e.kind === 'dbt-ref') out.set(e.id, e)
const upstream = new Map<string, AssetUpstream>()
for (const [assetId, writes] of producersOf) {
const runnableIds = [...new Set(writes.map((w) => w.source))]
if (runnableIds.length !== 1) {
upstream.set(assetId, { multiple: true })
continue
}
const runnableId = runnableIds[0]
// Every asset trigger of a script is one cause, "an input changed".
const triggers = [
...new Map(
(triggersOf.get(runnableId) ?? []).map((t) => [
t.kind === 'trigger-asset' ? 'asset' : t.source,
t
])
).values()
]
if (triggers.length > 1) {
upstream.set(assetId, { multiple: true })
continue
}
const t = triggers[0]
const tNode = t && nodeById.get(t.source)
upstream.set(assetId, {
multiple: false,
runnableId,
trigger: !t
? undefined
: t.kind === 'trigger-asset'
? { kind: 'asset' }
: { nodeId: t.source, kind: tNode?.data?.kind, data: tNode?.data }
})
}
const nodeIds = new Set(nodes.filter((n) => n.type === 'asset').map((n) => n.id))
return { nodeIds, edges: [...out.values()], upstream }
}