Struct mz_persist::indexed::background::Maintainer

pub struct Maintainer<B> {
    blob: Arc<BlobCache<B>>,
    metrics: Arc<Metrics>,
    key_val_data_max_len: Option<usize>,
}

A runtime for background asynchronous maintenance of stored data.

Fields

blob: Arc<BlobCache<B>>

metrics: Arc<Metrics>

key_val_data_max_len: Option<usize>

Implementations

impl<B: Blob> Maintainer<B>

pub fn new(blob: BlobCache<B>, metrics: Arc<Metrics>) -> Self

Returns a new Maintainer.

impl<B: Blob + Send + Sync + 'static> Maintainer<B>

pub async fn compact_trace(
    &self,
    req: CompactTraceReq
) -> Result<CompactTraceRes, Error>

Asynchronously runs the requested compaction on the tokio blocking work pool.

fn compact_trace_blocking(
    handle: &Handle,
    blob: Arc<BlobCache<B>>,
    metrics: Arc<Metrics>,
    req: CompactTraceReq,
    key_val_data_max_len: Option<usize>
) -> Result<CompactTraceRes, Error>

Physically and logically compact two trace batches together

This function performs trace compaction with bounded memory usage by:

1. Only ever keeping one BlobTraceBatchPart in memory from each of the two batches being compacted at a time.
2. Only keeping one ColumnarRecords worth of merged data in memory at a time.
3. Performing roughly a linear merge on the two trace batch parts and as we read data from each trace batch, figuring out the upper bound on data that can be compacted, consolidating and merging that, and then placing that in a ColumnarRecords to await being written out as an output BlobTraceBatchPart.

(3). is best explained with an example. If we are compacting two trace batches, A which has two parts, one with keys [0, 10) and one with keys [10, 20) and B has three parts, with keys [0, 15), [15, 30) and [30, 45) respectively then while compacting when we observe the first batches from A and B which have keys [0, 10) and [0, 15) respectively, we cannot merge all of both batches together.

We can only merge the subset of keys in [0, 10), as its possible that subsequent parts in A have relevant keys in [10, 15).

TODO(rkhaitan): We don’t do the real linear merge as compaction requires us to both forward times and consolidate multiple updates at the forwarded times. I believe that doing so is possible in linear time, but doing so in a single pass was complicated enough that it is left for future work.

async fn load_trace_batch_part(
    blob: &BlobCache<B>,
    key: &str,
    updates: &mut Vec<((Vec<u8>, Vec<u8>), u64, i64)>
) -> Result<(), Error>

Read the data from the trace batch part at key into updates.

fn drain_leq_threshold(
    updates: Vec<((Vec<u8>, Vec<u8>), u64, i64)>,
    buffer: &mut Vec<((Vec<u8>, Vec<u8>), u64, i64)>,
    threshold: &(Vec<u8>, Vec<u8>)
) -> Vec<((Vec<u8>, Vec<u8>), u64, i64)>

Drain all records from updates with a (key, val) <= threshold into buffer.

TODO: this could be replaced with a drain_filter if that wasn’t experimental.

async fn write_trace_batch_part(
    blob: &BlobCache<B>,
    desc: Description<u64>,
    updates: ColumnarRecords,
    index: u64
) -> Result<(String, u64), Error>

Write a BlobTraceBatchPart containing updates into Blob.

Returns the key and size in bytes for the trace batch part.

Trait Implementations

impl<B: Debug> Debug for Maintainer<B>

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

Formats the value using the given formatter.