Implementation:ArroyoSystems Arroyo Datastream Optimizers

Overview

Datastream Optimizers provides optimization passes over Arroyo's logical dataflow graph. The primary optimizer is the ChainingOptimizer, which merges forward-connected operators that share the same parallelism into chained execution units.

Description

The module defines:

• Optimizer trait: The base interface for dataflow graph optimizers:
  • optimize_once(graph: &mut DiGraph<LogicalNode, LogicalEdge>): Apply a single optimization pass, returning true if the graph was modified.
  • optimize(graph: &mut DiGraph<LogicalNode, LogicalEdge>): Apply the optimization repeatedly until no further changes are made (fixed-point iteration).

• ChainingOptimizer: Implements the Optimizer trait to chain operators. The optimization identifies pairs of nodes where:
  • The source node has exactly one outgoing edge to the target
  • The target node has exactly one incoming edge from the source
  • The edge type is Forward
  • Both nodes have the same parallelism
  • When found, the source node's operator chain is extended with the target node's operator chain, preserving the target node's edges to downstream nodes.

• remove_in_place: A helper function that removes a node from the graph, reconnecting all its incoming edges to all its outgoing edges. Used internally during graph optimization.

The chaining optimization reduces inter-operator communication overhead by executing compatible operators within the same thread and avoiding serialization between them.

Usage

The ChainingOptimizer is applied to the logical dataflow graph after the plan-to-graph conversion is complete, as a post-processing step before the graph is distributed to workers.

Code Reference

Source Location

crates/arroyo-datastream/src/optimizers.rs

Signature

pub trait Optimizer {
    fn optimize_once(graph: &mut DiGraph<LogicalNode, LogicalEdge>) -> bool;

    fn optimize(graph: &mut DiGraph<LogicalNode, LogicalEdge>) {
        while Self::optimize_once(graph) {}
    }
}

pub struct ChainingOptimizer;

impl Optimizer for ChainingOptimizer {
    fn optimize_once(graph: &mut DiGraph<LogicalNode, LogicalEdge>) -> bool
}

pub fn remove_in_place(graph: &mut DiGraph<LogicalNode, LogicalEdge>, index: NodeIndex)

Import

use arroyo_datastream::optimizers::{Optimizer, ChainingOptimizer, remove_in_place};

I/O Contract

Inputs

Name	Type	Description
graph	&mut DiGraph<LogicalNode, LogicalEdge>	Mutable reference to the dataflow graph to optimize

Outputs

Name	Type	Description
modified	bool	Whether the graph was changed (from optimize_once)
graph	DiGraph<LogicalNode, LogicalEdge>	The optimized graph (modified in-place)

Usage Examples

use arroyo_datastream::optimizers::{Optimizer, ChainingOptimizer};

// Apply chaining optimization to a dataflow graph
ChainingOptimizer::optimize(&mut graph);

// Or apply a single pass
let changed = ChainingOptimizer::optimize_once(&mut graph);

Related Pages

• ArroyoSystems_Arroyo_Datastream_Types - Core datastream types including LogicalNode and LogicalEdge
• ArroyoSystems_Arroyo_Plan_Graph_Builder - Produces the graph that this optimizer processes