Persist state across workers

How functions that span multiple worker processes — buffering functions, distributed aggregates, COPY writers — share state across them.

Prerequisites

A buffering function or aggregate.

Three kinds of state — don’t confuse them

Go’s SDK draws the line more sharply than the concept does, because each kind lives somewhere different:

Kind	Where it lives	Lifetime
Scan state — a generator’s cursor	The `S` type parameter of `TypedTableFunc[S]`, gob-encoded into the stream token	One scan
Execution storage — `params.Storage`	`ExecutionStorage`, scoped by `execution_id`	One query
Attach storage — `params.Storage.AttachStore(scope)`	`AttachStore`, scoped by the attached catalog	Across queries, until `DETACH`

Picking the wrong scope is a quiet bug: attach-scoped state shared between two concurrent queries will interleave, and execution-scoped state vanishes before the next query can read it.

Execution storage

params.Storage is what a buffering function’s phases share. Its primitives are an append-log and a queue:

Method	Use it for
`StateAppend(key, value)`	Append one entry; returns its index. The sink’s tool.
`StateLogScan(key, afterID, limit)`	Read the log back in order. `-1, 0` means “from the beginning, no limit”.
`StateLogClear()`	Drop everything under this execution.
`QueuePush` / `QueuePop`	Hand work items between worker processes.
`QueuePushBatches` / `QueuePopBatch`	The same, for Arrow batches.

// Sink: append a partial. Runs per batch, in parallel across DuckDB threads.
if _, err := params.Storage.StateAppend(countsKey, buf[:]); err != nil {
  return nil, err
}

// Source: read every partial back and reduce.
entries, err := params.Storage.StateLogScan(countsKey, -1, 0)

Append, don't read-modify-write

The sink runs in parallel. StateAppend is a log, so concurrent appends cannot lose each other the way a get-then-put would. Reduce in Combine or Finalize, where you are alone.

Attach storage

AttachStore is a key–value store scoped to the attached catalog, so it outlives a query. Reach it through params.Storage.AttachStore(scope):

// In Process, open it from the scan's attach scope:
store, err := params.Storage.AttachStore(params.AttachScope)
if err != nil {
  return err
}
if err := store.Put([]byte("cursor"), []byte("page"), []byte("42")); err != nil {
  return err
}

BindParams carries a shorthand — params.AttachStore() — so bind-time code that only needs the current catalog’s scope can skip the argument. ProcessParams has no such method; pass params.AttachScope explicitly there.

It offers Put / Get / Scan / Drain, range and namespace deletes, and atomic counters (CounterAdd, CounterGet, CounterSet, CounterDelete).

Counters are attach-scoped, not execution-scoped

CounterAdd is genuinely atomic and looks like the obvious tool for a parallel sink — but it lives on AttachStore, so two concurrent scans of the same catalog would share one counter and corrupt each other’s totals. For per-query accumulation use the execution-scoped append-log instead; that is what the buffering example does.

Values are bytes

Everything stored is []byte, so serialize first. For Arrow payloads the SDK provides vgi.SerializeRecordBatch and vgi.DeserializeRecordBatch; for small scalars, pack them yourself:

var buf [8]byte
binary.LittleEndian.PutUint64(buf[:], uint64(batch.NumRows()))

The sink and the source may run in different processes, so a mismatch between how you write and how you read is invisible until a query returns a wrong answer. Keep the encoding in one place.

Choosing a backend

By default a worker always uses local SQLite, at the per-user state path, with no configuration — every worker in the deployment shares one WAL-mode database. That is the whole story unless you ask for something else.

Env-driven selection is opt-in, and costs one line

vgi.NewWorker does not read VGI_WORKER_SHARED_STORAGE. Setting that variable against a worker built the obvious way changes nothing — it is honoured only by the resolve sub-package, which you wire in yourself:

import "github.com/Query-farm/vgi-go/vgi/storage/resolve"

storage, err := resolve.FromEnv()
if err != nil {
  log.Fatalf("resolve storage backend: %v", err)
}
w := vgi.NewWorker(
  vgi.WithFunctionStorage(storage),
  vgi.WithCatalogName("acme"),
)

resolve lives in its own sub-package deliberately: it imports every backend, and a worker that doesn’t need env-driven selection shouldn’t pay for those imports.

With resolve wired in, VGI_WORKER_SHARED_STORAGE accepts:

Value	Backend
`sqlite` (default, or unset)	Local SQLite at the per-user state path.
`memory`	In-process SQLite at `:memory:`. Process-local with no cross-process coordination — single-process only, so not usable for a buffering function spanning workers.
`cloudflare-do`	Cloudflare Worker + Durable Object. Requires `VGI_CF_DO_URL`, optionally `VGI_CF_DO_TOKEN`.

You can also skip the env var entirely and pass a backend directly with vgi.WithFunctionStorage.

Next steps

The shape that needs this most → Function patterns → Buffering.
Exact methods → State storage.