Struct DagExecutor 

pub struct DagExecutor { /* private fields */ }

Ring 0 DAG executor for event processing.

Processes events through a finalized StreamingDag in topological order using the pre-computed RoutingTable for O(1) dispatch.

Construction

let dag = DagBuilder::new()
    .source("src", schema.clone())
    .operator("transform", schema.clone())
    .connect("src", "transform")
    .sink_for("transform", "out", schema.clone())
    .build()?;

let mut executor = DagExecutor::from_dag(&dag);
executor.register_operator(transform_id, Box::new(my_operator));
executor.process_event(src_id, event)?;
let outputs = executor.take_sink_outputs(out_id);

Implementations

impl DagExecutor

pub fn from_dag(dag: &StreamingDag) -> Self

Creates a new executor from a finalized StreamingDag.

Allocates all per-node state (input queues, runtimes, sink buffers) up front in Ring 2. The hot path (process_event) is allocation-free.

Arguments

• dag - A finalized StreamingDag topology

pub fn register_operator(&mut self, node: NodeId, operator: Box<dyn Operator>)

Registers an operator for a node.

Nodes without registered operators act as passthrough: events are forwarded to downstream nodes unchanged. This is the default for source and sink nodes.

Arguments

• node - The node ID to register the operator for
• operator - The operator implementation

pub fn process_event( &mut self, source_node: NodeId, event: Event, ) -> Result<(), DagError>

Processes an event from a source node through the entire DAG.

The event is enqueued at the source node, then all nodes are processed in topological order. Events produced by operators are routed to downstream nodes via the RoutingTable. Sink outputs are collected and can be retrieved via take_sink_outputs().

Arguments

• source_node - The source node to inject the event
• event - The event to process

Errors

Returns DagError::NodeNotFound if the source node is out of bounds.

pub fn process_watermark( &mut self, source_node: NodeId, watermark: i64, ) -> Result<(), DagError>

Advances watermark generators for all nodes in topological order.

Called when an external source provides a watermark (e.g., via Source::watermark()). Propagates the watermark through the DAG so downstream operators can use it for late-data filtering and window triggering.

Arguments

• source_node - The source node that originated the watermark
• watermark - The watermark timestamp in milliseconds

Errors

Returns DagError::NodeNotFound if source_node is out of bounds.

pub fn fire_timers(&mut self, current_time: i64)

Fires all expired timers for each node in topological order.

Polls each node’s TimerService for timers with timestamp <= current_time, calls Operator::on_timer() for each fired timer, and routes the resulting outputs downstream.

This is used to trigger window closures after watermark advancement.

Arguments

• current_time - The current event-time (typically the latest watermark)

pub fn take_sink_outputs(&mut self, sink_node: NodeId) -> Vec<Event>

Takes collected sink outputs for a given sink node.

Returns all events that reached this sink during prior process_event calls, draining the internal buffer.

pub fn take_all_sink_outputs(&mut self) -> FxHashMap<NodeId, Vec<Event>>

Takes all sink outputs across all sink nodes.

pub fn metrics(&self) -> &DagExecutorMetrics

Returns a reference to the executor metrics.

pub fn node_metrics(&self) -> &[OperatorNodeMetrics]

Returns per-operator node metrics (only available with dag-metrics feature).

pub fn reset_metrics(&mut self)

Resets all executor metrics to zero.

pub fn source_nodes(&self) -> &[NodeId]

Returns the source node IDs.

pub fn sink_nodes(&self) -> &[NodeId]

Returns the sink node IDs.

pub fn node_type(&self, node: NodeId) -> Option<DagNodeType>

Returns the node type for a given node ID.

pub fn checkpoint(&self) -> FxHashMap<NodeId, OperatorState>

Checkpoints all registered operators.

Returns a map of NodeId to OperatorState for all nodes that have registered operators.

pub fn restore( &mut self, states: &FxHashMap<NodeId, OperatorState>, ) -> Result<(), DagError>

Restores operator state from a checkpoint snapshot.

Iterates the provided states and calls operator.restore() on each registered operator.

Errors

Returns DagError::RestoreFailed if any operator fails to restore.

pub fn checkpoint_watermarks(&self) -> DagWatermarkCheckpoint

Checkpoints the watermark tracker state.

pub fn restore_watermarks(&mut self, checkpoint: &DagWatermarkCheckpoint)

Restores watermark tracker state from a checkpoint.

pub fn inject_events(&mut self, node_id: NodeId, events: Vec<Event>)

Injects events into a node’s input queue.

Used during recovery to repopulate queues with buffered events.

pub fn input_count(&self, node_id: NodeId) -> usize

Returns the number of incoming edges for a node.

pub fn process_checkpoint_barrier( &mut self, _barrier: &CheckpointBarrier, ) -> FxHashMap<NodeId, OperatorState>

Snapshots all registered operators in topological order.

Takes the barrier for consistency (future use with epoch tracking). In the synchronous single-threaded executor, topological ordering guarantees upstream-first snapshots.

Trait Implementations

impl Debug for DagExecutor

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.