Aggregate functions

On this page

Per-group accumulation — update, combine, finalize — via the AggregateFunction interface.

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

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

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

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

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

source

func (s *Int64Slice) At(i int) int64

At returns the i-th entry.

method Len

source

func (s *Int64Slice) Len() int

Len returns the number of entries.

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

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

source

func unpackWindowPartition(req AggregateWindowInitRequestWire) (*WindowPartition, error)

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

source

func (s *aggregateStorage) bucket(funcName string, execID []byte, shardKey string) *stateBucket

method ensureOpen

source

func (s *aggregateStorage) ensureOpen() (FunctionStorage, error)

ensureOpen resolves and caches the backend on first use.

function newAggregateStorage

source

func newAggregateStorage() *aggregateStorage

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

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

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

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

source

func (b *stateBucket) deleteWindowPartition(partitionID int64) error

method getConstArgs

source

func (b *stateBucket) getConstArgs() ([]byte, error)

method getWindowPartition

source

func (b *stateBucket) getWindowPartition(partitionID int64) ([]byte, error)

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

source

func (b *stateBucket) putConstArgs(args []byte) error

method putWindowPartition

source

func (b *stateBucket) putWindowPartition(partitionID int64, payload []byte) error

method saveStates

source

func (b *stateBucket) saveStates(states map[int64][]byte) error

saveStates writes (gid, bytes) pairs.

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

source

type streamingSessionStore struct {
mu    sync.Mutex
sessions  map[string]*streamingSession
}

Methods

method drop

source

func (s *streamingSessionStore) drop(execID []byte) *streamingSession

method get

source

func (s *streamingSessionStore) get(execID []byte) *streamingSession

method put

source

func (s *streamingSessionStore) put(execID []byte, sess *streamingSession)

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

source

func decodeWindowPartitionPayload(data []byte) (windowPartitionPayload, error)

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

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

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

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

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

source

func buildSingleScalarArray(mem memory.Allocator, dt arrow.DataType, v interface{}) (arrow.Array, error)

function encodeWindowPartitionPayload

source

func encodeWindowPartitionPayload(req AggregateWindowInitRequestWire, windowState []byte) []byte

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

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

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

source

func uniqueInt64(in []int64) []int64

function unpackAllValid

source

func unpackAllValid(data []byte, numCols int) []bool

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

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

source

func unpackFrameStats(data []byte) [2][2]int64

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