Struct QueryMemoryLedger 

pub struct QueryMemoryLedger { /* private fields */ }

Implementations

impl QueryMemoryLedger

pub fn new(budget_bytes: NonZeroUsize) -> Self

pub fn new_with_budget_option( budget_bytes: NonZeroUsize, budget_option: &'static str, ) -> Self

Names the configuration option or runtime grant supplying this limit.

pub fn account( &self, class: QueryMemoryClass, owner: impl Into<Arc<str>>, budget_bytes: NonZeroUsize, ) -> QueryMemoryAccount

pub fn account_with_budget_option( &self, class: QueryMemoryClass, owner: impl Into<Arc<str>>, budget_bytes: NonZeroUsize, budget_option: &'static str, ) -> QueryMemoryAccount

Creates an account with an explicit origin for its effective byte limit.

pub fn snapshot(&self) -> QueryMemoryLedgerSnapshot

Examples found in repository

examples/spill_resource_probe.rs (line 199)

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,
        },
        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 QueryMemoryLedger

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 QueryMemoryLedger

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

Formats the value using the given formatter.