Struct ExecutionMemoryConfig 

pub struct ExecutionMemoryConfig {

    pub query_memory_bytes: NonZeroUsize,
    pub batch_rows: NonZeroUsize,
    pub batch_payload_bytes: NonZeroUsize,
    pub blocking_operator_bytes: NonZeroUsize,
    pub graph_expansion_budget: Option<GraphExpansionBudget>,
    pub max_spill_bytes: NonZeroU64,
    pub max_spill_runs: NonZeroUsize,
    pub max_total_spill_bytes: NonZeroU64,
    pub max_total_spill_runs: NonZeroUsize,
    pub min_spill_free_bytes: NonZeroU64,
    pub spill_free_space_probe_interval_bytes: NonZeroU64,
    pub spill_orphan_grace_period: Duration,
    pub spill_directory: PathBuf,
}

Fields

query_memory_bytes: NonZeroUsize

Maximum tracked resident bytes owned by one query across all operators.

batch_rows: NonZeroUsize

Maximum row count in an executor-owned transfer batch.

batch_payload_bytes: NonZeroUsize

Maximum estimated resident bytes in an executor-owned transfer batch.

blocking_operator_bytes: NonZeroUsize

Maximum estimated bytes retained for blocking state, including sort, aggregation, deduplication, graph traversal state and selected top-K. Source reads use the shared query budget independently of this cap.

graph_expansion_budget: Option<GraphExpansionBudget>

Optional row and cumulative payload limits for each seeded expansion. None adds no cumulative cap; live allocations still obey the operator, batch and shared query budgets, including runtime-admitted reservations.

max_spill_bytes: NonZeroU64

Maximum cumulative serialized spill bytes, including merge passes.

max_spill_runs: NonZeroUsize

Maximum cumulative spill runs created, including merge passes.

max_total_spill_bytes: NonZeroU64

Maximum live spill bytes shared by all queries using the spill directory.

max_total_spill_runs: NonZeroUsize

Maximum live spill runs shared by all queries using the spill directory.

min_spill_free_bytes: NonZeroU64

Free space preserved on the filesystem containing the spill directory.

spill_free_space_probe_interval_bytes: NonZeroU64

Reserved bytes between filesystem free-space probes for one spill pool.

spill_orphan_grace_period: Duration

Minimum age before a spill file from an earlier process is removed.

spill_directory: PathBuf

Directory governed as one shared spill pool.

Implementations

impl ExecutionMemoryConfig

pub fn spill_pool_snapshot(&self) -> Result<SpillPoolSnapshot>

Returns process-wide spill usage for this configured directory.

Examples found in repository

examples/spill_resource_probe.rs (line 201)

fn query(
    index: usize,
    rows: usize,
    width: usize,
    blocking: usize,
    barrier: &Barrier,
    governor: Option<&RuntimeGovernor>,
) -> Result<Measurement> {
    let directory =
        std::env::temp_dir().join(format!("hawdb-spill-probe-{}-{index}", std::process::id()));
    let memory = ExecutionMemoryConfig {
        query_memory_bytes: NonZeroUsize::new(16 * 1024 * 1024).unwrap(),
        blocking_operator_bytes: NonZeroUsize::new(blocking).unwrap(),
        batch_payload_bytes: NonZeroUsize::new(64 * 1024).unwrap(),
        max_spill_runs: NonZeroUsize::new(4096).unwrap(),
        max_total_spill_runs: NonZeroUsize::new(16384).unwrap(),
        max_spill_bytes: NonZeroU64::new(1024 * 1024 * 1024).unwrap(),
        min_spill_free_bytes: NonZeroU64::MIN,
        spill_directory: directory.clone(),
        ..ExecutionMemoryConfig::default()
    };
    let permit = governor.map(|governor| {
        governor
            .try_admit(RuntimeWorkRequest::foreground_query(16 * 1024 * 1024, 0))
            .unwrap()
    });
    let task = permit
        .as_ref()
        .map(|permit| permit.bind_task_context(RuntimeTaskContext::default()));
    let ledger = QueryMemoryLedger::for_query(&memory, task.as_ref())?;
    let observer = QueryExecutionObserver::default();
    let catalog = Catalog::default();
    let items = [SortItem {
        key: SortKey::Column("key".into()),
        direction: SortDirection::Asc,
    }];
    let mut source = Rows { count: rows, width };
    let mut measurement = Measurement::default();
    let mut previous = None;
    barrier.wait();
    barrier.wait();
    stream_sort_batches(
        &PhysicalPlan::EmptyExec,
        &items,
        &mut source,
        BatchExecutionContext {
            catalog: &catalog,
            memory: &memory,
            memory_ledger: &ledger,
            task_context: task.as_ref(),
            observer: &observer,
            graph: None,
        },
        ExecutionLimit::unlimited(),
        &mut |batch| {
            for binding in batch {
                let Value::Int(key) = binding.values["key"] else {
                    panic!("missing key")
                };
                let Value::Int(ordinal) = binding.values["ordinal"] else {
                    panic!("missing ordinal")
                };
                assert!(
                    previous.is_none_or(|previous| previous < (key, ordinal)),
                    "stable ordering changed"
                );
                previous = Some((key, ordinal));
                let Value::List(payload) = &binding.values["payload"] else {
                    panic!("missing payload")
                };
                let Value::Map(payload) = &payload[0] else {
                    panic!("missing payload map")
                };
                let Value::String(body) = &payload["body"] else {
                    panic!("missing body")
                };
                assert_eq!(body.len(), width);
                measurement.checksum = measurement.checksum.wrapping_add(ordinal as u64);
                measurement.rows += 1;
            }
            Ok(BatchControl::Continue)
        },
    )?;
    assert_eq!(measurement.rows, rows);
    assert_eq!(measurement.checksum, (rows as u64) * (rows as u64 - 1) / 2);
    assert_eq!(ledger.snapshot().used_bytes, 0);
    assert_eq!(memory.spill_pool_snapshot()?.active_bytes, 0);
    measurement.ledger_peak = ledger.snapshot().peak_bytes;
    for report in observer.into_reports().blocking_memory {
        measurement.spill_bytes += report.spilled_bytes;
        measurement.runs += report.spill_run_count;
    }
    if directory.exists() {
        std::fs::remove_dir_all(directory).unwrap();
    }
    Ok(measurement)
}

Trait Implementations

impl Clone for ExecutionMemoryConfig

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for ExecutionMemoryConfig

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

Formats the value using the given formatter.

impl Default for ExecutionMemoryConfig

fn default() -> Self

Returns the “default value” for a type.

impl Eq for ExecutionMemoryConfig

impl PartialEq for ExecutionMemoryConfig

fn eq(&self, other: &Self) -> bool

Equality operator ==.

fn ne(&self, other: &Rhs) -> bool

Inequality operator !=.

impl StructuralPartialEq for ExecutionMemoryConfig