Implementation:ArroyoSystems Arroyo Datastream Types

Overview

Datastream Types defines the core data types for Arroyo's streaming datastream representation, including window types, WASM behavior definitions, aggregator configurations, and the default sink factory.

Description

The module defines:

• WindowType: The central windowing enum used throughout the planner and runtime:
  • Tumbling { width: Duration }: Fixed-size non-overlapping windows
  • Sliding { width: Duration, slide: Duration }: Overlapping windows with configurable slide
  • Instant: Per-event windows (no accumulation)
  • Session { gap: Duration }: Activity-based windows that close after a gap
  • Derives: Clone, Encode, Decode, Serialize, Deserialize, PartialEq, Eq, Hash, PartialOrd

• WASM types:
  • WasmBehavior: Enum (Map, OptMap, Filter, Timestamp, KeyBy) defining WASM function roles
  • WasmDef: WASM function definition with name, key/value args and types, return type, and body
  • WasmUDF: Combines WasmBehavior with WasmDef

• Aggregator types:
  • SlidingWindowAggregator: Configuration for sliding window aggregation with bin merging, in-memory add/remove functions
  • TumblingWindowAggregator: Configuration for tumbling window aggregation with bin merging
  • TumblingTopN: Top-N computation within tumbling windows
  • SlidingAggregatingTopN: Top-N with sliding window aggregation
  • NonWindowAggregator: Non-windowed aggregation with expiration TTL

• Other types:
  • OffsetMode: Earliest or Latest consumer offset for sources
  • ImpulseSpec: Delay-based or events-per-second impulse source configuration
  • WindowAgg: Window aggregate operation (Count, Max, Min, Sum, Expression)

• default_sink: Factory function returning the default sink connector (Preview or Stdout based on configuration).

• duration_to_syn_expr: Utility to convert a Duration to a syn::Expr for code generation.

Usage

These types are used throughout the Arroyo planner, runtime, and codegen layers. WindowType is particularly central, appearing in aggregate extensions, join validation, and plan rewriting.

Code Reference

Source Location

crates/arroyo-datastream/src/lib.rs

Signature

#[derive(Clone, Encode, Decode, Serialize, Deserialize, PartialEq, Eq, Hash, PartialOrd)]
pub enum WindowType {
    Tumbling { width: Duration },
    Sliding { width: Duration, slide: Duration },
    Instant,
    Session { gap: Duration },
}

#[derive(Copy, Clone, Debug, Eq, PartialEq, Encode, Decode, Serialize, Deserialize)]
pub enum OffsetMode {
    Earliest,
    Latest,
}

#[derive(Copy, Clone, Debug, Encode, Decode, Serialize, Deserialize, PartialEq)]
pub enum ImpulseSpec {
    Delay(Duration),
    EventsPerSecond(f32),
}

pub fn default_sink() -> api::ConnectorOp

pub fn duration_to_syn_expr(duration: Duration) -> syn::Expr

Import

use arroyo_datastream::{WindowType, OffsetMode, ImpulseSpec, default_sink};

I/O Contract

Inputs

Name	Type	Description
width	Duration	Window width for tumbling or sliding windows
slide	Duration	Slide interval for sliding windows
gap	Duration	Inactivity gap for session windows

Outputs

Name	Type	Description
WindowType	enum	Configured window type for use in planner and runtime
ConnectorOp	struct	Default sink connector operation (preview or stdout)

Usage Examples

use arroyo_datastream::WindowType;
use std::time::Duration;

// Creating window types
let tumbling = WindowType::Tumbling { width: Duration::from_secs(60) };
let sliding = WindowType::Sliding {
    width: Duration::from_secs(300),
    slide: Duration::from_secs(60),
};
let session = WindowType::Session { gap: Duration::from_secs(30) };
let instant = WindowType::Instant;

// Debug formatting
println!("{:?}", tumbling);  // TumblingWindow(1m)
println!("{:?}", sliding);   // SlidingWindow(size: 5m, slide: 1m)

Related Pages

• ArroyoSystems_Arroyo_Aggregate_Extension - Uses WindowType for aggregate window configuration
• ArroyoSystems_Arroyo_Join_Planner - Uses WindowType for join window validation
• ArroyoSystems_Arroyo_Datastream_Optimizers - Optimizer passes on the datastream graph