Aggregate functions

The deep reference for the aggregate shape — the exact contracts of initial_state, update, combine, and finalize, plus dynamic output types, multi-argument aggregates, and the streaming-partitioned opt-in for unbounded windowed queries.

New to aggregates?

Start with the runnable example in Function patterns → Aggregate; this page is the deep reference. For how the phases fit into a query’s overall execution, see Function lifecycle. The full method signatures live in the aggregate_function API reference.

Aggregate functions accumulate input rows into per-group state and produce one result per group, powering SQL like SELECT my_agg(col) FROM t GROUP BY category. Each DuckDB callback (bind, update, combine, finalize, destructor) maps to one unary RPC. Per-group state lives in FunctionStorage (SQLite-backed), keyed by a globally unique group_id assigned by C++ from a shared atomic counter, so IDs never collide across parallel threads. Because state is serialized to bytes rather than held in memory, the design is HTTP-transport compatible.

State class

The state is the function’s first type parameter — AggregateFunction[SumState] — and is serialized to bytes between every RPC call. The framework asks it for exactly two things, serialize_to_bytes() and deserialize_from_bytes() (the StreamStateCodec protocol), and treats the result as opaque.

The usual answer is a dataclass extending ArrowSerializableDataclass, which writes both methods for you. Every field needs an ArrowType annotation so the wire type is explicit:

@dataclass(kw_only=True)
class AvgState(ArrowSerializableDataclass):
  total: Annotated[float, ArrowType(pa.float64())] = 0.0
  count: Annotated[int, ArrowType(pa.int64())] = 0

Use simple scalar fields (int, float, str, bytes) for efficient serialization. Provide defaults that represent the identity element of the aggregation (0 for sum, empty for concatenation) so initial_state() can return a bare instance.

Packing the state yourself

Arrow is not required, and for a small state it is the expensive choice. An aggregate serializes once per group, per batch, and a one-row Arrow stream pays for a schema message, a batch message and an end-of-stream marker whatever the payload — so the framing cost scales with cardinality. Two counters pack into 16 bytes:

@dataclass(kw_only=True)
class SumState:
  # Same two counters, packed as little-endian int64s.
  total: int = 0
  count: int = 0

  _STRUCT: ClassVar[struct.Struct] = struct.Struct("<qq")

  def serialize_to_bytes(self) -> bytes:
      return self._STRUCT.pack(self.total, self.count)

  @classmethod
  def deserialize_from_bytes(cls, data: bytes) -> "SumState":
      total, count = cls._STRUCT.unpack(data)
      return cls(total=total, count=count)

This also lets a Python worker match the state encoding a sibling VGI implementation in another language already uses.

The codec must round-trip exactly

T.deserialize_from_bytes(s.serialize_to_bytes()) has to equal s for every state the aggregate can produce, including initial_state(). The framework cannot check that, and a lossy codec surfaces as wrong aggregate results rather than as an error. A TState with neither method is rejected at class-definition time, so the mistake that is catchable fails where you declared it rather than deep in the persistence path.

Method contracts

The four required classmethods form the accumulation pipeline. (finalize and update are declared with *args/**kwargs on the base class — you give them concrete, Param-annotated signatures in your subclass.)

`initial_state(params) -> TState`

Returns the identity element for a new group. Called lazily — the first time a group_id is encountered during update(), not during DuckDB’s C++ initialize(). The framework pre-populates the states dict for you, so you rarely call this directly.

`update(states, group_ids, ...columns) -> None`

Accumulates a batch of input rows into per-group state. Mutate states in place; the framework persists every modified state to FunctionStorage after the call returns.

Parameter	Type	Meaning
`states`	`dict[int, TState]`	Pre-populated with `initial_state()` for every new group_id in this batch.
`group_ids`	`pa.Int64Array`	Parallel to each column array — identifies the group each row belongs to.
input columns	`pa.Array`	Declared via `Param` annotations (see Input parameters).

A vectorized update groups within the batch first, then folds each group’s partial into the stored state:

@classmethod
def update(cls, states, group_ids, value: Annotated[pa.Int64Array, Param(doc="Column to sum")]) -> None:
  table = pa.table({"gid": group_ids, "value": value})
  grouped = table.group_by("gid").aggregate([("value", "sum")])
  for i in range(grouped.num_rows):
      gid = grouped.column("gid")[i].as_py()
      val = grouped.column("value_sum")[i].as_py()
      if val is not None:
          states[gid] = SumState(total=states[gid].total + val)

`combine(source, target, params) -> TState`

Merges two partial states produced by parallel workers, called during DuckDB’s hash-aggregate combine phase. source is removed after the call; the returned state replaces target. combine must be associative and commutative — DuckDB calls it in an unspecified order across threads.

@classmethod
def combine(cls, source: SumState, target: SumState, params) -> SumState:
  return SumState(total=source.total + target.total)

For multi-field state, combine each field with its own algebra — sum the sums, sum the counts, concatenate the lists.

`finalize(group_ids, states, params) -> RecordBatch`

Produces the final result. Must return a RecordBatch with one row per group_id, in the same order as the group_ids argument. Annotate the return with Returns(arrow_type) to declare the static output type:

@classmethod
def finalize(
  cls, group_ids: pa.Int64Array, states: dict[int, SumState], params
) -> Annotated[pa.RecordBatch, Returns(pa.float64())]:
  results = [states[gid.as_py()].total for gid in group_ids]
  return pa.record_batch({"result": pa.array(results, type=pa.float64())})

`on_bind(params) -> BindResponse` (optional)

Override for dynamic output types or bind-time validation. params is an AggregateBindParams carrying args, input_schema, settings, secrets, and auth_context. The default implementation derives the output schema from the Returns(...) annotation on finalize(); you only need on_bind() when the type isn’t statically known (see Dynamic output type) or when you want to validate arguments or resolve secrets at plan time.

Meta options

class Meta:
  name = "vgi_my_agg"                                          # SQL function name
  description = "Description for catalog"                      # optional
  null_handling = NullHandling.DEFAULT                        # DEFAULT or SPECIAL
  order_dependent = OrderDependence.ORDER_DEPENDENT           # for order-sensitive aggs
  distinct_dependent = DistinctDependence.DISTINCT_DEPENDENT  # DISTINCT changes the result

Setting	Effect
`NullHandling.DEFAULT`	NULL inputs are skipped — never passed to `update()`.
`NullHandling.SPECIAL`	NULL inputs are passed through (needed for `COUNT(*)`).
`OrderDependence.ORDER_DEPENDENT`	Result depends on input order (e.g. `LISTAGG`).
`DistinctDependence.DISTINCT_DEPENDENT`	A `DISTINCT` modifier changes the result.

Input parameters

Input columns are declared on update() with Param annotations — the same pattern as ScalarFunction.compute(). Multiple Params give you a multi-argument (e.g. weighted) aggregate; each is delivered as a column array parallel to group_ids:

@classmethod
def update(
  cls,
  states: dict[int, MyState],
  group_ids: pa.Int64Array,
  value: Annotated[pa.DoubleArray, Param(doc="Values")],
  weight: Annotated[pa.DoubleArray, Param(doc="Weights")],
) -> None:
  ...

Constant parameters (ConstParam)

For parameters that are constant across all rows (e.g. a percentile threshold), use ConstParam. These are constant-folded at bind time and stored in FunctionStorage:

percentile: Annotated[float, ConstParam("Percentile (0-1)", phase="finalize")] = 0.5

The phase controls when the constant is injected:

Phase	Injected in `update()`	Injected in `finalize()`
`“all”` (default)	Yes	Yes
`“update”`	Yes	No
`“finalize”`	No	Yes

Use phase="finalize" to avoid serializing a large constant on every update batch — it’s only loaded when finalize() needs it. In finalize(), read constant values from params.args.positional:

pct = params.args.positional[0].as_py() if params.args and params.args.positional else 0.5

Varargs

For aggregates accepting a variable number of columns, use Param(varargs=True); the parameter receives a list of arrays. SQL: SELECT vgi_sum_all(a, b, c) FROM t GROUP BY category.

@classmethod
def update(
  cls,
  states: dict[int, MyState],
  group_ids: pa.Int64Array,
  columns: Annotated[pa.Array, Param(doc="Columns to sum", varargs=True)],
) -> None:
  for i in range(len(group_ids)):
      gid = group_ids[i].as_py()
      for col in columns:
          val = col[i].as_py()
          if val is not None:
              states[gid].total += float(val)

Dynamic output type (ANY)

When the output type depends on the input, annotate finalize() with Returns() (no arrow type) and override on_bind() to derive the schema from params.input_schema. The catalog reports this column as a dynamic any type until bind resolves it:

class GenericSum(AggregateFunction[GenericSumState]):
  @classmethod
  def on_bind(cls, params, **kwargs):
      if params.input_schema is not None:
          input_type = params.input_schema.field(0).type
          return BindResponse(output_schema=pa.schema([("result", input_type)]))
      return BindResponse(output_schema=pa.schema([("result", pa.float64())]))

  @classmethod
  def finalize(cls, group_ids, states, params) -> Annotated[pa.RecordBatch, Returns()]:
      output_type = params.output_schema.field(0).type if params.output_schema else pa.float64()
      results = [states[gid.as_py()].total for gid in group_ids]
      return pa.record_batch({"result": pa.array(results, type=output_type)})

Streaming-partitioned aggregates (opt-in)

When to reach for this

For most analytics, pre-aggregating the input in plain SQL is the cleaner answer than a custom streaming path — collapse fills per period in a CTE, then run a normal windowed aggregate over the pre-aggregate. The streaming_partitioned opt-in is for the cases where pre-aggregation isn’t viable: per-fill running views, very high per-partition cardinality, or aggregates whose state isn’t algebraically reducible.

For OVER (PARTITION BY ... ORDER BY ...) queries against unbounded inputs (e.g. running aggregates across years of trade history), the standard windowed path materializes each partition in DuckDB memory before the aggregate sees it — fine for bounded data, but it OOMs at scale. Setting streaming_partitioned = True on the function’s Meta routes eligible queries through a custom physical operator: input chunks pipe directly to the worker, which maintains concurrent per-partition state in a hash map and emits one cumulative-snapshot output per input row. Memory is bounded by partitions × state_per_partition, not by row count.

class MyRunningAgg(AggregateFunction[MyState]):
  class Meta:
      name = "my_running_agg"
      streaming_partitioned = True   # opt-in

  @classmethod
  def streaming_open(cls, params: ProcessParams[None]) -> dict[str, Any]:
      # Cross-partition session state; lives in an in-process cache keyed
      # by execution_id for the session's duration.
      return {"partition_states": {}}

  @classmethod
  def streaming_chunk(
      cls,
      chunk: pa.RecordBatch,
      streaming_state: dict[str, Any],
      partition_key_count: int,
      order_key_count: int,
      params: ProcessParams[None],
  ) -> pa.Array:
      # Column layout: [partition_key_cols..., order_key_cols..., value_cols...]
      # Return one output value per input row (cumulative snapshot at that row).
      ...

  @classmethod
  def streaming_close(cls, streaming_state, params) -> None:
      # Cleanup hook, called once per session. Default: no-op.
      ...

The extension’s optimizer rule decides eligibility and requires: streaming_partitioned = True; a cumulative frame (UNBOUNDED PRECEDING AND CURRENT ROW, or the implicit cumulative frame from ORDER BY alone); no EXCLUDE, DISTINCT, FILTER (WHERE ...), or aggregate-arg ORDER BY; and no const-arg parameters (v1 limitation). Queries that don’t satisfy all of these fall back to the standard windowed path automatically. The streaming path is additive — it does not replace update/combine/finalize, which still service GROUP BY queries normally.

Registration

Register aggregates alongside scalar and table functions; the framework detects AggregateFunction subclasses and registers them with the correct catalog function type:

worker = Worker(functions=[SumFunction, AvgFunction])

Example functions

See vgi/examples/aggregate.py for complete implementations:

Function	Demonstrates
`CountFunction`	Nullary aggregate (no inputs), `NullHandling.SPECIAL`
`SumFunction`	Single input, basic grouping
`AvgFunction`	Multi-field state (sum + count)
`WeightedSumFunction`	Multiple input columns
`ListAggFunction`	Order-dependent aggregate
`PercentileFunction`	`ConstParam` with `phase=“finalize”`
`GenericSumFunction`	ANY type, dynamic output via `on_bind()`
`SumAllFunction`	Varargs aggregate

Next steps

A runnable aggregate → Function patterns → Aggregate.
How the phases are driven → Function lifecycle.
Where combine state lives → Persist state across workers.
Exact signatures → API Reference: aggregate_function.