17 Chapter 16: Local And Streaming Execution

The distributed chapters explain the cluster path in detail: job graphs, drivers, workers, task regions, streams, and shuffles. This chapter fills in two execution views that deserve their own concise treatment:

• local execution, where Sail directly asks DataFusion to execute a physical plan,
• structured streaming, where Sail rewrites a logical plan into a flow-event plan before physical execution.

Both use the same JobRunner abstraction, which is exactly why the abstraction is valuable.

17.1 Code Map

17.2 One Trait, Two Execution Modes

The JobRunner trait hides the execution backend:

#[tonic::async_trait]
pub trait JobRunner: StateObservable<JobRunnerObserver> + Send + Sync + 'static {
    async fn execute(
        &self,
        ctx: &SessionContext,
        plan: Arc<dyn ExecutionPlan>,
    ) -> Result<SendableRecordBatchStream>;

    async fn stop(&self, history: oneshot::Sender<JobRunnerHistory>);
}

The local and cluster runners implement the same method. Protocol code does not need to know which mode is active. It retrieves JobService from the session and calls:

service.runner().execute(ctx, plan).await

That line is one of Sail’s key architecture compression points.

17.3 Local Execution

LocalJobRunner is intentionally small:

let plan = trace_execution_plan(plan, options)?;
Ok(execute_stream(plan, ctx.task_ctx())?)

The runner wraps the plan with telemetry tracing and then delegates to DataFusion’s execute_stream. DataFusion handles partition execution inside the process, and Sail receives a SendableRecordBatchStream.

Local mode is not a lesser engine. It is the same planning path with a simpler execution backend. That makes it useful for:

• development,
• compatibility tests,
• small deployments,
• debugging physical plans before cluster concerns enter.

17.4 Cluster Execution

ClusterJobRunner sends the physical plan to the driver actor:

self.driver.send(DriverEvent::ExecuteJob {
    plan,
    context: ctx.task_ctx(),
    result: tx,
}).await?;

The driver builds a job graph and eventually returns a stream. From the caller’s perspective, local and cluster modes both produce SendableRecordBatchStream.

The difference is below the trait boundary.

17.5 Streaming Starts As A Logical Rewrite

Streaming is not a different protocol. A streaming query still enters through Spark Connect, resolves to a logical plan, and becomes a physical plan. The key difference is that Sail rewrites the logical plan before physical planning:

let plan = if is_streaming_plan(&plan)? {
    rewrite_streaming_plan(plan)?
} else {
    plan
};

The rewriter turns ordinary plan nodes into streaming-aware extension nodes where needed. For example:

• batch table scans over streaming sources become StreamSourceWrapperNode,
• RangeNode can be wrapped by StreamSourceAdapterNode,
• filters and limits become streaming filter/limit nodes,
• sinks and display paths use collectors and barriers.

17.6 Stream Sources

A streaming source implements StreamSource:

#[async_trait::async_trait]
pub trait StreamSource: Send + Sync + fmt::Debug {
    fn data_schema(&self) -> SchemaRef;

    async fn scan(
        &self,
        state: &dyn Session,
        projection: Option<&Vec<usize>>,
        filters: &[Expr],
        limit: Option<usize>,
    ) -> Result<Arc<dyn ExecutionPlan>>;
}

The source returns an ExecutionPlan, not a raw stream. That keeps it inside the DataFusion physical-planning model. Current concrete examples include Sail’s rate and socket sources.

17.7 Flow Event Schema

Streaming records carry more than user columns. Sail prepends flow-event fields:

_marker
_retracted
<user columns...>

_marker is for control messages. _retracted distinguishes normal insert events from retraction/delete events.

This design lets streaming physical operators process one Arrow RecordBatch shape while preserving event semantics. It also gives future stateful operators a place to represent retract-mode updates.

17.8 Streaming Physical Nodes

The streaming logical nodes are planned by ExtensionPhysicalPlanner, just like other custom nodes. Their physical counterparts live under crates/sail-physical-plan/src/streaming/.

The important nodes are:

BarrierNode and BarrierExec provide checkpoint-like coordination points.

17.9 Streaming Query Lifecycle

Spark Connect exposes streaming operations such as start, stop, status, and await. Sail tracks running queries with StreamingQuery and StreamingQueryManager.

The lifecycle uses asynchronous coordination primitives:

• watch channels for stopped/error state,
• oneshot channels for stop signals,
• JoinHandle for the background task.

The manager lives in SparkSessionState, so streaming query state is scoped to the Spark session.

17.10 Current Boundaries

Streaming support is real, but not all Spark streaming semantics are complete. Readers should distinguish:

• the architecture, which is present and coherent,
• the feature surface, which is still evolving.

The flow-event schema, streaming rewriter, and query manager are the architectural foundation. Full stateful aggregations, event-time triggers, and continuous-mode coverage are areas to verify against the current code before making claims.

17.11 Takeaways

Local execution and cluster execution share JobRunner. Streaming and batch execution share the planning/execution stack below a logical rewrite. That is the pattern to preserve: new execution behavior should enter through clear boundaries, not by bypassing the common plan and stream model.

Navigation: Previous: Chapter 15, Custom Nodes And Optimizers | Next: Chapter 17, Testing Spark Compatibility | Reader Guide