Skip to content
Query.Farm
Talk with Us

Function patterns

Each of the five VGI function shapes in Java, 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 built the worker with ./gradlew installDist.

All five register into one worker below, which is what the examples project ships — so a single ATTACH serves every query on this page.

Which shape do I need?

Section titled “Which shape do I need?”
If your function…UseBase type
maps each row independentlyScalarScalarFn
produces rows from argumentsTableCountdownTableFunction / TableFunction
transforms a relation as it streamsTable-in-outTableInOutFunction
folds rows to one value per groupAggregateAggregateFunction
needs every row before it can answerBufferingTableBufferingFunction

One worker, five functions

Section titled “One worker, five functions”
AllInOneWorker.java 
// VGI-Java example: one worker serving all five function kinds at once.
//
// This is the artifact the quickstart and the integration test attach to. It
// registers every example function under the catalog `demo`, so a single ATTACH
// exposes upper_case (scalar), numbers (table), echo (table-in-out), vgi_sum
// (aggregate), and collect (buffering).
//
//   ./gradlew installDist
//   ATTACH 'demo' AS demo (TYPE vgi,
//       LOCATION 'launch:/abs/path/build/install/vgi-java-examples/bin/vgi-java-examples');
package farm.query.vgi.examples;

import farm.query.vgi.Worker;

/** A single worker process exposing one function of each kind. */
public final class AllInOneWorker {

    public static void main(String[] args) {
        Worker w = Worker.builder()
                .catalogName("demo")
                .catalogComment("VGI-Java introductory examples")
                .registerScalar(new ScalarExample())          // upper_case
                .registerTable(new TableExample())             // numbers (parallel-safe via storage)
                .registerTableInOut(new TableInOutExample())   // echo
                .registerAggregate(new AggregateExample())     // vgi_sum
                .registerTableBuffering(new BufferingExample());// collect
        CatalogExample.register(w);                            // catalog: schema, table, view, macros
        w.runFromArgs(args);
    }
}

ATTACH 'demo' (TYPE vgi, LOCATION 'launch:/abs/path/bin/demo');

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.

Extend ScalarFn and write one compute(). Parameter annotations derive the signature, the output type and the dispatch — there is no spec to keep in sync by hand.

ScalarExample.java 
// VGI-Java example: a scalar function.
//
// A scalar function maps each input row to one output row. You extend
// `ScalarFn` and write a single `compute()` method; the framework reads its
// parameter annotations to derive the SQL signature, the output type, and the
// per-batch dispatch. There is no schema boilerplate to write by hand.
//
// Run it on its own:
//   ./gradlew runScalar --args="--unix /tmp/scalar.sock --idle-timeout 60"
// then from Haybarn:
//   ATTACH 'demo' AS demo (TYPE vgi, LOCATION 'launch:/abs/path/bin/runScalar');
//   SELECT demo.upper_case('hello');   -- HELLO
package farm.query.vgi.examples;

import farm.query.vgi.Worker;
import farm.query.vgi.scalar.ScalarFn;
import farm.query.vgi.scalar.Vector;
import org.apache.arrow.vector.VarCharVector;

import java.nio.charset.StandardCharsets;
import java.util.Locale;

/** {@code upper_case(value VARCHAR) -> VARCHAR}: ASCII/Unicode uppercase. */
public final class ScalarExample extends ScalarFn {

    @Override public String name() { return "upper_case"; }
    @Override public String description() { return "Uppercase a string"; }

    // One `@Vector` input column + one trailing (unannotated) output vector.
    // The framework allocates `result`, sized to the batch row count, and
    // writes whatever you put into it back across the wire.
    //
    // Parameter rules in one breath:
    //   @Vector  -> a per-row input column (the Arrow vector type is the SQL type)
    //   @Const   -> a bind-time constant arg (long/double/String/boolean/byte[])
    //   @Setting -> a session setting (SET demo.foo = ...)
    //   last unannotated vector = the output (framework-allocated)
    public void compute(@Vector VarCharVector value, VarCharVector result) {
        int rows = value.getValueCount();
        result.allocateNew();
        for (int i = 0; i < rows; i++) {
            if (value.isNull(i)) { result.setNull(i); continue; }
            String up = new String(value.get(i), StandardCharsets.UTF_8).toUpperCase(Locale.ROOT);
            byte[] bytes = up.getBytes(StandardCharsets.UTF_8);
            result.setSafe(i, bytes, 0, bytes.length);
        }
    }

    public static void main(String[] args) {
        Worker.builder()
                .catalogName("demo")
                .registerScalar(new ScalarExample())
                .runFromArgs(args);   // handles --unix / --http / --idle-timeout / stdio
    }
}

SELECT demo.upper_case('hello');

Output

upper_case('hello')
HELLO
Four parameter roles

@Vector is a per-row input column (the Arrow vector type is the SQL type), @Const a bind-time constant, @Setting a session setting, and the last unannotated vector is the output, allocated for you. @Vector(any=true) FieldVector accepts any type; @Vector(varargs=true) List<FieldVector> takes the rest.

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. The function object is shared across scan threads; the per-thread cursor is a producer.

TableExample.java 
// VGI-Java example: a table function (a set-returning generator), parallel-safe.
//
// A table function produces rows. You extend `CountdownTableFunction` — a base
// for "emit rows in fixed-size batches" generators — and declare the output
// schema plus a producer. The base gives you the `count` positional arg and the
// `batch_size := 2048` named arg for free. The producer's `produceTick()` is
// called repeatedly: emit one batch per call, then call `out.finish()`.
//
// PARALLELISM. `maxWorkers()` lets DuckDB scan this function on several threads.
// Each thread gets its OWN producer, so a naive producer that counted from 0
// would re-emit the whole range once per thread. The fix: coordinate. Every
// parallel producer of one scan shares the same execution_id, hence the same
// `params.storage()` (a BoundStorage). An atomic counter there is a single cursor
// they all draw disjoint chunks from, so the union covers 0..count-1 exactly once.
//
//   ATTACH 'demo' AS demo (TYPE vgi, LOCATION 'launch:/abs/path/bin/runTable');
//   SELECT * FROM demo.numbers(5);                                   -- 0,1,2,3,4
//   SELECT count(*), count(DISTINCT n) FROM demo.numbers(10000000);  -- 10000000, 10000000
package farm.query.vgi.examples;

import farm.query.vgi.Worker;
import farm.query.vgi.function.FunctionMetadata;
import farm.query.vgi.function.ParameterExtractor;
import farm.query.vgi.pushdown.FilterApplier;
import farm.query.vgi.storage.BoundStorage;
import farm.query.vgi.table.CountdownTableFunction;
import farm.query.vgi.table.TableInitParams;
import farm.query.vgi.table.TableProducerState;
import farm.query.vgi.types.Schemas;
import farm.query.vgirpc.CallContext;
import farm.query.vgirpc.OutputCollector;
import farm.query.vgirpc.wire.Allocators;
import org.apache.arrow.vector.BigIntVector;
import org.apache.arrow.vector.VectorSchemaRoot;
import org.apache.arrow.vector.types.pojo.Schema;

import java.nio.charset.StandardCharsets;

/** {@code numbers(count BIGINT, batch_size := 2048) -> n BIGINT}, scanned in parallel. */
public final class TableExample extends CountdownTableFunction {

    private static final Schema OUTPUT_SCHEMA = Schemas.of(Schemas.nullable("n", Schemas.INT64));

    // The shared cursor: a counter in a user namespace of params.storage(). All
    // parallel producers of one scan address the same (namespace, key).
    private static final byte[] CURSOR_NS = "cursor".getBytes(StandardCharsets.UTF_8);
    private static final byte[] CURSOR_KEY = new byte[0];

    @Override public String name() { return "numbers"; }

    // Allow up to 4 parallel scan threads. Safe BECAUSE the producer coordinates
    // through storage (see produceTick); without that this would duplicate rows.
    @Override public long maxWorkers() { return 4L; }

    // Default to 2048 rows per batch — DuckDB's STANDARD_VECTOR_SIZE — so each
    // emitted batch lines up with one of the engine's vectors.
    @Override protected long defaultBatchSize() { return 2048L; }

    @Override public FunctionMetadata metadata() {
        // withPushdown(projection, filter, limit): accept LIMIT pushdown so a
        // `LIMIT 5` stops the scan early instead of materializing everything.
        return FunctionMetadata.describe("Generate the integers 0..count-1")
                .withPushdown(false, true, false)
                .withCategories("generator");
    }

    @Override protected Schema outputSchema() { return OUTPUT_SCHEMA; }

    @Override public TableProducerState createProducer(TableInitParams params) {
        // `count` is the positional arg; `batch_size` is the named arg the base
        // class declares (default 2048). `params.storage()` is scoped to this
        // scan's execution_id — the scope every parallel worker shares.
        ParameterExtractor p = ParameterExtractor.of(params.arguments());
        long count = p.positional(0, "count").asLong().required();
        long batchSize = p.named("batch_size").asLong().ge(1).orElse(2048L);
        return new NumbersState(count, batchSize,
                FilterApplier.from(params.pushdownFilters(), params.joinKeys()),
                params.storage());
    }

    /** Per-execution producer state. One instance per scan worker; they coordinate
     *  through the shared `storage` counter. */
    public static final class NumbersState extends TableProducerState {
        public long count;
        public long batchSize;
        public FilterApplier filters;
        public BoundStorage storage;

        public NumbersState() {}
        NumbersState(long count, long batchSize, FilterApplier filters, BoundStorage storage) {
            this.count = count; this.batchSize = batchSize; this.filters = filters; this.storage = storage;
        }

        @Override public void produceTick(OutputCollector out, CallContext ctx) {
            // Atomically reserve the next [start, start+batchSize) chunk. counterAdd
            // returns the post-add value, so concurrent calls from other workers
            // get non-overlapping chunks. When the cursor passes `count`, we're done.
            long claimedEnd = storage.counterAdd(CURSOR_NS, CURSOR_KEY, batchSize);
            long start = claimedEnd - batchSize;
            if (start >= count) { out.finish(); return; }
            int n = (int) Math.min(batchSize, count - start);

            VectorSchemaRoot root = VectorSchemaRoot.create(OUTPUT_SCHEMA, Allocators.root());
            BigIntVector v = (BigIntVector) root.getVector("n");
            v.allocateNew(n);
            for (int i = 0; i < n; i++) v.set(i, start + i);
            v.setValueCount(n);
            root.setRowCount(n);
            out.emit(filters.apply(root));   // emit() takes ownership of the (filtered) root
        }
    }

    public static void main(String[] args) {
        Worker.builder()
                .catalogName("demo")
                .registerTable(new TableExample())
                .runFromArgs(args);
    }
}

SELECT * FROM demo.numbers(5);
Input
count
5
Output
n
0
1
2
3
4

See the tutorial for why maxWorkers() above 1 needs the producers to coordinate.

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.

TableInOutExample.java 
// VGI-Java example: a table-in-out (TIO) function.
//
// A TIO function consumes a relation and streams a relation back — a row-by-row
// (really batch-by-batch) transform. DuckDB feeds you input batches; you emit
// output batches. Use it for streaming reshapes, enrichment, or filtering that
// you'd rather express in Java than SQL.
//
// This example is the canonical `echo`: output schema == input schema, every
// input batch passes through unchanged. `PassthroughTIOFunction` supplies the
// "output schema = input schema" bind, so you only write the exchange.
//
//   ATTACH 'demo' AS demo (TYPE vgi, LOCATION 'launch:/abs/path/bin/runTableInOut');
//   SELECT * FROM demo.echo((SELECT * FROM range(3) t(x)));   -- 0,1,2
package farm.query.vgi.examples;

import farm.query.vgi.Worker;
import farm.query.vgi.function.ArgSpec;
import farm.query.vgi.function.FunctionMetadata;
import farm.query.vgi.tableinout.PassthroughTIOFunction;
import farm.query.vgi.tableinout.TableInOutExchangeState;
import farm.query.vgi.tableinout.TableInOutInitParams;
import farm.query.vgirpc.AnnotatedBatch;
import farm.query.vgirpc.CallContext;
import farm.query.vgirpc.OutputCollector;
import farm.query.vgirpc.wire.Allocators;
import org.apache.arrow.vector.FieldVector;
import org.apache.arrow.vector.VectorSchemaRoot;
import org.apache.arrow.vector.util.TransferPair;

import java.util.ArrayList;
import java.util.List;

/** {@code echo(data TABLE) -> *}: passes every input batch through unchanged. */
public final class TableInOutExample extends PassthroughTIOFunction {

    @Override public String name() { return "echo"; }

    @Override public FunctionMetadata metadata() {
        return FunctionMetadata.describe("Emit each input batch unchanged")
                .withCategories("utility");
    }

    // Declare the single table-valued argument. TIO functions take a relation.
    @Override public List<ArgSpec> argumentSpecs() {
        return List.of(ArgSpec.table("data", 0));
    }

    @Override public TableInOutExchangeState createExchange(TableInOutInitParams params) {
        return new EchoState();
    }

    /** One exchange instance per execution; `onInputBatch` runs per input batch. */
    public static final class EchoState extends TableInOutExchangeState {
        @Override
        public void onInputBatch(AnnotatedBatch input, OutputCollector out, CallContext ctx) {
            // Transfer the input vectors into a fresh root before emitting.
            //
            // Why not just `out.emit(input.root())`? The framework close()s each
            // emitted root after writing it. The input root is owned by the
            // reader and reused for the NEXT batch — closing it would corrupt the
            // stream. TransferPair moves the buffers into a root we own, leaving
            // the reader intact. (TransferPair, not a row copy, also preserves
            // dictionary-encoded children.)
            VectorSchemaRoot in = input.root();
            List<FieldVector> outVectors = new ArrayList<>();
            for (FieldVector v : in.getFieldVectors()) {
                TransferPair tp = v.getTransferPair(Allocators.root());
                tp.transfer();
                outVectors.add((FieldVector) tp.getTo());
            }
            VectorSchemaRoot copy = new VectorSchemaRoot(outVectors);
            copy.setRowCount(in.getRowCount());
            out.emit(copy);   // emit() takes ownership; do not close `copy` yourself
        }
    }

    public static void main(String[] args) {
        Worker.builder()
                .catalogName("demo")
                .registerTableInOut(new TableInOutExample())
                .runFromArgs(args);
    }
}

SELECT n FROM demo.echo((SELECT * FROM demo.numbers(3)));
Input
n
0
1
2
Output
n
0
1
2

Aggregate

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

Folds many rows down into a single value per group.

Partial aggregation with a cross-process state combine: DuckDB decides how many workers run and in what order their partials merge, so the combine must be associative and commutative.

AggregateExample.java 
// VGI-Java example: an aggregate function.
//
// An aggregate collapses many rows into one value per group. VGI aggregates are
// built for DuckDB's *parallel, partial* aggregation model, so you implement
// four pieces:
//
//   newState()  — a fresh, empty accumulator
//   update()    — fold a batch of rows into the per-group accumulators
//   combine()   — merge two partial accumulators (parallel workers / spill)
//   finalize()  — write a group's accumulator out as the result value
//
// The `State` is `Serializable` because partials may cross process boundaries
// when DuckDB parallelizes the aggregation. Keep it small.
//
//   ATTACH 'demo' AS demo (TYPE vgi, LOCATION 'launch:/abs/path/bin/runAggregate');
//   SELECT g, demo.vgi_sum(v) FROM (VALUES (1,10),(1,20),(2,5)) t(g,v) GROUP BY g;
//   -- 1 -> 30, 2 -> 5
package farm.query.vgi.examples;

import farm.query.vgi.Worker;
import farm.query.vgi.aggregate.AggregateFunction;
import farm.query.vgi.function.FunctionSpec;
import farm.query.vgi.types.Schemas;
import org.apache.arrow.vector.BigIntVector;
import org.apache.arrow.vector.FieldVector;
import org.apache.arrow.vector.VectorSchemaRoot;
import org.apache.arrow.vector.types.pojo.Schema;

import java.io.Serializable;
import java.util.List;
import java.util.Map;

/** {@code vgi_sum(value BIGINT) -> BIGINT}: sum per group, overflow-checked. */
public final class AggregateExample implements AggregateFunction<AggregateExample.State> {

    /** Per-group accumulator. Serializable: partials may be merged across workers. */
    public static final class State implements Serializable {
        private static final long serialVersionUID = 1L;
        long total;
    }

    private static final Schema OUTPUT_SCHEMA =
            new Schema(List.of(Schemas.nullable("result", Schemas.INT64)));

    private static final FunctionSpec SPEC = FunctionSpec.builder("vgi_sum")
            .description("Sum integer values")
            .arg("value", Schemas.INT64)
            .build();

    @Override public FunctionSpec spec() { return SPEC; }
    @Override public Schema outputSchema() { return OUTPUT_SCHEMA; }
    @Override public State newState() { return new State(); }

    // Fold one input batch into the accumulators. `groupIds[i]` is the group of
    // row i; states.computeIfAbsent mints an accumulator the first time a group
    // is seen in this partition.
    @Override
    public void update(Map<Long, State> states, long[] groupIds, VectorSchemaRoot input) {
        FieldVector v = input.getFieldVectors().get(0);
        if (!(v instanceof BigIntVector b)) return;
        int rows = input.getRowCount();
        try {
            for (int i = 0; i < rows; i++) {
                if (b.isNull(i)) continue;
                State s = states.computeIfAbsent(groupIds[i], k -> new State());
                s.total = Math.addExact(s.total, b.get(i));
            }
        } catch (ArithmeticException e) {
            throw new IllegalArgumentException("vgi_sum: int64 overflow", e);
        }
    }

    // Merge a partial (`source`) produced by another worker into `target`.
    @Override
    public void combine(State target, State source) {
        target.total = Math.addExact(target.total, source.total);
    }

    // Write one group's final value into the output column at `rowIndex`.
    @Override
    public void finalize(FieldVector result, int rowIndex, State state) {
        ((BigIntVector) result).setSafe(rowIndex, state.total);
    }

    public static void main(String[] args) {
        Worker.builder()
                .catalogName("demo")
                .registerAggregate(new AggregateExample())
                .runFromArgs(args);
    }
}

SELECT g, demo.vgi_sum(v) AS total
FROM (VALUES (1,10),(1,20),(2,5)) t(g,v)
GROUP BY g ORDER BY g;

Output

g total
1 30
2 5

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. Three phases: process (the sink, per batch and parallel), combine (once, on the coordinator), and a finalize producer (the source).

BufferingExample.java 
// VGI-Java example: a table-buffering (Sink + Source) function.
//
// A buffering function must see ALL input before it produces ANY output — think
// sort, top-k, or whole-relation aggregation. Unlike a TIO function (which emits
// per input batch), it has a three-phase lifecycle:
//
//   process()                — Sink: stash each input batch, return a state_id
//   combine()                — once at end-of-input: group state_ids into the
//                              finalize streams the Source will drain
//   createFinalizeProducer() — Source: emit the buffered rows back out
//
// State is stashed in `params.storage()` — a durable, execution-scoped key/value
// + append-log store — so buffering survives even when DuckDB spreads the Sink
// across parallel workers. This example is the canonical "collect every batch,
// replay it during finalize" (a passthrough that happens to fully buffer).
//
//   ATTACH 'demo' AS demo (TYPE vgi, LOCATION 'launch:/abs/path/bin/runBuffering');
//   SELECT * FROM demo.collect((SELECT * FROM range(3) t(x)));   -- 0,1,2
package farm.query.vgi.examples;

import farm.query.vgi.Worker;
import farm.query.vgi.buffering.BufferingFinalizeProducer;
import farm.query.vgi.buffering.TableBufferingCombineParams;
import farm.query.vgi.buffering.TableBufferingFinalizeParams;
import farm.query.vgi.buffering.TableBufferingFunction;
import farm.query.vgi.buffering.TableBufferingProcessParams;
import farm.query.vgi.function.FunctionMetadata;
import farm.query.vgi.function.FunctionSpec;
import farm.query.vgi.internal.BatchUtil;
import farm.query.vgi.internal.SchemaUtil;
import farm.query.vgi.protocol.BindResponse;
import farm.query.vgi.storage.FunctionStorage;
import farm.query.vgi.table.TableProducerState;
import farm.query.vgi.tableinout.TableInOutBindParams;
import farm.query.vgirpc.CallContext;
import farm.query.vgirpc.OutputCollector;
import farm.query.vgirpc.wire.Allocators;
import org.apache.arrow.vector.VectorSchemaRoot;
import org.apache.arrow.vector.types.pojo.Schema;

import java.nio.charset.StandardCharsets;
import java.util.List;

/** {@code collect(data TABLE) -> *}: buffers every input batch, replays them. */
public final class BufferingExample implements TableBufferingFunction {

    // A namespace + key naming the append-log we buffer batches into.
    private static final byte[] NS = "buf".getBytes(StandardCharsets.UTF_8);
    private static final byte[] KEY = new byte[0];

    private static final FunctionSpec SPEC = FunctionSpec.builder("collect")
            .metadata(FunctionMetadata.describe("Buffer all input, then replay it")
                    .withCategories("utility"))
            .table("data")
            .build();

    @Override public FunctionSpec spec() { return SPEC; }

    // Output schema = input schema (passthrough).
    @Override public BindResponse onBind(TableInOutBindParams params) {
        Schema in = params.inputSchema();
        Schema out = (in == null || in.getFields().isEmpty()) ? new Schema(List.of()) : in;
        return BindResponse.forSchema(SchemaUtil.serializeSchema(out));
    }

    // Sink: append this batch's IPC bytes to the log, return our execution id as
    // the state_id (every batch of one execution shares the same log).
    @Override public byte[] process(VectorSchemaRoot batch, TableBufferingProcessParams params) {
        params.storage().stateAppend(NS, KEY, BatchUtil.writeSingleBatch(batch));
        return params.executionId();
    }

    // Combine: one output stream, keyed by the execution id.
    @Override public List<byte[]> combine(List<byte[]> stateIds, TableBufferingCombineParams params) {
        return List.of(params.executionId());
    }

    // Source: drain the log one buffered batch per tick.
    @Override public TableProducerState createFinalizeProducer(TableBufferingFinalizeParams params) {
        return new ReplayProducer(params);
    }

    private static final class ReplayProducer extends BufferingFinalizeProducer {
        private long afterId = -1;   // log cursor; -1 = before the first entry

        ReplayProducer(TableBufferingFinalizeParams params) { super(params); }

        @Override public void produceTick(OutputCollector out, CallContext ctx) {
            List<FunctionStorage.LogEntry> rows = storage().stateLogScan(NS, KEY, afterId, 1);
            if (rows.isEmpty()) { out.finish(); return; }
            FunctionStorage.LogEntry e = rows.get(0);
            VectorSchemaRoot full = BatchUtil.readSingleBatch(e.value(), Allocators.root());
            emitProjected(full, out);   // narrows to projected cols + applies filters
            full.close();
            afterId = e.id();
        }
    }

    public static void main(String[] args) {
        Worker.builder()
                .catalogName("demo")
                .registerTableBuffering(new BufferingExample())
                .runFromArgs(args);
    }
}

SELECT n FROM demo.collect((SELECT * FROM demo.numbers(4)));
Input
n
0
1
2
3
Output
n
0
1
2
3
Call `storage()`, not a captured field

BufferingFinalizeProducer exposes storage as a method. That is not style: after an HTTP continuation the producer is deserialized with a null storage view, and the accessor re-binds it from (executionId, attachId). A field read cannot, so a producer that captured one works locally and fails the moment the transport resumes it.

This example used to read the field. It compiled against the SDK it was written for and stopped compiling later — which is the good outcome; the bad one would have been an NPE in production.

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.

Next steps

Section titled “Next steps”