Implementation:ArroyoSystems Arroyo Plan Graph Builder

Overview

Plan Graph Builder translates DataFusion logical plans into Arroyo's physical dataflow graph representation. It implements a TreeNodeVisitor that traverses the rewritten logical plan and constructs a DiGraph of LogicalNode and LogicalEdge elements.

Description

The module defines:

• PlanToGraphVisitor: A tree node visitor that maintains a directed graph (DiGraph<LogicalNode, LogicalEdge>), a mapping of output schemas per node, a memoization map of named nodes, and a traversal stack for tracking parent-child relationships during the visit.

• Planner: An internal helper that wraps DataFusion's DefaultPhysicalPlanner with a custom ArroyoExtensionPlanner. It provides sync_plan() for synchronously converting logical plans to physical plans (using a dedicated thread with an enlarged stack in debug mode to avoid overflow). It also provides serialization methods for physical plans and expressions to protobuf format.

• ArroyoExtensionPlanner: Implements DataFusion's ExtensionPlanner to handle Arroyo-specific logical nodes (Debezium unrolling/encoding, memory execution, unnest) during physical planning.

The visitor processes each ArroyoExtension node by calling its plan_node() method to obtain a LogicalNode and associated LogicalEdge connections, then adds them to the graph.

Usage

This module is invoked after the logical plan rewriting phase to produce the final dataflow graph that is compiled and distributed to workers.

Code Reference

Source Location

crates/arroyo-planner/src/builder.rs

Signature

pub(crate) struct PlanToGraphVisitor<'a> {
    graph: DiGraph<LogicalNode, LogicalEdge>,
    output_schemas: HashMap<NodeIndex, ArroyoSchemaRef>,
    named_nodes: HashMap<NamedNode, NodeIndex>,
    traversal: Vec<Vec<NodeIndex>>,
    planner: Planner<'a>,
}

impl<'a> PlanToGraphVisitor<'a> {
    pub fn new(schema_provider: &'a ArroyoSchemaProvider, session_state: &'a SessionState) -> Self
}

pub(crate) struct Planner<'a> {
    schema_provider: &'a ArroyoSchemaProvider,
    planner: DefaultPhysicalPlanner,
    session_state: &'a SessionState,
}

impl<'a> Planner<'a> {
    pub(crate) fn new(schema_provider: &'a ArroyoSchemaProvider, session_state: &'a SessionState) -> Self
    pub(crate) fn sync_plan(&self, plan: &LogicalPlan) -> Result<Arc<dyn ExecutionPlan>>
}

Import

use crate::builder::{PlanToGraphVisitor, Planner, NamedNode};

I/O Contract

Inputs

Name	Type	Description
logical_plan	&LogicalPlan	Rewritten DataFusion logical plan containing Arroyo extensions
schema_provider	&ArroyoSchemaProvider	Schema provider with table definitions and planning options
session_state	&SessionState	DataFusion session state for physical planning

Outputs

Name	Type	Description
graph	DiGraph<LogicalNode, LogicalEdge>	Directed acyclic graph of operators and edges for the dataflow
output_schemas	HashMap<NodeIndex, ArroyoSchemaRef>	Output schema for each node in the graph

Usage Examples

let mut visitor = PlanToGraphVisitor::new(&schema_provider, &session_state);
rewritten_plan.visit_with_subqueries(&mut visitor)?;
let graph = visitor.graph;

Related Pages

• ArroyoSystems_Arroyo_Planner_Extensions - ArroyoExtension trait that nodes implement for plan_node()
• ArroyoSystems_Arroyo_Plan_Rewriter - Logical plan rewriting that produces the input for this builder
• ArroyoSystems_Arroyo_Physical_Planner - Physical execution plan types used during conversion