Struct CorrectionV1

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

A collection holding persist_sink updates.

The CorrectionV1 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.

growth_dampener: usize

Configuration for ConsolidatingVec driving the growth rate down from doubling.

Implementations

impl<D> CorrectionV1<D>

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

Construct a new CorrectionV1 instance.

fn update_metrics(&mut self, new_size: LengthAndCapacity)

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

impl<D: Data> CorrectionV1<D>

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

Insert a batch of updates.

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

Insert a batch of updates, after negating their diffs.

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

Insert a batch of updates.

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

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

Consolidate and return updates within the given bounds.

Panics

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

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

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 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 CorrectionV1<D>

fn drop(&mut self)

Executes the destructor for this type.