Struct Consolidator

pub(crate) struct Consolidator<T, D, Sort: RowSort<T, D>> {

    context: String,
    cfg: FetchConfig,
    shard_id: ShardId,
    sort: Sort,
    blob: Arc<dyn Blob>,
    metrics: Arc<Metrics>,
    shard_metrics: Arc<ShardMetrics>,
    read_metrics: Arc<ReadMetrics>,
    runs: Vec<VecDeque<(ConsolidationPart<T, D>, usize)>>,
    filter: FetchBatchFilter<T>,
    budget: usize,
    lower_bound: Option<LowerBound<T>>,
    drop_stash: Option<StructuredUpdates>,
}

A tool for incrementally consolidating a persist shard.

The naive way to consolidate a Persist shard would be to fetch every part, then consolidate the whole thing. We’d like to improve on that in two ways:

• Concurrency: we’d like to be able to start consolidating and returning results before every part is fetched. (And continue fetching while we’re doing other work.)
• Memory usage: we’d like to limit the number of parts we have in memory at once, dropping parts that are fully consolidated and fetching parts just a little before they’re needed.

This interface supports this by consolidating in multiple steps. Each call to Self::next will do some housekeeping work – prefetching needed parts, dropping any unneeded parts – and return an iterator over a consolidated subset of the data. To read an entire dataset, the client should call next until it returns None, which signals all data has been returned… but it’s also free to abandon the instance at any time if it eg. only needs a few entries.

Fields

context: String

cfg: FetchConfig

shard_id: ShardId

sort: Sort

blob: Arc<dyn Blob>

metrics: Arc<Metrics>

shard_metrics: Arc<ShardMetrics>

read_metrics: Arc<ReadMetrics>

runs: Vec<VecDeque<(ConsolidationPart<T, D>, usize)>>

filter: FetchBatchFilter<T>

budget: usize

lower_bound: Option<LowerBound<T>>

An optional exclusive lower bound for the KVTs that this consolidator will return. This can be used to filter out KVTs that are not strictly larger than the bound when “resuming” a consolidation run during incremental compaction.

drop_stash: Option<StructuredUpdates>

Implementations

impl<T, D, Sort> Consolidator<T, D, Sort>

where T: Timestamp + Codec64 + Lattice, D: Codec64 + Semigroup + Ord, Sort: RowSort<T, D>,

pub fn new( context: String, cfg: FetchConfig, shard_id: ShardId, sort: Sort, blob: Arc<dyn Blob>, metrics: Arc<Metrics>, shard_metrics: Arc<ShardMetrics>, read_metrics: ReadMetrics, filter: FetchBatchFilter<T>, lower_bound: Option<LowerBound<T>>, prefetch_budget_bytes: usize, ) -> Self

Create a new Self instance with the given prefetch budget. This budget is a “soft limit” on the size of the parts that the consolidator will fetch… we’ll try and stay below the limit, but may burst above it if that’s necessary to make progress.

impl<T, D, Sort> Consolidator<T, D, Sort>

where T: Timestamp + Codec64 + Lattice + Sync, D: Codec64 + Semigroup + Ord, Sort: RowSort<T, D>,

pub fn enqueue_run( &mut self, desc: &Description<T>, run_meta: &RunMeta, parts: impl IntoIterator<Item = RunPart<T>>, )

Add another run of data to be consolidated.

To ensure consolidation, every tuple in this run should be larger than any tuple already returned from the iterator. At the moment, this invariant is not checked. The simplest way to ensure this is to enqueue every run before any calls to next.

fn push_run(&mut self, run: VecDeque<(ConsolidationPart<T, D>, usize)>)

fn trim(&mut self)

Tidy up: discard any empty parts, and discard any runs that have no parts left.

fn iter(&mut self) -> Option<ConsolidatingIter<'_, T, D>>

Return an iterator over the next consolidated chunk of output, if there’s any left.

Requirement: at least the first part of each run should be fetched and nonempty.

async fn unblock_progress(&mut self) -> Result<()>

We don’t need to have fetched every part to make progress, but we do at least need to have fetched some parts: in particular, parts at the beginning of their runs which may include the smallest remaining KVT.

Returns success when we’ve successfully fetched enough parts to be able to make progress.

pub(crate) async fn next( &mut self, ) -> Result<Option<impl Iterator<Item = ((ArrayIdx<'_>, Option<ArrayIdx<'_>>), T, D)>>>

Wait until data is available, then return an iterator over the next consolidated chunk of output. If this method returns None, that all the data has been exhausted and the full consolidated dataset has been returned.

fn chunk(&mut self, max_len: usize, max_bytes: usize) -> Option<Part>

pub(crate) async fn next_chunk( &mut self, max_len: usize, max_bytes: usize, ) -> Result<Option<Part>>

Wait until data is available, then return an iterator over the next consolidated chunk of output. If this method returns None, that all the data has been exhausted and the full consolidated dataset has been returned.

fn live_bytes(&self) -> usize

The size of the data that we might be holding concurrently in memory. While this is normally kept less than the budget, it may burst over it temporarily, since we need at least one part in every run to continue making progress.

pub(crate) fn start_prefetches(&mut self) -> Option<usize>

Returns None if the budget was exhausted, or Some(remaining_bytes) if it is not.

Trait Implementations

impl<T: Debug, D: Debug, Sort: Debug + RowSort<T, D>> Debug for Consolidator<T, D, Sort>

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

Formats the value using the given formatter.

impl<T, D, Sort: RowSort<T, D>> Drop for Consolidator<T, D, Sort>

fn drop(&mut self)

Executes the destructor for this type.