Skip to content

function-patterns.md

Latest commit

 

History

History
255 lines (201 loc) · 10 KB

function-patterns.md

File metadata and controls

255 lines (201 loc) · 10 KB
Raw
description How to write each of the five VGI function patterns: scalar, table, table-in-out, aggregate, and buffering — with a complete runnable worker for each.

Function patterns

What this is: a recipe showing each of the five VGI function patterns, with a complete, runnable worker for each.
Who it's for: developers who've finished the tutorial and want to know which pattern fits their problem.

Prerequisites

  • You can build and run a worker (see the tutorial).
  • Python 3.13+, uv, and a DuckDB-compatible engine (Haybarn or stock DuckDB).

Which pattern do I need?

Pattern Shape Use it when… SQL
Scalar 1 row → 1 row you transform each row independently SELECT f(col) FROM t
Table args → N rows you generate rows from arguments SELECT * FROM f(args)
Table-in-out N rows → M rows you reshape/filter a streamed table SELECT * FROM f((SELECT …))
Aggregate grouped rows → 1 row/group you accumulate per GROUP BY group SELECT f(col) FROM t GROUP BY k
Buffering stream → [state] → stream you must see every row first (sort, top-k, full reduction) SELECT * FROM f((SELECT …))

Scalar

![scalar shape](../assets/kinds/scalar.svg){ .kind-banner__glyph }
`1 row → 1 value`{ .kind-banner__formula }

Runs on each row independently and returns a single value — a pure per-row transform. { .kind-banner__desc }

One row in, one row out. Operate on the whole column with pyarrow.compute; the Annotated types are the schema.

--8<-- "examples/calc_scalar_worker.py"
ATTACH 'calc' (TYPE vgi, LOCATION 'uv run calc_scalar_worker.py');
SELECT calc.double(n) FROM (VALUES (1), (2), (3)) AS t(n);

Scalars aren't just for numbers — any column type works. A string transform looks the same; here compute joins Hello, + name + ! across the column:

--8<-- "examples/greeting_scalar_worker.py"
SELECT greetings.greeting(name) FROM (VALUES ('Alice'), ('Bob')) AS t(name);

Table

![table shape](../assets/kinds/table.svg){ .kind-banner__glyph }
`args → N rows`{ .kind-banner__formula }

A table-valued source: scalar arguments in, a whole set of rows out. { .kind-banner__desc }

Generate rows from arguments, no input table. Declare a typed args dataclass and a FIXED_SCHEMA; process emits batches until out.finish(). The @bind_fixed_schema / @init_single_worker decorators wire up the common single-worker lifecycle. (This is the full tutorial worker — the scalar double plus the series generator.)

--8<-- "examples/calc_worker.py"
ATTACH 'calc' (TYPE vgi, LOCATION 'uv run calc_worker.py');
SELECT * FROM calc.series(3);

Streaming with state

The tutorial's series emits every row in a single process call. That's fine for small results, but process is actually called repeatedly until you call out.finish() — so for large output you emit a bounded chunk per call and remember your place in a small state object. The state extends ArrowSerializableDataclass so it survives HTTP state round-trips:

--8<-- "examples/series_streaming_worker.py"
SELECT * FROM calc.series(1000000);   -- streamed CHUNK rows per process() call

Table-in-out

![table-in-out shape](../assets/kinds/table-in-out.svg){ .kind-banner__glyph }
`N rows → M rows`{ .kind-banner__formula }

Consumes a relation and streams a transformed relation back, batch by batch. { .kind-banner__desc }

Stream an input table through, batch by batch, emitting transformed output. on_bind declares the output schema; process receives each input batch and emits results. Here we keep only rows whose value column is positive.

--8<-- "examples/filter_worker.py"
ATTACH 'filters' (TYPE vgi, LOCATION 'uv run filter_worker.py');
SELECT * FROM filters.filter_positive((SELECT * FROM my_table));

Aggregate

![aggregate shape](../assets/kinds/aggregate.svg){ .kind-banner__glyph }
`N rows → 1 value`{ .kind-banner__formula }

Folds many rows down into a single value per group. { .kind-banner__desc }

Accumulate input rows into per-group state, then emit one row per group. Aggregates are driven by DuckDB's GROUP BY and run in three phases — update (fold a batch into per-group state), combine (merge partial states across parallel workers), and finalize (state → output row).

--8<-- "examples/sum_worker.py"
ATTACH 'aggregates' (TYPE vgi, LOCATION 'uv run sum_worker.py');
SELECT category, aggregates.vgi_sum(value) FROM t GROUP BY category;

??? info "State must be serializable" Table generators and aggregates keep state between calls. A state type must provide serialize_to_bytes() and deserialize_from_bytes() so the framework can carry it across HTTP round-trips or persist it between aggregate calls. ArrowSerializableDataclass implements these methods for you and is the usual starting point. Custom state types can implement the same [StreamStateCodec][vgi.function.StreamStateCodec] contract without inheriting from ArrowSerializableDataclass. See State Class for both approaches.

Buffering

![buffering shape](../assets/kinds/buffering.svg){ .kind-banner__glyph }
`stream → [state] → stream`{ .kind-banner__formula }

Holds every input row in state before emitting — the basis for sorts, top-k, and full-stream reductions. { .kind-banner__desc }

When a function must see the whole input before it can produce any output — a global sort, top-k, or a full reduction — use a buffering function. Unlike table-in-out (which emits per input batch), it runs in three phases: process (the sink — stash each batch, return a state_id), combine (reduce all the partials once), and finalize (the source — stream the result out). Because the phases can run in different worker processes, state can't live in memory — it goes in params.storage, a shared store keyed to this call (its execution_id). See Persist state across workers for the backends behind it.

--8<-- "examples/row_count_worker.py"
ATTACH 'buffers' (TYPE vgi, LOCATION 'uv run row_count_worker.py');
SELECT * FROM buffers.row_count((SELECT * FROM big_table));

Getting the input in order

The sink runs in parallel across DuckDB threads, so process sees batches in no particular order. When output depends on input order — a running total, a row pattern match — set Meta.requires_input_batch_index = True and each process call receives params.batch_index, a monotonic ordinal you can sort the buffered batches by in finalize.

Not every source can supply one (a base table scan can; range() and VALUES cannot). When it cannot, the extension serializes the sink and numbers the batches itself, so a worker always gets a valid index and callers never have to wrap their input. This example reports the indices it saw, one row per buffered batch:

--8<-- "examples/batch_index_worker.py"
ATTACH 'bi' (TYPE vgi, LOCATION 'uv run batch_index_worker.py');
SELECT * FROM bi.batch_indexes((SELECT * FROM range(5000))) ORDER BY batch_index;

The alternative, Meta.sink_order_dependent = True, delivers ordered input by forcing a single-threaded sink. It works with any source but gives up parallel ingest, which is usually the bulk of a buffering query's wall time. The two flags are mutually exclusive.

??? info "Buffering vs. table-in-out" Both consume a relation, but a table-in-out function emits per input batch and never holds the whole input — use it for streaming transforms (filter, enrich, reshape). Reach for buffering only when output genuinely depends on every row. See Persist state across workers for the storage backends params.storage uses.

Next steps