Struct StreamingExecutionEngine 

pub struct StreamingExecutionEngine { /* private fields */ }

Streaming execution engine

Drives a single root executor (or multiple partition executors) via direct pull: open() → loop next() → close().

Partitioned roots retain their normal Gather semantics. Eligible local inputs may run through the bounded coordinator when max_workers > 1; all other trees retain the serial pull path.

An optional ExecutionRuntime can be attached to enable cancellation, memory tracking, and profiling. Cancel checking and profile recording are built into StreamingExecutor dispatch in executor.rs.

Implementations

impl StreamingExecutionEngine

pub fn new() -> Self

Create a new streaming execution engine

pub fn set_max_workers(&mut self, max_workers: usize)

Set the maximum number of worker threads for intra-query parallelism.

Values below one are normalised to the serial fallback. When a runtime is already attached and no shared scheduler is configured, this creates a per-query MorselWorkerPool.

pub fn max_workers(&self) -> usize

Returns the maximum number of worker threads.

pub fn set_spill_config(&mut self, config: SpillConfig)

Set the spill configuration for blocking operators.

pub fn set_max_buffered_chunks(&mut self, max_buffered_chunks: usize)

Set the bounded output capacity used by worker channels.

pub fn register_executor( &mut self, _executor_id: usize, executor: StreamingExecutor, )

Register the root executor.

If a runtime is already attached, it is recursively injected into the executor tree so that cancellation, profiling, and resource tracking work regardless of registration order.

pub fn register_partitioned_root( &mut self, partition_count: usize, executor: StreamingExecutor, ) -> Result<(), QueryError>

Register a root that owns local partition trees below one or more exchange nodes. Unlike register_partition_executors, execution still starts from a single root, while the recorded count remains available to profile and explain consumers.

pub fn register_partition_executors( &mut self, executors: Vec<StreamingExecutor>, )

Register partition executors (one per partition).

When set, [execute] will run each partition’s executor sequentially and combine the results. This replaces any single root executor. If a runtime is already attached, it is recursively injected into each partition executor.

pub fn build_partitioned_executor( &mut self, local_trees: Vec<StreamingExecutor>, gather_mode: GatherOperator, global_tree: Option<StreamingExecutor>, ) -> Result<(), QueryError>

Build and register a partitioned executor tree.

Wraps local_trees (one per partition) in a GatherOperator with the given gather_mode, then optionally attaches a global_tree on top. The resulting tree is set as the root executor so that normal [execute] and [into_stream] paths work unchanged.

When global_tree is Some, the tree is: global_tree → Gather(local[0], local[1], …)

When global_tree is None, only the gather is used as root.

pub fn build_partitioned_sort_executor( &mut self, local_trees: Vec<StreamingExecutor>, sort_expressions: Vec<Expression>, sort_directions: Vec<SortDirection>, limit: Option<usize>, ) -> Result<(), QueryError>

Build the only supported partitioned sort shape: LocalSort(spec) × N -> Gather::MergeSort(spec, limit).

This entry point prevents callers from feeding arbitrary, unsorted local streams to the k-way merge implementation. limit, when set, is applied globally after merge order is established.

pub fn build_partitioned_join_executor( &mut self, left_local_trees: Vec<StreamingExecutor>, right_local_trees: Vec<StreamingExecutor>, global_join: StreamingExecutor, ) -> Result<(), QueryError>

Build a semantically-correct partitioned join shape: Gather(left local trees) -> GlobalJoin <- Gather(right local trees).

This is a gather exchange rather than a hash/range shuffle. It keeps every join variant correct while the engine remains single-threaded; a future distributed exchange may replace either gather independently.

pub fn partition_count(&self) -> usize

Returns the number of registered partition executors.

pub fn set_runtime(&mut self, runtime: Arc<ExecutionRuntime>)

Attach an execution runtime (for cancellation, profiling, memory tracking). Also propagates the runtime recursively into all operators. If max_workers > 1 and no shared scheduler is configured, creates a per-query MorselWorkerPool.

pub fn runtime(&self) -> Option<&Arc<ExecutionRuntime>>

Return a reference to the attached runtime, if any.

pub fn open_root(&mut self) -> Result<(), QueryError>

Open the root executor (used by ResultStream).

pub fn next_chunk_from_root(&mut self) -> Result<Option<DataChunk>, QueryError>

Pull the next chunk from the root executor (used by ResultStream).

pub fn stop_root(&mut self) -> Result<(), QueryError>

Stop the root executor (signal upstream to stop producing).

Used by ResultStream before close to allow operators to stop upstream production early.

pub fn close_root(&mut self) -> Result<(), QueryError>

Close the root executor (used by ResultStream).

pub fn execute(self) -> Result<ResultStream, QueryError>

Execute the query and return a streaming result handle.

This is the default execution path. Results are delivered chunk-at-a-time via ResultStream, enabling:

• Low first-chunk latency (no need to wait for full materialization)
• Constant memory usage regardless of result size
• Cooperative cancellation mid-stream

For cases that require full materialization (e.g. EXPLAIN/PROFILE diagnostics, test assertions), use execute_collected.

Partitioned Execution

When partition executors are registered, partitions are executed sequentially and the collected chunks are wrapped in a stream. Formal parallelism runs through the Gather-based root.

pub fn execute_collected(&mut self) -> Result<Vec<DataChunk>, QueryError>

Execute and collect all chunks into memory (explicit materialization).

Use this when you need the full result set upfront. For streaming consumption, prefer execute.

pub fn into_stream(self) -> Result<ResultStream, QueryError>

Convert this engine into a ResultStream for chunk-at-a-time consumption.

Requires a runtime to have been set (via [set_runtime]). Returns an error if no runtime has been attached.

Trait Implementations

impl Default for StreamingExecutionEngine

fn default() -> Self

Returns the “default value” for a type.