Skip to content
Query.Farm
Talk with Us

Aggregate functions

On this page

Per-group accumulation: update, combine, finalize.

struct AggregateBindParams

Section titled “struct AggregateBindParams”
source
type AggregateBindParams struct {
Args    *Arguments
InputSchema  *arrow.Schema
Settings  map[string]interface{}
Secrets    map[string]map[string]interface{}
Auth    *vgirpc.AuthContext
}

Description

AggregateBindParams holds the parameters passed to AggregateFunction.OnBind.

interface AggregateFunction

Section titled “interface AggregateFunction”
source
type AggregateFunction interface {
// Name is the SQL-visible function name.
Name() string
// Metadata describes function attributes (stability, null-handling, etc.).
Metadata() FunctionMetadata
// ArgumentSpecs declares the input columns and any const params.
ArgumentSpecs() []ArgSpec
// OnBind resolves the output schema. Must return a single-column schema.
OnBind(params *AggregateBindParams) (*BindResponse, error)
// NewState returns a fresh state pointer for a new group. The pointer
// type is what the dispatcher uses for gob register/serialize.
NewState(params *AggregateProcessParams) interface{}
// Update accumulates rows into per-group state. The states map is
// pre-populated with existing states, but groups that aren't yet in
// the map MUST be created on demand by the function — call NewState
// only when the row genuinely contributes (e.g. non-null value with
// NullHandlingDefault). Groups never inserted into states stay absent
// from storage so finalize() returns NULL.
Update(states map[int64]interface{}, groupIDs *Int64Slice, columns []arrow.Array, params *AggregateProcessParams) error
// Combine merges source into target, returning the new target state.
Combine(source, target interface{}, params *AggregateProcessParams) (interface{}, error)
// Finalize produces one result row per group_id. groupIDs entries map
// to states[gid] which may be nil if the group was never updated
// (NULL-only inputs with NullHandlingDefault).
Finalize(groupIDs []int64, states map[int64]interface{}, params *AggregateProcessParams) (arrow.RecordBatch, error)
}

Description

AggregateFunction is the interface every aggregate implements.

State is held on the worker side keyed by group_id; the Go SDK gob-serializes it across RPCs for cross-call continuity. Implementations register a concrete state type via NewState (returning a fresh zero value) so the dispatcher can reflect on it.

struct AggregateProcessParams

Section titled “struct AggregateProcessParams”
source
type AggregateProcessParams struct {
Args    *Arguments
OutputSchema  *arrow.Schema
Settings  map[string]interface{}
Secrets    map[string]map[string]interface{}
Auth    *vgirpc.AuthContext
// AttachOpaqueData is the catalog the function was invoked under (nil for ad-hoc calls).
AttachOpaqueData  []byte
}

Description

AggregateProcessParams is shared by Update/Combine/Finalize/Window callbacks.

interface AggregateWindowFunction

Section titled “interface AggregateWindowFunction”
source
type AggregateWindowFunction interface {
AggregateFunction
// WindowInit can derive optional per-partition state. Returns nil if
// no derived state is needed; otherwise the value is gob-serialized
// and passed back to Window.
WindowInit(partition *WindowPartition, params *AggregateProcessParams) (interface{}, error)
// Window computes the aggregate value for one output row.
// Subframes are usually a single (begin,end) pair; up to 3 for
// EXCLUDE TIES/GROUP. Returns a Go scalar matching the output schema.
Window(rid int64, subframes [][2]int64, partition *WindowPartition, windowState interface{}, params *AggregateProcessParams) (interface{}, error)
}

Description

AggregateWindowFunction is implemented by aggregates that also support SQL OVER windowing. Functions opt in by also setting Metadata.SupportsWindow.

struct Int64Slice

Section titled “struct Int64Slice”
source
type Int64Slice struct {
Data []int64
}

Description

Int64Slice wraps an int64 slice to ease passing to Update without importing arrow types in user code. The underlying array is borrowed.

Methods

method At

Section titled “method At”
source
func (s *Int64Slice) At(i int) int64

At returns the i-th entry.

method Len

Section titled “method Len”
source
func (s *Int64Slice) Len() int

Len returns the number of entries.

interface StreamingAggregateFunction

Section titled “interface StreamingAggregateFunction”
source
type StreamingAggregateFunction interface {
AggregateFunction
// StreamingOpen prepares cross-partition state for the session. The
// returned value is opaque to the framework — it's threaded back into
// StreamingChunk and StreamingClose.
StreamingOpen(params *AggregateProcessParams) (interface{}, error)
// StreamingChunk processes one input chunk. Implementations must return
// an arrow.Array of the same length as `chunk.NumRows()`. The first
// `partitionKeyCount` columns of `chunk` are the partition keys; the
// next `orderKeyCount` are order keys; remaining columns are the
// function's value arguments.
StreamingChunk(state interface{}, chunk arrow.RecordBatch, partitionKeyCount, orderKeyCount int, params *AggregateProcessParams) (arrow.Array, error)
// StreamingClose drops the session. Always called once per execution_id.
StreamingClose(state interface{}, params *AggregateProcessParams) error
}

Description

StreamingAggregateFunction is the optional interface an aggregate may implement to participate in DuckDB’s streaming-window optimizer rule. When eligible (cumulative frame, no EXCLUDE/DISTINCT/FILTER), DuckDB pipes input chunks to the worker and expects an output array of the same length as the input.

Functions that don’t implement this still work as windowed aggregates via the standard aggregate_window callbacks; the streaming path is purely an optimization.

struct WindowPartition

Section titled “struct WindowPartition”
source
type WindowPartition struct {
// Inputs is the partition's input columns (excludes reserved group_id).
Inputs  arrow.RecordBatch
// RowCount is the number of rows in the partition.
RowCount  int64
// FilterMask is a per-row boolean from FILTER (WHERE ...); nil when absent.
FilterMask  []bool
// FrameStats is the optimizer's per-partition frame statistics:
// ((begin_delta, end_delta), (begin_delta, end_delta)).
FrameStats  [2][2]int64
// AllValid[i] is true if input column i has no nulls.
AllValid  []bool
// OutputSchema is the function's resolved output schema.
OutputSchema  *arrow.Schema
}

Description

WindowPartition is the partition data available during windowed evaluation.

Methods

function unpackWindowPartition

Section titled “function unpackWindowPartition”
source
func unpackWindowPartition(req AggregateWindowInitRequestWire) (*WindowPartition, error)

struct aggregateStorage

Section titled “struct aggregateStorage”
source
type aggregateStorage struct {
mu  sync.Mutex
back  FunctionStorage
resolve  func() (FunctionStorage, error)  // set by the Worker
}

Description

aggregateStorage is a thin shim that lazily resolves a FunctionStorage backend from a setter and exposes the bucket(funcName, execID) pattern the protocol layer uses.

Methods

method bucket

Section titled “method bucket”
source
func (s *aggregateStorage) bucket(funcName string, execID []byte, shardKey string) *stateBucket

method ensureOpen

Section titled “method ensureOpen”
source
func (s *aggregateStorage) ensureOpen() (FunctionStorage, error)

ensureOpen resolves and caches the backend on first use.

function newAggregateStorage

Section titled “function newAggregateStorage”
source
func newAggregateStorage() *aggregateStorage

method setResolver

Section titled “method setResolver”
source
func (s *aggregateStorage) setResolver(r func() (FunctionStorage, error))

setResolver wires a lazy backend resolver. Called once by the Worker so the FunctionStorage is shared with ExecutionStorage and any future backends (Cloudflare DO, etc.).

struct stateBucket

Section titled “struct stateBucket”
source
type stateBucket struct {
storage    *aggregateStorage
functionName  string
executionID  []byte
// shardKey routes per logical ATTACH (att-<hex uuid>) for the CfDo backend;
// "" for non-sharding backends. Derived by the handler from the request's
// unwrapped attach UUID.
shardKey  string
}

Description

stateBucket binds aggregate operations to one (function_name, execution_id). Mirrors the original surface so aggregate_protocol.go and aggregate_helpers.go don’t have to change.

Methods

method backend

Section titled “method backend”
source
func (b *stateBucket) backend() (FunctionStorage, error)

backend resolves the shared backend and, for a remote-sharding backend (CfDo), pins it to this bucket’s shard key — so aggregate state routes to the same Durable Object as the execution’s other storage.

method clear

Section titled “method clear”
source
func (b *stateBucket) clear() error

clear drops aggregate state and window partition rows for this execution_id. Const args are intentionally left behind: they’re small, keyed by (execution_id, function_name), and reaped by the FunctionStorage’s TTL sweep. Matches vgi-python which also has no per-call const-args clear.

method deleteWindowPartition

Section titled “method deleteWindowPartition”
source
func (b *stateBucket) deleteWindowPartition(partitionID int64) error

method getConstArgs

Section titled “method getConstArgs”
source
func (b *stateBucket) getConstArgs() ([]byte, error)

method getWindowPartition

Section titled “method getWindowPartition”
source
func (b *stateBucket) getWindowPartition(partitionID int64) ([]byte, error)

method loadStates

Section titled “method loadStates”
source
func (b *stateBucket) loadStates(gids []int64) (map[int64][]byte, error)

loadStates fetches all states for the given group_ids. Returns a map keyed by group_id; gids absent from the result are not yet stored. Matches the pre-shared-backend surface so callers don’t change.

method putConstArgs

Section titled “method putConstArgs”
source
func (b *stateBucket) putConstArgs(args []byte) error

method putWindowPartition

Section titled “method putWindowPartition”
source
func (b *stateBucket) putWindowPartition(partitionID int64, payload []byte) error

method saveStates

Section titled “method saveStates”
source
func (b *stateBucket) saveStates(states map[int64][]byte) error

saveStates writes (gid, bytes) pairs.

struct streamingSession

Section titled “struct streamingSession”
source
type streamingSession struct {
fn      StreamingAggregateFunction
state      interface{}
outputSchema    *arrow.Schema
partitionKeyCount  int
orderKeyCount    int
args      *Arguments
settings    map[string]interface{}
secrets      map[string]map[string]interface{}
attachOpaqueData  []byte
}

Description

streamingSession holds per-execution_id state for a streaming-aggregate invocation. The session is removed in handleAggregateStreamingClose.

struct streamingSessionStore

Section titled “struct streamingSessionStore”
source
type streamingSessionStore struct {
mu    sync.Mutex
sessions  map[string]*streamingSession
}

Methods

method drop

Section titled “method drop”
source
func (s *streamingSessionStore) drop(execID []byte) *streamingSession

method get

Section titled “method get”
source
func (s *streamingSessionStore) get(execID []byte) *streamingSession

method put

Section titled “method put”
source
func (s *streamingSessionStore) put(execID []byte, sess *streamingSession)

struct windowPartitionPayload

Section titled “struct windowPartitionPayload”
source
type windowPartitionPayload struct {
PartitionBatch  []byte
OutputSchema  []byte
FilterMask  []byte
FrameStats  []byte
AllValid  []byte
RowCount  int64
WindowState  []byte  // optional gob-encoded WindowInit return value
}

Methods

function decodeWindowPartitionPayload

Section titled “function decodeWindowPartitionPayload”
source
func decodeWindowPartitionPayload(data []byte) (windowPartitionPayload, error)

function EnsureState

Section titled “function EnsureState”
source
func EnsureState[T any](states map[int64]interface{}, gid int64, newFn func() *T) *T

EnsureState is a helper for Update implementations: returns the existing state for gid, or creates one via newFn() and registers it. Functions should call this only when the row genuinely contributes (e.g. non-null with NullHandlingDefault) — groups never inserted stay absent from storage so finalize() returns NULL.

function PartitionKey

Section titled “function PartitionKey”
source
func PartitionKey(chunk arrow.RecordBatch, partitionKeyCount, i int) uint64

PartitionKey is a helper to derive a stable hash key from the partition-key columns at row i. Use it from StreamingChunk implementations to look up per-partition state.

function buildBatchResult

Section titled “function buildBatchResult”
source
func buildBatchResult(values []interface{}, outputSchema *arrow.Schema) (arrow.RecordBatch, error)

buildBatchResult builds a count-row RecordBatch from a slice of scalar values.

function buildScalarColumn

Section titled “function buildScalarColumn”
source
func buildScalarColumn(mem memory.Allocator, dt arrow.DataType, values []interface{}) (arrow.Array, error)

buildScalarColumn appends each value (or null) into a column of the given type.

function buildScalarResultBatch

Section titled “function buildScalarResultBatch”
source
func buildScalarResultBatch(value interface{}, outputSchema *arrow.Schema) (arrow.RecordBatch, error)

buildScalarResultBatch wraps a single scalar value in a one-row RecordBatch matching outputSchema (which must have exactly one field).

function buildSingleScalarArray

Section titled “function buildSingleScalarArray”
source
func buildSingleScalarArray(mem memory.Allocator, dt arrow.DataType, v interface{}) (arrow.Array, error)

function encodeWindowPartitionPayload

Section titled “function encodeWindowPartitionPayload”
source
func encodeWindowPartitionPayload(req AggregateWindowInitRequestWire, windowState []byte) []byte

function gobDecodeState

Section titled “function gobDecodeState”
source
func gobDecodeState(data []byte) (interface{}, error)

gobDecodeState deserializes bytes into a state value via gob’s interface machinery. The concrete type must have been previously registered with gob.Register.

function gobEncodeState

Section titled “function gobEncodeState”
source
func gobEncodeState(state interface{}) ([]byte, error)

gobEncodeState serializes a state value (passed by interface, typically a pointer to a registered struct) to bytes.

function splitGroupIDColumn

Section titled “function splitGroupIDColumn”
source
func splitGroupIDColumn(batch arrow.RecordBatch) (int, []int64, []arrow.Array, error)

splitGroupIDColumn returns the group_id column index, the int64 group_ids, and the remaining input columns (excluding group_id).

function uniqueInt64

Section titled “function uniqueInt64”
source
func uniqueInt64(in []int64) []int64

function unpackAllValid

Section titled “function unpackAllValid”
source
func unpackAllValid(data []byte, numCols int) []bool

unpackAllValid reads numCols bytes — one per input column — into a []bool.

function unpackBoolMask

Section titled “function unpackBoolMask”
source
func unpackBoolMask(packed []byte, rowCount int64) []bool

unpackBoolMask converts the C++ extension’s packed-bit boolean encoding into a flat []bool of length rowCount.

function unpackFrameStats

Section titled “function unpackFrameStats”
source
func unpackFrameStats(data []byte) [2][2]int64

unpackFrameStats reads ((begin_delta, end_delta), (begin_delta, end_delta)) from 4×int64 little-endian bytes.