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 work that doesn't map to SQL: calling an external API, wrangling files, arbitrary glue. When such a node produces tabular data for downstream steps, write it straight into DuckLake (or a data table) from the same script — see "Ingesting from Python / TypeScript" below — 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.

Ingesting from Python / TypeScript

A node that fetches data in code (an API, a SaaS export, a scraper) writes the rows directly into their DuckLake table with wmill.ducklake(), or into a data table with wmill.datatable(). DuckLake already stores every table as parquet files in the workspace object storage, so never stage the raw payload first — no intermediate JSON / CSV / parquet file in S3 and no second "load the raw file" node. One ingestion node, one table.

Pass the rows as a query argument (a list of dicts / array of objects is sent as JSON) and let SQL unnest them; declare the column types so the table gets a real schema:

# pipeline
# on schedule
import wmill

def main():
    rows = fetch_charges()  # list[dict]
    lake = wmill.ducklake()
    lake.query("""
        CREATE OR REPLACE TABLE stripe_charges AS
        SELECT unnest(from_json($rows, '[{"id":"VARCHAR","amount":"BIGINT","created":"BIGINT"}]'), recursive := true)
    """, rows=rows).fetch()

// pipeline
// on schedule
import * as wmill from "windmill-client"

export async function main() {
  const rows = await fetchCharges()
  const lake = wmill.ducklake()
  await lake`CREATE OR REPLACE TABLE stripe_charges AS
    SELECT unnest(from_json(${rows}, '[{"id":"VARCHAR","amount":"BIGINT","created":"BIGINT"}]'), recursive := true)`.fetch()
}

• Use INSERT INTO <table> SELECT ... instead of CREATE OR REPLACE to append (an incremental pull), with a one-time CREATE TABLE IF NOT EXISTS first.
• Assign the client to a variable (lake = wmill.ducklake()) and keep the SQL a string literal (Python) or a template without sql.raw (TS): that is what lets the graph record the node's write to ducklake://main/<table>. A chained wmill.ducklake().query(...) or SQL built in a variable records no output, so the downstream edge never forms.
• wmill.ducklake("<name>") targets a non-default catalog; wmill.datatable("<name>") writes a Postgres data table the same way (Python db.query(sql, *args) with $1 placeholders, TS the same template form) — use it only when the data belongs in Postgres (see above).
• The downstream node triggers on the table itself: -- on ducklake://main/stripe_charges.

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 which DuckLake catalogs the workspace has before you build. wmill ducklake list lists them. Drafting the annotated scripts does not require storage, but the pipeline can't ingest, materialize, or read its assets until it exists. So if there is 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:

• // on <ref> — declares an execution-DAG input (what triggers/feeds this node). <ref> is either:
  • an asset URI (the node runs when that asset is produced upstream): ducklake://main/orders, datatable://main/users, $res:f/folder/my_resource, volume://name/path, or an S3 object (see the S3 storage-form rule below).
  • a native trigger kind: schedule, webhook, email, kafka, mqtt, amqp, nats, postgres, sqs, gcp, or data_upload (a user-uploaded S3 file). For these the actual trigger row (cron, topic, …) is created separately; the annotation only declares the binding. data_upload is special: there is no trigger row — the node instead declares an S3Object input parameter fed by the auto-generated upload picker; it never hard-codes a key. Any language can be the data_upload node:
    • Python (has import wmill): def main(file: wmill.S3Object): then wmill.load_s3_file(file); TS (has import * as wmill from "windmill-client"): export async function main(file: wmill.S3Object). Qualify the type as wmill.S3Object (or add from wmill import S3Object / import { S3Object } from "windmill-client") — a bare S3Object is undefined.
    • DuckDB takes the s3object arg via a -- $<name> (s3object) declaration and reads it directly, so a single DuckDB node can ingest and materialize:

      -- pipeline
      -- on data_upload
      -- materialize ducklake://main/raw_uploads
      -- $file (s3object)
      SELECT * FROM read_csv($file)
      

• Outputs are inferred from what the body writes — a CREATE TABLE, a wmill.writeS3File(...), a DuckLake/datatable write. To declare a managed output explicitly, use // materialize <asset-uri>.
• Optional badges: // partitioned <daily|hourly|weekly|monthly|dynamic>, // freshness <duration> (e.g. 1h), // tag <worker-tag>, // retry <count> [delay], // data_test <kind> ... (managed DuckLake targets only — deploy rejects it beside a dbt:// one), // measure <name> = <agg> [where <pred>] and // dimension <name> = <expr> (see "Declared metrics" below).

S3 object wiring (storage form matters)

An s3:// URI's first slashes select the storage, not part of the key — get this wrong and the producer/consumer edge silently won't connect:

• s3:///<key> (triple slash, empty first segment) = the default workspace storage. A downstream node reading or triggering on that object uses s3:///<key> — e.g. DuckDB -- on s3:///orders/2024.parquet and read_parquet('s3:///orders/2024.parquet').
• s3://<storage>/<key> (double slash, non-empty first segment) = a named secondary storage called <storage> — so s3://ingest/x means storage ingest, key x, NOT key ingest/x. Only use this when the object genuinely lives in a configured secondary storage; never invent a bucket/storage name for a default-storage object (it breaks the edge).

To make the producer side visible to lineage, a Python/TS node MUST pass the S3Object form, not a bare key string: Python wmill.write_s3_file(wmill.S3Object(s3="<key>"), data) (or the import-free dict {"s3": "<key>"}), TS wmill.writeS3File({ s3: "<key>" }, data). That records the default-storage asset /<key>, which a downstream s3:///<key> reader connects to (same key both sides). A bare write_s3_file("<key>", ...) records no asset and produces no edge. Add storage="<name>" only for a named secondary storage.

The key must be a string literal — the graph parser is static and cannot follow a variable, f-string, or computed path, so write_s3_file(wmill.S3Object(s3=key_var), ...) records no edge. Inline the literal (s3="events/user_events.parquet") on both the writing and reading node. The same rule applies to every asset URI in an annotation or SDK call (ducklake://, datatable://, s3://): write them literally, not via a variable.

Materialize (the managed output)

A MANAGED // materialize is DuckDB-only, and its target must be a DuckLake table (ducklake://<name>/<table>). Deploy rejects a ducklake:// // materialize on any other language (python3, bun, postgresql). For a non-DuckDB node writing the lake, do not use // materialize — write the table itself with wmill.ducklake() (see "Ingesting from Python / TypeScript") and let the output be inferred. Use duckdb when a node should materialize a DuckLake table.

The one target ANY language may declare (except a dbt script, whose writes come from its manifest) is a warehouse relation: // materialize manual dbt://<warehouse>/<schema>/<name>, where <warehouse> is a warehouse the workspace configures under Settings → dbt. manual is the only mode it has — nothing generates warehouse DDL, so the node issues its own write (a postgresql CREATE TABLE / INSERT, an SDK load, …) and the annotation records the outcome. Use it on an ingestion node whose output a dbt project reads as a source: the declared relation and the dbt model land on ONE graph node, and a downstream // on dbt://<warehouse>/<schema>/<name> fires when the ingestion node completes.

A managed // materialize ducklake://<name>/<table> tells the runtime to write the node's output table for you: write the body as a single SELECT and the runtime wraps it in the create/replace — do not also write your own CREATE TABLE / INSERT. (The opposite holds for the dbt:// target above: there the node writes its own DDL and the strategies below do not apply.) Write strategy:

• no option → replace the whole table each run (full refresh; the only mode whose output columns may change);
• // materialize <uri> append → INSERT-append rows (incremental);
• // materialize <uri> key=<col> → merge/upsert on <col>.

// materialize manual <uri> opts out of managed writes — the script writes its own DDL and the annotation only records the output asset for lineage.

materialize pairs with partitioning for incremental pipelines: a // partitioned <daily|hourly|weekly|monthly|dynamic> node runs once per partition (append/merge into a fixed-schema table). The {partition} token, usable in any asset URI and in the body SQL, is replaced at run time by the current partition's identity string:

• To filter the source to the active slice on a time grain, use the runtime-injected macro: WHERE wm_partition(<ts_col>) = {partition}. wm_partition(ts) buckets a timestamp in exactly the identity format the runtime uses for daily/hourly/weekly/monthly, so it always matches; never hand-write a strftime format.
• Do NOT write = TIMESTAMP {partition}: the identity string is not a valid timestamp literal for hourly/weekly/monthly and errors at run time.
• For dynamic partitioning the identity is the caller-supplied key (not a timestamp, no macro), so filter on it directly: WHERE <your_key_col> = {partition}.

materialize is an output declaration on a node — not a command. There is no "materialize run".

Declared metrics (measure / dimension)

On a node that materializes a DuckLake table, // measure <name> = <aggregate> [where <predicate>] names the canonical way to aggregate that table (e.g. // measure revenue = sum(amount) where not is_refund), and // dimension <name> = <expr> names a way to slice it (e.g. // dimension region = region, // dimension month = date_trunc('month', ordered_at)). They execute nothing: they are catalogued at deploy so the editor and other agents reuse the definition instead of re-deriving it and silently disagreeing.

• Keep the predicate in the where clause rather than folding it into the aggregate: it is rendered as <agg> FILTER (WHERE <pred>), which is what lets two measures with different predicates share one GROUP BY.
• DuckLake-only, and only meaningful next to // materialize.
• Declare one when a number carries a judgement call someone else would get wrong (refunds excluded, test rows dropped, which column is the amount); do NOT blanket every table with measures — an obvious count(*) earns nothing.
• To use a metric another node declares, read that node and reuse its exact expression rather than guessing it.

How to build one

1. Put every node in the same folder: f/<folder>/<name>. The folder is the pipeline.
2. Write each node as its own 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. Don't deploy nodes unless the user asks to. A pipeline only "runs" once its scripts are deployed and their triggers exist.

Locally:

• create each node with wmill script new f/<folder>/<name> <language>, then write its body;
• wmill pipeline show <folder> --local draws the graph from your working tree — check that every edge you meant to form is there;
• wmill pipeline dev <folder> live-previews the pipeline, and wmill pipeline run <folder> --local runs the cascade from local files without deploying (--dry-run prints the plan first);
• a trigger such as // on schedule only declares the binding: the schedule or trigger itself is created separately (see the schedules and triggers skills).

Example (DuckDB → DuckLake, scheduled ingest + downstream transform)

Node f/sales/orders_ingest (runs on a schedule, materializes a DuckLake table):

-- pipeline
-- on schedule
-- materialize ducklake://main/orders
SELECT * FROM read_csv('s3:///raw/orders/*.csv')

Node f/sales/orders_daily (runs when orders is produced, writes a rollup):

-- pipeline
-- on ducklake://main/orders
-- materialize ducklake://main/orders_daily
SELECT date_trunc('day', ts) AS day, count(*) AS n
FROM ducklake.main.orders GROUP BY 1