2. Your first table function

Step two: add a table function. A scalar function transforms a column; a table function produces rows, so it is called in a FROM clause. About 10 minutes.

args → N rows

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

The worker

TableExample.java

TableExample.java Copy Copied!

// 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);
    }
}

Four things to notice:

• CountdownTableFunction is a base for generators. Extend it and you get the count positional argument and a batch_size := 2048 named argument for free — 2048 being DuckDB’s standard vector size, so each batch fills one of its chunks.
• produceTick is the pull loop. DuckDB calls it repeatedly; emit one batch per call and call out.finish() when there is nothing left.
• The producer is per-thread; storage is per-scan. That distinction is the whole point of the next section.
• The schema is built once, at class level, because it never varies.

Call it

Copy Copied!

ATTACH 'demo' (TYPE vgi, LOCATION 'launch:/abs/path/bin/demo');
SELECT * FROM demo.numbers(5);

Output

n
0
1
2
3
4

Parallel scans, and the trap underneath them

maxWorkers() returns 4, so DuckDB may scan this function on four threads at once. Each thread gets its own producer. A producer that simply counted from 0 to count would therefore emit the entire range once per thread — four times the rows, no error, no warning.

This example avoids that by coordinating rather than counting. Every parallel producer in one scan shares an execution_id, and therefore the same params.storage(). An atomic counter there is a single cursor they all draw disjoint chunks from:

Copy Copied!

private static final byte[] CURSOR_NS = "cursor".getBytes(StandardCharsets.UTF_8);
private static final byte[] CURSOR_KEY = new byte[0];

The union covers 0..count-1 exactly once, which a query can check rather than assume:

Copy Copied!

SELECT count(*) AS rows, count(DISTINCT n) AS distinct_n, sum(n) AS total
FROM demo.numbers(1000000);

Output

rows	distinct_n	total
1000000	1000000	499999500000

A million rows, a million distinct values, and the sum a closed form agrees with — across four threads.

Raising `maxWorkers` without coordinating multiplies your output

This is worth stating on its own, because the failure is silent and the fix is not obvious. If your producer derives its rows from the arguments alone, leave maxWorkers() at 1. Raise it only once the producers agree on who emits what — through params.storage(), a work queue, or a partition column.

The Go SDK’s version of this shows the same failure concretely: four workers turned series(10) into forty rows.

Why “producer” and not just “function”? Why a producer?

A table function is pulled, not called: the engine asks the worker for output until it signals completion. produceTick is the pull handler. The function object is shared across threads and must stay safe to call concurrently; the producer is per-scan-thread state, which is where a cursor belongs.

Next steps

• The other three shapes → Function patterns.
• When each callback fires → Function lifecycle.
• Exact contracts → Table functions.