Aggregate functions

On this page

Per-group accumulation: update, combine, finalize.

trait AggregateFunction

source

pub trait AggregateFunction: Send + Sync {
  fn name(&self) -> &str;
  fn metadata(&self) -> FunctionMetadata;
  fn argument_specs(&self) -> Vec<ArgSpec>;
  fn on_bind(&self, params: &AggregateBindParams) -> Result<BindResponse>;
  fn initial_state(&self) -> Vec<u8>;
  fn update(
      &self,
      states: &mut HashMap<i64, Vec<u8>>,
      group_ids: &Int64Array,
      columns: &[ArrayRef],
  ) -> Result<()>;
  fn combine(&self, target: Vec<u8>, source: Vec<u8>) -> Result<Vec<u8>>;
  fn finalize(
      &self,
      output_schema: &SchemaRef,
      group_ids: &Int64Array,
      states: &[Option<Vec<u8>>],
  ) -> Result<RecordBatch>;
  fn window(
      &self,
      _partition: &RecordBatch,
      _output_schema: &SchemaRef,
      _frames: &[Vec<(i64, i64)>],
      _filter_mask: Option<&[bool]>,
  ) -> Result<arrow_array::ArrayRef> {
      Err(RpcError::runtime_error(
          "window() not supported by this aggregate",
      ))
  }

  fn streaming_chunk(
      &self,
      _chunk: &RecordBatch,
      _partition_key_count: usize,
      _order_key_count: usize,
      _states: &mut HashMap<Vec<u8>, Vec<u8>>,
  ) -> Result<ArrayRef> {
      Err(RpcError::runtime_error(
          "streaming_chunk() not supported by this aggregate",
      ))
  }

  fn finalize_with_args(
      &self,
      output_schema: &SchemaRef,
      group_ids: &Int64Array,
      states: &[Option<Vec<u8>>],
      _args: &crate::arguments::Arguments,
  ) -> Result<RecordBatch> {
      self.finalize(output_schema, group_ids, states)
  }
}

Description

An aggregate VGI function.

struct AggregateBindParams

source

pub struct AggregateBindParams {
  pub arguments: Arguments,
  pub input_schema: Option<SchemaRef>,
  pub settings: Settings,
  pub secrets: crate::secrets::Secrets,
}

Description

Parameters for aggregate_bind.

struct BoundAggregate

source

pub struct BoundAggregate {
  execution_id: Bytes,
  output_schema: SchemaRef,
  raw_output_schema: Bytes,
  function_name: String,
  schema_name: Option<String>,
}

Description

A bound aggregate execution.

Methods

method execution_id

source

pub fn execution_id(&self) -> &Bytes

The worker-minted id for this aggregation.

Two executions of the same function have different ids; passing one execution’s id to [VgiClient::aggregate_combine] is how parallel partial aggregates are merged.

method output_schema

source

pub fn output_schema(&self) -> &SchemaRef

The result schema, resolved at bind.

constant GROUP_COLUMN_NAME

source

pub const GROUP_COLUMN_NAME: &str = “__vgi_group_id”;

Description

The reserved group-id column prepended to UPDATE input batches.

constant GROUP_COLUMN_NAME

source

pub const GROUP_COLUMN_NAME: &str = “__vgi_group_id”;

Description

The group-id column every update batch must carry.

Must match vgi::aggregate::GROUP_COLUMN_NAME on the worker side.

struct StreamingAggregate

source

pub struct StreamingAggregate {
  execution_id: Bytes,
  function_name: String,
  schema_name: Option<String>,
}

Description

An open streaming-aggregate session.

The streaming protocol skips client-side partition materialisation entirely: input chunks go straight to the worker, which answers each with a same-length output batch.

Methods

method execution_id

source

pub fn execution_id(&self) -> &Bytes

The worker’s session token.

struct WindowPartition

source

pub struct WindowPartition {
  execution_id: Bytes,
  partition_id: i64,
  function_name: String,
  schema_name: Option<String>,
}

Description

A window partition cached on the worker.

Window evaluation is two-phase: the whole partition is shipped once, then each output row’s frames are evaluated against it. Dropping this releases nothing by itself — call [VgiClient::window_destroy], which the worker needs to free the cached partition.

Methods

method partition_id

source

pub fn partition_id(&self) -> i64

Which partition this is, within its execution.

function with_group_ids

source

pub fn with_group_ids(group_ids: &[i64], values: &RecordBatch) -> Result<RecordBatch>

Description

Prepend the group-id column to a batch of value columns.

The worker splits the batch back apart on the column name, so a caller that builds the batch by hand must use [GROUP_COLUMN_NAME].