Choose a function pattern

Choose by lifecycle, not by the name of the operation:

Shape	SQL shape	Use when
`ScalarFn` / `IScalarFunction`	one row → one value	Rows are independent and vectorizable.
`ITableFunction`	arguments → rows	The worker produces a relation.
`ITableInOutFunction`	rows → rows, streaming	Each input batch can produce output without seeing later batches.
`IAggregateFunction`	rows → one value per group	DuckDB owns grouping; the worker owns mergeable group state.
`ITableBufferingFunction`	rows → rows, after all input	Output depends on the complete relation: global sort, collect, write, or deduplicate.

Streaming table-in/out

EchoFunction.cs

using Apache.Arrow;
using QueryFarm.Vgi.Table;
using QueryFarm.Vgi.TableInOut;
using QueryFarm.VgiRpc.Streaming;

namespace QueryFarm.Vgi.DocsExamples;

public sealed class EchoFunction : ITableInOutFunction
{
    public string Name => "echo";

    public string Description => "Return each input batch unchanged";

    public Schema ArgumentsSchema { get; } = new([TableArgFields.Table("data")], metadata: null);

    public Schema OutputSchema { get; } = new([], metadata: null);

    public Schema ResolveOutputSchema(TableInOutBindParams bindParams) => bindParams.InputSchema;

    public ITableInOutProcessor CreateProcessor(TableInOutInitParams initParams) => new Processor();

    private sealed class Processor : ITableInOutProcessor
    {
        public void Process(RecordBatch input, OutputCollector output) => output.Emit(input);
    }
}

The output schema may depend on the input schema. Each processor belongs to one input substream; do not use instance fields for query-wide state.

Aggregate

SumFunction.cs

using Apache.Arrow;
using Apache.Arrow.Types;
using QueryFarm.Vgi.Aggregate;

namespace QueryFarm.Vgi.DocsExamples;

public sealed class SumFunction : IAggregateFunction
{
    public string Name => "sum_int64";

    public string Description => "Sum BIGINT values";

    public Schema ArgumentsSchema { get; } = new(
        [new Field("value", Int64Type.Default, nullable: true)],
        metadata: null);

    public Schema OutputSchema { get; } = new(
        [new Field("result", Int64Type.Default, nullable: true)],
        metadata: null);

    public void Update(
        RecordBatch inputColumns,
        long[] groupIds,
        Dictionary<long, byte[]> states,
        AggregateCallParams callParams)
    {
        var values = (Int64Array)inputColumns.Column(0);
        for (var row = 0; row < groupIds.Length; row++)
        {
            if (values.IsNull(row))
            {
                continue;
            }

            var groupId = groupIds[row];
            var current = states.TryGetValue(groupId, out var state) ? BitConverter.ToInt64(state) : 0;
            states[groupId] = BitConverter.GetBytes(current + values.GetValue(row)!.Value);
        }
    }

    public byte[] Combine(byte[] source, byte[]? target, AggregateCallParams callParams) =>
        BitConverter.GetBytes(BitConverter.ToInt64(source) + (target is null ? 0 : BitConverter.ToInt64(target)));

    public IArrowArray Finalize(
        long[] groupIds,
        byte[]?[] states,
        Schema outputSchema,
        AggregateCallParams callParams)
    {
        var result = new Int64Array.Builder();
        foreach (var state in states)
        {
            if (state is null)
            {
                result.AppendNull();
            }
            else
            {
                result.Append(BitConverter.ToInt64(state));
            }
        }

        return result.Build();
    }
}

State is opaque byte[], which makes it portable across processes. Preserve SQL null semantics: the absence of state can differ from a serialized zero.

Buffering

CollectFunction.cs

using Apache.Arrow;
using Apache.Arrow.Types;
using QueryFarm.Vgi.Buffering;
using QueryFarm.Vgi.Table;
using QueryFarm.Vgi.TableInOut;
using QueryFarm.VgiRpc.Streaming;

namespace QueryFarm.Vgi.DocsExamples;

public sealed class CollectFunction : ITableBufferingFunction
{
    private const string StorageNamespace = "collect";
    private const string StorageKey = "partial-sums";

    public string Name => "collect_sum";

    public string Description => "Buffer all input before returning its sum";

    public Schema ArgumentsSchema { get; } = new([TableArgFields.Table("data")], metadata: null);

    public Schema OutputSchema { get; } = new(
        [new Field("sum", Int64Type.Default, nullable: false)],
        metadata: null);

    public byte[] Process(RecordBatch batch, TableBufferingProcessParams processParams)
    {
        var values = (Int64Array)batch.Column(0);
        long partialSum = 0;
        for (var row = 0; row < values.Length; row++)
        {
            if (!values.IsNull(row))
            {
                partialSum += values.GetValue(row)!.Value;
            }
        }

        processParams.Storage.Append(StorageNamespace, StorageKey, BitConverter.GetBytes(partialSum));
        return processParams.ExecutionId;
    }

    public IReadOnlyList<byte[]> Combine(
        IReadOnlyList<byte[]> stateIds,
        TableBufferingCombineParams combineParams) => [combineParams.ExecutionId];

    public ITableFunctionProducer CreateFinalizeProducer(
        byte[] finalizeStateId,
        TableBufferingFinalizeParams finalizeParams) => new Producer(finalizeParams);

    private sealed class Producer(TableBufferingFinalizeParams finalizeParams) : ITableFunctionProducer
    {
        private bool _finished;

        public void Produce(OutputCollector output)
        {
            if (_finished)
            {
                output.Finish();
                return;
            }

            var total = finalizeParams.Storage
                .ScanLog(StorageNamespace, StorageKey)
                .Sum(state => BitConverter.ToInt64(state));
            var values = new Int64Array.Builder().Append(total).Build();
            output.Emit(new RecordBatch(finalizeParams.OutputSchema, [values], 1));
            output.Finish();
            _finished = true;
        }
    }
}

Process is the sink, Combine decides the final output streams, and the final producer is the source. Store derived summaries when possible; storing complete Arrow batches costs more memory and I/O.

The shared function lifecycle explains when each phase runs.