Struct ExecutionRuntime 

pub struct ExecutionRuntime {

    pub memory_budget: MemoryBudget,
    pub worker_pool: Arc<Mutex<Option<Arc<dyn TaskScheduler>>>>,
    pub max_buffered_chunks: AtomicUsize,
    pub spill_manager: Arc<Mutex<Option<Arc<SpillManager>>>>,
    pub storage: Option<Arc<RwLock<dyn QueryStorage>>>,
    pub search: SearchContext,
    pub state_arenas: Vec<Mutex<StateArenaSet>>,
    pub parameter_values: Option<Arc<HashMap<String, Value>>>,
    pub session_variable_values: Option<Arc<HashMap<String, Value>>>,
    pub arena: Option<Arc<Mutex<Arena>>>,
    pub feedback_history: Option<Arc<QueryFeedbackHistory>>,
    pub macro_manager: Option<Arc<MacroManager>>,
    pub type_alias_manager: Option<Arc<TypeAliasManager>>,
    pub sequence_manager: Option<Arc<SequenceManager>>,
    pub cte_tables: Arc<Mutex<HashMap<String, Vec<Vec<Value>>>>>,
    /* private fields */
}

Per-query execution runtime shared across all operators.

Centralises cancellation, memory tracking, profiling, resource lifecycle, and query-registration so that operators do not each carry ad-hoc context.

unified cancellation via CancelToken has been removed; operators check CancelToken::is_cancelled.

Fields

memory_budget: MemoryBudget

Per-query memory budget for blocking operators.

worker_pool: Arc<Mutex<Option<Arc<dyn TaskScheduler>>>>

Query-level morsel worker pool for dynamic partition execution Kept for backward compatibility. Created when max_workers > 1 and no shared_scheduler is set; None means serial fallback. Behind a Mutex so the engine can set the pool after construction.

max_buffered_chunks: AtomicUsize

Per-partition output channel capacity for parallel exchange/gather.

spill_manager: Arc<Mutex<Option<Arc<SpillManager>>>>

Spill manager for offloading operator data to disk.

storage: Option<Arc<RwLock<dyn QueryStorage>>>

Storage client for this query execution. Moved here from OperatorSpec so that the physical plan tree is truly immutable and cacheable without sharing storage handles.

search: SearchContext

state_arenas: Vec<Mutex<StateArenaSet>>

Per-partition operator state arenas.

Indexed by partition_id so parallel workers do not contend on a single lock. state_arenas[0] serves global / non-partitioned operators.

Operators create/read/update their typed state during open() / next() / close(), indexed by [PhysicalOperatorId] stored in OperatorBase::physical_operator_id.

parameter_values: Option<Arc<HashMap<String, Value>>>

Runtime parameter name→value map, bound at materialization time. Operators read this to resolve Expression::Parameter references.

session_variable_values: Option<Arc<HashMap<String, Value>>>

Runtime session variable name→value snapshot, bound at materialization time. Operators read this to resolve Expression::SessionVariable references.

arena: Option<Arc<Mutex<Arena>>>

Per-query bumpalo arena for executor temporary allocations.

feedback_history: Option<Arc<QueryFeedbackHistory>>

Shared query feedback history for collecting execution statistics.

Injected by the materializer from the query bindings; when set, the execution instance records estimated-vs-actual operator feedback here after execution completes (stats feedback loop).

macro_manager: Option<Arc<MacroManager>>

Macro catalog manager (user-defined macros), shared engine-wide.

Injected by the materializer from the query bindings; read by DDL operators executing CREATE/DROP MACRO and SHOW MACROS.

type_alias_manager: Option<Arc<TypeAliasManager>>

Type-alias catalog manager (user-defined types), shared engine-wide.

Injected by the materializer from the query bindings; read by DDL operators executing CREATE/DROP TYPE.

sequence_manager: Option<Arc<SequenceManager>>

Sequence catalog manager (named sequences), shared engine-wide.

Injected by the materializer from the query bindings; read by DDL operators executing CREATE/ALTER/DROP SEQUENCE and by the curr_val / next_val expression functions.

cte_tables: Arc<Mutex<HashMap<String, Vec<Vec<Value>>>>>

Working tables of enclosing recursive-CTE fixpoints, keyed by the mangled CTE tag (crate::cte::mangle_cte_name).

Written by the fixpoint operator before each step iteration and cleared afterwards; read by CTE scan sources. Serial execution only (fixpoint fragments are never partitioned).

Implementations

impl ExecutionRuntime

pub fn new( query_id: QueryIdentity, memory_budget: MemoryBudget, storage: Option<Arc<RwLock<dyn QueryStorage>>>, search: SearchContext, ) -> Self

Create a new execution runtime with the given query identity, memory budget, and optional storage client.

pub fn default_budget() -> Self

Create a runtime with default settings (query_id = 0, default memory budget, no storage).

pub fn query_id(&self) -> QueryIdentity

pub fn assign_query_id(&self, id: u64)

Override the query ID number after construction.

The factory initialises query_id.query_id to 0; the API layer assigns the real server-side ID before the handle is returned to the caller.

pub fn set_query_manager(&self, qm: Arc<QueryManager>)

Attach a QueryManager so that KILL QUERY, finish tracking, and progress forwarding work.

Uses interior mutability (&self) because the runtime is shared with the executor tree before the assembly attaches the manager, so &mut access is unavailable at that point.

pub fn finish_guard(&self) -> Option<QueryFinishGuard>

Register this query with the attached QueryManager and return a QueryFinishGuard that marks it finished on drop.

Returns None when no QueryManager is attached (non-fatal).

pub fn set_query_registry(&self, registry: Arc<QueryRegistry>, qid: QueryId)

Attach a QueryRegistry and the allocated QueryId.

Interior mutability via Mutex: the runtime is shared with the executor tree before registration, so &mut access is not available.

pub fn registry_query_id(&self) -> Option<QueryId>

Return the registry-allocated query ID, if set.

pub fn set_session_controller(&self, ctrl: Arc<SessionTransactionController>)

Set the session-level transaction controller for transaction commands.

pub fn session_controller(&self) -> Option<Arc<SessionTransactionController>>

Return the session-level transaction controller, if set.

pub fn set_parameter_values(&mut self, values: Arc<HashMap<String, Value>>)

Set the parameter name→value map for this execution instance.

pub fn parameter_values(&self) -> Option<Arc<HashMap<String, Value>>>

Return the parameter name→value map, if bound.

pub fn set_session_variable_values( &mut self, values: Arc<HashMap<String, Value>>, )

Set the session variable snapshot for this execution instance.

pub fn session_variable_values(&self) -> Option<Arc<HashMap<String, Value>>>

Return the session variable snapshot, if bound.

pub fn set_macro_manager(&mut self, manager: Option<Arc<MacroManager>>)

Set the macro catalog manager for this execution instance.

pub fn macro_manager(&self) -> Option<Arc<MacroManager>>

Return the macro catalog manager, if bound.

pub fn set_type_alias_manager(&mut self, manager: Option<Arc<TypeAliasManager>>)

Set the type-alias catalog manager for this execution instance.

pub fn type_alias_manager(&self) -> Option<Arc<TypeAliasManager>>

Return the type-alias catalog manager, if bound.

pub fn set_sequence_manager(&mut self, manager: Option<Arc<SequenceManager>>)

Set the sequence catalog manager for this execution instance.

pub fn sequence_manager(&self) -> Option<Arc<SequenceManager>>

Return the sequence catalog manager, if bound.

pub fn set_cte_table(&self, mangled_tag: &str, rows: Vec<Vec<Value>>)

Publish a recursive-CTE working table for the duration of one step iteration. Overwrites any previous table under the same mangled tag.

pub fn cte_table(&self, mangled_tag: &str) -> Option<Vec<Vec<Value>>>

Read the current working table published under a mangled CTE tag.

pub fn clear_cte_table(&self, mangled_tag: &str)

Remove the working table published under a mangled CTE tag.

pub fn set_transaction_scope(&mut self, scope: TransactionScope)

Set the transaction scope for this execution.

pub fn transaction_scope(&self) -> Option<&TransactionScope>

Return the current transaction scope, if any.

pub fn cancel_token_v2(&self) -> CancelToken

Return the typed CancelToken for cooperative cancellation.

pub fn note_transaction_conflict(&self)

Propagate a write-conflict failure of the current transaction.

Called by write operators when a storage access fails with a conflict-classified error (write-write conflict, rollback-only). The session controller is marked rollback-only so a later COMMIT is rejected, and the shared cancellation token is cancelled with the typed CancelReason::TransactionConflict reason — terminating any pipeline stages of the same transaction that have not started yet.

pub fn set_cancel_token(&self, token: CancelToken)

Adopt an externally-owned cancellation token.

Called at instantiation so the runtime, the query registry, and the request-scoped [QueryContext] share one token (single source for KILL QUERY / cancellation). Interior mutability via Mutex because the executor tree holds an Arc clone of the runtime by the time instantiation wires the token.

pub fn cancel_token(&self) -> CancelToken

Shared cancellation token for cooperative checks by operators and I/O.

pub fn is_cancelled(&self) -> bool

Check whether the query has been cancelled (token or deadline).

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

Return an error if the query has been cancelled.

pub fn cancel_with_reason(&self, reason: CancelReason)

Cancel this query with a typed reason.

Sets the CancelToken, marks the query as Killed in the attached QueryManager, and cancels the registry entry (if configured).

pub fn cancel(&self)

Legacy cancel (no typed reason). Delegates to [cancel_with_reason] with CancelReason::UserKill.

pub fn cancel_on_deadline(&self)

Enable deadline-based cancellation.

pub fn set_deadline(&mut self, deadline: Option<Instant>)

Set or clear a deadline.

pub fn profile(&self) -> &Arc<ProfileBoard>

pub fn register_operator( &self, op_profile: &OperatorProfile, ) -> Arc<ProfileEntry>

Register an operator profile entry (called during open()).

pub fn state_arena_for( &self, partition_id: Option<usize>, ) -> &Mutex<StateArenaSet>

Return a StateArenaSet mutex for the given partition.

Non-partitioned operators (partition_id == None) always map to arena 0. Partitioned operators map to (partition_id + 1), with fallback to the last arena when the partition count was under-estimated.

pub fn set_partition_count(&mut self, count: usize)

Set the number of partition arenas (must be ≥ 1).

pub fn profile_start(&self)

Record that execution has started (profile timing).

pub fn profile_end(&self)

Record that execution has ended.

pub fn profile_add_rows(&self, count: u64)

Add rows to the profile counter.

pub fn set_progress_rows_interval(&self, rows_interval: u64)

Row cadence for query-progress notifications (0 disables).

Set alongside set_query_manager by the assembly that owns both the executor and the QueryManager. Zero by default: unconfigured runtimes never emit and pay a single relaxed atomic load per batch.

pub fn set_progress_identity(&self, session_id: i64, query_id: i64)

Identity reported in progress notifications, in the QueryManager id space. A negative query id suppresses emission until the assembly provides the real mapping between the two query identity schemes.

pub fn resource_owner(&self) -> &Arc<Mutex<ResourceOwner>>

pub fn on_cleanup<F>(&self, f: F)

where F: FnOnce() + Send + 'static,

Register a cleanup callback.

pub fn release_resources(&self)

Release all owned resources.

pub fn set_worker_pool(&self, pool: Option<MorselWorkerPool>)

Set the morsel worker pool for this query.

When a shared scheduler is available via set_shared_scheduler, this call is ignored — the shared scheduler’s pool is used instead. Legacy support: creates a per-query pool when no shared scheduler is configured.

pub fn set_shared_scheduler(&self, scheduler: Option<Arc<SharedScheduler>>)

Set the engine-level shared scheduler for this query.

When set, all parallel execution uses the shared worker pool instead of per-query threads. The scheduler’s Arc<dyn TaskScheduler> is injected into worker_pool so existing consumer code paths (Exchange / Gather operators) continue to work unchanged.

Takes priority over any per-query pool that may have been set.

pub fn get_shared_scheduler(&self) -> Option<Arc<SharedScheduler>>

Return the shared scheduler, if set.

pub fn effective_worker_pool(&self) -> Option<Arc<dyn TaskScheduler>>

Return the effective worker pool — either from the shared scheduler, or from the per-query pool, or None for serial fallback.

pub fn set_spill_manager(&self, manager: Option<Arc<SpillManager>>)

Set the spill manager for this query execution.

pub fn get_spill_manager(&self) -> Option<Arc<SpillManager>>

Access the spill manager.

pub fn arena(&self) -> Option<&Arc<Mutex<Arena>>>

Return a reference to the bumpalo arena, if configured.

pub fn columnar_stats(&self) -> Arc<ColumnarStats>

Return the columnar fast-path counters shared with produced chunks.

pub fn columnar_policy(&self) -> Option<Arc<ColumnarPolicy>>

Return the shared columnar layout policy, if injected.

pub fn set_columnar_policy(&mut self, policy: Option<Arc<ColumnarPolicy>>)

Inject the shared columnar layout policy (owned by the optimizer engine; set by the materializer from the query bindings).

pub fn columnar_override(&self) -> QueryColumnarOverride

Per-query columnar override for this runtime.

pub fn set_columnar_override(&mut self, query_override: QueryColumnarOverride)

Pin the columnar decision for this query only.

pub fn flush_columnar_stats_to_policy(&self)

Merge this query’s columnar hit/miss counts into the shared policy.

Called once when the query finishes (materialized, streaming, and discard paths) so the adaptive gate learns across queries.

pub fn reset_arena(&self)

Reset the bumpalo arena, freeing all temporary allocations.

pub fn set_max_buffered_chunks(&self, chunks: usize)

Set the per-partition output channel capacity for parallel operators.

Trait Implementations

impl Debug for ExecutionRuntime

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

Formats the value using the given formatter.