From 032300e28eba9f8e790f16e894bb00fff22eb296 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sat, 1 Aug 2026 13:58:05 +0200 Subject: [PATCH] feat: run dbt projects as a first-class Windmill runtime (#10326) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix: mount only the engine in the dbt jail, reject shadowed and malformed args Review round 42. The jail mounted the whole dbt cache directory, whose siblings of the engine are `repos/` and `packages/` — other workspaces' private checkouts and package trees, kept apart by cache key rather than by permissions. A jailed project could read them. It now mounts the engine's own directory, which the provisioner names; verified from inside the jail that `repos/`, `packages/` and `state/` are invisible while the engine stays usable. A `{{ placeholder }}` may no longer take the name of a run argument this runtime defines. It was silently dropped from the signature, so a descriptor like `value: "{{ select }}"` deployed and then could not be run at all: the built-in `select` is an array and the interpolation needs a scalar. Refused at parse, so the deploy says so. A `vars` override that is not an object is refused rather than ignored. Argument-schema validation is opt-in, so a string or an array silently ran the descriptor's own vars — against a different schema or alias than the caller asked for. `select` and `exclude` already refused theirs. * feat(dbt): the project is the script's module bundle, not a git checkout A dbt script now carries its whole dbt project as its module bundle. The descriptor is the script content; ` @@ -712,6 +721,10 @@ {/if} +{#if dbtRun} + +{/if} + {#if result_stream && result == undefined}
diff --git a/frontend/src/lib/components/HighlightCode.svelte b/frontend/src/lib/components/HighlightCode.svelte index 73c9a08c4a..bb51851003 100644 --- a/frontend/src/lib/components/HighlightCode.svelte +++ b/frontend/src/lib/components/HighlightCode.svelte @@ -26,7 +26,10 @@ interface Props { code?: string - language: Script['language'] | 'bunnative' | 'frontend' | 'json' | undefined + // `sql` is the dialect-agnostic option: a dbt model's SQL is compiled by + // whichever adapter the project targets, so naming one dialect would be a + // guess. Every dialect below highlights through the same grammar anyway. + language: Script['language'] | 'bunnative' | 'frontend' | 'json' | 'sql' | undefined highlightLanguage?: LanguageType | undefined lines?: boolean className?: string @@ -56,7 +59,9 @@ ? 'opacity-0 group-hover:opacity-100 focus-within:opacity-100 transition-opacity duration-150' : '' - function getLang(lang: Script['language'] | 'bunnative' | 'frontend' | 'json' | undefined) { + function getLang( + lang: Script['language'] | 'bunnative' | 'frontend' | 'json' | 'sql' | undefined + ) { switch (lang) { case 'python3': return python @@ -76,6 +81,8 @@ return javascript case 'graphql': return graphql + case 'sql': + return sql case 'mysql': return sql case 'postgresql': diff --git a/frontend/src/lib/components/ScriptBuilder.svelte b/frontend/src/lib/components/ScriptBuilder.svelte index bd30cd6909..f209907537 100644 --- a/frontend/src/lib/components/ScriptBuilder.svelte +++ b/frontend/src/lib/components/ScriptBuilder.svelte @@ -454,6 +454,8 @@ language: 'bun' } } + } else if (script.language === 'dbt') { + seedDbtProject() } const restarter = scheduleRestartSync(userDraftPath, { waitForContent: true }) initContent(script.language, script.kind, template).finally(() => restarter.markContentReady()) @@ -971,6 +973,34 @@ function handleDeployTrigger(_trigger: Trigger) {} + // A dbt script's modules ARE its dbt project, and the runtime refuses one + // without a `dbt_project.yml`. Seeded from BOTH entry points — the empty-script + // bootstrap and the language picker — because reaching dbt by switching an + // existing draft otherwise produces a script that cannot deploy or run. + // Existing modules are left alone: switching away and back must not discard a + // project the user has already grown. + function seedDbtProject() { + // Keyed on the project file rather than on "has any modules at all": a + // draft that grew modules under another language carries none of what dbt + // needs, and the worker refuses a bundle with no `dbt_project.yml` — so + // that draft reached dbt in a state it could neither deploy nor run. + if (script.modules?.['dbt_project.yml']) return + script.modules = { + 'dbt_project.yml': { + content: + 'name: my_dbt_project\nversion: "1.0"\nprofile: my_dbt_project\nmodels:\n my_dbt_project:\n +materialized: view\n', + language: 'dbt' + }, + 'models/example.sql': { + content: 'select 1 as id\n', + language: 'dbt' + }, + // Last, so anything already written wins: the previous language's + // helper files are inert to dbt and are the user's to remove. + ...(script.modules ?? {}) + } + } + function onScriptLanguageTrigger(lang: 'docker' | 'bunnative' | ScriptLang) { if (lang == 'docker') { template = 'docker' @@ -983,6 +1013,9 @@ // initContent(language, script.kind, template) script.language = language + if (language === 'dbt') { + seedDbtProject() + } } function onSummaryChange(value: string) { diff --git a/frontend/src/lib/components/ScriptEditor.svelte b/frontend/src/lib/components/ScriptEditor.svelte index 9a65fc9d2b..e112f92b3b 100644 --- a/frontend/src/lib/components/ScriptEditor.svelte +++ b/frontend/src/lib/components/ScriptEditor.svelte @@ -19,6 +19,10 @@ import { isWorkflowAsCode } from '$lib/components/graph/wacToFlow' import WacDiagram from '$lib/components/graph/WacDiagram.svelte' import { Pane, Splitpanes } from 'svelte-splitpanes' + import DbtProjectPanel, { + dbtFileLang, + dbtModelSelector + } from '$lib/components/dbt/DbtProjectPanel.svelte' import SchemaForm from './SchemaForm.svelte' import PowerShellCommonParams from './PowerShellCommonParams.svelte' import LogPanel from './scriptEditor/LogPanel.svelte' @@ -354,9 +358,32 @@ editor?.setCode(editorCode) } + // Whether the open file is tested as a runnable of its own. A `__mod` helper + // is; a dbt project's files are not — the run is always the project's, so the + // arguments shown, edited and logged must be the descriptor's, not an empty + // per-module set the request would ignore. + let onModuleArgs = $derived(activeModuleTab !== null && lang !== 'dbt') + + // The selector a Test would build with, when the open file is a model. Macros, + // analyses and singular tests are `.sql` too and none is selectable by name, + // so those fall back to running the project. + let dbtSelected = $derived.by(() => { + const open = activeModuleTab + if (lang !== 'dbt' || !open) return undefined + const selector = dbtModelSelector(modules ?? {}, open) + // The label drops whichever extension the selector matched, so a Python + // model reads `Build my_model` rather than `Build my_model.py`. + const name = open.split('/').pop()!.replace(/\.(sql|py)$/, '') + return selector ? { selector, name } : undefined + }) + let effectiveLang = $derived( activeModuleTab && modules?.[activeModuleTab] - ? (modules[activeModuleTab].language as Preview['language']) + ? lang === 'dbt' + ? // Every dbt module is stored as `dbt`; the extension is what says + // whether this file is SQL, YAML or a seed. + dbtFileLang(activeModuleTab) + : (modules[activeModuleTab].language as Preview['language']) : lang ) @@ -373,7 +400,15 @@ return isTsWac || isPyWac }) let supportsModules = $derived((lang === 'bun' || lang === 'python3') && isWacV2) - let mainFileName = $derived('script.' + langToExt(scriptLangToEditorLang(lang))) + // A dbt script's content is the descriptor and its modules are the project. + // A tree rather than the module tab strip: a project has folders and dozens + // of files, which a strip cannot show. + let isDbt = $derived(lang === 'dbt') + let mainFileName = $derived( + isDbt + ? 'wm_dbt.yaml' + : 'script.' + langToExt(scriptLangToEditorLang(lang)) + ) let modulePathInput = $state('') let showAddModulePopover = $state(false) @@ -428,13 +463,28 @@ bunnative: ['.ts'] } + // A dbt project's files are dbt's own, not Windmill modules: models and tests + // are `.sql`, schemas and the project file `.yml`, seeds `.csv`, docs `.md`. + // `.py` because dbt Python models are first-class on Snowflake, BigQuery and + // Databricks, and the CLI already bundles one; refusing to CREATE one here + // was the only place that restriction existed. + const DBT_MODULE_EXTENSIONS = ['.sql', '.py', '.yml', '.yaml', '.csv', '.md'] let allowedModuleExtensions = $derived( - lang - ? (LANG_MODULE_EXTENSIONS[lang] ?? Object.keys(ALL_MODULE_EXTENSIONS)) - : Object.keys(ALL_MODULE_EXTENSIONS) + lang === 'dbt' + ? DBT_MODULE_EXTENSIONS + : lang + ? (LANG_MODULE_EXTENSIONS[lang] ?? Object.keys(ALL_MODULE_EXTENSIONS)) + : Object.keys(ALL_MODULE_EXTENSIONS) ) function inferModuleLang(filePath: string): ScriptModule['language'] | undefined { + // Every file of a dbt project is stored as `dbt`, whatever its extension: + // they are the project's, and dbt is what reads them. + if (lang === 'dbt') { + return DBT_MODULE_EXTENSIONS.some((e) => filePath.endsWith(e)) + ? ('dbt' as ScriptModule['language']) + : undefined + } for (const [ext, moduleLang] of Object.entries(ALL_MODULE_EXTENSIONS)) { if (filePath.endsWith(ext)) return moduleLang } @@ -442,6 +492,12 @@ } function getModuleDefaultContent(filePath: string): string { + if (lang === 'dbt') { + // A model that compiles on its own, so a new file is runnable before it + // is edited; anything else starts empty rather than with a guess at + // which dbt schema it is. + return filePath.endsWith('.sql') ? `select 1 as id\n` : '' + } if (filePath.endsWith('.py')) { return `def hello() -> str:\n return "world"\n` } else if (filePath.endsWith('.ts')) { @@ -474,8 +530,19 @@ return '' } + /// The descriptor is the script's CONTENT, not a module. A module at that same + /// path would be a second, independent value for one file: the export writes + /// the content there, and the bundle would emit over it. + function reservedDbtPath(path: string): string | undefined { + return lang === 'dbt' && path.trim() === 'wm_dbt.yaml' + ? `wm_dbt.yaml is the descriptor, edited from the tree — it cannot also be a file` + : undefined + } + function validateModulePath(path: string): string { if (!path.trim()) return '' + const reserved = reservedDbtPath(path) + if (reserved) return reserved const moduleLang = inferModuleLang(path) if (!moduleLang) { const exts = allowedModuleExtensions.join(', ') @@ -525,6 +592,8 @@ function validateRenameModulePath(newPath: string, oldPath: string): string { if (!newPath.trim()) return '' + const reserved = reservedDbtPath(newPath) + if (reserved) return reserved const moduleLang = inferModuleLang(newPath) if (!moduleLang) { const exts = allowedModuleExtensions.join(', ') @@ -838,16 +907,30 @@ // Flush module edits back to modules map before running preview flushModuleContent() - const testCode = activeModuleTab !== null ? editorCode : code - const testLang = activeModuleTab !== null ? effectiveLang : lang - const rawTestArgs = - activeModuleTab !== null - ? testPanelArgs - : selectedTab === 'preprocessor' || kind === 'preprocessor' - ? { _ENTRYPOINT_OVERRIDE: 'preprocessor', ...(args ?? {}) } - : (args ?? {}) - const testSchema = activeModuleTab !== null ? testPanelSchema : schema + // A dbt run is always the project's, whichever file is open: `dbt build` + // takes the whole bundle, and testing one model in isolation is not a + // thing dbt does. + const onModule = onModuleArgs + const testCode = onModule ? editorCode : code + const testLang = onModule ? effectiveLang : lang + const rawTestArgs = onModule + ? testPanelArgs + : selectedTab === 'preprocessor' || kind === 'preprocessor' + ? { _ENTRYPOINT_OVERRIDE: 'preprocessor', ...(args ?? {}) } + : (args ?? {}) + const testSchema = onModule ? testPanelSchema : schema const testArgs = await processSecretArgs(rawTestArgs, testSchema, opWs) + // Testing with a model open builds THAT model: `dbt build --select ` + // is dbt's own inner loop, and running the whole project to check one file + // is the thing a dbt developer never does. Its tests come along, because + // `build` interleaves them. + if (dbtSelected) { + testArgs.command = { + ...((testArgs.command as object) ?? {}), + label: 'build', + select: [dbtSelected.selector] + } + } if (showPsCommonParams) { for (const [k, v] of Object.entries(psCommonParams)) { if (v !== undefined && v !== false && v !== '') { @@ -891,7 +974,10 @@ } }, undefined, - activeModuleTab !== null ? undefined : modules, + // A `__mod` helper is tested alone, so its siblings are left out. A dbt + // project cannot be: the bundle IS the project, and without it the run + // finds no `dbt_project.yml` whichever file happens to be open. + onModule ? undefined : modules, undefined, timeout ) @@ -1041,6 +1127,11 @@ async function inferModuleSchema() { if (activeModuleTab === null) return + // A dbt project's files are not independently runnable: a model is SQL dbt + // compiles, not a script with arguments. Inferring some would put another + // language's parameters (a `.sql` model reads as Postgres) in the run form + // beside the descriptor's own. + if (lang === 'dbt') return try { await inferArgs(effectiveLang, editorCode, testPanelSchema) injectPartitionArg(testPanelSchema, testPanelArgs, effectiveLang, editorCode) @@ -2241,7 +2332,7 @@ { if (e.detail) { - if (activeModuleTab !== null) { + if (onModuleArgs) { testPanelArgs = e.detail } else { args = e.detail @@ -2259,7 +2350,7 @@ bind:clientHeight={schemaHeight} > {#key argsRender} - {#if activeModuleTab !== null} + {#if onModuleArgs} - Test + + {dbtSelected ? `Build ${dbtSelected.name}` : 'Test'} {/snippet} @@ -2371,7 +2464,7 @@ previewIsLoading={debugMode ? $debugState.running && !$debugState.stopped : testIsLoading} {editor} {diffEditor} - args={activeModuleTab !== null ? testPanelArgs : args} + args={onModuleArgs ? testPanelArgs : args} {showCaptures} customUi={customUi?.previewPanel} showCustomResultPanel={showDebugPanel} @@ -2505,7 +2598,34 @@ {/snippet} {#snippet editorContent()} -
+
+ {#if isDbt} + (p === null ? switchToMain() : switchToModule(p))} + onDelete={removeModule} + > + {#snippet addFile()} + + {#snippet trigger()} +
+ +
+ {/snippet} + {#snippet content({ close })} + {@render addModuleForm(close)} + {/snippet} +
+ {/snippet} +
+ {/if} {#if supportsModules}
{/if} -
+
{#if assets?.length} diff --git a/frontend/src/lib/components/apps/editor/inlineScriptsPanel/EmptyInlineScript.svelte b/frontend/src/lib/components/apps/editor/inlineScriptsPanel/EmptyInlineScript.svelte index edaefe947a..214ab3e31d 100644 --- a/frontend/src/lib/components/apps/editor/inlineScriptsPanel/EmptyInlineScript.svelte +++ b/frontend/src/lib/components/apps/editor/inlineScriptsPanel/EmptyInlineScript.svelte @@ -7,7 +7,7 @@ import { inferArgs } from '$lib/infer' import { initialCode } from '$lib/script_helpers' import { emptySchema } from '$lib/utils' - import { defaultScriptLanguages, getScriptByPath, processLangs } from '$lib/scripts' + import { defaultScriptLanguages, getScriptByPath, processInlineLangs } from '$lib/scripts' import { Building, GitFork, Globe2 } from 'lucide-svelte' import { createEventDispatcher } from 'svelte' @@ -88,7 +88,7 @@ } let langs = $derived( - processLangs(undefined, $defaultScripts?.order ?? Object.keys(defaultScriptLanguages)) + processInlineLangs(undefined, $defaultScripts?.order ?? Object.keys(defaultScriptLanguages)) .map((l) => [defaultScriptLanguages[l], l]) .filter( (x) => diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte index 3e5e9d8cef..8a893e3e84 100644 --- a/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte @@ -18,10 +18,15 @@ import PanToNode from './PanToNode.svelte' import InitialFitView from './InitialFitView.svelte' import { layoutAssetGraph } from './assetGraphLayout' - import { computeMutedReadKeys } from './resolveGraph' + import { computeMutedReadKeys, dbtAssociations } from './resolveGraph' import { buildDownstreamMap } from './graphTraversal' import { buildLineageDownstreamMap } from './boundedCascade' - import type { AssetGraphResponse, AssetGraphSelection, NativeTriggerKind } from './types' + import type { + AssetGraphResponse, + AssetGraphSelection, + AssetRunState, + NativeTriggerKind + } from './types' import type { RunnableRunState } from './activeRunnables.svelte' import type { AssetKind } from '$lib/gen' import { NODE } from '$lib/components/graph/util' @@ -179,6 +184,11 @@ * When a node's nonce changes, it flashes a fading green background — its * producer just recomputed it. Driven by the replay player frame-by-frame. */ recomputedAssetIds?: ReadonlyMap + /** What a run is currently doing to each relation, keyed `asset::` + * like every other per-asset map here. + * Distinct from `recomputedAssetIds`, which is a one-shot pulse: this is + * the state a node holds until the run moves it. */ + assetRunStatus?: ReadonlyMap /** Let the wheel zoom the canvas (and swallow the page scroll while doing * so). Default true for the full-height editor/player. Set false when the * canvas is embedded inline inside a scrollable container, so a wheel @@ -215,6 +225,7 @@ viewportFitKey = '', highlightActiveRun = false, recomputedAssetIds, + assetRunStatus, scrollZoom = true }: Props = $props() @@ -244,6 +255,7 @@ | 'data-test' | 'macro' | 'test-dependency' + | 'dbt-ref' unsaved?: boolean // Muted read edge: a ducklake/s3 input read every run whose (default) // auto cascade trigger is suppressed by `// mute` / `// mute all`. @@ -287,6 +299,26 @@ // by node id across producers). const addedTestNodes = new Set() + // A dbt script owns every relation its project materializes. Drawing that + // as one edge per model buries the lineage that matters — `ref()` between + // models, and native consumers — under a fan-out that grows with the + // project, so the association is carried by the model's badge and its + // hover/click highlight instead. Only the DRAWING is dropped: the + // producer rows still drive cascade dispatch and "who produced this". + const dbtRunnableIds = new Set( + g.runnables.filter((r) => r.dbt).map((r) => `${r.usage_kind}:${r.path}`) + ) + // Association only — the canvas deliberately draws no edge for it. + const { ownerByAsset: dbtOwnerByAsset, writesByOwner: dbtWritesByOwner } = dbtAssociations( + g.runnables, + g.edges + ) + // Per-relation dbt description, to tell a project's own declared source + // from a relation another script materializes. + const dbtAssetProvenance = new Map( + g.assets.filter((a) => a.dbt).map((a) => [`asset:${a.kind}:${a.path}`, a.dbt!]) + ) + const hasAddNode = onAddPipelineScript != null if (hasAddNode) { nodes.push({ @@ -395,6 +427,7 @@ path: a.path, fork_materialization: a.fork_materialization, derived_from: a.derived_from, + dbt: a.dbt, onAddScript: onAddScriptForAsset, pathPrefix, defaultPathSuffix, @@ -404,7 +437,32 @@ producerFailed, // Bumped by the replay player when this asset's producer just // recomputed it — the node flashes green and fades. - recomputePulse: recomputedAssetIds?.get(assetId) + recomputePulse: recomputedAssetIds?.get(assetId), + // What the run in view is doing to this relation right now. + runStatus: assetRunStatus?.get(assetId)?.status, + runRowCount: assetRunStatus?.get(assetId)?.rowCount, + // The dbt project that materializes this relation, related by + // badge rather than by an edge — so only when that node is on + // this graph. The run page and the pipeline page both hide it, + // and passing handlers anyway makes the chip advertise a click + // that resolves to nothing. + ...(dbtOwnerByAsset.has(assetId) + ? { + onDbtHover: (on: boolean) => (dbtHoverId = on ? assetId : undefined), + onDbtSelect: () => { + // `runnable::` — the id shape `build` uses. + const owner = model.dbtOwnerByAsset.get(assetId) + const [kind, ...rest] = owner?.split(':') ?? [] + if (kind && rest.length) { + onselect?.({ + kind: 'runnable', + runnable_kind: kind as 'script' | 'flow', + path: rest.join(':') + }) + } + } + } + : {}) } }) } @@ -475,6 +533,8 @@ tag: r.tag, retry: r.retry, macros: r.macros, + dbt: r.dbt, + onDbtHover: (on: boolean) => (dbtHoverId = on ? rid : undefined), unsaved: r.unsaved ?? false, // Same dispatch the asset node uses, only routed when the // runnable is a script (the page handler short-circuits @@ -522,10 +582,28 @@ // (`// mute` / `// mute all` opted the default auto trigger out). Gated // on pipeline scripts inside the helper (non-pipeline reads never derive). const mutedReadKeys = computeMutedReadKeys(g.edges, g.triggers, g.runnables) + // A dbt script owns every relation of its project. Drawing that as one + // edge per model buries the lineage that matters (`ref()` between models, + // and native consumers) under a fan-out that grows with the project — so + // the association is carried by the node badge and its hover/click + // highlight instead. Only the DRAWING is dropped: the producer rows still + // drive cascade dispatch and "who produced this". for (const e of g.edges) { const runnableId = `${e.runnable_kind}:${e.runnable_path}` const assetId = `asset:${e.asset_kind}:${e.asset_path}` const access = e.access_type ?? 'r' + // A dbt project's own relations are related by badge, not by edges: its + // writes are the fan-out, and its declared sources already reach its + // models through the `ref()` edges, so both would be noise. + // + // A read of a relation ANOTHER script builds is different — that is how + // two selections of one project compose (decision 6), it carries the + // cascade, and no `ref()` edge survives the split to stand in for it. + // Kept, or the two halves render as disconnected islands. + if (dbtRunnableIds.has(runnableId)) { + const isOwnSource = dbtAssetProvenance.get(assetId)?.resource_type === 'source' + if (access === 'w' || access === 'rw' || isOwnSource) continue + } if (access === 'w' || access === 'rw') { // Data tests assert on the `// materialize` target, which is always // a ducklake asset (v1 enforces this), so only the ducklake @@ -621,6 +699,17 @@ }) } + // dbt `ref()` lineage: model → model inside one project. The dbt script + // writes every one of them, so without these the canvas shows a flat + // fan-out from the script and loses the project's actual shape. + const assetNodeIds = new Set(g.assets.map((a) => `asset:${a.kind}:${a.path}`)) + for (const de of g.dbt_edges ?? []) { + const from = `asset:dbt:${de.from_asset_path}` + const to = `asset:dbt:${de.to_asset_path}` + if (!assetNodeIds.has(from) || !assetNodeIds.has(to)) continue + edges.push({ id: `dbtref:${from}->${to}`, source: from, target: to, kind: 'dbt-ref' }) + } + // 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 @@ -787,11 +876,24 @@ } } - return { nodes, edges } + return { nodes, edges, dbtOwnerByAsset, dbtWritesByOwner } } let model = $derived(build(graph)) + // 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 + // project node, so the association survives the pointer leaving. + let dbtHoverId = $state(undefined) + let dbtEmphasisIds = $derived.by(() => { + if (!dbtHoverId) return new Set() + const owned = model.dbtWritesByOwner.get(dbtHoverId) + if (owned) return new Set([dbtHoverId, ...owned]) + const owner = model.dbtOwnerByAsset.get(dbtHoverId) + return owner ? new Set([dbtHoverId, owner]) : new Set() + }) + let selectedId = $derived.by(() => { if (!selection) return undefined return selection.kind === 'asset' @@ -895,12 +997,13 @@ else if (boundPick.bounded.has(n.id)) boundClass = 'wm-bound-in' else if (!boundPick.eligible.has(n.id)) boundClass = 'wm-bound-dim' } + const dbtClass = dbtEmphasisIds.has(n.id) ? 'wm-dbt-linked' : undefined return { id: n.id, type: n.type, position: { x: p.x + xCenter + xShift, y: p.y + 40 }, data: n.data, - class: boundClass ?? runClass ?? assetClass, + class: boundClass ?? dbtClass ?? runClass ?? assetClass, selected: n.id === selectedId, // All nodes non-draggable: the layout is sugiyama-computed, // dragging would fight the reactive re-layout. Selection is @@ -1064,6 +1167,21 @@ label = 'test needs' labelStyle = 'fill: rgb(217 119 6); font-size: 10px; font-weight: 600;' break + case 'dbt-ref': + // model → model inside one dbt project. Orange, matching the + // dbt badges, and dashed because the edge is dbt's own lineage + // rather than a Windmill read/write the cascade acts on. + style = 'stroke: rgb(234 88 12); stroke-width: 1.25px;' + strokeDasharray = '4 3' + markerColor = 'rgb(234 88 12)' + label = 'ref' + labelStyle = 'fill: rgb(234 88 12); font-size: 10px; font-weight: 600;' + // Same rule the pipeline uses for a running script: the edges + // touching what is happening animate. Here the unit of work is + // the model, so the edges feeding the one dbt is building move, + // and the flow reads in DAG order as it advances. + animated = assetRunStatus?.get(e.target)?.status === 'running' + break default: style = '' } @@ -1254,6 +1372,11 @@ /* Activity-panel emphasis — soft, monochromatic, less prominent than the blue details selection above. Hover is a thin neutral ring (transient); pinning an expanded run is a soft-blue ring. */ + /* A dbt project node and the models it materializes, related by badge + rather than by edges — hovering either lights up the whole set. */ + :global(.svelte-flow__node.wm-dbt-linked .drop-shadow-sm) { + @apply outline outline-2 outline-orange-400/80; + } :global(.svelte-flow__node.wm-run-hover .drop-shadow-sm) { @apply outline outline-1 outline-gray-400 dark:outline-gray-500; } diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte index d48ecbe09e..2c5e5fd97e 100644 --- a/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte @@ -27,7 +27,9 @@ import { inferArgs } from '$lib/infer' import { emptySchema, sendUserToast } from '$lib/utils' import type { Schema } from '$lib/common' - import type { AssetGraphSelection, PipelineMode } from './types' + import type { AssetGraphSelection, DbtAssetProvenance, PipelineMode } from './types' + import HighlightCode from '$lib/components/HighlightCode.svelte' + import DbtIcon from '$lib/components/icons/DbtIcon.svelte' import PipelineScriptView from './PipelineScriptView.svelte' import { parsePipelineAnnotations, @@ -154,6 +156,9 @@ // resolved graph). Drives the transitive column-lineage trace shown for a // selected materialized asset. selectionColumnGraph?: ColumnLineageGraph + /** dbt provenance of the selected relation, when a dbt project + * materializes it — carries the model's own SQL. */ + selectionDbt?: DbtAssetProvenance // Whether the selected ducklake asset's schema can evolve (whole-table // `replace` producer). Forwarded to the Schema tab: version history when // true, a single fixed-schema view when false. Defaults to true (unknown). @@ -284,6 +289,7 @@ onScriptRemoved, selectionProducers = [], selectionColumnGraph, + selectionDbt, schemaCanEvolve = true, selectionForkMaterialization = undefined, schemaContractContext = undefined, @@ -426,6 +432,20 @@ ) ) + // Where the selected model's file sits on disk: the producing script's + // module folder holds the dbt project verbatim, so this is the path a + // `wmill sync pull` writes and the one to edit. + let dbtBundlePath = $derived.by(() => { + const file = selectionDbt?.original_file_path + if (!file) return undefined + // A relation may have several script producers, and nothing here says which + // of them is the dbt project this model came from. Prefixing the wrong one + // names a `__dbt` folder that does not exist, so an ambiguous relation shows + // the path inside the project alone. + const scripts = selectionProducers.filter((p) => p.kind === 'script') + return scripts.length === 1 ? `${scripts[0].path}__dbt/${file}` : file + }) + // Bound from ScriptEditor — populated by inferAssets on every code // change. Forwarded to the page so the canvas can re-derive write // edges as the user edits the body (e.g. renaming a CREATE TABLE @@ -1216,6 +1236,25 @@
{/key} + {:else if selectionDbt?.raw_code} + +
+
+ + {dbtBundlePath ?? selectionDbt.unique_id} + read-only · edit locally +
+
+ +
+
{:else}
No inline preview yet for {selection.asset_kind}. Use the producer/consumer arrows diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetNode.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetNode.svelte index 999559a0c7..815db36b18 100644 --- a/frontend/src/lib/components/assets/AssetGraph/AssetNode.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/AssetNode.svelte @@ -14,13 +14,17 @@ Loader2, Plus, ShieldCheck, - ShieldAlert + ShieldAlert, + CheckCircle2, + XCircle } from 'lucide-svelte' import type { ScriptLang } from '$lib/gen' import { enterpriseLicense, workspaceStore } from '$lib/stores' import { sendUserToast } from '$lib/utils' import { PIPELINE_LANGUAGES } from './pipelineLanguages' import type { PipelineOutputKind } from './pipelineTemplates' + import type { DbtAssetProvenance } from './types' + import DbtIcon from '$lib/components/icons/DbtIcon.svelte' // Shape used for both the data prop and the run callback. Drafts carry // `content` / `language` so the page-level run handler can dispatch to @@ -44,6 +48,15 @@ // "current view of " marker so it reads as a derived node, not an // unrelated table. derived_from?: string + // dbt provenance when this warehouse table is a dbt node: which model + // it is, how dbt materializes it, its tags and its generic tests. + dbt?: DbtAssetProvenance + /** Hovering the dbt chip emphasizes the project node that + * materializes this model — the association the graph deliberately + * does not draw as an edge. */ + onDbtHover?: (on: boolean) => void + /** Clicking it selects that project node. */ + onDbtSelect?: () => void onAddScript?: ( asset: { kind: AssetKind; path: string }, language: ScriptLang, @@ -78,6 +91,11 @@ // producer just recomputed it. A change triggers a one-shot green // fade so a freshly-written table stands out as the run progresses. recomputePulse?: number + // What the run being viewed is doing to this relation. dbt records it + // per model as it walks the DAG, so the graph moves with the run + // instead of only settling once the job ends. + runStatus?: 'running' | 'materialized' | 'failed' + runRowCount?: number | null } // SvelteFlow injects this on the node component when the user clicks // the node. Combined with our own `hovered` state to drive the @@ -133,6 +151,40 @@ let showAdd = $derived(data.onAddScript != undefined) + // dbt badge. `materialized` is dbt's own word rather than the Windmill + // strategy because `view` and `ephemeral` have no strategy, and showing the + // dbt word keeps the node legible to someone reading their own project. + let dbtLabel = $derived(data.dbt?.materialized ?? data.dbt?.resource_type) + let dbtTitle = $derived.by(() => { + const d = data.dbt + if (!d) return '' + const lines = [`dbt ${d.resource_type}: ${d.unique_id}`] + if (d.materialized) { + const strategy = d.materialize_strategy ? ` -> ${d.materialize_strategy}` : '' + lines.push(`materialized: ${d.materialized}${strategy}`) + } + if (d.tags?.length) lines.push(`tags: ${d.tags.join(', ')}`) + for (const t of d.data_tests ?? []) { + const col = t.column ? ` on ${t.column}` : '' + lines.push(`test ${t.kind}${col}${t.severity ? ` [${t.severity}]` : ''}`) + } + const cols = Object.entries(d.columns ?? {}) + if (cols.length) { + lines.push( + `columns: ${cols.map(([c, desc]) => (desc ? `${c} (${desc})` : c)).join(', ')}` + ) + } + if (d.freshness) { + const f = d.freshness as Record + const window = (k: string) => + f[k]?.count != null ? `${k.replace('_after', '')} after ${f[k].count}${f[k].period?.[0] ?? ''}` : '' + const windows = ['warn_after', 'error_after'].map(window).filter(Boolean) + if (windows.length) lines.push(`freshness: ${windows.join(', ')}`) + } + if (d.description) lines.push(d.description) + return lines.join('\n') + }) + // Data-test outcome badge. Only guarded assets show it. The write's fate on a // failing test differs by edition — surface which one applies so a shared // parent/fork table name can't hide a silently-published bad version. @@ -200,6 +252,39 @@ class={`shrink-0 ml-2 mr-2 ${selected ? 'text-accent' : 'text-blue-600 dark:text-blue-400'}`} size="14px" /> + {#if data.runStatus} + + + {#if data.runStatus === 'running'} + + {:else if data.runStatus === 'failed'} + + {:else} + + {/if} + + + {#if data.runStatus === 'materialized' && data.runRowCount != undefined} + + {Intl.NumberFormat().format(data.runRowCount)} + + {/if} + {/if} {formatShortAssetPath(asset)} @@ -225,6 +310,34 @@ fork {/if} + + {#if data.dbt} + + {/if} diff --git a/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte b/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte index 7db35c2353..ec28b7c784 100644 --- a/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte @@ -16,8 +16,7 @@ AssetGraphResponse, AssetGraphSelection, NativeTriggerKind, - PipelineMode - } from './types' + PipelineMode, DbtAssetProvenance } from './types' import type { AssetKind, Script, ScriptLang } from '$lib/gen' import type { RunnableRunState, PipelineEvent } from './activeRunnables.svelte' import type { PipelineOutputKind } from './pipelineTemplates' @@ -76,6 +75,7 @@ localScriptsVersion, selectionProducers = [], selectionColumnGraph, + selectionDbt, schemaCanEvolve = true, selectionForkMaterialization = undefined, schemaContractContext = undefined, @@ -181,6 +181,8 @@ selectionProducers?: Array<{ kind: 'script' | 'flow'; path: string; unsaved?: boolean }> /** Transitive column-lineage trace for a selected ducklake asset (route page). */ selectionColumnGraph?: ColumnLineageGraph + /** dbt provenance of the selected relation — carries its SQL. */ + selectionDbt?: DbtAssetProvenance schemaCanEvolve?: boolean /** Fork workspaces: data-environment state of the selected ducklake asset (route page). */ selectionForkMaterialization?: 'fork' | 'deferred' @@ -512,6 +514,7 @@ selection={activeDraft ? undefined : editor.selection} selectionProducers={activeDraft ? [] : selectionProducers} {selectionColumnGraph} + {selectionDbt} {schemaCanEvolve} {selectionForkMaterialization} {schemaContractContext} diff --git a/frontend/src/lib/components/assets/AssetGraph/RunnableNode.svelte b/frontend/src/lib/components/assets/AssetGraph/RunnableNode.svelte index d8a977539b..c260bc2e5b 100644 --- a/frontend/src/lib/components/assets/AssetGraph/RunnableNode.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/RunnableNode.svelte @@ -18,6 +18,7 @@ XCircle, Zap } from 'lucide-svelte' + import DbtIcon from '$lib/components/icons/DbtIcon.svelte' import { twMerge } from 'tailwind-merge' import { preventDefault, stopPropagation } from 'svelte/legacy' import type { GraphUsageKind } from './types' @@ -45,6 +46,13 @@ // Macros this script provides (deployed/drafted `// macros` library). // Non-empty renders the ƒ chip marking the node as a macro library. macros?: { name: string; params: string; is_table: boolean }[] + // Set on a dbt script: the number of models the project materializes. + // One runnable node stands for the whole project, so the count is what + // tells it apart from a single-output script. + dbt?: { model_count: number } + /** Hovering the project badge emphasizes every model it + * materializes — the fan-out the graph deliberately omits. */ + onDbtHover?: (on: boolean) => void // Last-run status + run count observed this session (from the // folder queue poll). Undefined until the first observed run. runState?: RunnableRunState @@ -188,7 +196,19 @@ -
(hover = true)} onmouseleave={() => (hover = false)}> + +
{ + hover = true + data.onDbtHover?.(true) + }} + onmouseleave={() => { + hover = false + data.onDbtHover?.(false) + }} +> + {#if onDelete && node.path !== 'dbt_project.yml'} + + {/if} +
+ {/if} + {/each} +{/snippet} + +
+
+ {scriptPath}__dbt/ +
+ {fileCount + 1} + {@render addFile?.()} +
+
+
+ + + {@render branch(tree, 0)} +
+ {#if fileCount === 0} +
+ No project yet. Copy one in and push it: +
cp -r my-dbt-project/. {scriptPath}__dbt/
+wmill sync push
+
+ {/if} +
diff --git a/frontend/src/lib/components/dbt/DbtRunGraph.svelte b/frontend/src/lib/components/dbt/DbtRunGraph.svelte new file mode 100644 index 0000000000..7b9cbee07b --- /dev/null +++ b/frontend/src/lib/components/dbt/DbtRunGraph.svelte @@ -0,0 +1,769 @@ + + +{#snippet sqlPane()} + {#if selectedIsForeign} +
+ Another dbt project in this workspace also materializes this relation, and the graph keeps one + project's model per relation — so the SQL shown here would not be this run's. Open that + project's own run to see it. +
+ {:else if selectedDbt?.raw_code} +
+
+ {selectedDbt.original_file_path ?? selectedDbt.unique_id} + {#if selectedDbt.materialized} + {selectedDbt.materialized} + {/if} + + {#if selectedDbt.resource_type === 'model'} + {#if showRows && preview && !('error' in preview)} + + {:else} + + {/if} + {/if} + {#if selectedRelation} + + {/if} + + + read-only · edit in the script + +
+
+ {#if showRows && preview} + {#if 'error' in preview} +
{preview.error}
+ + {:else if 'pending' in preview} +
+ + Running `dbt show` — this is a job, so it waits on a worker and the engine. +
+ {:else} + {@const cols = Object.keys(preview.rows[0] ?? {})} + {#if cols.length === 0} +
The model returned no rows.
+ {:else} + + + + {#each cols as c (c)} + + {/each} + + + + {#each preview.rows as row, i (i)} + + {#each cols as c (c)} + + {/each} + + {/each} + +
{c}
{cellText(row[c])}
+
+ {preview.rows.length} rows in {(preview.tookMs / 1000).toFixed(1)}s{preview.node + ? ` · ${preview.node}` + : ''} +
+ {/if} + {/if} + {:else} + + {/if} +
+
+ {/if} +{/snippet} + +{#if resumable} +
+ {(run?.totals?.error ?? 0) > 0 ? `${run?.totals?.error} failed` : ''}{(run?.totals?.error ?? + 0) > 0 && (run?.totals?.skipped ?? 0) > 0 + ? ', ' + : ''}{(run?.totals?.skipped ?? 0) > 0 ? `${run?.totals?.skipped} skipped` : ''}. Rebuild only + those with dbt retry, instead of the whole project. + + + +
+{/if} + +{#if loading} +
+ Loading the model graph +
+{:else if failed} +
Could not load the model graph.
+{:else if !graph} +
+ {#if ranTestsOnly} + This run selected tests alone, so it built no models. A dbt test is an assertion rather than a + relation, so it has no node here — the models it asserts against belong to the runs that build + them. Its results are in the table below. + {:else} + This dbt script has no models in the asset graph. A project that brings its own + profiles.yml without naming a + profile.warehouse has no warehouse identity to key them on. + {/if} +
+{:else} +
+ {#if relationDrift > 0} +
+ {relationDrift} + {relationDrift === 1 ? 'model has' : 'models have'} been renamed or moved since this run — + {relationDrift === 1 ? 'its node shows' : 'their nodes show'} today's relation, not the one this + run wrote. +
+ {/if} + {#if goneSinceRun > 0} +
+ {goneSinceRun} + {goneSinceRun === 1 ? 'model' : 'models'} this run built {goneSinceRun === 1 ? 'is' : 'are'} + no longer in the project — renamed or removed since, so + {goneSinceRun === 1 ? 'it is' : 'they are'} not drawn. +
+ {/if} +
+ (selection = s)} + showMinimap={false} + scrollZoom={false} + /> +
+ {@render sqlPane()} +
+{/if} diff --git a/frontend/src/lib/components/dbt/DbtRunResult.svelte b/frontend/src/lib/components/dbt/DbtRunResult.svelte new file mode 100644 index 0000000000..34d7886b30 --- /dev/null +++ b/frontend/src/lib/components/dbt/DbtRunResult.svelte @@ -0,0 +1,146 @@ + + +
+
+ {#each [{ k: 'success', label: 'passed', cls: 'text-green-600 dark:text-green-400' }, { k: 'warn', label: 'warned', cls: 'text-yellow-600 dark:text-yellow-400' }, { k: 'error', label: 'failed', cls: 'text-red-600 dark:text-red-400' }, { k: 'skipped', label: 'skipped', cls: 'text-secondary' }] as t (t.k)} + {@const n = (totals as Record)[t.k] ?? 0} + {#if n > 0} + {n} {t.label} + {/if} + {/each} + of {totals.total ?? nodes.length} nodes + + {run.command ?? 'build'} · {run.engine ?? ''} + {run.engine_version ?? ''} + +
+ + {#if nodes.length > 0} +
+ + + + + + + + + + + + {#each nodes as node (node.unique_id)} + {@const s = split(node.unique_id)} + {@const r = rank(node.status, node.outcome)} + + + + + + + + {/each} + +
NodeKindRelationRowsTime
+
+ + {#if r === 0} + + {:else if r === 1} + + {:else if r === 2} + + {:else} + + {/if} + + {s.name} + {#if node.message && r < 2} + {node.message} + {/if} +
+ {#if node.message && r < 2} +
+ {node.message} +
+ {/if} +
+ + {#if s.kind === 'test'} + + {/if} + {s.kind} + + + {fmtRelation(node.relation_name) ?? ''} + + {node.rows_affected ?? ''} + + {fmtTime(node.execution_time)} +
+
+ {#if hasTests} +
+ A test's severity decides the outcome: dbt's own warn surfaces + without failing the job. +
+ {/if} + {/if} +
diff --git a/frontend/src/lib/components/dbt/parseDbtRun.test.ts b/frontend/src/lib/components/dbt/parseDbtRun.test.ts new file mode 100644 index 0000000000..47d6073ef7 --- /dev/null +++ b/frontend/src/lib/components/dbt/parseDbtRun.test.ts @@ -0,0 +1,193 @@ +import { describe, it, expect } from 'vitest' +import { + parseDbtRun, + relationOutcome, + splitRelation, + statusRank, + splitUniqueId, + nodeSelector +} from './parseDbtRun' + +const run = { + engine: 'dbt-core-1x', + engine_version: '1.12.0', + command: 'build', + totals: { total: 1, success: 1, error: 0, warn: 0, skipped: 0 }, + nodes: [{ unique_id: 'model.p.customers', status: 'success' }] +} + +describe('parseDbtRun', () => { + it('takes a successful run as-is', () => { + expect(parseDbtRun(run)?.engine).toBe('dbt-core-1x') + }) + + // The worker puts the same JSON in the error message after the exit-status + // line, and this is the case worth rendering: the failing node is what the + // user came for. + it('recovers the run from a failed job’s error message', () => { + const failed = { + error: { + name: 'ExecutionErr', + message: `execution error:\nNon-zero exit status for dbt build: 1\n\n${JSON.stringify(run)}` + } + } + expect(parseDbtRun(failed)?.totals?.total).toBe(1) + }) + + // The failures worth reading are the ones whose message carries braces of its + // own — a Jinja template, the compiled SQL, an adapter's own JSON — and the + // payload is appended after all of it. + it('finds the run past braces in the error text', () => { + const failed = { + error: { + name: 'ExecutionErr', + message: + 'execution error:\nCompilation Error in model x\n {{ ref("missing") }} depends on {"a": 1}\n' + + `Non-zero exit status for dbt build: 1\n\n${JSON.stringify(run)}` + } + } + expect(parseDbtRun(failed)?.totals?.total).toBe(1) + }) + + // `{nodes, totals}` alone is a shape an ordinary script can return, and it + // would then be rendered as somebody's dbt run. + it('does not claim an ordinary result that happens to have nodes and totals', () => { + expect(parseDbtRun({ nodes: [], totals: {} })).toBeUndefined() + expect(parseDbtRun({ engine: 'v8', nodes: [], totals: {} })).toBeUndefined() + }) + + it('accepts every engine the worker stamps', () => { + for (const engine of ['dbt-core-1x', 'dbt-core-2x', 'fusion']) { + expect(parseDbtRun({ ...run, engine })?.engine).toBe(engine) + } + }) + + // The payload carries one object per node, so a scan bounded by brace COUNT + // gives up on an ordinary project — a few hundred nodes, tests included — and + // silently loses the per-model table on exactly the runs it exists for. + it('finds the run in a payload with hundreds of nodes', () => { + const big = { + ...run, + totals: { total: 400, success: 399, error: 1, warn: 0, skipped: 0 }, + nodes: Array.from({ length: 400 }, (_, i) => ({ + unique_id: `model.p.m${i}`, + status: i === 0 ? 'error' : 'success' + })) + } + const failed = { + error: { + name: 'ExecutionErr', + message: + 'execution error:\nCompilation Error {{ ref("x") }} {"a": 1}\n\n' + + JSON.stringify(big, null, 2) + } + } + expect(parseDbtRun(failed)?.nodes?.length).toBe(400) + }) + + it('is undefined for anything unparseable', () => { + expect(parseDbtRun(undefined)).toBeUndefined() + expect(parseDbtRun('a string')).toBeUndefined() + expect(parseDbtRun({ error: { message: 'failed with {not json' } })).toBeUndefined() + }) +}) + +describe('statusRank', () => { + // dbt counts `partial success` in totals.error and a retry redoes it, so + // ranking it as a pass would contradict the job's own outcome. + it('ranks partial success with the failures', () => { + expect(statusRank('partial success')).toBe(statusRank('error')) + expect(statusRank('PARTIAL SUCCESS')).toBe(0) + }) + + // The worker publishes `unknown` for a status it does not recognise and counts + // it in `totals.error`; ranking it as a pass drew a green check on a node the + // same result called an error. + // The worker counts `no_op` in totals.skipped — dbt built nothing for that + // node — so a green check would claim a run that never happened. + it('ranks a no-op with the skips, not with the passes', () => { + expect(statusRank('no-op', 'no_op')).toBe(statusRank('skipped', 'skipped')) + expect(statusRank('success', 'no_op')).toBe(2) + }) + + it('ranks an unknown outcome with the failures, not with the passes', () => { + expect(statusRank('some-future-dbt-status', 'unknown')).toBe(0) + expect(statusRank('success', 'unknown')).toBe(0) + }) + + it('orders failed before warned before skipped before passed', () => { + expect( + ['success', 'skipped', 'warn', 'error'].sort((a, b) => statusRank(a) - statusRank(b)) + ).toEqual(['error', 'warn', 'skipped', 'success']) + }) +}) + +describe('splitUniqueId', () => { + it('splits kind from name and drops a generic test’s uniqueness hash', () => { + expect(splitUniqueId('model.jaffle.stg_orders')).toEqual({ + kind: 'model', + name: 'stg_orders' + }) + expect(splitUniqueId('test.jaffle.not_null_orders_id.4e687af8d0')).toEqual({ + kind: 'test', + name: 'not_null_orders_id' + }) + // A model whose name contains a dot keeps it: only tests carry the hash. + expect(splitUniqueId('model.jaffle.a.b').name).toBe('a.b') + }) +}) + +describe('relationOutcome', () => { + it('agrees with the worker classifier on every status it names', () => { + expect(relationOutcome('started')).toBe('running') + for (const s of ['success', 'pass', 'PASS', ' Success ']) { + expect(relationOutcome(s)).toBe('materialized') + } + // `partial success` built the relation and then failed its tests; the + // worker records it failed, so the colour must agree. + for (const s of ['error', 'fail', 'runtime error', 'partial success', 'PARTIAL SUCCESS']) { + expect(relationOutcome(s)).toBe('failed') + } + // Nothing was built, so nothing is coloured. + for (const s of ['warn', 'skipped', 'no-op', 'something new']) { + expect(relationOutcome(s)).toBeUndefined() + } + }) +}) + +describe('splitRelation', () => { + it('keeps a period that lives inside a quoted identifier', () => { + // The backend supports it, so rendering it as `v2.orders` names a + // relation that does not exist. + expect(splitRelation('"wh"."analytics.v2"."orders"')).toEqual(['wh', 'analytics.v2', 'orders']) + expect(splitRelation('"db"."schema"."name"')).toEqual(['db', 'schema', 'name']) + expect(splitRelation('db.schema.name')).toEqual(['db', 'schema', 'name']) + // BigQuery backticks and T-SQL brackets quote too. + expect(splitRelation('`proj`.`data.set`.`t`')).toEqual(['proj', 'data.set', 't']) + expect(splitRelation('[db].[my.schema].[t]')).toEqual(['db', 'my.schema', 't']) + }) + + // Every one of these dialects escapes its delimiter by doubling it. Dropping + // the pair renames the relation, and the manifest keeps the real spelling — + // so the run's status would be recorded against a key no graph node has. + it('keeps a delimiter the identifier escaped by doubling', () => { + expect(splitRelation('"wh"."schema"."a""b"')).toEqual(['wh', 'schema', 'a"b']) + expect(splitRelation('`proj`.`da``ta`.`t`')).toEqual(['proj', 'da`ta', 't']) + expect(splitRelation('[db].[my]]schema].[t]')).toEqual(['db', 'my]schema', 't']) + }) +}) + +describe('nodeSelector', () => { + // Verified against dbt-core 1.12, dbt-core 2.0.0-alpha.5 and fusion + // 2.0.0-preview.202: the intersection resolves to the one node whatever the + // project's `model-paths` is, while a path-derived FQN resolves to nothing + // as soon as that root is more than one segment deep. + it('intersects the name with its package, wherever the model sits', () => { + expect(nodeSelector('model.jaffle_shop.fct_orders')).toBe('fct_orders,package:jaffle_shop') + }) + + // Ambiguous across packages, but a selector dbt resolves rather than rejects. + it('falls back to the bare name without a package', () => { + expect(nodeSelector('fct_orders')).toBe('fct_orders') + }) +}) diff --git a/frontend/src/lib/components/dbt/parseDbtRun.ts b/frontend/src/lib/components/dbt/parseDbtRun.ts new file mode 100644 index 0000000000..aecd3dcf03 --- /dev/null +++ b/frontend/src/lib/components/dbt/parseDbtRun.ts @@ -0,0 +1,247 @@ +export type DbtNode = { + unique_id: string + status: string + /** Windmill's stable word for the same result, published beside dbt's own. + * Preferred wherever a decision is made: `status` is dbt's vocabulary and + * dbt may rename it. */ + outcome?: DbtOutcome + execution_time?: number + rows_affected?: number + relation_name?: string + message?: string +} + +export type DbtRun = { + engine?: string + engine_version?: string + command?: string + totals?: { total?: number; success?: number; error?: number; warn?: number; skipped?: number } + nodes?: DbtNode[] + /** The arguments the run actually used, as submitted. A `dbt retry` restores + * the failed run's arguments inside the worker, so the retry job's own args + * name only the run it resumed — this is the sole way to recover what it + * really ran with. */ + invocation_args?: Record +} + +/** The engines the worker stamps on a result. This is the discriminator: a + * `{nodes, totals}` shape alone is one an ordinary script can return, and it + * would then be rendered as somebody's dbt run. */ +const ENGINES = ['dbt-core-1x', 'dbt-core-2x', 'fusion'] + +function asDbtRun(v: unknown): DbtRun | undefined { + if (!v || typeof v !== 'object') return undefined + const o = v as Record + return ENGINES.includes(o.engine as string) && + Array.isArray(o.nodes) && + o.totals != undefined && + typeof o.totals === 'object' + ? (o as DbtRun) + : undefined +} + +/** + * The dbt invocation a job result describes, if it describes one. + * + * On success the result IS the run. On failure the worker puts the same JSON in + * the error message after the exit-status line, and that is the case worth + * rendering: the failing node is what the user came for. + */ +export function parseDbtRun(result: any): DbtRun | undefined { + const direct = asDbtRun(result) + if (direct) return direct + const msg = result?.error?.message + if (typeof msg !== 'string') return undefined + // The payload is appended pretty-printed, so its `{` is the only one at COLUMN + // ZERO — everything nested is indented, and dbt's own error text carries its + // braces mid-line. Counting braces instead needs a cap that a real project + // blows: forwards on an error full of them, backwards on one `{` per node. + for (const line of lineStarts(msg)) { + if (msg[line] !== '{') continue + try { + const run = asDbtRun(JSON.parse(msg.slice(line))) + if (run) return run + } catch { + // A `{` alone on a line inside the error text; the payload is later. + } + } + return undefined +} + +/** Index of the first character of each line, the payload's own `{` among them. */ +function* lineStarts(s: string): Generator { + let at = 0 + while (at !== -1 && at < s.length) { + yield at + const next = s.indexOf('\n', at) + at = next === -1 ? -1 : next + 1 + } +} + +/** Ordering rank of a node's status: 0 failed, 1 warned, 2 skipped, 3 passed. + * + * Both `unknown` and `no_op` rank where the RESULT counts them, not where the + * default would put them: the worker counts `unknown` in `totals.error` and + * `no_op` in `totals.skipped`, so falling through to 3 drew a green check on a + * node the same result called an error, and on one it never built. */ +export function statusRank(status: string, outcome?: DbtOutcome): number { + switch (outcome ?? classifyStatus(status)) { + case 'failed': + case 'unknown': + return 0 + case 'warned': + return 1 + case 'skipped': + case 'no_op': + return 2 + default: + return 3 + } +} + +/** The worker's stable vocabulary for a node result, published as `outcome`. */ +export type DbtOutcome = + | 'started' + | 'passed' + | 'failed' + | 'warned' + | 'skipped' + | 'no_op' + | 'unknown' + +/** + * dbt's node status, reduced to the outcomes the UI distinguishes. + * + * Only for results that predate `outcome`, or for the live event stream, which + * carries dbt's word alone. Anything holding a node from a job result should + * read `outcome` instead — that is the field the worker publishes precisely so + * this mapping is not the contract. + */ +function classifyStatus( + status: string +): 'started' | 'passed' | 'failed' | 'warned' | 'skipped' | 'other' { + // `partial success` is dbt's word for a node that built but whose tests + // failed. The worker counts it in `totals.error` and a retry redoes it, so + // showing it green would contradict the job's own outcome. + switch (status.trim().toLowerCase()) { + case 'started': + return 'started' + case 'success': + case 'pass': + return 'passed' + case 'error': + case 'fail': + case 'runtime error': + case 'partial success': + return 'failed' + case 'warn': + return 'warned' + case 'skipped': + return 'skipped' + default: + return 'other' + } +} + +/** + * The kind and name behind a dbt `unique_id`, which dbt builds as + * `..`. A generic test's name carries a trailing + * hash dbt adds for uniqueness; it is noise in a run summary. + */ +export function splitUniqueId(uniqueId: string): { kind: string; name: string } { + const parts = uniqueId.split('.') + const kind = parts[0] ?? '' + let name = parts.slice(2).join('.') + if (kind === 'test') name = name.replace(/\.[0-9a-f]{6,}$/, '') + return { kind, name: name || uniqueId } +} + +/** + * What a node's status says happened to the relation it builds, or `undefined` + * when it says nothing. + * + * Mirrors the worker's `classify_status`, and must keep mirroring it: the two + * decide the same thing about the same string, one for the record it writes and + * one for the colour drawn over it. `warn`, `skipped` and `no-op` leave the + * relation untouched, so they get no colour rather than a misleading one. + */ +export function relationOutcome( + status: string, + outcome?: DbtOutcome +): 'running' | 'materialized' | 'failed' | undefined { + switch (outcome ?? classifyStatus(status)) { + case 'started': + return 'running' + case 'passed': + return 'materialized' + case 'failed': + return 'failed' + // `warn`, `skipped` and `no-op` say nothing about the relation: nothing + // was written, so its state is whatever the last run left. + default: + return undefined + } +} + +/** + * dbt's `relation_name` split into its parts, honouring quoting. + * + * Mirrors the worker's `split_relation`: `"`, `` ` `` and `[` open a quoted + * identifier, and a `.` inside one is part of the name. Splitting on every `.` + * turns `"wh"."analytics.v2"."orders"` into a relation called `orders` in a + * schema called `v2` — a table that does not exist. + */ +export function splitRelation(relation: string): string[] { + const parts: string[] = [] + let current = '' + let quote: string | undefined + for (let i = 0; i < relation.length; i++) { + const c = relation[i] + if (quote !== undefined) { + const close = quote === '[' ? ']' : quote + if (c === close) { + // Doubled, which is how each of these dialects escapes its own + // delimiter: one literal character, not the end of the identifier. + if (relation[i + 1] === close) { + current += close + i++ + } else { + quote = undefined + } + } else current += c + } else if (c === '"' || c === '`' || c === '[') { + quote = c + } else if (c === '.') { + parts.push(current) + current = '' + } else { + current += c + } + } + parts.push(current) + return parts.map((p) => p.trim()) +} + +/** + * A selector naming exactly one node: `,package:`. + * + * The comma is dbt's intersection operator, so this reads "the node whose name + * is `` and whose package is ``" — one node, since dbt refuses + * two models of one name inside a package. A bare name would match the leaf of + * every package's FQN, and a package can ship a model whose name the project + * also uses. + * + * Not the FQN (`..`), which cannot be rebuilt from + * `original_file_path`: how many leading segments are the resource root is + * `model-paths`, and dropping exactly one turns `src/models/marts/orders.sql` + * into `pkg.models.marts.orders`, which dbt's matcher — equal lengths, from the + * front — resolves to nothing at all. + * + * Without a package, the bare name — ambiguous across packages, but a selector + * dbt resolves rather than one it rejects. + */ +export function nodeSelector(uniqueId: string): string { + const { name } = splitUniqueId(uniqueId) + const pkg = uniqueId.split('.')[1] + return pkg ? `${name},package:${pkg}` : name +} diff --git a/frontend/src/lib/components/dbt/previewRows.ts b/frontend/src/lib/components/dbt/previewRows.ts new file mode 100644 index 0000000000..48ff56ec4c --- /dev/null +++ b/frontend/src/lib/components/dbt/previewRows.ts @@ -0,0 +1,68 @@ +import { JobService } from '$lib/gen' + +/** A model's rows, as `dbt show` returns them. */ +export type DbtPreview = + | { pending: true } + | { rows: Record[]; node?: string; tookMs: number } + | { error: string } + +/** + * Preview one model's rows by running its own project's `dbt show`. + * + * A job, not a query: the rows come from the warehouse through the project's + * profile, with its vars and its adapter, which is the only place that knows how + * to resolve `ref()` and where the relation actually lives. `show` is therefore + * not a run-form command — it is what a table's preview is made of, here and on + * the run page's graph. + * + * `stillWanted` is asked before each poll and before the result is used, so a + * preview outlives neither the page that asked for it nor a navigation. + */ +export async function previewDbtRows(opts: { + workspace: string + scriptPath: string + /** Pins the preview to a deployed version, for a graph showing that version. */ + scriptHash?: string | number + /** One node: a model name, or `package.model` where two packages share one. */ + model: string + /** The run's own vars, so a descriptor with a required `{{ }}` var resolves. */ + vars?: Record + limit?: number + /** Extra top-level arguments — a run's `{{ placeholder }}` values. */ + args?: Record + stillWanted?: () => boolean +}): Promise { + const { workspace, scriptPath, scriptHash, model, vars, limit, args, stillWanted } = opts + const startedAt = Date.now() + const requestBody = { + ...(args ?? {}), + command: { label: 'show', vars: vars ?? {}, model, limit: limit ?? 25 } + } + try { + // By HASH whenever the caller pins one: the SQL on screen is that version's, + // and running the deployed one would show today's rows under it — or fail + // outright for a model since removed. + const id = scriptHash + ? await JobService.runScriptByHash({ + workspace, + hash: String(scriptHash), + requestBody + }) + : await JobService.runScriptByPath({ workspace, path: scriptPath, requestBody }) + // Polled rather than awaited: a preview is a job, and its engine may need + // provisioning on a cold worker. + for (let i = 0; i < 90; i++) { + await new Promise((r) => setTimeout(r, 1000)) + if (stillWanted && !stillWanted()) return undefined + const done = await JobService.getCompletedJobResultMaybe({ workspace, id }) + if (!done.completed) continue + const res = done.result as { node?: string; show?: Record[] } | undefined + return done.success && res?.show + ? { rows: res.show, node: res.node, tookMs: Date.now() - startedAt } + : { error: 'The preview job failed — open it from Runs for the detail.' } + } + return { error: 'The preview is still running; open it from Runs.' } + } catch (e) { + return { error: e instanceof Error ? e.message : String(e) } + } +} diff --git a/frontend/src/lib/components/flows/content/FlowInputs.svelte b/frontend/src/lib/components/flows/content/FlowInputs.svelte index 335a37b0c7..e275df3976 100644 --- a/frontend/src/lib/components/flows/content/FlowInputs.svelte +++ b/frontend/src/lib/components/flows/content/FlowInputs.svelte @@ -12,7 +12,7 @@ import { Check, Code, Zap } from 'lucide-svelte' import SuspendDrawer from './SuspendDrawer.svelte' import { defaultScripts } from '$lib/stores' - import { defaultScriptLanguages, processLangs } from '$lib/scripts' + import { defaultScriptLanguages, processInlineLangs } from '$lib/scripts' import type { SupportedLanguage } from '$lib/common' import DefaultScripts from '$lib/components/DefaultScripts.svelte' import type { FlowBuilderWhitelabelCustomUi } from '$lib/components/custom_ui' @@ -48,7 +48,7 @@ let filter = $state('') let langs = $derived( - processLangs(undefined, $defaultScripts?.order ?? Object.keys(defaultScriptLanguages)) + processInlineLangs(undefined, $defaultScripts?.order ?? Object.keys(defaultScriptLanguages)) .map((l) => [defaultScriptLanguages[l], l]) .filter( (x) => $defaultScripts?.hidden == undefined || !$defaultScripts.hidden.includes(x[1]) diff --git a/frontend/src/lib/components/flows/content/FlowInputsQuick.svelte b/frontend/src/lib/components/flows/content/FlowInputsQuick.svelte index dc2cba93d4..7a55ca92fc 100644 --- a/frontend/src/lib/components/flows/content/FlowInputsQuick.svelte +++ b/frontend/src/lib/components/flows/content/FlowInputsQuick.svelte @@ -6,7 +6,7 @@ + + + + + + + + + + + diff --git a/frontend/src/lib/components/worker_group.ts b/frontend/src/lib/components/worker_group.ts index d0391b17c2..31d0cee7db 100644 --- a/frontend/src/lib/components/worker_group.ts +++ b/frontend/src/lib/components/worker_group.ts @@ -59,7 +59,8 @@ export const defaultTags = [ 'java', 'ruby', 'rlang', - 'duckdb' + 'duckdb', + 'dbt' // for related places search: ADD_NEW_LANG ] /** Strip cache_clear, null/undefined values, empty arrays and empty objects from a worker group config. */ diff --git a/frontend/src/lib/components/workspaceSettings/DbtSettings.svelte b/frontend/src/lib/components/workspaceSettings/DbtSettings.svelte new file mode 100644 index 0000000000..d472ce6d08 --- /dev/null +++ b/frontend/src/lib/components/workspaceSettings/DbtSettings.svelte @@ -0,0 +1,173 @@ + + + + + + Where dbt projects in this workspace run. A project names a warehouse by name in its descriptor (profile.warehouse) and reaches + {DEFAULT_WAREHOUSE} when it names none, so a project carries no + connection of its own. The name is also what its tables are keyed on in the asset graph (dbt://{DEFAULT_WAREHOUSE}/schema/table), so two projects on one warehouse share their nodes. Each entry points at a resource, and + configuring one here is what makes it available: anyone who may run a dbt script builds with it + and reads its models, without being granted the resource, the same bargain workspace object + storage makes. + + + + + + Name + Resource + Target + + + + + {#each dbtSettings.warehouses as warehouse, i (i)} + + + + + + + + + + + + + + + + + + onDiscard?.()} + saveLabel="Save dbt warehouses" +/> diff --git a/frontend/src/lib/infer.ts b/frontend/src/lib/infer.ts index 1e0a36af50..db32bf0274 100644 --- a/frontend/src/lib/infer.ts +++ b/frontend/src/lib/infer.ts @@ -42,7 +42,8 @@ import initRustParser, { parse_rust } from 'windmill-parser-wasm-rust' import initYamlParser, { parse_assets_ansible, parse_ansible, - parse_ansible_delegate + parse_ansible_delegate, + parse_dbt } from 'windmill-parser-wasm-yaml' import initCSharpParser, { parse_csharp } from 'windmill-parser-wasm-csharp' import initNuParser, { parse_nu } from 'windmill-parser-wasm-nu' @@ -521,6 +522,9 @@ export async function inferArgs( } catch { inferedSchema = parseRSignatureFallback(code) } + } else if (language == 'dbt') { + await initWasmYaml() + inferedSchema = JSON.parse(parse_dbt(code)) // for related places search: ADD_NEW_LANG } else { return null diff --git a/frontend/src/lib/mcpEndpointTools.ts b/frontend/src/lib/mcpEndpointTools.ts index 70dabb01f6..f67c0415f0 100644 --- a/frontend/src/lib/mcpEndpointTools.ts +++ b/frontend/src/lib/mcpEndpointTools.ts @@ -709,7 +709,7 @@ export const mcpEndpointTools: EndpointTool[] = [ }, "language": { "type": "string", - "description": "Possible values: python3, deno, go, bash, powershell, postgresql, mysql, bigquery, snowflake, mssql, oracledb, graphql, nativets, bun, php, rust, ansible, csharp, nu, java, ruby, rlang, duckdb, bunnative" + "description": "Possible values: python3, deno, go, bash, powershell, postgresql, mysql, bigquery, snowflake, mssql, oracledb, graphql, nativets, bun, php, rust, ansible, csharp, nu, java, ruby, rlang, duckdb, bunnative, dbt" }, "kind": { "type": "string", @@ -1218,7 +1218,7 @@ export const mcpEndpointTools: EndpointTool[] = [ }, "language": { "type": "string", - "description": "Possible values: python3, deno, go, bash, powershell, postgresql, mysql, bigquery, snowflake, mssql, oracledb, graphql, nativets, bun, php, rust, ansible, csharp, nu, java, ruby, rlang, duckdb, bunnative" + "description": "Possible values: python3, deno, go, bash, powershell, postgresql, mysql, bigquery, snowflake, mssql, oracledb, graphql, nativets, bun, php, rust, ansible, csharp, nu, java, ruby, rlang, duckdb, bunnative, dbt" }, "tag": { "type": "string" @@ -1250,7 +1250,7 @@ export const mcpEndpointTools: EndpointTool[] = [ }, "language": { "type": "string", - "description": "Possible values: python3, deno, go, bash, powershell, postgresql, mysql, bigquery, snowflake, mssql, oracledb, graphql, nativets, bun, php, rust, ansible, csharp, nu, java, ruby, rlang, duckdb, bunnative" + "description": "Possible values: python3, deno, go, bash, powershell, postgresql, mysql, bigquery, snowflake, mssql, oracledb, graphql, nativets, bun, php, rust, ansible, csharp, nu, java, ruby, rlang, duckdb, bunnative, dbt" }, "lock": { "type": "string", diff --git a/frontend/src/lib/script_helpers.ts b/frontend/src/lib/script_helpers.ts index 91e310f5c9..178f4474e7 100644 --- a/frontend/src/lib/script_helpers.ts +++ b/frontend/src/lib/script_helpers.ts @@ -1345,6 +1345,43 @@ main <- function( return(toJSON(result, auto_unbox = TRUE)) } ` + +// A dbt script is a whole dbt project: the descriptor below is the script's +// content, and the project's own files live in its module bundle (the +// `