mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-09-05 16:03:47 +00:00
feat(ai): improve data-pipeline building in AI sessions (prompt + evals + e2e) (#10270)
* test(ai_evals): pipeline coverage for AI sessions + editor e2e Add a complex incremental DuckLake pipeline case, harden the two-node case, and encode the declarative pipeline contract (`-- on` triggers, `-- materialize` + bare SELECT) in the pipeline judgeChecklists so the LLM judge stops false-negativing correct nodes. Add a deterministic Playwright e2e that seeds annotated pipeline scripts and asserts the /pipeline/<folder> editor derives the lineage DAG. Fixes WIN-2229 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * feat(ai): steer pipeline chat to DuckDB+materialize and warn on missing storage The pipeline authoring prompt (getPipelinePrompt, used by the global/session chat and the /pipeline editor) was neutral on language choice and said nothing about storage readiness. Default it to duckdb materializing into DuckLake unless the work specifically needs postgres/data-tables or bun/python, and add a storage prerequisites section: a DuckLake pipeline needs workspace object storage + a DuckLake catalog, so warn when none is configured and give role-appropriate next steps (admin: workspace settings; others: ask an admin). Drafting is not blocked. A/B on sonnet (global pipeline cases): no regression, +639 finalContext tokens. Fixes WIN-2229 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * test(ai_evals): address review - e2e teardown, async-edge note, merge-mode hedge Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * chore(ai): address review nits - drop phantom list_ducklakes tool ref, trim narration comments Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * feat(ai): add list_ducklakes chat tool for pipeline storage readiness The pipeline counterpart to list_datatables: lists the workspace's configured DuckLake catalogs so the chat can detect the storage prerequisite before building a DuckLake pipeline and warn with role-appropriate next steps when none exists (drafting stays unblocked). Wired into the global tool set and referenced from the pipeline authoring prompt. In the eval run all three pipeline cases called it unprompted with no build regression. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * chore(ai): fix DuckDB annotation syntax in duckdb-default section (-- not //) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
07d4b674f1
commit
0819641f3a
@@ -1484,9 +1484,15 @@
|
||||
- deploy_workspace_item
|
||||
- delete_workspace_item
|
||||
judgeChecklist:
|
||||
# A pipeline node is DECLARATIVE: triggers are declared by `-- on <ref>`
|
||||
# annotations (the trigger row is created separately) and a `-- materialize`
|
||||
# output is a MANAGED write where the body is a bare SELECT that the runtime
|
||||
# wraps in the create/replace. Do not expect a separate trigger config or a
|
||||
# hand-written CREATE TABLE / INSERT — those would be wrong for a materialize node.
|
||||
- builds a data pipeline node as a script (not a flow)
|
||||
- marks the script as a pipeline member with the pipeline annotation in the script's comment syntax (`-- pipeline` for a DuckDB/SQL node, not `// pipeline`)
|
||||
- declares a schedule trigger and writes its output to a managed DuckLake table
|
||||
- declares the schedule trigger with the `-- on schedule` annotation comment (this annotation is the correct and complete way a pipeline node binds a schedule; no separate trigger configuration is expected)
|
||||
- declares the managed DuckLake output with `-- materialize ducklake://<table>` and writes the body as a bare SELECT (materialize is a managed write, so the node correctly does NOT hand-write its own CREATE TABLE / INSERT)
|
||||
- leaves the result as an AI draft and does not deploy or save it
|
||||
|
||||
- id: global-test-pipeline-two-node-chain
|
||||
@@ -1500,6 +1506,12 @@
|
||||
maxTurns: 14
|
||||
validate:
|
||||
draftCountAtLeast: 2
|
||||
requiredDrafts:
|
||||
- type: script
|
||||
pathStartsWith: f/evals/global/
|
||||
valueIncludes:
|
||||
- pipeline
|
||||
- ducklake
|
||||
forbiddenDrafts:
|
||||
- type: flow
|
||||
pathStartsWith: f/evals/global/
|
||||
@@ -1511,12 +1523,65 @@
|
||||
- deploy_workspace_item
|
||||
- delete_workspace_item
|
||||
judgeChecklist:
|
||||
# Pipeline nodes are declarative: `-- on <ref>` binds inputs/triggers and
|
||||
# `-- materialize ducklake://<table>` is a managed write whose body is a bare
|
||||
# SELECT. Do not expect hand-written CREATE TABLE / INSERT on a materialize node.
|
||||
- creates two data pipeline nodes as scripts (not a flow) in f/evals/global
|
||||
- both scripts carry the pipeline annotation in their comment syntax (`-- pipeline` for DuckDB/SQL nodes, not `// pipeline`)
|
||||
- the first ingests orders into a DuckLake table
|
||||
- the second reads that same table and writes a daily rollup, wired to the first step's output asset
|
||||
- the first ingests orders into a DuckLake table (a `-- materialize ducklake://<table>` output with a bare SELECT body is correct; no hand-written CREATE TABLE / INSERT is expected)
|
||||
- the second reads that same table via `-- on ducklake://<that-table>` and materializes a daily rollup table, wiring it to the first step's output asset
|
||||
- leaves both as AI drafts without deploying
|
||||
|
||||
- id: global-test-pipeline-complex-incremental
|
||||
prompt: |-
|
||||
Build a data pipeline in the `f/evals/global` folder for our web shop's
|
||||
orders. It has three steps:
|
||||
1. On a schedule, ingest the raw order CSVs under `s3://raw/orders/` into a
|
||||
managed DuckLake table.
|
||||
2. An incremental daily rollup: read that raw orders table and, on each run,
|
||||
append just the current day's order count and total revenue into a second
|
||||
DuckLake table. It should process one day at a time, not rebuild the whole
|
||||
table every run.
|
||||
3. A final step that reads the daily rollup table and exports the latest data
|
||||
as a Parquet file to `s3://reports/` for the BI team.
|
||||
Wire each step to the previous step's output so they form one pipeline. Keep
|
||||
everything as AI drafts — don't deploy or save.
|
||||
initial: ai_evals/fixtures/frontend/global/initial/user_admin_evals_folder.json
|
||||
runtime:
|
||||
maxTurns: 18
|
||||
validate:
|
||||
draftCountAtLeast: 3
|
||||
requiredDrafts:
|
||||
- type: script
|
||||
pathStartsWith: f/evals/global/
|
||||
valueIncludes:
|
||||
- pipeline
|
||||
- ducklake
|
||||
forbiddenDrafts:
|
||||
- type: flow
|
||||
pathStartsWith: f/evals/global/
|
||||
toolExpect:
|
||||
requiredToolsUsed:
|
||||
- write_script
|
||||
forbiddenToolsUsed:
|
||||
- write_flow
|
||||
- deploy_workspace_item
|
||||
- delete_workspace_item
|
||||
judgeChecklist:
|
||||
# Pipeline nodes are declarative: `-- on <ref>` binds inputs/triggers, and a
|
||||
# DuckLake `-- materialize` output is a managed write whose body is a bare SELECT
|
||||
# (the runtime performs the create/replace/append/merge). Do not expect a
|
||||
# separate trigger config or hand-written CREATE TABLE / INSERT on a
|
||||
# materialize node. S3/Parquet output is NOT materialize: the body writes it.
|
||||
- builds the pipeline as three independent scripts (not a flow) in f/evals/global
|
||||
- every node carries the pipeline annotation in its own comment syntax (`-- pipeline` for DuckDB/SQL nodes, not `// pipeline`)
|
||||
- step 1 binds a schedule with `-- on schedule` and declares a managed DuckLake output with `-- materialize ducklake://<table>` and a bare SELECT body (no separate trigger config or hand-written CREATE TABLE is expected)
|
||||
- "step 2 is incremental: each run adds only that day's rows to a second DuckLake table rather than rebuilding the whole table every run (e.g. an `append` or `key=<col>` merge materialize mode, not a full replace). Selecting the day via the `-- partitioned daily` + `{partition}` / `wm_partition(...)` idiom is the idiomatic form, but an equivalent current-day filter also satisfies this; a full-refresh/replace of the whole table does not"
|
||||
- step 2 reads the same DuckLake table step 1 writes (via `-- on ducklake://<that-table>`), wiring it to step 1's output asset
|
||||
- step 3 reads the daily rollup table and exports it as a Parquet file to S3
|
||||
- does not misuse `-- materialize` for the S3 Parquet export (materialize is DuckLake-only; the S3 output is written by the script body, e.g. a DuckDB COPY or an SDK write)
|
||||
- leaves all three nodes as AI drafts without deploying or saving
|
||||
|
||||
- id: global-path5-create-folder-then-draft
|
||||
prompt: |-
|
||||
Create a new shared folder called "analytics" for our data work, then draft a
|
||||
|
||||
@@ -0,0 +1,103 @@
|
||||
import { test, expect, Page } from '@playwright/test'
|
||||
|
||||
// The pipeline surface an AI session builds into: session pipeline tools emit
|
||||
// annotated scripts (`-- pipeline`, `-- on <asset>`, `-- materialize <asset>`),
|
||||
// and the /pipeline/<folder> editor derives the lineage DAG from those
|
||||
// annotations alone. This test seeds the scripts an AI-built DuckLake pipeline
|
||||
// would produce and asserts the editor renders every derived node, asset, and
|
||||
// edge (including the missing-schedule-trigger edge): a deterministic check of
|
||||
// the graph the AI session relies on, without a live model in the loop.
|
||||
|
||||
const WORKSPACE = 'admins'
|
||||
|
||||
declare const process: any
|
||||
|
||||
function uniqueSuffix(project: string): string {
|
||||
// Per-project suffix so the three browser projects don't collide on the
|
||||
// shared dev instance when Playwright runs them in parallel.
|
||||
return `${process.env.TEST_UNIQUE_ID ?? 'local'}_${project}`
|
||||
}
|
||||
|
||||
async function seedScript(page: Page, path: string, content: string, summary: string) {
|
||||
const res = await page.request.post(`/api/w/${WORKSPACE}/scripts/create`, {
|
||||
data: { path, summary, description: '', content, language: 'duckdb', schema: {} }
|
||||
})
|
||||
expect(res.ok(), `seed ${path}: ${res.status()} ${await res.text()}`).toBeTruthy()
|
||||
}
|
||||
|
||||
test.describe('Pipeline editor', () => {
|
||||
// Track what the test seeds so afterAll can remove it (keeps the shared dev/CI
|
||||
// instance from accumulating a folder + scripts per run).
|
||||
let seeded: { folder: string; scripts: string[] } | undefined
|
||||
|
||||
test.afterAll(async ({ request }) => {
|
||||
if (!seeded) return
|
||||
for (const path of seeded.scripts) {
|
||||
await request.post(`/api/w/${WORKSPACE}/scripts/delete/p/${path}`).catch(() => {})
|
||||
}
|
||||
await request.delete(`/api/w/${WORKSPACE}/folders/delete/${seeded.folder}`).catch(() => {})
|
||||
})
|
||||
|
||||
test('derives the DAG from annotated pipeline scripts', async ({ page }, testInfo) => {
|
||||
const suffix = uniqueSuffix(testInfo.project.name)
|
||||
const folder = `pipeline_e2e_${suffix}`
|
||||
const ingest = `f/${folder}/orders_ingest`
|
||||
const daily = `f/${folder}/orders_daily`
|
||||
const ordersTbl = `main/orders_${suffix}`
|
||||
const dailyTbl = `main/orders_daily_${suffix}`
|
||||
seeded = { folder, scripts: [ingest, daily] }
|
||||
|
||||
// Folder may already exist from a prior run; only fail on the seeds.
|
||||
await page.request.post(`/api/w/${WORKSPACE}/folders/create`, { data: { name: folder } })
|
||||
|
||||
await seedScript(
|
||||
page,
|
||||
ingest,
|
||||
[
|
||||
'-- pipeline',
|
||||
'-- on schedule',
|
||||
`-- materialize ducklake://${ordersTbl}`,
|
||||
"SELECT * FROM read_csv('s3://raw/orders/*.csv')"
|
||||
].join('\n'),
|
||||
'Ingest orders'
|
||||
)
|
||||
await seedScript(
|
||||
page,
|
||||
daily,
|
||||
[
|
||||
'-- pipeline',
|
||||
`-- on ducklake://${ordersTbl}`,
|
||||
`-- materialize ducklake://${dailyTbl}`,
|
||||
`SELECT date_trunc('day', ts) AS day, count(*) AS n FROM ducklake.${ordersTbl.replace('/', '.')} GROUP BY 1`
|
||||
].join('\n'),
|
||||
'Daily rollup'
|
||||
)
|
||||
|
||||
await page.goto(`/pipeline/${folder}`)
|
||||
|
||||
await expect(page.getByRole('heading', { name: 'Pipeline', level: 1 })).toBeVisible()
|
||||
await expect(page.getByText('2 scripts', { exact: false })).toBeVisible()
|
||||
|
||||
// A long path truncates in the node label, so match the leaf name; the full
|
||||
// paths are asserted on the edge labels below.
|
||||
await expect(page.getByText('orders_ingest').first()).toBeVisible()
|
||||
await expect(page.getByText('orders_daily').first()).toBeVisible()
|
||||
|
||||
// Two edges resolve asynchronously after the initial graph fetch (the s3 read
|
||||
// is detected from the SQL body at deploy time; the missing-schedule edge is
|
||||
// synthesized client-side by the page's per-script annotation sweep), so a
|
||||
// cold-CI failure here points at that async timing, not a missing edge.
|
||||
const edges = [
|
||||
`Edge from asset:s3object:raw/orders/*.csv to script:${ingest}`,
|
||||
`Edge from script:${ingest} to asset:ducklake:${ordersTbl}`,
|
||||
`Edge from asset:ducklake:${ordersTbl} to script:${daily}`,
|
||||
`Edge from script:${daily} to asset:ducklake:${dailyTbl}`,
|
||||
`Edge from trigger:schedule:missing:${ingest} to script:${ingest}`
|
||||
]
|
||||
for (const name of edges) {
|
||||
// Edges are SVG <g> groups (no visible box of their own), so assert they
|
||||
// are rendered into the graph rather than in-viewport visible.
|
||||
await expect(page.getByRole('group', { name, exact: true }).first()).toBeAttached()
|
||||
}
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,53 @@
|
||||
import { beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
const { listMock } = vi.hoisted(() => ({ listMock: vi.fn() }))
|
||||
|
||||
vi.mock('./shared', () => ({
|
||||
createToolDef: (_schema: unknown, name: string, description: string) => ({
|
||||
type: 'function',
|
||||
function: { name, description, parameters: {} }
|
||||
})
|
||||
}))
|
||||
|
||||
vi.mock('$lib/gen', () => ({
|
||||
WorkspaceService: { listDucklakes: listMock }
|
||||
}))
|
||||
|
||||
import { getDucklakeTools } from './ducklakeTools'
|
||||
|
||||
function createToolCallbacks() {
|
||||
return { setToolStatus: vi.fn(), removeToolStatus: vi.fn() }
|
||||
}
|
||||
|
||||
function run(name: string, args: Record<string, unknown> = {}) {
|
||||
const tool = getDucklakeTools().find((entry) => entry.def.function.name === name)
|
||||
if (!tool) throw new Error(`${name} tool not found`)
|
||||
return tool.fn({
|
||||
args,
|
||||
workspace: 'test-workspace',
|
||||
helpers: {},
|
||||
toolCallbacks: createToolCallbacks(),
|
||||
toolId: `tool-${name}`
|
||||
})
|
||||
}
|
||||
|
||||
beforeEach(() => listMock.mockReset())
|
||||
|
||||
describe('list_ducklakes', () => {
|
||||
it('returns the configured catalog names', async () => {
|
||||
listMock.mockResolvedValue(['main', 'analytics'])
|
||||
const result = await run('list_ducklakes')
|
||||
expect(listMock).toHaveBeenCalledWith({ workspace: 'test-workspace' })
|
||||
expect(JSON.parse(result)).toEqual({ ducklakes: ['main', 'analytics'] })
|
||||
})
|
||||
|
||||
it('explains the storage prerequisite with role-appropriate steps when none exist', async () => {
|
||||
listMock.mockResolvedValue([])
|
||||
const result = await run('list_ducklakes')
|
||||
expect(result).toContain('No DuckLake catalogs are configured')
|
||||
expect(result).toContain('Workspace settings → Object Storage')
|
||||
expect(result).toContain('ask a workspace admin')
|
||||
// Drafting is not blocked: the message must say the scripts can still be drafted.
|
||||
expect(result).toContain('still draft the pipeline scripts')
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,62 @@
|
||||
import { z } from 'zod'
|
||||
import { WorkspaceService } from '$lib/gen'
|
||||
import { createToolDef, type Tool } from './shared'
|
||||
|
||||
/**
|
||||
* Workspace-scoped DuckLake readiness tool, the pipeline counterpart to
|
||||
* `list_datatables` in `datatableTools.ts`.
|
||||
*
|
||||
* A data pipeline materializes DuckLake tables and reads/writes S3 assets, which
|
||||
* only work once the workspace has object storage + a DuckLake catalog
|
||||
* configured. This tool lets the chat detect that prerequisite (and warn with
|
||||
* role-appropriate next steps) instead of silently producing a pipeline that
|
||||
* cannot run. It is a plain read gated only by workspace membership, so it needs
|
||||
* no app context and belongs in the global tool set.
|
||||
*/
|
||||
|
||||
/** List the names of the DuckLake catalogs configured in the workspace. */
|
||||
export async function listDucklakes(workspace: string): Promise<string[]> {
|
||||
return await WorkspaceService.listDucklakes({ workspace })
|
||||
}
|
||||
|
||||
const NO_DUCKLAKES_CONFIGURED_MESSAGE =
|
||||
'No DuckLake catalogs are configured in this workspace. A data pipeline that materializes DuckLake tables or reads/writes S3 assets cannot run until object storage and a DuckLake catalog are set up. ' +
|
||||
'You can still draft the pipeline scripts, but tell the user how to enable it by role: a workspace admin adds object storage under Workspace settings → Object Storage (S3/Azure/GCS), then a DuckLake catalog on top of it; a user without admin rights should ask a workspace admin. ' +
|
||||
"Do not assume a default 'main' DuckLake exists."
|
||||
|
||||
const listDucklakesSchema = z.object({})
|
||||
const listDucklakesToolDef = createToolDef(
|
||||
listDucklakesSchema,
|
||||
'list_ducklakes',
|
||||
'List the DuckLake catalogs configured in this workspace, by name. Call this before building or deploying a data pipeline that materializes DuckLake tables or reads/writes S3 assets: if it returns none, the workspace has no object storage + DuckLake configured and the pipeline cannot run until a workspace admin sets it up. Returns names only.'
|
||||
)
|
||||
|
||||
/** The workspace DuckLake tools, for registration in global mode. */
|
||||
export function getDucklakeTools(): Tool<{}>[] {
|
||||
return [
|
||||
{
|
||||
def: listDucklakesToolDef,
|
||||
fn: async ({ workspace, toolId, toolCallbacks }) => {
|
||||
toolCallbacks.setToolStatus(toolId, { content: 'Listing DuckLake catalogs...' })
|
||||
try {
|
||||
const ducklakes = await listDucklakes(workspace)
|
||||
if (ducklakes.length === 0) {
|
||||
toolCallbacks.setToolStatus(toolId, {
|
||||
content:
|
||||
'No DuckLake configured — set up object storage + DuckLake in workspace settings'
|
||||
})
|
||||
return NO_DUCKLAKES_CONFIGURED_MESSAGE
|
||||
}
|
||||
toolCallbacks.setToolStatus(toolId, {
|
||||
content: `Listed ${ducklakes.length} DuckLake catalog(s)`
|
||||
})
|
||||
return JSON.stringify({ ducklakes }, null, 2)
|
||||
} catch (e) {
|
||||
const errorMsg = `Error listing DuckLake catalogs: ${e instanceof Error ? e.message : String(e)}`
|
||||
toolCallbacks.setToolStatus(toolId, { content: errorMsg, error: errorMsg })
|
||||
return errorMsg
|
||||
}
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -96,6 +96,7 @@ import { searchDocsTool, readDocsPageTool } from '../docs/core'
|
||||
import { createDbSchemaTool } from '../script/core'
|
||||
import type { ContextElement } from '../context'
|
||||
import { getDatatableTools } from '../datatableTools'
|
||||
import { getDucklakeTools } from '../ducklakeTools'
|
||||
import { fileTools } from '../files/fileTools'
|
||||
import type { AttachedFilesStore } from '../files/attachedFiles.svelte'
|
||||
import { artifactTools } from '../artifacts/artifactTools'
|
||||
@@ -3493,6 +3494,8 @@ export const globalTools: Tool<{}>[] = [
|
||||
},
|
||||
// Workspace-scoped datatable tools (unrestricted: no whitelist, no creation policy)
|
||||
...getDatatableTools(),
|
||||
// Workspace DuckLake readiness (storage prerequisite check for pipelines)
|
||||
...getDucklakeTools(),
|
||||
// Read-only tools over files the user attached to the conversation
|
||||
...fileTools,
|
||||
// Search + call access to the backend API endpoint catalog, for operations
|
||||
|
||||
@@ -755,6 +755,24 @@ export const PIPELINE_BASE = `# Data pipeline authoring
|
||||
|
||||
A **data pipeline** is NOT a flow. A flow is one runnable that orchestrates steps internally. A data pipeline is a set of **independent scripts**, each deployed on its own, that form a DAG by reading and writing shared **storage assets** (DuckLake tables, data tables, S3 objects, volumes, resources) and by declaring execution **triggers**. The pipeline is visualized and edited at \`/pipeline/<folder>\`; every node is a normal workspace script that happens to carry pipeline annotations. When the user asks for a "data pipeline" (or to "ingest / transform / materialize" data across steps), build pipeline-annotated scripts — do NOT build a flow.
|
||||
|
||||
## Default to DuckDB + DuckLake
|
||||
|
||||
A pipeline node that produces a table should almost always be a **\`duckdb\`** node that materializes its output into a **DuckLake** table with \`-- materialize ducklake://<name>/<table>\` (in a DuckDB node the annotation uses SQL \`--\` comment syntax; write the body as a bare \`SELECT\` and let the runtime do the write). DuckLake is the default lakehouse store for pipelines and is the shape the pipeline editor is built around, so prefer it unless the work specifically calls for something else:
|
||||
|
||||
- \`postgresql\` / data tables — only for row-level, OLTP-style mutations against an existing Postgres data table (frequent single-row upserts/updates, transactional reads that an app queries live).
|
||||
- \`bun\` / \`python3\` — only for non-tabular work that doesn't map to SQL: calling an external API, wrangling files, arbitrary glue. When such a node still produces tabular data for downstream steps, land it in DuckLake (write it with the wmill SDK / ducklake helpers) rather than inventing a parallel store.
|
||||
|
||||
Do not spread a pipeline across postgres, S3, and DuckLake when one DuckLake lake would do; a consistent DuckLake lakehouse is the goal.
|
||||
|
||||
## Storage prerequisites
|
||||
|
||||
A DuckLake pipeline only runs once the workspace has **object storage** (S3 / Azure Blob / GCS) **and a DuckLake catalog** configured — DuckLake tables and \`s3://\` assets can't be materialized or read without it. Check with the \`list_ducklakes\` tool before you build (it returns the configured DuckLake catalogs, or none). Drafting the annotated scripts does not require storage, but the pipeline can't ingest, materialize, or read its assets until it exists. So if \`list_ducklakes\` returns none (or the user hits "storage not configured" errors), say so and give the right next step **by role**:
|
||||
|
||||
- a workspace **admin** sets it up in Workspace settings → Object Storage (add an S3/Azure/GCS storage), then adds a DuckLake catalog on top of it;
|
||||
- anyone **without admin rights** should ask a workspace admin to configure object storage + a DuckLake catalog.
|
||||
|
||||
Never hand back a DuckLake pipeline that cannot run without flagging the missing storage and pointing to who sets it up.
|
||||
|
||||
## What makes a script a pipeline node
|
||||
|
||||
A script joins the pipeline when its source begins with the \`pipeline\` annotation as a top-of-file comment, **written in the script's own comment syntax** — \`//\` for TS/JS (bun), \`--\` for SQL (DuckDB/Postgres), \`#\` for Python/Bash. So it's \`-- pipeline\` in a DuckDB node, \`# pipeline\` in a Python node, \`// pipeline\` in a bun node. Every annotation below uses that same prefix (the \`//\` shown is the TS form). All other wiring is expressed as annotation comments near the top of the file:
|
||||
@@ -784,7 +802,7 @@ A script joins the pipeline when its source begins with the \`pipeline\` annotat
|
||||
## How to build one in chat
|
||||
|
||||
1. Put every node in the **same folder**: \`f/<folder>/<name>\`. The folder is the pipeline.
|
||||
2. Author each node as a **script draft** with \`write_script\` (or \`edit_script\`), language chosen for the work: \`duckdb\` or \`postgresql\` for SQL-shaped data work, \`bun\`/\`python3\` for general transforms. SQL-heavy lakehouse steps usually use \`duckdb\`.
|
||||
2. Author each node as a **script draft** with \`write_script\` (or \`edit_script\`). Default to \`duckdb\` materializing into DuckLake (see "Default to DuckDB + DuckLake" above); pick \`postgresql\`, \`bun\`, or \`python3\` only when that section says the work calls for it.
|
||||
3. Start each body with \`// pipeline\`, then the \`// on\` input declarations, then the transform that writes the output.
|
||||
4. **Chain nodes by asset URI**: read an upstream node's output asset, then \`// on <that-same-uri>\` in the downstream node so the edge forms. Reuse exact asset paths from existing nodes rather than inventing parallel ones.
|
||||
5. Leave nodes as drafts unless the user asks to deploy. A pipeline only "runs" once its scripts are deployed and their triggers exist.
|
||||
|
||||
@@ -2,6 +2,24 @@
|
||||
|
||||
A **data pipeline** is NOT a flow. A flow is one runnable that orchestrates steps internally. A data pipeline is a set of **independent scripts**, each deployed on its own, that form a DAG by reading and writing shared **storage assets** (DuckLake tables, data tables, S3 objects, volumes, resources) and by declaring execution **triggers**. The pipeline is visualized and edited at `/pipeline/<folder>`; every node is a normal workspace script that happens to carry pipeline annotations. When the user asks for a "data pipeline" (or to "ingest / transform / materialize" data across steps), build pipeline-annotated scripts — do NOT build a flow.
|
||||
|
||||
## Default to DuckDB + DuckLake
|
||||
|
||||
A pipeline node that produces a table should almost always be a **`duckdb`** node that materializes its output into a **DuckLake** table with `-- materialize ducklake://<name>/<table>` (in a DuckDB node the annotation uses SQL `--` comment syntax; write the body as a bare `SELECT` and let the runtime do the write). DuckLake is the default lakehouse store for pipelines and is the shape the pipeline editor is built around, so prefer it unless the work specifically calls for something else:
|
||||
|
||||
- `postgresql` / data tables — only for row-level, OLTP-style mutations against an existing Postgres data table (frequent single-row upserts/updates, transactional reads that an app queries live).
|
||||
- `bun` / `python3` — only for non-tabular work that doesn't map to SQL: calling an external API, wrangling files, arbitrary glue. When such a node still produces tabular data for downstream steps, land it in DuckLake (write it with the wmill SDK / ducklake helpers) rather than inventing a parallel store.
|
||||
|
||||
Do not spread a pipeline across postgres, S3, and DuckLake when one DuckLake lake would do; a consistent DuckLake lakehouse is the goal.
|
||||
|
||||
## Storage prerequisites
|
||||
|
||||
A DuckLake pipeline only runs once the workspace has **object storage** (S3 / Azure Blob / GCS) **and a DuckLake catalog** configured — DuckLake tables and `s3://` assets can't be materialized or read without it. Check with the `list_ducklakes` tool before you build (it returns the configured DuckLake catalogs, or none). Drafting the annotated scripts does not require storage, but the pipeline can't ingest, materialize, or read its assets until it exists. So if `list_ducklakes` returns none (or the user hits "storage not configured" errors), say so and give the right next step **by role**:
|
||||
|
||||
- a workspace **admin** sets it up in Workspace settings → Object Storage (add an S3/Azure/GCS storage), then adds a DuckLake catalog on top of it;
|
||||
- anyone **without admin rights** should ask a workspace admin to configure object storage + a DuckLake catalog.
|
||||
|
||||
Never hand back a DuckLake pipeline that cannot run without flagging the missing storage and pointing to who sets it up.
|
||||
|
||||
## What makes a script a pipeline node
|
||||
|
||||
A script joins the pipeline when its source begins with the `pipeline` annotation as a top-of-file comment, **written in the script's own comment syntax** — `//` for TS/JS (bun), `--` for SQL (DuckDB/Postgres), `#` for Python/Bash. So it's `-- pipeline` in a DuckDB node, `# pipeline` in a Python node, `// pipeline` in a bun node. Every annotation below uses that same prefix (the `//` shown is the TS form). All other wiring is expressed as annotation comments near the top of the file:
|
||||
@@ -31,7 +49,7 @@ A script joins the pipeline when its source begins with the `pipeline` annotatio
|
||||
## How to build one in chat
|
||||
|
||||
1. Put every node in the **same folder**: `f/<folder>/<name>`. The folder is the pipeline.
|
||||
2. Author each node as a **script draft** with `write_script` (or `edit_script`), language chosen for the work: `duckdb` or `postgresql` for SQL-shaped data work, `bun`/`python3` for general transforms. SQL-heavy lakehouse steps usually use `duckdb`.
|
||||
2. Author each node as a **script draft** with `write_script` (or `edit_script`). Default to `duckdb` materializing into DuckLake (see "Default to DuckDB + DuckLake" above); pick `postgresql`, `bun`, or `python3` only when that section says the work calls for it.
|
||||
3. Start each body with `// pipeline`, then the `// on` input declarations, then the transform that writes the output.
|
||||
4. **Chain nodes by asset URI**: read an upstream node's output asset, then `// on <that-same-uri>` in the downstream node so the edge forms. Reuse exact asset paths from existing nodes rather than inventing parallel ones.
|
||||
5. Leave nodes as drafts unless the user asks to deploy. A pipeline only "runs" once its scripts are deployed and their triggers exist.
|
||||
|
||||
Reference in New Issue
Block a user