diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte index 77f9ff516a..951fa573d8 100644 --- a/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte @@ -230,13 +230,11 @@ nodeFixes?: ReadonlyMap /** "Archive" for someone who can only archive the script the delete removes. */ assetDeleteVerb?: 'Delete' | 'Archive' - /** Makes a script run after each write to an asset it reads, or stop. */ + /** Makes scripts run after each write to an asset they read, or stop. `reads` + * decides whether stopping also mutes the asset. */ onEdgeAction?: (a: { action: 'listen' | 'stop' - script: string - asset: { kind: AssetKind; path: string } - /** Whether the script reads the asset, which decides whether stopping mutes it. */ - reads?: boolean + targets: Array<{ script: string; asset: { kind: AssetKind; path: string }; reads: boolean }> }) => void } let { @@ -305,29 +303,51 @@ } // What the hovered edge lets you change: a plain read can become a trigger, and // an asset trigger can be dropped. Both rewrite the script's `// on` header. - let hoveredEdgeAction = $derived.by(() => { + // What the hovered edge lets you change: a script that only reads the upstream + // asset can rerun on each write to it, and one it triggers can stop. On an + // assets-only edge, those are the scripts folded into it; both rewrite `// on`. + type EdgeTarget = { script: string; asset: { kind: AssetKind; path: string }; reads: boolean } + let hoveredEdgeActions = $derived.by(() => { const e = hoveredEdge && onEdgeAction && view.edges.find((x) => x.id === hoveredEdge!.id) - if (!e || (e.kind !== 'lineage-read' && e.kind !== 'trigger-asset')) return undefined - const asset = view.nodes.find((n) => n.id === e.source)?.data - const script = view.nodes.find((n) => n.id === e.target)?.data - if (!asset?.asset_kind || script?.runnable_kind !== 'script' || !script.in_pipeline) - return undefined - const target = { kind: asset.asset_kind as AssetKind, path: asset.path as string } - const reads = view.edges.some( - (x) => x.kind === 'lineage-read' && x.source === e.source && x.target === e.target - ) - return e.kind === 'lineage-read' - ? { - label: 'Rerun on each write', - detail: `${listenChangeText(target)} in ${script.path}; only its annotation header changes. Saved as a draft.`, - run: () => onEdgeAction?.({ action: 'listen', script: script.path, asset: target }) - } - : { - label: 'Stop rerunning on writes', - detail: `${stopListeningChangeText(target, reads)} in ${script.path}; only its annotation header changes. Saved as a draft.`, - run: () => - onEdgeAction?.({ action: 'stop', script: script.path, asset: target, reads }) - } + if (!e) return [] + const assetData = view.nodes.find((n) => n.id === e.source)?.data + if (!assetData?.asset_kind) return [] + const asset = { kind: assetData.asset_kind as AssetKind, path: assetData.path as string } + const scriptIds = + e.kind === 'asset-flow' + ? (e.via ?? []) + : e.kind === 'lineage-read' || e.kind === 'trigger-asset' + ? [e.target] + : [] + const listen: EdgeTarget[] = [] + const stop: EdgeTarget[] = [] + for (const id of scriptIds) { + const script = model.nodes.find((n) => n.id === id)?.data + if (script?.runnable_kind !== 'script' || !script.in_pipeline) continue + const has = (kind: string) => + model.edges.some((x) => x.kind === kind && x.source === e.source && x.target === id) + const target = { script: script.path as string, asset, reads: has('lineage-read') } + if (has('trigger-asset')) stop.push(target) + else if (target.reads) listen.push(target) + } + const where = (ts: EdgeTarget[]) => + ts.length === 1 ? ts[0].script : `${ts.length} scripts: ${ts.map((t) => t.script).join(', ')}` + const out: Array<{ label: string; detail: string; run: () => void }> = [] + if (listen.length) { + out.push({ + label: 'Rerun on each write', + detail: `${listenChangeText(asset)} in ${where(listen)}; only the annotation header changes. Saved as a draft.`, + run: () => onEdgeAction?.({ action: 'listen', targets: listen }) + }) + } + if (stop.length) { + out.push({ + label: 'Stop rerunning on writes', + detail: `${stopListeningChangeText(asset, stop.some((t) => t.reads))} in ${where(stop)}; only the annotation header changes. Saved as a draft.`, + run: () => onEdgeAction?.({ action: 'stop', targets: stop }) + }) + } + return out }) let hoveredEdgeText = $derived.by(() => { const e = hoveredEdge && view.edges.find((x) => x.id === hoveredEdge!.id) @@ -1194,10 +1214,19 @@ : 0 }) : undefined + // The scripts and triggers folded into this node, whose errors it reports. + const scriptIds = u ? (u.multiple ? u.runnableIds : [u.runnableId]) : [] + const foldedIds = [ + ...scriptIds, + ...m.edges + .filter((e) => e.kind === 'trigger-native' && scriptIds.includes(e.target)) + .map((e) => e.source) + ] return { ...data, upstream, width, + foldedIds, ...(toDelete ? { deleteCount: toDelete.trigger ? 2 : 1, @@ -1391,7 +1420,14 @@ } } : n.data - const data = fix ? { ...withUpstream, fix } : withUpstream + const foldedFixes = ((n.data.foldedIds ?? []) as string[]) + .map((id) => toFix(nodeFixes?.get(id))) + .filter((f) => f != undefined) + const data = { + ...withUpstream, + ...(fix ? { fix } : {}), + ...(foldedFixes.length ? { foldedFixes } : {}) + } return { id: n.id, type: n.type, @@ -1841,7 +1877,7 @@ {#if hoveredEdge && hoveredEdgeText} {@const d = hoveredEdgeText} - {@const action = hoveredEdgeAction} + {@const actions = hoveredEdgeActions}
{d.object}{/if} {#if d.note}
{d.note}
{/if} - {#if action} + {#each actions as action (action.label)}
{action.detail}
- {/if} + {/each}
{/if} diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetNode.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetNode.svelte index a7bebf7c6c..2c7a0bdbc9 100644 --- a/frontend/src/lib/components/assets/AssetGraph/AssetNode.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/AssetNode.svelte @@ -152,6 +152,8 @@ upstream?: AssetUpstreamChips /** What is wrong with the node, and how to fix it: replaces the "+". */ fix?: NodeFix + /** Assets-only: errors of the scripts and triggers folded into this node. */ + foldedFixes?: NodeFix[] /** The assets-only node for the scripts that build no asset. */ noAsset?: boolean /** Card width in the assets-only view, sized to its text. */ @@ -355,7 +357,14 @@ {/snippet} {#snippet upstreamRow()} - {#if upstream} + {#if data.foldedFixes?.length} + + + {:else if upstream} {@const trigger = producer?.trigger} {@const chipClass = twMerge( 'truncate rounded-md border px-1.5 py-0.5 text-3xs leading-none font-normal', diff --git a/frontend/src/lib/components/assets/AssetGraph/NodeFixButton.svelte b/frontend/src/lib/components/assets/AssetGraph/NodeFixButton.svelte index 7520e4c51a..89112eca00 100644 --- a/frontend/src/lib/components/assets/AssetGraph/NodeFixButton.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/NodeFixButton.svelte @@ -39,7 +39,16 @@ red fill already says something is wrong. import { Button } from '$lib/components/common' import { Wrench } from 'lucide-svelte' - let { fix }: { fix: NodeFix } = $props() + let { + fix, + label = 'Fix', + chip = false + }: { + fix: NodeFix + label?: string + /** A small chip in a node's text, rather than the pill on its edge. */ + chip?: boolean + } = $props() let open = $state(false) let anchorEl: HTMLElement | undefined = $state() @@ -60,12 +69,21 @@ red fill already says something is wrong. contentClasses="p-3 w-96" > {#snippet trigger()} - - - Fix - + {#if chip} + + + {label} + + {:else} + + + {label} + + {/if} {/snippet} {#snippet content()}
diff --git a/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte b/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte index 293a40b83e..4370b27dc1 100644 --- a/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte @@ -181,9 +181,7 @@ assetDeleteVerb?: 'Delete' | 'Archive' onEdgeAction?: (a: { action: 'listen' | 'stop' - script: string - asset: { kind: AssetKind; path: string } - reads?: boolean + targets: Array<{ script: string; asset: { kind: AssetKind; path: string }; reads: boolean }> }) => void onRunProducer?: (producer: RunProducer) => Promise idlePane?: Snippet diff --git a/frontend/src/lib/components/assets/AssetGraph/PipelineInsertMenu.svelte b/frontend/src/lib/components/assets/AssetGraph/PipelineInsertMenu.svelte index 34f4780b7a..6374526e51 100644 --- a/frontend/src/lib/components/assets/AssetGraph/PipelineInsertMenu.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/PipelineInsertMenu.svelte @@ -122,14 +122,20 @@ }) let selected = $state(buildEmptySelected()) let menuOpen = $state(false) + // Opened for a given output: the output is not the user's to pick. + let outputLocked = $state(false) $effect(() => { if (!openSignal) return untrack(() => { + outputLocked = !!presetOutput if (presetOutput) { resetWizard() selected.outputId = presetOutput.kind config.asset = { ...presetOutput } config.tableEdited = true + // Named after what it writes, in the characters a script path allows. + const name = presetOutput.table.replace(/[^\w-]+/g, '_').replace(/^_+|_+$/g, '') + if (name) selected.scriptPath = name } if (scheduleFor) { resetWizard() @@ -206,15 +212,13 @@ // The path grammar of `Path.svelte`: word segments, no empty or trailing ones. const SCRIPT_NAME_RE = /^[\w-]+(\/[\w-]+)*$/ let scriptNameValid = $derived(SCRIPT_NAME_RE.test(selected.scriptPath.trim())) - let assetValid = $derived( - !!config.asset?.store && TABLE_NAME_RE.test(config.asset.table.trim()) - ) + let assetValid = $derived(!!config.asset?.store && TABLE_NAME_RE.test(config.asset.table.trim())) let stepValid = $derived( currentConfigStep === 'schedule' ? config.validCron && - !!config.schedule.trim() && - !config.schedulePathError && - config.advancedDefaults === 'loaded' + !!config.schedule.trim() && + !config.schedulePathError && + config.advancedDefaults === 'loaded' : currentConfigStep === 'asset' ? assetValid : scriptNameValid && !!selected.triggerId && !!selected.language && !!selected.outputId @@ -336,7 +340,14 @@ contentClasses={twMerge( 'p-0 bg-surface overflow-hidden relative transition-height', // The first step's columns (w-56 + w-48 + w-80), held across steps. - singleKind ? 'w-[32rem]' : 'w-[46rem]', + // Without the output column when the output is preset (w-80 fewer). + singleKind + ? outputLocked + ? 'w-[12rem]' + : 'w-[32rem]' + : outputLocked + ? 'w-[26rem]' + : 'w-[46rem]', currentConfigStep === 'schedule' ? 'h-[27rem]' : currentConfigStep === 'asset' @@ -364,23 +375,23 @@ {/snippet} {#snippet content({ close })}
- {#if currentConfigStep} - {@render configStepSection(currentConfigStep, close)} - {:else} -
-
- {@render topSection()} + {#if currentConfigStep} + {@render configStepSection(currentConfigStep, close)} + {:else} +
+
+ {@render topSection()} +
+
+ {@render bottomSection(close)} +
-
- {@render bottomSection(close)} -
-
- {/if} + {/if}
{/snippet} @@ -437,7 +448,9 @@ 'flex flex-col gap-1 p-2 overflow-auto transition-opacity w-48', selected.triggerId ? '' : 'opacity-20' )} - {@attach arrowTabNav({ onKeyDown: selectAndAdvanceTo(() => outputEl) })} + {@attach arrowTabNav({ + onKeyDown: selectAndAdvanceTo(() => (outputLocked ? pathEl : outputEl)) + })} >
Language
{#each languages as l} @@ -447,6 +460,9 @@ unifiedSize="sm" btnClasses="justify-start" selected={isSelected} + disabled={outputLocked && + !!selected.outputId && + !compatibleOutputKinds(l.lang).includes(selected.outputId)} onClick={() => { selected.language = l.lang const _compatibleOutputKinds = compatibleOutputKinds(l.lang) @@ -461,34 +477,36 @@ {/each}
-
pathEl, { timeout: 50 }) })} - > -
Output asset
- {#each visibleOutputKinds.length ? visibleOutputKinds : PIPELINE_OUTPUT_KINDS as k} - {@const isSelected = selected.outputId === k.id} - - {/each} -
+ {#if !outputLocked} +
pathEl, { timeout: 50 }) })} + > +
Output asset
+ {#each visibleOutputKinds.length ? visibleOutputKinds : PIPELINE_OUTPUT_KINDS as k} + {@const isSelected = selected.outputId === k.id} + + {/each} +
+ {/if} {/snippet} {#snippet bottomSection(close: () => void)} @@ -555,14 +573,14 @@ >{scheduleFor ? 'Create schedule' : !isLastStep - ? step === 0 - ? 'Configure' - : 'Next' - : hasAiPrompt - ? 'Generate' - : selectedKind?.configuredAfterCreate + ? step === 0 ? 'Configure' - : 'Create'} {/snippet} @@ -613,12 +631,12 @@ Saved as a draft schedule and deployed along with the pipeline.
- +
diff --git a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte index b2639707d2..8cb9f58c67 100644 --- a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte @@ -1339,6 +1339,23 @@ return true } + // Edits several scripts' headers at once, after the user has seen the list: a + // change to one edge or node must not silently rewrite scripts they didn't pick. + let multiEdit = $state<{ scripts: string[]; what: string; run: () => Promise } | undefined>( + undefined + ) + function editScripts( + scripts: string[], + what: string, + apply: (script: string) => (content: string) => string + ) { + const run = async () => { + for (const s of scripts) await editScriptHeader(s, apply(s), what) + } + if (scripts.length > 1) multiEdit = { scripts, what, run } + else void run() + } + const scriptName = (p: string) => p.split('/').pop() ?? p const isTableKind = (k: AssetKind): k is 'ducklake' | 'datatable' => k === 'ducklake' || k === 'datatable' @@ -1412,15 +1429,12 @@ actions.push({ label: one ? `Stop ${scriptName(one)} running on writes` : 'Stop them running on writes', detail: `${stopListeningChangeText(asset, reads)} in ${subscribers.join(', ')}; only the annotation header changes. Saved as a draft.`, - run: async () => { - for (const s of subscribers) { - await editScriptHeader( - s, - (c) => stopListeningToAsset(c, asset, readsAsset(s, asset)), - 'Stopped running on writes' - ) - } - } + run: () => + editScripts( + subscribers, + 'Stopped running on writes', + (s) => (c) => stopListeningToAsset(c, asset, readsAsset(s, asset)) + ) }) return actions } @@ -3218,15 +3232,29 @@ assetDeleteVerb={canHardDeleteScripts ? 'Delete' : 'Archive'} onEdgeAction={isOperator ? undefined - : ({ action, script, asset, reads }) => - editScriptHeader( - script, - (c) => - action === 'listen' - ? listenToAsset(c, asset) - : stopListeningToAsset(c, asset, reads ?? false), - action === 'listen' ? 'Now reruns on each write' : 'Stopped rerunning on writes' - )} + : ({ + action, + targets + }: { + action: 'listen' | 'stop' + targets: Array<{ + script: string + asset: { kind: AssetKind; path: string } + reads: boolean + }> + }) => { + const byScript = new Map(targets.map((t) => [t.script, t] as const)) + editScripts( + [...byScript.keys()], + action === 'listen' ? 'Now reruns on each write' : 'Stopped rerunning on writes', + (s) => (c) => { + const t = byScript.get(s)! + return action === 'listen' + ? listenToAsset(c, t.asset) + : stopListeningToAsset(c, t.asset, t.reads) + } + ) + }} onRunProducer={mode === 'edit' ? handleRunProducer : undefined} onRequestEdit={isOperator ? undefined : () => setMode('edit')} canRunByPath={openScriptHasDataUpload} @@ -3356,6 +3384,30 @@ + { + const m = multiEdit + multiEdit = undefined + void m?.run() + }} + onCanceled={() => (multiEdit = undefined)} +> + {#if multiEdit} +
+

This will edit the annotation header of the following scripts. Each is saved as a draft.

+
    + {#each multiEdit.scripts as s (s)} +
  • {s}
    • + {/each} +
+
+ {/if} + +