diff --git a/frontend/src/lib/components/assets/AssetGraph/resolveGraph.test.ts b/frontend/src/lib/components/assets/AssetGraph/resolveGraph.test.ts index 34291bf297..2663c52e28 100644 --- a/frontend/src/lib/components/assets/AssetGraph/resolveGraph.test.ts +++ b/frontend/src/lib/components/assets/AssetGraph/resolveGraph.test.ts @@ -1290,7 +1290,36 @@ describe('pipelineNodeErrors', () => { } ] }) - expect([...pipelineNodeErrors(g, new Map()).untriggered]).toEqual(['f/x/stray']) + expect([...pipelineNodeErrors(g, new Map()).untriggered.keys()]).toEqual(['f/x/stray']) + }) + + it('flags a script only started by reads of assets nothing writes', () => { + const g = baseGraph({ + assets: [{ kind: 'ducklake', path: 'main/raw' }], + runnables: [{ path: 'f/x/reader', usage_kind: 'script', in_pipeline: true }] as any, + edges: [ + { + runnable_path: 'f/x/reader', + runnable_kind: 'script', + asset_kind: 'ducklake', + asset_path: 'main/raw', + access_type: 'r' + } + ], + triggers: [ + { + trigger_kind: 'asset', + asset_kind: 'ducklake', + asset_path: 'main/raw', + runnable_kind: 'script', + runnable_path: 'f/x/reader' + } + ] + }) + expect(pipelineNodeErrors(g, new Map()).untriggered.get('f/x/reader')).toBe('unwritten-inputs') + expect( + pipelineNodeErrors(g, new Map(), new Set(['ducklake:main/raw'])).untriggered.size + ).toBe(0) }) it('clears a subscription the workspace writes outside the pipeline', () => { diff --git a/frontend/src/lib/components/assets/AssetGraph/resolveGraph.ts b/frontend/src/lib/components/assets/AssetGraph/resolveGraph.ts index c9ec0530bf..8c48e4d0fa 100644 --- a/frontend/src/lib/components/assets/AssetGraph/resolveGraph.ts +++ b/frontend/src/lib/components/assets/AssetGraph/resolveGraph.ts @@ -327,8 +327,12 @@ export type PipelineNodeErrors = { unwritten: Map /** Script path → the unwritten asset it runs on (its first, when several). */ waitsOn: Map - /** Pipeline scripts nothing starts: no trigger of any kind. */ - untriggered: Set + /** Pipeline scripts nothing starts → why: no trigger at all, or only asset + * triggers on assets nothing writes (an auto-triggering read included). */ + untriggered: Map + /** Assets some asset trigger waits on that nothing in the pipeline writes: the + * ones to look up across the workspace for `writtenElsewhere`. */ + unwrittenTriggerAssets: Set } /** @@ -385,17 +389,43 @@ export function pipelineNodeErrors( } } } - // A pipeline script nothing starts runs only by hand. dbt projects and macro - // libraries are never started on their own, so they are left out. - const triggered = new Set(graph.triggers.map((t) => t.runnable_path)) - const untriggered = new Set() + // A pipeline script nothing starts runs only by hand. An asset trigger starts + // it only if something writes that asset, which covers the automatic trigger + // of a ducklake/s3 read too. dbt projects and macro libraries are never + // started on their own, so they are left out. + const assetKey = (t: { asset_kind: AssetKind; asset_path: string }) => + `${t.asset_kind}:${t.asset_path}` + const unwrittenTriggerAssets = new Set() + const hasTrigger = new Set() + const started = new Set() + for (const t of graph.triggers) { + hasTrigger.add(t.runnable_path) + if (t.trigger_kind !== 'asset') started.add(t.runnable_path) + else if (written.has(assetKey(t))) started.add(t.runnable_path) + else unwrittenTriggerAssets.add(assetKey(t)) + } + for (const k of unwritten.keys()) unwrittenTriggerAssets.add(k) + const untriggered: PipelineNodeErrors['untriggered'] = new Map() for (const r of graph.runnables) { if (r.usage_kind !== 'script' || !r.in_pipeline || r.dbt || r.macros?.length) continue - if (triggered.has(r.path) || scriptErrors.has(r.path)) continue - untriggered.add(r.path) - scriptErrors.set(r.path, 'nothing starts it: it has no trigger') + if (started.has(r.path) || scriptErrors.has(r.path)) continue + const why = hasTrigger.has(r.path) ? 'unwritten-inputs' : 'no-trigger' + untriggered.set(r.path, why) + scriptErrors.set( + r.path, + why === 'no-trigger' + ? 'nothing starts it: it has no trigger' + : 'nothing starts it: nothing writes the assets it runs after' + ) + } + return { + assets: errors, + scripts: scriptErrors, + unwritten, + waitsOn, + untriggered, + unwrittenTriggerAssets } - return { assets: errors, scripts: scriptErrors, unwritten, waitsOn, untriggered } } function referencedAssetKeys( diff --git a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte index 9351026a7e..e2773cd516 100644 --- a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte @@ -1408,13 +1408,15 @@ ] }) } - for (const path of nodeErrors.untriggered) { + for (const [path, why] of nodeErrors.untriggered) { const unsaved = displayGraph.runnables.some((r) => r.path === path && r.unsaved) fixes.set(`script:${path}`, { explainer: [ 'Nothing starts ', scriptLink(path), - ': it has no trigger, so it only runs by hand.' + why === 'no-trigger' + ? ': it has no trigger, so it only runs by hand.' + : ': it only runs after writes to tables it reads, and nothing writes them, so it only runs by hand.' ], actions: [ { @@ -1700,10 +1702,13 @@ } return out }) - // Subscribed assets nothing in the pipeline writes, checked against the whole - // workspace: an asset trigger fires on a write from anywhere. + // Assets an asset trigger waits on (explicit `on`, or an auto-triggering read) + // that nothing in the pipeline writes, checked against the whole workspace: an + // asset trigger fires on a write from anywhere. let unwrittenInPipeline = $derived( - [...pipelineNodeErrors(shownGraph, explicitOnByPath).unwritten.keys()].sort().join('\n') + [...pipelineNodeErrors(shownGraph, explicitOnByPath).unwrittenTriggerAssets] + .sort() + .join('\n') ) let writtenElsewhere = resource( [() => $workspaceStore, () => unwrittenInPipeline],