Function patterns

Each of the five VGI function shapes in Go, 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 into the binary its ATTACH names:

go mod tidy
go build -o calcscalar .      # scalar
go build -o calc .            # table (and the scalar again)
go build -o filterworker .    # table-in-out
go build -o sumworker .       # aggregate
go build -o rowcountworker .  # buffering

Each worker is its own main package, so build them from their own directories. LOCATION is resolved relative to the directory the engine was started in.

Which shape do I need?

Scalar1 row → 1 value

Tableargs → N rows

Table-in-outN rows → M rows

AggregateN rows → 1 value

Bufferingstream → [state] → stream

Pattern	Go interface	Use it when…	SQL
Scalar	`ScalarFunction`	you transform each row independently	`SELECT f(col) FROM t`
Table	`TableFunction`	you generate rows from arguments	`SELECT * FROM f(args)`
Table-in-out	`TableInOutFunction`	you reshape or filter a streamed relation	`SELECT * FROM f((SELECT …))`
Aggregate	`AggregateFunction`	you accumulate per `GROUP BY` group	`SELECT f(col) FROM t GROUP BY k`
Buffering	`TableBufferingFunction`	you must see every row first (sort, top-k, full reduction)	`SELECT * FROM f((SELECT …))`

Your function value is shared across calls

A function is registered once and used for every call, so treat the receiver as immutable — the examples all use an empty struct for that reason. Anything mutable belongs in the per-scan state (NewState) or in params.Storage, not in a field on your type. Under the Unix, TCP and HTTP transports the worker serves concurrent connections; stdio is serial, which is exactly why a data race here can survive local testing and appear only in deployment.

Typed adapters vs. raw interfaces

Three shapes ship a Typed* variant — TypedScalarFunc, TypedTableFunc, TypedTableInOutFunc — adapted by AsScalarFunction / AsTableFunction / AsTableInOutFunction. Note what each is parameterised by: TypedScalarFunc[A] takes your arguments struct (a scalar has no state at all), while TypedTableFunc[S] and TypedTableInOutFunc[S] take your state. Both state-carrying adapters gob-register it for you; only AsTableFunction additionally validates that it is encodable at registration. Aggregates and buffering functions implement their interface directly.

Scalar

One row in, one row out. MapColumn walks the input column and builds the output, propagating nulls; the row count out must equal the row count in.

calcscalar/main.go

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

// Command calcscalar is the worker built in step 1 of the vgi-go 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.
//
//	go build -o calcscalar .
//	# then, in a Haybarn shell:
//	ATTACH 'calc' (TYPE vgi, LOCATION './calcscalar');
//	SELECT calc.double(21);
package main

import (
	"context"
	"flag"
	"log"

	"github.com/Query-farm/vgi-go/vgi"
	"github.com/apache/arrow-go/v18/arrow"
	"github.com/apache/arrow-go/v18/arrow/array"
)

// doubleArgs declares the function's arguments. The `vgi:"..."` tags are the
// whole signature: they drive both the ArgumentSpecs the catalog advertises and
// the runtime binding of values into this struct, so the SQL signature and the
// Go type can never drift apart.
type doubleArgs struct {
	// No type= needed: the Go field type infers the Arrow type (int64 here).
	// When you do override it, the tag takes ARROW names ("int64"), not SQL
	// names ("bigint") — an unrecognised name silently falls back to VARCHAR.
	N int64 `vgi:"pos=0,const=false,doc=Value to double"`
}

// DoubleFn doubles each value in its input column.
type DoubleFn struct{}

func (*DoubleFn) Name() string { return "double" }

func (*DoubleFn) Metadata() vgi.FunctionMetadata {
	fortyTwo := "42"
	return vgi.FunctionMetadata{
		Description: "Doubles a BIGINT",
		Stability:   vgi.StabilityConsistent,
		ReturnType:  arrow.PrimitiveTypes.Int64,
		Examples: []vgi.CatalogExample{
			{SQL: "SELECT calc.double(21)", Description: "Doubles a literal", ExpectedOutput: &fortyTwo},
		},
	}
}

// OnBindTyped runs once per query, before any data moves. Returning the output
// type here is what lets DuckDB plan the query.
func (*DoubleFn) OnBindTyped(_ *doubleArgs, _ *vgi.BindParams) (*vgi.BindResponse, error) {
	return vgi.BindResult(arrow.PrimitiveTypes.Int64)
}

// ProcessTyped runs per input batch. MapColumn walks column 0 and builds the
// output column, preserving nulls — the row count out must equal the row count
// in, which is the contract that makes this a *scalar* function.
func (*DoubleFn) ProcessTyped(_ context.Context, _ *doubleArgs, params *vgi.ProcessParams, batch arrow.RecordBatch) (arrow.RecordBatch, error) {
	return vgi.MapColumn(params, batch, 0, array.NewInt64Builder,
		func(col arrow.Array, i int) int64 {
			return vgi.GetInt64Value(col, i) * 2
		})
}

// NewDouble returns the registration-ready function. Tests use this too, so the
// wiring under test is the wiring that ships.
func NewDouble() vgi.ScalarFunction {
	return vgi.AsScalarFunction[doubleArgs](&DoubleFn{})
}

func main() {
	httpMode := flag.Bool("http", false, "serve over HTTP instead of stdio")
	logFlags := vgi.RegisterLoggingFlags(flag.CommandLine)
	flag.Parse()
	if err := logFlags.Apply(); err != nil {
		log.Fatalf("logging flags: %v", err)
	}

	w := vgi.NewWorker(
		vgi.WithCatalogName("calc"),
		vgi.WithCatalogComment("Tutorial worker: a single scalar function"),
	)
	w.RegisterScalar(NewDouble())

	if *httpMode {
		if err := w.RunHttp("127.0.0.1:0"); err != nil {
			log.Fatal(err)
		}
		return
	}
	w.RunStdio()
}

ATTACH 'calc' (TYPE vgi, LOCATION './calcscalar');
SELECT calc.double(n) FROM (VALUES (1), (2), (3)) AS t(n);

Input

n
1
2
3

Output

double(n)
2
4
6

Table

Generate rows from arguments, with no input relation. Arguments are read once in NewState; Process is then called repeatedly until the state reports nothing remains.

calc/main.go

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

// Command calc is the worker built across the vgi-go 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.
//
//	go build -o calc .
//	# then, in a Haybarn shell:
//	ATTACH 'calc' (TYPE vgi, LOCATION './calc');
//	SELECT calc.double(21);
//	SELECT * FROM calc.series(3);
package main

import (
	"context"
	"flag"
	"log"

	"github.com/Query-farm/vgi-go/vgi"
	"github.com/Query-farm/vgi-rpc-go/vgirpc"
	"github.com/apache/arrow-go/v18/arrow"
	"github.com/apache/arrow-go/v18/arrow/array"
)

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

type doubleArgs struct {
	N int64 `vgi:"pos=0,const=false,doc=Value to double"`
}

// DoubleFn doubles each value in its input column.
type DoubleFn struct{}

func (*DoubleFn) Name() string { return "double" }

func (*DoubleFn) Metadata() vgi.FunctionMetadata {
	return vgi.FunctionMetadata{
		Description: "Doubles a BIGINT",
		Stability:   vgi.StabilityConsistent,
		ReturnType:  arrow.PrimitiveTypes.Int64,
	}
}

func (*DoubleFn) OnBindTyped(_ *doubleArgs, _ *vgi.BindParams) (*vgi.BindResponse, error) {
	return vgi.BindResult(arrow.PrimitiveTypes.Int64)
}

func (*DoubleFn) ProcessTyped(_ context.Context, _ *doubleArgs, params *vgi.ProcessParams, batch arrow.RecordBatch) (arrow.RecordBatch, error) {
	return vgi.MapColumn(params, batch, 0, array.NewInt64Builder,
		func(col arrow.Array, i int) int64 { return vgi.GetInt64Value(col, i) * 2 })
}

// NewDouble returns the registration-ready scalar function.
func NewDouble() vgi.ScalarFunction {
	return vgi.AsScalarFunction[doubleArgs](&DoubleFn{})
}

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

// seriesOutputSchema is fixed, so OnBind can return it without inspecting the
// call. A table function that shapes its output from its arguments would build
// the schema here instead.
var seriesOutputSchema = arrow.NewSchema([]arrow.Field{
	{Name: "n", Type: arrow.PrimitiveTypes.Int64},
}, nil)

type seriesArgs struct {
	Count int64 `vgi:"pos=0,ge=0,doc=How many numbers to generate"`
}

// seriesState is the per-scan cursor. Embedding BatchState gives the generator
// helpers the remaining/batch-size bookkeeping; anything else the function needs
// between Process calls goes alongside it.
type seriesState struct {
	vgi.BatchState
}

// SeriesFn generates the integers 0..count-1.
type SeriesFn struct{}

// Compile-time proof that the shape is satisfied. Cheap, and it turns a missing
// method into a build error rather than a registration-time surprise.
var _ vgi.TypedTableFunc[seriesState] = (*SeriesFn)(nil)

func (*SeriesFn) Name() string { return "series" }

func (*SeriesFn) Metadata() vgi.FunctionMetadata {
	return vgi.FunctionMetadata{
		Description: "Generates the integers 0..count-1",
		Stability:   vgi.StabilityConsistent,
	}
}

func (*SeriesFn) ArgumentSpecs() []vgi.ArgSpec { return vgi.DeriveArgSpecs(seriesArgs{}) }

func (*SeriesFn) OnBind(_ *vgi.BindParams) (*vgi.BindResponse, error) {
	return vgi.BindSchema(seriesOutputSchema)
}

// NewState runs once per scan, after bind. This is where arguments are read:
// they are fixed for the whole scan, so decoding them per batch would be waste.
func (*SeriesFn) NewState(params *vgi.ProcessParams) (*seriesState, error) {
	var args seriesArgs
	if err := vgi.BindArgs(params.Args, &args); err != nil {
		return nil, err
	}
	return &seriesState{BatchState: vgi.NewBatchState(args.Count, 1024)}, nil
}

// Process is called repeatedly until the state reports no rows remain — that is
// what makes a table function a *generator*. GenerateBatch handles the chunking
// and signals completion for you; the callback only fills `size` rows.
func (*SeriesFn) Process(_ context.Context, _ *vgi.ProcessParams, state *seriesState, out *vgirpc.OutputCollector) error {
	return vgi.GenerateBatch(&state.BatchState, out, func(size int64) ([]arrow.Array, error) {
		start := state.Index
		return []arrow.Array{
			vgi.BuildInt64Array(size, func(i int64) int64 { return start + i }),
		}, nil
	})
}

// NewSeries returns the registration-ready table function.
func NewSeries() vgi.TableFunction {
	return vgi.AsTableFunction[seriesState](&SeriesFn{})
}

func main() {
	httpMode := flag.Bool("http", false, "serve over HTTP instead of stdio")
	logFlags := vgi.RegisterLoggingFlags(flag.CommandLine)
	flag.Parse()
	if err := logFlags.Apply(); err != nil {
		log.Fatalf("logging flags: %v", err)
	}

	w := vgi.NewWorker(
		vgi.WithCatalogName("calc"),
		vgi.WithCatalogComment("Tutorial worker: a scalar and a table function"),
	)
	w.RegisterScalar(NewDouble())
	w.RegisterTable(NewSeries())

	if *httpMode {
		if err := w.RunHttp("127.0.0.1:0"); err != nil {
			log.Fatal(err)
		}
		return
	}
	w.RunStdio()
}

ATTACH 'calc' (TYPE vgi, LOCATION './calc');
SELECT * FROM calc.series(3);

Input

count
3

Output

n
0
1
2

Table-in-out

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

filter/main.go

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

// Command filter is the table-in-out example for the vgi-go 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.
//
//	go build -o filterworker .
//	# then, in a Haybarn shell:
//	ATTACH 'filters' (TYPE vgi, LOCATION './filterworker');
//	SELECT * FROM filters.filter_positive((SELECT * FROM t));
package main

import (
	"context"
	"flag"
	"log"

	"github.com/Query-farm/vgi-go/vgi"
	"github.com/Query-farm/vgi-rpc-go/vgirpc"
	"github.com/apache/arrow-go/v18/arrow"
	"github.com/apache/arrow-go/v18/arrow/array"
	"github.com/apache/arrow-go/v18/arrow/memory"
)

var filterOutputSchema = arrow.NewSchema([]arrow.Field{
	{Name: "value", Type: arrow.PrimitiveTypes.Int64, Nullable: true},
}, nil)

// FilterPositiveFn keeps only rows whose value is greater than zero.
type FilterPositiveFn struct{}

var _ vgi.TypedTableInOutFunc[struct{}] = (*FilterPositiveFn)(nil)

func (*FilterPositiveFn) Name() string { return "filter_positive" }

func (*FilterPositiveFn) Metadata() vgi.FunctionMetadata {
	return vgi.FunctionMetadata{
		Description: "Keeps only rows whose value is positive",
		Stability:   vgi.StabilityConsistent,
	}
}

// ArgumentSpecs declares the input relation. A TABLE argument is written as an
// explicit ArgSpec rather than a struct tag: the struct-tag path binds scalar
// values into fields, and a relation is not a value — it arrives as the stream
// of batches Process is called with.
func (*FilterPositiveFn) ArgumentSpecs() []vgi.ArgSpec {
	return []vgi.ArgSpec{
		{Name: "data", Position: 0, ArrowType: "table", Doc: "Rows to filter"},
	}
}

func (*FilterPositiveFn) OnBind(_ *vgi.BindParams) (*vgi.BindResponse, error) {
	return vgi.BindSchema(filterOutputSchema)
}

// NewState returns the per-scan state. This function is stateless across
// batches — each one is decided on its own — so the state type is empty.
func (*FilterPositiveFn) NewState(_ *vgi.ProcessParams) (*struct{}, error) {
	return &struct{}{}, nil
}

// keepPositive is the predicate, kept separate from the plumbing so the rule can
// be read — and tested — on its own.
func keepPositive(col arrow.Array, rows int) []int64 {
	var kept []int64
	for row := 0; row < rows; row++ {
		if col.IsNull(row) {
			continue // a NULL is not positive
		}
		if v := vgi.GetInt64Value(col, row); v > 0 {
			kept = append(kept, v)
		}
	}
	return kept
}

// Process is called once per input batch. Emitting fewer rows than arrived is
// the whole point of the shape; emitting none is fine too.
func (*FilterPositiveFn) Process(_ context.Context, _ *vgi.ProcessParams, _ *struct{},
	batch arrow.RecordBatch, out *vgirpc.OutputCollector,
) error {
	kept := keepPositive(batch.Column(0), int(batch.NumRows()))

	mem := memory.NewGoAllocator()
	b := array.NewInt64Builder(mem)
	defer b.Release()
	b.AppendValues(kept, nil)

	arr := b.NewArray()
	defer arr.Release()
	// A zero-row batch is legitimate — this batch simply had nothing to keep.
	return out.Emit(array.NewRecordBatch(filterOutputSchema, []arrow.Array{arr}, int64(arr.Len())))
}

// Finalize runs once after the last input batch. A streaming filter holds nothing
// back, so it has nothing to flush.
func (*FilterPositiveFn) Finalize(_ context.Context, _ *vgi.ProcessParams, _ *struct{}) ([]arrow.RecordBatch, error) {
	return nil, nil
}

// NewFilterPositive returns the registration-ready function.
func NewFilterPositive() vgi.TableInOutFunction {
	return vgi.AsTableInOutFunction[struct{}](&FilterPositiveFn{})
}

func main() {
	httpMode := flag.Bool("http", false, "serve over HTTP instead of stdio")
	logFlags := vgi.RegisterLoggingFlags(flag.CommandLine)
	flag.Parse()
	if err := logFlags.Apply(); err != nil {
		log.Fatalf("logging flags: %v", err)
	}

	w := vgi.NewWorker(
		vgi.WithCatalogName("filters"),
		vgi.WithCatalogComment("Documentation example: a streaming row filter"),
	)
	w.RegisterTableInOut(NewFilterPositive())

	if *httpMode {
		if err := w.RunHttp("127.0.0.1:0"); err != nil {
			log.Fatal(err)
		}
		return
	}
	w.RunStdio()
}

ATTACH 'filters' (TYPE vgi, LOCATION './filterworker');
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

A TABLE argument needs an explicit ArgSpec

vgi:"..." struct tags bind scalar values into fields, and a relation is not a value — it arrives as the stream of batches Process is called with. So a table argument is written out longhand:

func (*FilterPositiveFn) ArgumentSpecs() []vgi.ArgSpec {
  return []vgi.ArgSpec{
      {Name: "data", Position: 0, ArrowType: "table", Doc: "Rows to filter"},
  }
}

Miss this and the function is un-callable rather than subtly wrong, which at least fails loudly.

Aggregate

Accumulate rows into per-group state, then emit one row per group. The four-phase split — NewState, Update, Combine, Finalize — is what lets DuckDB run the accumulation in parallel and merge the partials afterwards.

sum/main.go

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

// Command sum is the aggregate example for the vgi-go 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:
//
//   - NewState  — 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.
//
//     go build -o sumworker .
//     # then, in a Haybarn shell:
//     ATTACH 'agg' (TYPE vgi, LOCATION './sumworker');
//     SELECT category, agg.vgi_sum(value) FROM t GROUP BY category;
package main

import (
	"encoding/gob"
	"flag"
	"fmt"
	"log"

	"github.com/Query-farm/vgi-go/vgi"
	"github.com/apache/arrow-go/v18/arrow"
	"github.com/apache/arrow-go/v18/arrow/array"
	"github.com/apache/arrow-go/v18/arrow/memory"
)

// SumState is the per-group accumulator. It is serialized between phases —
// Update and Combine can run in different processes — so keep it small and
// keep it a plain struct of exported fields.
type SumState struct {
	Total int64
}

// Register the state with gob.
//
// Aggregate state is gob-encoded between phases. Through v0.21.0 nothing does
// this for you: the typed adapters (AsTableFunction, AsTableInOutFunction) call
// gob.Register(new(S)) themselves, but RegisterAggregate takes the interface
// directly and has no adapter. An unregistered state compiles, attaches, and
// then fails on the first GROUP BY with:
//
//	encoding aggregate state: gob: type not registered for interface: main.SumState
//
// Later versions register it from NewState, at which point this line becomes a
// harmless no-op — registering the same type twice is fine — so it is safe to
// keep either way.
func init() { gob.Register(&SumState{}) }

type sumArgs struct {
	Value int64 `vgi:"pos=0,const=false,doc=Column to sum"`
}

// SumFn sums a BIGINT column per group.
type SumFn struct{}

var _ vgi.AggregateFunction = (*SumFn)(nil)

func (*SumFn) Name() string { return "vgi_sum" }

func (*SumFn) Metadata() vgi.FunctionMetadata {
	return vgi.FunctionMetadata{
		Description: "Sums a BIGINT column per group",
		Stability:   vgi.StabilityConsistent,
		// NULLs are skipped rather than treated as zero, matching SQL's SUM.
		NullHandling:      vgi.NullHandlingDefault,
		ReturnType:        arrow.PrimitiveTypes.Int64,
		OrderDependent:    vgi.OrderDependenceNotDependent,
		DistinctDependent: vgi.DistinctDependenceNotDependent,
	}
}

func (*SumFn) ArgumentSpecs() []vgi.ArgSpec { return vgi.DeriveArgSpecs(sumArgs{}) }

func (*SumFn) OnBind(_ *vgi.AggregateBindParams) (*vgi.BindResponse, error) {
	return vgi.BindSchema(arrow.NewSchema([]arrow.Field{
		{Name: "result", Type: arrow.PrimitiveTypes.Int64},
	}, nil))
}

func (*SumFn) NewState(*vgi.AggregateProcessParams) interface{} { return &SumState{} }

// Update folds one batch into per-group state.
//
// A group is only created when a row genuinely contributes. Skipping a NULL
// without calling EnsureState is what makes SUM over an all-NULL group return
// NULL rather than 0 — the group never reaches storage, and Finalize sees no
// state for it.
func (*SumFn) Update(states map[int64]interface{}, gids *vgi.Int64Slice, columns []arrow.Array, _ *vgi.AggregateProcessParams) error {
	if len(columns) == 0 {
		return fmt.Errorf("vgi_sum: missing value column")
	}
	col, ok := columns[0].(*array.Int64)
	if !ok {
		return fmt.Errorf("vgi_sum: value column is %T, expected int64", columns[0])
	}
	for i := 0; i < gids.Len(); i++ {
		if col.IsNull(i) {
			continue
		}
		s := vgi.EnsureState(states, gids.At(i), func() *SumState { return &SumState{} })
		s.Total += col.Value(i)
	}
	return nil
}

// Combine merges two partials for the same group. It must be associative and
// commutative: DuckDB decides how many workers run and in what order they merge.
func (*SumFn) Combine(source, target interface{}, _ *vgi.AggregateProcessParams) (interface{}, error) {
	return &SumState{Total: source.(*SumState).Total + target.(*SumState).Total}, nil
}

// Finalize emits exactly one row per group id, in the order given. A group with
// no state contributed nothing, so it emits NULL.
func (*SumFn) Finalize(gids []int64, states map[int64]interface{}, _ *vgi.AggregateProcessParams) (arrow.RecordBatch, error) {
	mem := memory.NewGoAllocator()
	b := array.NewInt64Builder(mem)
	defer b.Release()
	for _, gid := range gids {
		if s, ok := states[gid].(*SumState); ok && s != nil {
			b.Append(s.Total)
		} else {
			b.AppendNull()
		}
	}
	arr := b.NewArray()
	defer arr.Release()
	schema := arrow.NewSchema([]arrow.Field{{Name: "result", Type: arrow.PrimitiveTypes.Int64, Nullable: true}}, nil)
	return array.NewRecordBatch(schema, []arrow.Array{arr}, int64(len(gids))), nil
}

func main() {
	httpMode := flag.Bool("http", false, "serve over HTTP instead of stdio")
	logFlags := vgi.RegisterLoggingFlags(flag.CommandLine)
	flag.Parse()
	if err := logFlags.Apply(); err != nil {
		log.Fatalf("logging flags: %v", err)
	}

	w := vgi.NewWorker(
		vgi.WithCatalogName("agg"),
		vgi.WithCatalogComment("Documentation example: a distributed aggregate"),
	)
	w.RegisterAggregate(&SumFn{})

	if *httpMode {
		if err := w.RunHttp("127.0.0.1:0"); err != nil {
			log.Fatal(err)
		}
		return
	}
	w.RunStdio()
}

ATTACH 'agg' (TYPE vgi, LOCATION './sumworker');
SELECT category, agg.vgi_sum(value) AS total
FROM (VALUES (0, 10), (0, 5), (1, 1), (1, 2), (1, 3)) AS t(category, value)
GROUP BY category ORDER BY category;

Input

category	value
0	10
0	5
1	1
1	2
1	3

Output

category	total
0	15
1	6

State is gob-encoded between phases

Update, Combine and Finalize may run in different processes, so per-group state round-trips through gob. Since v0.22.0 RegisterAggregate registers the concrete type for you, by asking NewState for one — so keep state a plain struct of exported fields and it works.

Two cases still need the explicit line the example carries:

v0.21.0 and earlier, where nothing registered it and an unregistered state failed on the first GROUP BY with gob: type not registered for interface.
A NewState that dereferences its params, which the SDK cannot call speculatively.

gob.Register(&MyState{}) in an init() covers both and is a no-op otherwise, so it is safe to keep regardless of the version you target.

Create a group only when a row contributes

Update calls vgi.EnsureState after the NULL check, not before. That is what makes SUM over an all-NULL group return NULL rather than 0: the group is never created, never reaches storage, and Finalize emits a null for it. Creating it eagerly quietly changes the SQL semantics.

Buffering

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). Because the phases can run in different worker processes, state lives in params.Storage, scoped to the execution.

rowcount/main.go

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

// Command rowcount is the buffering example for the vgi-go 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
// memory between them. State goes in params.Storage, which is scoped to this
// execution and shared across the workers serving it.
//
//	go build -o rowcountworker .
//	# then, in a Haybarn shell:
//	ATTACH 'buffers' (TYPE vgi, LOCATION './rowcountworker');
//	SELECT * FROM buffers.row_count((SELECT * FROM big_table));
package main

import (
	"context"
	"encoding/binary"
	"flag"
	"log"

	"github.com/Query-farm/vgi-go/vgi"
	"github.com/apache/arrow-go/v18/arrow"
	"github.com/apache/arrow-go/v18/arrow/array"
	"github.com/apache/arrow-go/v18/arrow/memory"
)

var rowCountOutputSchema = arrow.NewSchema([]arrow.Field{
	{Name: "count", Type: arrow.PrimitiveTypes.Int64},
}, nil)

// countsKey names the append-log inside this execution's storage scope. It does
// not need to be unique across queries — params.Storage is already scoped by
// execution id, so two concurrent scans cannot see each other's entries.
var countsKey = []byte("row_counts")

// RowCountFn counts every input row and emits a single total.
type RowCountFn struct{}

var _ vgi.TableBufferingFunction = (*RowCountFn)(nil)

func (*RowCountFn) Name() string { return "row_count" }

func (*RowCountFn) Metadata() vgi.FunctionMetadata {
	return vgi.FunctionMetadata{
		Description: "Counts every input row and returns one total",
		Stability:   vgi.StabilityConsistent,
	}
}

func (*RowCountFn) ArgumentSpecs() []vgi.ArgSpec {
	return []vgi.ArgSpec{
		{Name: "data", Position: 0, ArrowType: "table", Doc: "Rows to count"},
	}
}

// OnBind resolves the output schema. It is one int64 column whatever the input
// looked like, so the schema is fixed rather than derived from the input.
func (*RowCountFn) OnBind(_ *vgi.BindParams) (*vgi.BindResponse, error) {
	return vgi.BindSchema(rowCountOutputSchema)
}

// Process is the sink. It runs per batch and in parallel across DuckDB threads,
// so it appends rather than read-modify-writes: StateAppend is a log, and
// concurrent appends cannot lose each other the way a get-then-put would.
//
// Returning the execution id says "my partial lives in this execution's scope".
func (*RowCountFn) Process(_ context.Context, params *vgi.ProcessParams, batch arrow.RecordBatch) ([]byte, error) {
	var buf [8]byte
	binary.LittleEndian.PutUint64(buf[:], uint64(batch.NumRows()))
	if _, err := params.Storage.StateAppend(countsKey, buf[:]); err != nil {
		return nil, err
	}
	return params.ExecutionID, nil
}

// Combine runs once, after every sink call, on the coordinator. Returning a
// single id means Finalize is called once and produces one stream; returning
// several would fan the source phase out.
func (*RowCountFn) Combine(_ context.Context, params *vgi.ProcessParams, _ [][]byte) ([][]byte, error) {
	return [][]byte{params.ExecutionID}, nil
}

// Finalize is the source. It reads the reduced state back and emits the result.
func (*RowCountFn) Finalize(_ context.Context, params *vgi.ProcessParams, _ []byte) ([]arrow.RecordBatch, error) {
	// -1 / 0 means "from the beginning, no limit".
	entries, err := params.Storage.StateLogScan(countsKey, -1, 0)
	if err != nil {
		return nil, err
	}
	var total int64
	for _, e := range entries {
		total += int64(binary.LittleEndian.Uint64(e.Value))
	}
	mem := memory.NewGoAllocator()
	b := array.NewInt64Builder(mem)
	defer b.Release()
	b.Append(total)
	arr := b.NewArray()
	defer arr.Release()
	return []arrow.RecordBatch{array.NewRecordBatch(rowCountOutputSchema, []arrow.Array{arr}, 1)}, nil
}

func main() {
	httpMode := flag.Bool("http", false, "serve over HTTP instead of stdio")
	logFlags := vgi.RegisterLoggingFlags(flag.CommandLine)
	flag.Parse()
	if err := logFlags.Apply(); err != nil {
		log.Fatalf("logging flags: %v", err)
	}

	w := vgi.NewWorker(
		vgi.WithCatalogName("buffers"),
		vgi.WithCatalogComment("Documentation example: a full-input reduction"),
	)
	w.RegisterTableBuffering(&RowCountFn{})

	if *httpMode {
		if err := w.RunHttp("127.0.0.1:0"); err != nil {
			log.Fatal(err)
		}
		return
	}
	w.RunStdio()
}

ATTACH 'buffers' (TYPE vgi, LOCATION './rowcountworker');
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

Append, don't read-modify-write

The sink runs in parallel across DuckDB threads. StateAppend is a log, so concurrent appends can’t lose each other the way a get-then-put would; Finalize sums the log. Counters exist too, but on AttachStore — they are scoped to the attach, not to one execution, so they are the wrong tool for per-query accumulation.

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.

Parallelism and `OnInit`

AsTableFunction and AsTableInOutFunction give your function a default OnInit that returns MaxWorkers: 1. That is why every example above is correct as written: DuckDB runs one scan thread, so one NewState and one Process loop.

To go parallel, implement the optional OnIniter interface — the adapter detects it with a type assertion and uses it instead of the default:

func (*SeriesFn) OnInit(params *vgi.InitParams) (*vgi.GlobalInitResponse, error) {
  return &vgi.GlobalInitResponse{MaxWorkers: 4}, nil
}

Raising MaxWorkers without partitioning multiplies your output

Each parallel scan thread gets its own NewState and its own Process loop. series builds its cursor from an argument, so four workers each generate the whole range:

SELECT count(*) AS rows, sum(n) AS total FROM calc.series(10);

Output

rows	total
40	180

Forty rows, not ten. Nothing errors — the query just quadruples. MaxWorkers: 1 is a safe default precisely because it is the only setting that needs no coordination.

The fix is to hand out work rather than let every worker derive it. params.Storage is scoped to the execution and shared by all of its workers, and QueuePush/QueuePop are the primitive for exactly this: OnInit runs once and pushes the shards, then each worker pops until the queue is empty.

func (*SeriesFn) OnInit(params *vgi.InitParams) (*vgi.GlobalInitResponse, error) {
  var args seriesArgs
  if err := vgi.BindArgs(params.Args, &args); err != nil {
      return nil, err
  }
  var items [][]byte
  for start := int64(0); start < args.Count; start += shardSize {
      items = append(items, encodeShard(start, min(start+shardSize, args.Count)))
  }
  if err := params.Storage.QueuePush(items); err != nil {
      return nil, err
  }
  return &vgi.GlobalInitResponse{MaxWorkers: 4}, nil
}

func (*SeriesFn) Process(_ context.Context, params *vgi.ProcessParams,
  state *seriesState, out *vgirpc.OutputCollector,
) error {
  if state.exhausted() {
      item, err := params.Storage.QueuePop()
      if err != nil {
          return err
      }
      if item == nil {
          return out.Finish()   // no shards left — this worker is done
      }
      state.load(item)
  }
  ...
}

Output

rows	total
10	45

Ten rows again, now produced by four workers. QueuePop returning nil is how a worker learns the scan is finished, which is why the pop lives in Process rather than NewState — a worker that claimed one shard up front would drop the rest if DuckDB started fewer threads than there are shards.

Next steps

The guided build → 1. Scalar function · 2. Table function.
When each callback fires → Function lifecycle.
Exact contracts → Package overview.

Function patterns

Which shape do I need?

Scalar

Table

Table-in-out

Aggregate

Buffering

Parallelism and OnInit

Next steps

Parallelism and `OnInit`