Struct mz_compute::sink::correction::Correction

pub(super) struct Correction<D> {
    updates: BTreeMap<Timestamp, ConsolidatingVec<D>>,
    since: Antichain<Timestamp>,
    total_size: LengthAndCapacity,
    metrics: SinkMetrics,
    worker_metrics: SinkWorkerMetrics,
}

A collection holding persist_sink updates.

The Correction data structure is purpose-built for the persist_sink::write_batches operator:

• It stores updates by time, to enable efficient separation between updates that should be written to a batch and updates whose time has not yet arrived.
• It eschews an interface for directly removing previously inserted updates. Instead, updates are removed by inserting them again, with negated diffs. Stored updates are continuously consolidated to give them opportunity to cancel each other out.
• It provides an interface for advancing all contained updates to a given frontier.

Fields

updates: BTreeMap<Timestamp, ConsolidatingVec<D>>

Stashed updates by time.

since: Antichain<Timestamp>

Frontier to which all update times are advanced.

total_size: LengthAndCapacity

Total length and capacity of vectors in updates.

Tracked to maintain metrics.

metrics: SinkMetrics

Global persist sink metrics.

worker_metrics: SinkWorkerMetrics

Per-worker persist sink metrics.

Implementations

impl<D> Correction<D>

pub fn new(metrics: SinkMetrics, worker_metrics: SinkWorkerMetrics) -> Self

Construct a new Correction instance.

fn update_metrics(&mut self, new_size: LengthAndCapacity)

Update persist sink metrics to the given new length and capacity.

impl<D: Data> Correction<D>

pub fn insert(&mut self, updates: Vec<(D, Timestamp, Diff)>)

Insert a batch of updates.

pub fn insert_negated(&mut self, updates: Vec<(D, Timestamp, Diff)>)

Insert a batch of updates, after negating their diffs.

fn insert_inner(&mut self, updates: Vec<(D, Timestamp, Diff)>)

Insert a batch of updates.

The given updates must all have been advanced by self.since.

pub fn updates_within( &mut self, lower: &Antichain<Timestamp>, upper: &Antichain<Timestamp>, ) -> impl Iterator<Item = (D, Timestamp, Diff)> + ExactSizeIterator + '_

Consolidate and return updates within the given bounds.

Panics

Panics if lower is not less than or equal to upper.

pub fn updates_before( &mut self, upper: &Antichain<Timestamp>, ) -> impl Iterator<Item = (D, Timestamp, Diff)> + ExactSizeIterator + '_

Consolidate and return updates before the given upper.

fn consolidate<R>(&mut self, range: R) -> usize

where R: RangeBounds<Timestamp>,

Consolidate the updates at the times in the given range.

Returns the number of updates remaining in the range afterwards.

pub fn since(&self) -> &Antichain<Timestamp>

Return the current since frontier.

pub fn advance_since(&mut self, since: Antichain<Timestamp>)

Advance the since frontier.

Panics

Panics if the given since is less than the current since frontier.

pub fn advance_by(&mut self, frontier: &Antichain<Timestamp>)

Advance all contained updates by the given frontier.

If the given frontier is empty, all remaining updates are discarded.

pub fn consolidate_at_since(&mut self)

Consolidate all updates at the current since.

Trait Implementations

impl<D> Drop for Correction<D>

fn drop(&mut self)

Executes the destructor for this type.