Skip to content
Query.Farm
Talk with Us

Function patterns

Each of the five VGI function shapes in TypeScript, with a complete, runnable worker for each — so you can find the shape that fits your problem. (Do the tutorial first.)

Session setup

Each section assumes you’re in a Haybarn shell that has loaded the extension once with INSTALL vgi FROM community; then LOAD vgi;, and that you’ve installed the SDK in the directory holding the worker:

bun add @query-farm/vgi @query-farm/apache-arrow @query-farm/vgi-rpc

There is no build step — LOCATION names the command that runs the file, and bun run executes TypeScript directly. LOCATION is resolved relative to the directory the engine was started in.

Which shape do I need?

Section titled “Which shape do I need?”
If your function…UseFactory
maps each row independentlyScalardefineScalarFunction
produces rows from argumentsTabledefineTableFunction
transforms a relation as it streamsTable-in-outdefineTableInOutFunction
folds rows to one value per groupAggregatedefineAggregate
needs every row before it can answerBufferingdefineTableBufferingFunction

Scalar

Section titled “Scalar”
scalar shape
1 row → 1 value

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

One value out per value in. No state, no finalize phase — compute gets a whole column and returns a column of the same length.

calcscalar.ts 
// Copyright 2025, 2026 Query Farm LLC - https://query.farm

// calcscalar is the worker built in step 1 of the vgi-typescript tutorial: one
// scalar function, served over stdio, callable from DuckDB as calc.double().
//
// A scalar function is the simplest shape — one row in, one value out, with no
// state and no finalize phase. DuckDB hands the worker a whole Arrow column and
// expects a column of the same length back.
//
//   bun run calcscalar.ts
//   # then, in a Haybarn shell:
//   ATTACH 'calc' (TYPE vgi, LOCATION 'bun run /abs/path/calcscalar.ts');
//   SELECT calc.double(21);

import { Worker, defineScalarFunction, int } from "@query-farm/vgi";

// `int` is the Int64 type alias. compute() therefore reads and returns bigint —
// DuckDB's BIGINT does not fit in a JS number, so the SDK never narrows it.
export const double = defineScalarFunction({
  name: "double",
  description: "Doubles a BIGINT",
  params: { n: int },
  returns: int,

  // compute runs once per input BATCH, not per row. Arguments are positional:
  // column 0 is the first argument, whatever the params key is called.
  compute: (batch) => {
    // A column is an erased `Iterable<unknown>` — the facade cannot know the
    // value type, because arrow-js and flechette parameterize differently. The
    // cast at the use site is the intended pattern, and under `strict` it is
    // required: without it `Array.from`'s mapper gets `unknown`.
    const ns = batch.getChildAt(0)! as Iterable<bigint | null>;
    // One output value per input row. null in → null out.
    return Array.from(ns, (v) => (v == null ? null : v * 2n));
  },
});

// Functions are served through a catalog, and the catalog's name is the name
// DuckDB ATTACHes. They must match: attaching under any other name fails.
export const worker = new Worker({
  catalog: { name: "calc", schemas: [{ name: "main", functions: [double] }] },
});

if (import.meta.main) worker.run();

ATTACH 'calc' (TYPE vgi, LOCATION 'bun run ./calcscalar.ts');
SELECT calc.double(21);

Output

double(21)
42

Table

Section titled “Table”
table shape
args → N rows

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

Generate rows from arguments, with no input relation. Arguments are read once in initialState; process is then called repeatedly until it calls out.finish().

calc.ts 
// Copyright 2025, 2026 Query Farm LLC - https://query.farm

// calc is the worker built across the vgi-typescript tutorial: one scalar
// function and one table function in a single catalog.
//
// The scalar `double` transforms a column in place. The table function `series`
// *generates* rows from an argument, so it is called in a FROM clause rather
// than an expression. One worker can serve any mix of shapes.
//
//   bun run calc.ts
//   # then, in a Haybarn shell:
//   ATTACH 'calc' (TYPE vgi, LOCATION 'bun run /abs/path/calc.ts');
//   SELECT calc.double(21);
//   SELECT * FROM calc.series(3);

import {
  Worker,
  defineScalarFunction,
  defineTableFunction,
  batchFromColumns,
  toSchema,
  int,
} from "@query-farm/vgi";

// ── scalar: double(n) ───────────────────────────────────────────────────────

export const double = defineScalarFunction({
  name: "double",
  description: "Doubles a BIGINT",
  params: { n: int },
  returns: int,
  compute: (batch) => {
    // Columns are erased `Iterable<unknown>`; cast at the use site, where the
    // declared params make the value type known.
    const ns = batch.getChildAt(0)! as Iterable<bigint | null>;
    return Array.from(ns, (v) => (v == null ? null : v * 2n));
  },
});

// ── table: series(count) ────────────────────────────────────────────────────

// The output schema is fixed, so it can be built once at module scope. A table
// function whose columns depend on its arguments would build it in onBind.
const seriesSchema = toSchema({ n: int });

const BATCH_SIZE = 1024;

export const series = defineTableFunction({
  name: "series",
  description: "Generates the integers 0..count-1",

  args: { count: int },
  argDocs: { count: "How many numbers to generate" },
  // Enforced at bind, not just advertised: series(-1) fails before any row is
  // produced, rather than looping or silently returning nothing.
  argConstraints: { count: { ge: 0 } },

  onBind: () => ({ outputSchema: seriesSchema }),

  // Runs once per scan, after bind. Arguments are fixed for the whole scan, so
  // this is where they are read — decoding them per batch would be waste.
  //
  // BigInt() is not decoration. A table function's `args` arrive as JS numbers
  // even for an int64 argument, while the same type reaches a *scalar*
  // function's columns as bigint. Normalize once here and the rest of the
  // function can do bigint arithmetic without a runtime "Invalid mix of BigInt
  // and other type" surprise on the first batch.
  initialState: ({ args }) => ({ i: 0n, count: BigInt(args.count) }),

  // process is the pull loop: DuckDB calls it repeatedly and consumes lazily.
  // Emit what you can, then finish() to signal end-of-stream. Nothing has to be
  // materialized up front, which is what makes a table function a *generator*.
  process: (_params, state, out) => {
    if (state.i >= state.count) return out.finish();
    const end = state.count - state.i > BigInt(BATCH_SIZE)
      ? state.i + BigInt(BATCH_SIZE)
      : state.count;
    const ns: bigint[] = [];
    for (let k = state.i; k < end; k++) ns.push(k);
    out.emit(batchFromColumns({ n: ns }, seriesSchema));
    state.i = end;
  },
});

export const worker = new Worker({
  catalog: {
    name: "calc",
    comment: "Tutorial worker: a scalar and a table function",
    schemas: [{ name: "main", functions: [double, series] }],
  },
});

if (import.meta.main) worker.run();

ATTACH 'calc' (TYPE vgi, LOCATION 'bun run ./calc.ts');
SELECT * FROM calc.series(3);
Input
count
3
Output
n
0
1
2

Table-in-out

Section titled “Table-in-out”
table-in-out shape
N rows → M rows

Consumes a relation and streams a transformed relation back, batch by batch.

Stream an input relation through, emitting per batch. Memory stays flat however large the scan is, because nothing is held back.

filter.ts 
// Copyright 2025, 2026 Query Farm LLC - https://query.farm

// filter is the table-in-out example for the vgi-typescript documentation.
//
// A table-in-out function consumes a relation and streams a transformed
// relation back, batch by batch. Unlike a scalar it may change the row count,
// and unlike a buffering function it never holds the whole input — each process
// call emits what it can from the batch in hand, which is what keeps memory
// flat over an arbitrarily large scan.
//
//   bun run filter.ts
//   # then, in a Haybarn shell:
//   ATTACH 'filters' (TYPE vgi, LOCATION 'bun run /abs/path/filter.ts');
//   SELECT * FROM filters.filter_positive((SELECT * FROM t));

import {
  Worker,
  defineTableInOutFunction,
  batchFromColumns,
  type TableInOutBindParams,
} from "@query-farm/vgi";

export const filterPositive = defineTableInOutFunction({
  name: "filter_positive",
  description: "Keeps only the rows whose `value` column is greater than zero",

  // No `args` entry declares the input relation. A table-in-out function's
  // TABLE argument is implicit: it arrives as the stream of batches process()
  // is called with, and its schema is on params.bindCall.input_schema.
  onBind: (params: TableInOutBindParams) => {
    const input = params.bindCall.input_schema;
    if (!input) throw new Error("filter_positive requires a table argument");
    // Output shape matches input shape — this function drops rows, not columns.
    return { outputSchema: input };
  },

  // Called once per input batch. Emit zero or more batches; returning without
  // emitting is how a batch is dropped entirely.
  process: (params, _state, batch, out) => {
    const values = batch.getChild("value");
    if (!values) throw new Error("expected a `value` column");

    // The column's width comes from the caller's relation, not from this
    // function, so `value` may arrive as a JS number (int32 and narrower) or a
    // bigint (int64). Comparing against 0 works for both — JS allows mixed
    // number/bigint *comparison*, just not mixed arithmetic — and the values
    // are handed back untouched, so the codec rebuilds the original type.
    const kept: (number | bigint)[] = [];
    for (let i = 0; i < batch.numRows; i++) {
      // get() returns `unknown` for the same reason — narrow it here.
      const v = values.get(i) as number | bigint | null;
      if (v != null && v > 0) kept.push(v);
    }

    // An empty batch is legal but pointless — skip the round trip.
    if (kept.length === 0) return;
    out.emit(batchFromColumns({ value: kept }, params.outputSchema));
  },
});

export const worker = new Worker({
  catalog: {
    name: "filters",
    comment: "Documentation example: a streaming table-in-out function",
    schemas: [{ name: "main", functions: [filterPositive] }],
  },
});

if (import.meta.main) worker.run();

ATTACH 'filters' (TYPE vgi, LOCATION 'bun run ./filter.ts');
SELECT * FROM filters.filter_positive((SELECT * FROM (VALUES (-2), (5), (0), (9), (-1)) AS t(value)));
Input
value
-2
5
0
9
-1
Output
value
5
9
The TABLE argument is implicit

Unlike the Go SDK, nothing declares the input relation. It arrives as the stream of batches process is called with, and its schema is on params.bindCall.input_schema at bind time — which is where a passthrough function gets its output schema from.

The corollary is that input_schema can be absent, and the SDK will not stop you: check it in onBind and throw a clear error, or the failure surfaces later as a null dereference.

Aggregate

Section titled “Aggregate”
aggregate shape
N rows → 1 value

Folds many rows down into a single value per group.

Four phases — initialState, update, combine, finalize — and the split is what lets DuckDB parallelise it. combine must be associative and commutative, because DuckDB decides how many workers run and in what order their partials merge.

sum.ts 
// Copyright 2025, 2026 Query Farm LLC - https://query.farm

// sum is the aggregate example for the vgi-typescript documentation.
//
// An aggregate folds many rows into one value per GROUP BY group. It runs in
// four phases, and the split is what lets DuckDB parallelise it:
//
//   - initialState — the identity value for a group (0 for a sum).
//   - update       — fold a batch of rows into per-group state. Runs in every
//                    worker, over that worker's share of the rows.
//   - combine      — merge two partial states for the same group.
//   - finalize     — turn state into one output row per group.
//
//   bun run sum.ts
//   # then, in a Haybarn shell:
//   ATTACH 'agg' (TYPE vgi, LOCATION 'bun run /abs/path/sum.ts');
//   SELECT category, agg.vgi_sum(value) FROM t GROUP BY category;

import { Worker, defineAggregate, batchFromColumns, int } from "@query-farm/vgi";

// Per-group accumulator. Unlike the Go and Python SDKs there is nothing to
// register: state stays inside this process, so it is an ordinary JS object.
interface SumState {
  total: bigint;
}

export const vgiSum = defineAggregate<{ value: bigint }, SumState>({
  name: "vgi_sum",
  description: "Sums a BIGINT column per group",
  args: { value: int },
  outputType: int,

  // DEFAULT means DuckDB skips NULL inputs, so update() never sees one. That
  // is what makes SUM over an all-NULL group return NULL rather than 0 — see
  // the lazy ensureState below.
  nullHandling: "DEFAULT",

  initialState: () => ({ total: 0n }),

  update: ({ groupIds, columns, ensureState }) => {
    const values = columns[0];
    for (let i = 0; i < groupIds.length; i++) {
      const v = values?.get(i);
      if (v == null) continue;
      // Allocate only when a row genuinely contributes. A group that never
      // reaches ensureState has no state at finalize, and emits NULL.
      ensureState(groupIds[i]).total += typeof v === "bigint" ? v : BigInt(v);
    }
  },

  // Must be associative and commutative: DuckDB decides how many workers run
  // and in what order their partials merge.
  combine: (src, tgt) => ({ total: src.total + tgt.total }),

  // Emits exactly one row per group id, in the order given.
  finalize: ({ groupIds, states, outputSchema }) => {
    const results = groupIds.map((gid) => states.get(gid)?.total ?? null);
    return batchFromColumns({ result: results }, outputSchema);
  },
});

export const worker = new Worker({
  catalog: {
    name: "agg",
    comment: "Documentation example: a distributed aggregate",
    schemas: [{ name: "main", functions: [vgiSum] }],
  },
});

if (import.meta.main) worker.run();

ATTACH 'agg' (TYPE vgi, LOCATION 'bun run ./sum.ts');
CREATE TABLE t AS SELECT * FROM (VALUES ('a',1),('a',2),('b',10),('b',NULL),('c',NULL)) AS v(category, value);
SELECT category, agg.vgi_sum(value::BIGINT) AS total FROM t GROUP BY category ORDER BY category;

Output

category total
a 3
b 10
c
Allocate state lazily, and NULL semantics follow

nullHandling: "DEFAULT" means DuckDB never calls update for a NULL input. Combined with ensureState being called only when a row genuinely contributes, a group whose values are all NULL never reaches the state map — so finalize finds nothing for it and emits SQL NULL, matching built-in SUM. Group c above is that case.

Call ensureState unconditionally instead and the same group returns 0. The difference is one line, and it is the difference between matching SQL and not.

State is just an object

Aggregate state is an ordinary JS object with no serialization contract to satisfy — the Go SDK’s gob.Register step has no counterpart here, because the state stays inside the worker process. Buffering functions are the exception: their phases can be split across processes, so their state goes through params.storage explicitly.

Buffering

Section titled “Buffering”
buffering shape
stream → [state] → stream

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

When output depends on the whole input — a global sort, top-k, a full reduction — use a buffering function. It runs in three phases: process (the sink, per batch and parallel), combine (once, on the coordinator), and finalize (the source).

rowcount.ts 
// Copyright 2025, 2026 Query Farm LLC - https://query.farm

// rowcount is the buffering example for the vgi-typescript documentation.
//
// A buffering function is for the case where output depends on the WHOLE input
// — a global sort, a top-k, a full reduction. It runs in three phases:
//
//   - process  (sink)   — called per input batch, in parallel across DuckDB
//     threads. Stash what you need and return a state id.
//   - combine           — called once, on the coordinator, with every state id
//     the sink produced. Reduce them into the ids the source will drain.
//   - finalize (source) — called per finalize id, streaming the result out.
//
// The phases can run in different worker processes, so nothing may live in a
// module-level variable between them. State goes in params.storage, which is
// scoped to this execution and shared across the workers serving it.
//
//   bun run rowcount.ts
//   # then, in a Haybarn shell:
//   ATTACH 'buffers' (TYPE vgi, LOCATION 'bun run /abs/path/rowcount.ts');
//   SELECT * FROM buffers.row_count((SELECT * FROM big_table));

import {
  Worker,
  defineTableBufferingFunction,
  batchFromColumns,
  toSchema,
  int,
} from "@query-farm/vgi";

const countSchema = toSchema({ count: int });

const enc = new TextEncoder();
const NS = enc.encode("rowcount");
const KEY = enc.encode("");

// The finalize cursor. `emitted` makes the source phase a one-shot: it emits
// the total on the first tick and finishes on the second.
interface DrainState {
  emitted: boolean;
}

export const rowCount = defineTableBufferingFunction<Record<string, never>, DrainState>({
  name: "row_count",
  description: "Counts every row of the input relation",

  onBind: (params) => {
    if (!params.bindCall.input_schema) {
      throw new Error("row_count requires a table argument");
    }
    // Output is one BIGINT, whatever the input looked like.
    return { outputSchema: countSchema };
  },

  // The sink runs in parallel across DuckDB threads. stateAppend is an
  // append-only log, so concurrent appends cannot lose each other the way a
  // read-modify-write would. Return the state id this batch contributed to.
  process: async (batch, params) => {
    const n = new BigInt64Array([BigInt(batch.numRows)]);
    await params.storage.stateAppend(NS, KEY, new Uint8Array(n.buffer));
    return params.executionId;
  },

  // Runs once, on the coordinator. Here there is a single bucket to drain, so
  // it just names the execution; a top-k would reduce the partials first.
  combine: async (_stateIds, params) => [params.executionId],

  initialFinalizeState: () => ({ emitted: false }),

  // The source phase. Sum the log and emit one row.
  finalize: async (params, _finalizeId, state, out) => {
    if (state.emitted) return out.finish();

    let total = 0n;
    // -1 starts before the first entry; the limit is a page size, not a cap.
    let afterId = -1;
    for (;;) {
      const rows = await params.storage.stateLogScan(NS, KEY, afterId, 256);
      if (rows.length === 0) break;
      for (const [logId, value] of rows) {
        total += new BigInt64Array(
          value.buffer.slice(value.byteOffset, value.byteOffset + value.byteLength),
        )[0];
        afterId = logId;
      }
    }

    state.emitted = true;
    out.emit(batchFromColumns({ count: [total] }, countSchema));
  },
});

export const worker = new Worker({
  catalog: {
    name: "buffers",
    comment: "Documentation example: a buffering (sink → combine → source) function",
    schemas: [{ name: "main", functions: [rowCount] }],
  },
});

if (import.meta.main) worker.run();

ATTACH 'buffers' (TYPE vgi, LOCATION 'bun run ./rowcount.ts');
SELECT * FROM buffers.row_count((SELECT * FROM (VALUES (1), (2), (3), (4), (5)) AS t(x)));
Input
x
1
2
3
4
5
Output
count
5
A module-level variable will not survive the phases

The sink and source phases can run in different worker processes, so anything a let at module scope accumulates during process may simply not be there in finalize. State goes in params.storage, which is scoped to the execution and shared by every worker serving it.

stateAppend is an append-only log, which is what makes the parallel sink safe: concurrent appends cannot lose each other the way a read-modify-write would.

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. Reach for buffering only when output genuinely depends on every row. One observable difference: an empty input yields no rows from a buffering function rather than a zero — with nothing to sink, the source phase never runs.

Next steps

Section titled “Next steps”