Function patterns

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

Before you run these 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:

Copy Copied!

cargo build --release

LOCATION is resolved relative to the directory the engine was started in.

Which shape do I need?

If your function…	Use	Trait
maps each row independently	Scalar	ScalarFunction
produces rows from arguments	Table	TableFunction + TableProducer
transforms a relation as it streams	Table-in-out	TableInOutFunction
folds rows to one value per group	Aggregate	AggregateFunction
needs every row before it can answer	Buffering	TableBufferingFunction

Scalar

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 — process gets a whole column and returns a column of the same length.

calcscalar.rs

Copy Copied!

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

//! The worker built in step 1 of the vgi-rust 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.
//!
//! ```text
//! cargo build --release --bin calcscalar
//! # then, in a Haybarn shell:
//! ATTACH 'calc' (TYPE vgi, LOCATION './target/release/calcscalar');
//! SELECT calc.double(21);
//! ```

use std::sync::Arc;

use arrow_array::cast::AsArray;
use arrow_array::types::Int64Type;
use arrow_array::{Array, ArrayRef, Int64Array, RecordBatch};
use arrow_schema::DataType;
use vgi::catalog::CatalogModel;
use vgi::function::{ArgSpec, FunctionMetadata, ProcessParams, ScalarFunction};
use vgi::{Result, RpcError};

/// Doubles each value in its input column.
struct Double;

impl ScalarFunction for Double {
    /// The SQL name, qualified by the catalog: `calc.double(...)`.
    fn name(&self) -> &str {
        "double"
    }

    fn metadata(&self) -> FunctionMetadata {
        FunctionMetadata {
            description: "Doubles a BIGINT".to_string(),
            // The declared output type. A function whose output depends on its
            // input leaves this off and decides in on_bind instead.
            return_type: Some(DataType::Int64),
            ..Default::default()
        }
    }

    /// The signature. `column` means the value arrives per row; the type is
    /// named as an Arrow type string, so `int64` rather than SQL's `bigint`.
    fn argument_specs(&self) -> Vec<ArgSpec> {
        vec![ArgSpec::column("n", 0, "int64", "Value to double")]
    }

    /// Called once per input BATCH, not per row.
    fn process(&self, params: &ProcessParams, batch: &RecordBatch) -> Result<RecordBatch> {
        // The column arrives as the type the spec declared, so this downcast is
        // safe. See the tutorial for what to do when it might not be.
        let n = batch.column(0).as_primitive::<Int64Type>();

        // One output value per input row. Null in, null out — collecting from
        // Option is what carries that through.
        let out: Int64Array = (0..n.len())
            .map(|i| {
                if n.is_valid(i) {
                    Some(n.value(i) * 2)
                } else {
                    None
                }
            })
            .collect();

        RecordBatch::try_new(
            params.output_schema.clone(),
            vec![Arc::new(out) as ArrayRef],
        )
        .map_err(|e| RpcError::runtime_error(e.to_string()))
    }
}

fn main() {
    let mut worker = vgi::Worker::new();
    worker.register_scalar(Double);

    // Functions are served through a catalog, and its name is the name DuckDB
    // ATTACHes. They must match: attaching under any other name fails.
    worker.set_catalog(CatalogModel {
        name: "calc".to_string(),
        ..Default::default()
    });

    worker.run();
}

Copy Copied!

ATTACH 'calc' (TYPE vgi, LOCATION './target/release/calcscalar');
SELECT calc.double(21);

Output

double(21)
42

Table

args → N rows

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

Generate rows from arguments, with no input relation. The function is shared and immutable; the per-scan cursor is a separate TableProducer, pulled until it answers Ok(None).

calc.rs

Copy Copied!

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

//! The worker built across the vgi-rust 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.
//!
//! ```text
//! cargo build --release --bin calc
//! # then, in a Haybarn shell:
//! ATTACH 'calc' (TYPE vgi, LOCATION './target/release/calc');
//! SELECT calc.double(21);
//! SELECT * FROM calc.series(3);
//! ```

use std::sync::Arc;

use arrow_array::cast::AsArray;
use arrow_array::types::Int64Type;
use arrow_array::{Array, ArrayRef, Int64Array, RecordBatch};
use arrow_schema::{DataType, Field, Schema, SchemaRef};
use vgi::catalog::CatalogModel;
use vgi::function::{
    ArgSpec, BindParams, BindResponse, FunctionMetadata, ProcessParams, ScalarFunction,
};
use vgi::table_function::{TableFunction, TableProducer};
use vgi::vgi_rpc::OutputCollector;
use vgi::{Result, RpcError};

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

struct Double;

impl ScalarFunction for Double {
    fn name(&self) -> &str {
        "double"
    }
    fn metadata(&self) -> FunctionMetadata {
        FunctionMetadata {
            description: "Doubles a BIGINT".to_string(),
            return_type: Some(DataType::Int64),
            ..Default::default()
        }
    }
    fn argument_specs(&self) -> Vec<ArgSpec> {
        vec![ArgSpec::column("n", 0, "int64", "Value to double")]
    }
    fn process(&self, params: &ProcessParams, batch: &RecordBatch) -> Result<RecordBatch> {
        let n = batch.column(0).as_primitive::<Int64Type>();
        let out: Int64Array = (0..n.len())
            .map(|i| {
                if n.is_valid(i) {
                    Some(n.value(i) * 2)
                } else {
                    None
                }
            })
            .collect();
        RecordBatch::try_new(
            params.output_schema.clone(),
            vec![Arc::new(out) as ArrayRef],
        )
        .map_err(|e| RpcError::runtime_error(e.to_string()))
    }
}

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

const BATCH_SIZE: i64 = 1024;

/// The per-scan cursor. A table function is *pulled*: the engine calls
/// `next_batch` until it answers `None`, so whatever the function needs between
/// calls lives here rather than in the function itself (which is shared and
/// must stay `Sync`).
struct SeriesProducer {
    schema: SchemaRef,
    next: i64,
    count: i64,
}

impl TableProducer for SeriesProducer {
    fn next_batch(&mut self, _out: &mut OutputCollector) -> Result<Option<RecordBatch>> {
        if self.next >= self.count {
            // None is end-of-stream. Returning an empty batch instead would
            // loop forever.
            return Ok(None);
        }
        let end = (self.next + BATCH_SIZE).min(self.count);
        let col: ArrayRef = Arc::new((self.next..end).collect::<Int64Array>());
        self.next = end;
        RecordBatch::try_new(self.schema.clone(), vec![col])
            .map(Some)
            .map_err(|e| RpcError::runtime_error(e.to_string()))
    }
}

struct Series;

impl TableFunction for Series {
    fn name(&self) -> &str {
        "series"
    }
    fn metadata(&self) -> FunctionMetadata {
        FunctionMetadata {
            description: "Generates the integers 0..count-1".to_string(),
            ..Default::default()
        }
    }

    fn argument_specs(&self) -> Vec<ArgSpec> {
        // const_arg, not column: the value is fixed for the whole scan and read
        // at bind. with_ge(0.0) is enforced by the framework before any row is
        // produced, so series(-1) fails rather than returning nothing.
        vec![ArgSpec::const_arg("count", 0, "int64", "How many numbers to generate").with_ge(0.0)]
    }

    /// Runs once per query, before any data moves. The output shape is fixed
    /// here, so it never inspects the arguments; a function whose columns depend
    /// on its arguments would build the schema from `params` instead.
    fn on_bind(&self, _params: &BindParams) -> Result<BindResponse> {
        Ok(BindResponse {
            output_schema: Arc::new(Schema::new(vec![Field::new("n", DataType::Int64, true)])),
            opaque_data: Vec::new(),
        })
    }

    /// 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.
    fn producer(&self, params: &ProcessParams) -> Result<Box<dyn TableProducer>> {
        Ok(Box::new(SeriesProducer {
            schema: params.output_schema.clone(),
            next: 0,
            count: params.arguments.const_i64(0).unwrap_or(0),
        }))
    }
}

fn main() {
    let mut worker = vgi::Worker::new();
    worker.register_scalar(Double);
    worker.register_table(Series);
    worker.set_catalog(CatalogModel {
        name: "calc".to_string(),
        comment: Some("Tutorial worker: a scalar and a table function".to_string()),
        ..Default::default()
    });
    worker.run();
}

Copy Copied!

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

Input

count
3

→

Output

n
0
1
2

Table-in-out

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.rs

Copy Copied!

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

//! The table-in-out example for the vgi-rust 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 call
//! emits what it can from the batch in hand, which is what keeps memory flat
//! over an arbitrarily large scan.
//!
//! ```text
//! cargo build --release --bin filter
//! # then, in a Haybarn shell:
//! ATTACH 'filters' (TYPE vgi, LOCATION './target/release/filter');
//! SELECT * FROM filters.filter_positive((SELECT * FROM t));
//! ```

use arrow_array::cast::AsArray;
use arrow_array::types::Int64Type;
use arrow_array::{Array, BooleanArray, RecordBatch};
use vgi::catalog::CatalogModel;
use vgi::function::{ArgSpec, FunctionMetadata, ProcessParams};
use vgi::table_in_out::TableInOutFunction;
use vgi::{Result, RpcError};

struct FilterPositive;

impl TableInOutFunction for FilterPositive {
    fn name(&self) -> &str {
        "filter_positive"
    }

    fn metadata(&self) -> FunctionMetadata {
        FunctionMetadata {
            description: "Keeps only the rows whose `value` column is greater than zero"
                .to_string(),
            ..Default::default()
        }
    }

    /// The TABLE argument is declared like any other, with the Arrow type
    /// string `"table"`. Leave it out and the extension does not know this
    /// function takes a relation: the call fails with
    /// *"Table function cannot contain subqueries"*.
    fn argument_specs(&self) -> Vec<ArgSpec> {
        vec![ArgSpec::column("data", 0, "table", "Rows to filter")]
    }

    // on_bind is left at its default, which echoes the input schema. This
    // function drops rows, not columns, so the shapes match.

    /// Called once per input batch. Return zero or more batches; an empty Vec
    /// drops the batch entirely.
    fn process(&self, _params: &ProcessParams, batch: &RecordBatch) -> Result<Vec<RecordBatch>> {
        let col = batch
            .column_by_name("value")
            .ok_or_else(|| RpcError::value_error("expected a `value` column"))?;

        // The caller's relation decides the column's width, so normalize to
        // int64 rather than assuming. cast errors instead of guessing, which is
        // what you want here — a `value` column of strings is a caller mistake,
        // not something to paper over.
        let cast = arrow_cast::cast(col, &arrow_schema::DataType::Int64)
            .map_err(|e| RpcError::value_error(format!("`value` must be an integer: {e}")))?;
        let v = cast.as_primitive::<Int64Type>();

        let keep: BooleanArray = (0..v.len())
            .map(|i| Some(v.is_valid(i) && v.value(i) > 0))
            .collect();

        let filtered = arrow_select::filter::filter_record_batch(batch, &keep)
            .map_err(|e| RpcError::runtime_error(e.to_string()))?;

        // An empty batch is legal but pointless — skip the round trip.
        if filtered.num_rows() == 0 {
            return Ok(Vec::new());
        }
        Ok(vec![filtered])
    }
}

fn main() {
    let mut worker = vgi::Worker::new();
    worker.register_table_in_out(FilterPositive);
    worker.set_catalog(CatalogModel {
        name: "filters".to_string(),
        comment: Some("Documentation example: a streaming table-in-out function".to_string()),
        ..Default::default()
    });
    worker.run();
}

Copy Copied!

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

Declare the TABLE argument, or the call will not bind

The input relation is an ordinary ArgSpec with the Arrow type string "table":

Copy Copied!

fn argument_specs(&self) -> Vec<ArgSpec> {
  vec![ArgSpec::column("data", 0, "table", "Rows to filter")]
}

Leave it out and the extension does not know this function takes a relation. The failure is “Binder Error: Table function cannot contain subqueries” — which does not mention the worker, the function, or the missing spec, so it is worth recognizing on sight.

on_bind’s default echoes the input schema, which is what a row-dropping function like this one wants; override it when the output shape differs.

Aggregate

N rows → 1 value

Folds many rows down into a single value per group.

Four phases — initial_state, 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.rs

Copy Copied!

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

//! The aggregate example for the vgi-rust 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:
//!
//!   - `initial_state` — 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.
//!
//! ```text
//! cargo build --release --bin sum
//! # then, in a Haybarn shell:
//! ATTACH 'agg' (TYPE vgi, LOCATION './target/release/sum');
//! SELECT category, agg.vgi_sum(value) FROM t GROUP BY category;
//! ```

use std::collections::HashMap;
use std::sync::Arc;

use arrow_array::cast::AsArray;
use arrow_array::types::Int64Type;
use arrow_array::{Array, ArrayRef, Int64Array, RecordBatch};
use arrow_schema::{DataType, Field, Schema, SchemaRef};
use vgi::aggregate::{AggregateBindParams, AggregateFunction};
use vgi::catalog::CatalogModel;
use vgi::function::{ArgSpec, BindResponse, FunctionMetadata};
use vgi::{Result, RpcError};

/// Aggregate state is an opaque `Vec<u8>` — the framework moves it between
/// phases and possibly between processes, so it never inspects it. Encoding is
/// the function's job. Eight little-endian bytes is the whole of it here; a
/// richer state would reach for serde.
fn encode(total: i64) -> Vec<u8> {
    total.to_le_bytes().to_vec()
}

fn decode(bytes: &[u8]) -> Result<i64> {
    bytes
        .try_into()
        .map(i64::from_le_bytes)
        .map_err(|_| RpcError::runtime_error(format!("corrupt sum state: {} bytes", bytes.len())))
}

struct VgiSum;

impl AggregateFunction for VgiSum {
    fn name(&self) -> &str {
        "vgi_sum"
    }

    fn metadata(&self) -> FunctionMetadata {
        FunctionMetadata {
            description: "Sums a BIGINT column per group".to_string(),
            // DEFAULT means DuckDB skips NULL inputs, so update never sees one.
            // That is what makes SUM over an all-NULL group return NULL: the
            // group never appears in group_ids, so finalize is handed None.
            null_handling: Some("DEFAULT".to_string()),
            ..Default::default()
        }
    }

    fn argument_specs(&self) -> Vec<ArgSpec> {
        vec![ArgSpec::column("value", 0, "int64", "Column to sum")]
    }

    fn on_bind(&self, _params: &AggregateBindParams) -> Result<BindResponse> {
        Ok(BindResponse {
            output_schema: Arc::new(Schema::new(vec![Field::new(
                "result",
                DataType::Int64,
                true,
            )])),
            opaque_data: Vec::new(),
        })
    }

    fn initial_state(&self) -> Vec<u8> {
        encode(0)
    }

    /// `states` arrives pre-loaded with the initial state for every group id in
    /// this batch, so there is no "create if missing" step — just fold.
    fn update(
        &self,
        states: &mut HashMap<i64, Vec<u8>>,
        group_ids: &Int64Array,
        columns: &[ArrayRef],
    ) -> Result<()> {
        let values = columns
            .first()
            .ok_or_else(|| RpcError::value_error("vgi_sum: missing value column"))?
            .as_primitive::<Int64Type>();

        for i in 0..group_ids.len() {
            if !values.is_valid(i) {
                continue;
            }
            let gid = group_ids.value(i);
            let total = states
                .get(&gid)
                .map(|s| decode(s))
                .transpose()?
                .unwrap_or(0);
            states.insert(gid, encode(total + values.value(i)));
        }
        Ok(())
    }

    /// Must be associative and commutative: DuckDB decides how many workers run
    /// and in what order their partials merge.
    fn combine(&self, target: Vec<u8>, source: Vec<u8>) -> Result<Vec<u8>> {
        Ok(encode(decode(&target)? + decode(&source)?))
    }

    /// One row per group id, in the order given. `None` means the group never
    /// contributed a non-null row, which is SQL NULL.
    fn finalize(
        &self,
        output_schema: &SchemaRef,
        group_ids: &Int64Array,
        states: &[Option<Vec<u8>>],
    ) -> Result<RecordBatch> {
        let mut out: Vec<Option<i64>> = Vec::with_capacity(group_ids.len());
        for state in states.iter() {
            out.push(match state {
                Some(bytes) => Some(decode(bytes)?),
                None => None,
            });
        }
        let col: ArrayRef = Arc::new(out.into_iter().collect::<Int64Array>());
        RecordBatch::try_new(output_schema.clone(), vec![col])
            .map_err(|e| RpcError::runtime_error(e.to_string()))
    }
}

fn main() {
    let mut worker = vgi::Worker::new();
    worker.register_aggregate(VgiSum);
    worker.set_catalog(CatalogModel {
        name: "agg".to_string(),
        comment: Some("Documentation example: a distributed aggregate".to_string()),
        ..Default::default()
    });
    worker.run();
}

Copy Copied!

ATTACH 'agg' (TYPE vgi, LOCATION './target/release/sum');
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

State is bytes You choose the encoding

Aggregate state is an opaque Vec<u8>. The framework moves it between phases and possibly between processes, so it never inspects it — encoding is your job. Eight little-endian bytes covers a sum; anything richer reaches for serde. This is more explicit than Go (which uses gob) or TypeScript (where state is a plain object), and it is why a Rust aggregate has no registration step.

An all-NULL group returns NULL, and that is not an accident

null_handling: "DEFAULT" means DuckDB never calls update for a NULL input. A group whose values are all NULL therefore never appears in group_ids, so finalize is handed None for it and emits SQL NULL — matching built-in SUM. Group c above is that case.

Buffering

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_producer (the source).

rowcount.rs

Copy Copied!

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

//! The buffering example for the vgi-rust 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 an opaque state id.
//!   - `combine`                    — called once, on the coordinator, with every
//!     state id the sink produced. Reduce them into the ids the source drains.
//!   - `finalize_producer` (source) — called per finalize id, streaming the
//!     result out.
//!
//! The phases can run in different worker processes, so nothing may live in a
//! `static` between them. State goes in `params.storage`, scoped to the
//! execution and shared across the workers serving it.
//!
//! ```text
//! cargo build --release --bin rowcount
//! # then, in a Haybarn shell:
//! ATTACH 'buffers' (TYPE vgi, LOCATION './target/release/rowcount');
//! SELECT * FROM buffers.row_count((SELECT * FROM big_table));
//! ```

use std::sync::Arc;

use arrow_array::{ArrayRef, Int64Array, RecordBatch};
use arrow_schema::{DataType, Field, Schema, SchemaRef};
use vgi::buffering::{BufferingParams, TableBufferingFunction};
use vgi::catalog::CatalogModel;
use vgi::function::{ArgSpec, BindParams, BindResponse, FunctionMetadata};
use vgi::table_function::TableProducer;
use vgi::vgi_rpc::OutputCollector;
use vgi::{Result, RpcError};

const NS: &[u8] = b"rowcount";
const KEY: &[u8] = b"";

/// The source-phase producer: emits the total once, then ends the stream.
struct CountProducer {
    schema: SchemaRef,
    total: Option<i64>,
}

impl TableProducer for CountProducer {
    fn next_batch(&mut self, _out: &mut OutputCollector) -> Result<Option<RecordBatch>> {
        let Some(total) = self.total.take() else {
            return Ok(None);
        };
        let col: ArrayRef = Arc::new(Int64Array::from(vec![total]));
        RecordBatch::try_new(self.schema.clone(), vec![col])
            .map(Some)
            .map_err(|e| RpcError::runtime_error(e.to_string()))
    }
}

struct RowCount;

impl TableBufferingFunction for RowCount {
    fn name(&self) -> &str {
        "row_count"
    }

    fn metadata(&self) -> FunctionMetadata {
        FunctionMetadata {
            description: "Counts every row of the input relation".to_string(),
            ..Default::default()
        }
    }

    fn argument_specs(&self) -> Vec<ArgSpec> {
        vec![ArgSpec::column("data", 0, "table", "Rows to count")]
    }

    /// Output is one BIGINT, whatever the input looked like.
    fn on_bind(&self, _params: &BindParams) -> Result<BindResponse> {
        Ok(BindResponse {
            output_schema: Arc::new(Schema::new(vec![Field::new(
                "count",
                DataType::Int64,
                true,
            )])),
            opaque_data: Vec::new(),
        })
    }

    /// The sink runs in parallel across DuckDB threads. `append` is an
    /// append-only log, so concurrent appends cannot lose each other the way a
    /// read-modify-write would.
    fn process(&self, params: &BufferingParams, batch: &RecordBatch) -> Result<Vec<u8>> {
        let n = batch.num_rows() as i64;
        params
            .storage
            .append(&params.execution_id, NS, KEY, n.to_le_bytes().to_vec());
        Ok(params.execution_id.clone())
    }

    /// Runs once, on the coordinator. There is a single bucket to drain here,
    /// so it just names the execution; a top-k would reduce the partials first.
    fn combine(&self, params: &BufferingParams, _state_ids: &[Vec<u8>]) -> Result<Vec<Vec<u8>>> {
        Ok(vec![params.execution_id.clone()])
    }

    /// The source phase: sum the log and hand back a one-shot producer.
    fn finalize_producer(
        &self,
        params: &BufferingParams,
        _finalize_state_id: Vec<u8>,
    ) -> Result<Box<dyn TableProducer>> {
        let mut total = 0i64;
        // -1 starts before the first entry; the limit is a page size, not a cap.
        let mut after_id = -1i64;
        loop {
            let rows = params
                .storage
                .scan(&params.execution_id, NS, KEY, after_id, 256);
            if rows.is_empty() {
                break;
            }
            for (id, value) in rows {
                let bytes: [u8; 8] = value
                    .as_slice()
                    .try_into()
                    .map_err(|_| RpcError::runtime_error("corrupt row_count state"))?;
                total += i64::from_le_bytes(bytes);
                after_id = id;
            }
        }
        Ok(Box::new(CountProducer {
            schema: params.output_schema.clone(),
            total: Some(total),
        }))
    }
}

fn main() {
    let mut worker = vgi::Worker::new();
    worker.register_buffering(RowCount);
    worker.set_catalog(CatalogModel {
        name: "buffers".to_string(),
        comment: Some(
            "Documentation example: a buffering (sink → combine → source) function".to_string(),
        ),
        ..Default::default()
    });
    worker.run();
}

Copy Copied!

ATTACH 'buffers' (TYPE vgi, LOCATION './target/release/rowcount');
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 `static` will not survive the phases

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

append 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

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