Implementation:ArroyoSystems Arroyo Udaf Wrapper

Knowledge Sources	ArroyoSystems_Arroyo
Domains	Streaming, UDF, Aggregation
Last Updated	2026-02-08 08:00 GMT

Overview

Provides the ArroyoUdaf accumulator wrapper that adapts Arroyo's vector-argument scalar UDFs into DataFusion's UDAF (User-Defined Aggregate Function) framework by accumulating batches of array data before passing them to the underlying UDF.

Description

This module bridges Arroyo's UDF system with DataFusion's aggregate function infrastructure:

• UdafArg -- a per-argument accumulator that collects ArrayRef values across multiple batches. Provides concatenate_array to merge all accumulated arrays into a single ListArray for UDF invocation.
• ArroyoUdaf -- implements DataFusion's Accumulator trait. It stores multiple UdafArgs (one per function parameter), the output DataType, and a reference to the underlying SyncUdfDylib function.
• Accumulator methods:
  • update_batch -- appends each input array to its corresponding UdafArg.
  • evaluate -- concatenates all accumulated arrays per argument, then calls udf.invoke_udaf with the combined arrays. Returns scalar_none if no data was accumulated.
  • state -- serializes accumulated state as a vector of ListArray ScalarValues for distributed aggregation.
  • merge_batch -- merges partial state from other accumulators by extracting arrays from incoming ListArrays.
  • size -- estimates memory usage from all accumulated arrays.
• scalar_none -- a comprehensive helper function that maps any Arrow DataType to its corresponding null ScalarValue, covering all numeric types, timestamps, intervals, durations, strings, binary, lists, and decimals.
• list_from_arr -- helper to wrap an ArrayRef in a single-element ListArray.

Usage

ArroyoUdaf is created internally by the operator framework when a user-defined aggregate function is registered. It is not typically used directly by operator implementors.

Code Reference

Source Location

• Repository: ArroyoSystems_Arroyo
• File: crates/arroyo-operator/src/udfs.rs

Signature

#[derive(Debug, Clone)]
pub struct UdafArg {
    values: Vec<ArrayRef>,
    inner: FieldRef,
}

impl UdafArg {
    pub fn new(inner: FieldRef) -> Self;
}

#[derive(Debug)]
pub struct ArroyoUdaf {
    args: Vec<UdafArg>,
    output_type: Arc<DataType>,
    udf: Arc<SyncUdfDylib>,
}

impl ArroyoUdaf {
    pub fn new(args: Vec<UdafArg>, output_type: Arc<DataType>, udf: Arc<SyncUdfDylib>) -> Self;
}

impl Accumulator for ArroyoUdaf {
    fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()>;
    fn evaluate(&self) -> Result<ScalarValue>;
    fn size(&self) -> usize;
    fn state(&self) -> Result<Vec<ScalarValue>>;
    fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()>;
}

Import

use arroyo_operator::udfs::{ArroyoUdaf, UdafArg};

I/O Contract

Inputs

Name	Type	Required	Description
values	&[ArrayRef]	Yes	Input column arrays for each UDAF argument per batch
args	Vec<UdafArg>	Yes	Argument accumulators (one per function parameter)
udf	Arc<SyncUdfDylib>	Yes	The underlying vector-taking scalar UDF

Outputs

Name	Type	Description
ScalarValue	ScalarValue	The aggregated result from the UDF
Vec<ScalarValue>	Vec<ScalarValue>	Intermediate state as ListArray scalars for distributed merge

Usage Examples

use arroyo_operator::udfs::{ArroyoUdaf, UdafArg};
use arrow::datatypes::{DataType, Field};
use std::sync::Arc;

// Create a UDAF wrapping a vector-argument UDF
let arg = UdafArg::new(Arc::new(Field::new("values", DataType::Float64, false)));
let udaf = ArroyoUdaf::new(
    vec![arg],
    Arc::new(DataType::Float64),
    my_udf_dylib.clone(),
);

// Use as a DataFusion Accumulator
udaf.update_batch(&[array_ref])?;
let result = udaf.evaluate()?;

Related Pages

• Environment:ArroyoSystems_Arroyo_Rust_Runtime