Struct mz_compute::sink::materialized_view_v2::append::State

struct State {
    sink_id: GlobalId,
    persist_writer: WriteHandle<SourceData, (), Timestamp, Diff>,
    batches_frontier: Antichain<Timestamp>,
    lower: Antichain<Timestamp>,
    batch_description: Option<BatchDescription>,
    batches: Vec<Batch<SourceData, (), Timestamp, Diff>>,
}

State maintained by the append operator.

Fields

sink_id: GlobalId

persist_writer: WriteHandle<SourceData, (), Timestamp, Diff>

batches_frontier: Antichain<Timestamp>

The current input frontier of batches.

lower: Antichain<Timestamp>

The greatest observed lower from both descs and batches.

batch_description: Option<BatchDescription>

The batch description for lower, if any.

batches: Vec<Batch<SourceData, (), Timestamp, Diff>>

Batches received for lower.

Implementations

impl State

fn new( sink_id: GlobalId, persist_writer: WriteHandle<SourceData, (), Timestamp, Diff>, ) -> Self

fn trace<S: AsRef<str>>(&self, message: S)

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

async fn advance_lower(&mut self, frontier: Antichain<Timestamp>)

Advance the current lower.

Discards all currently stashed batches and batch descriptions, assuming that they are now invalid.

async fn absorb_batch_description(&mut self, desc: BatchDescription)

Absorb the given batch description into the state, provided it is not outdated.

async fn absorb_batch(&mut self, batch: ProtoBatch)

Absorb the given batch into the state, provided it is not outdated.

async fn maybe_append_batches(&mut self)

async fn append_batches( &mut self, desc: BatchDescription, ) -> Result<Antichain<Timestamp>, Antichain<Timestamp>>

Append the current batches to the output shard.

Returns whether the append was successful or not, and the current shard upper in either case.

This method advances the shard upper to the batch lower if necessary. This is the mechanism that brings the shard upper to the sink as-of when appending the initial batch.

An alternative mechanism for bringing the shard upper to the sink as-of would be making a single append at operator startup. The reason we are doing it here instead is that it simplifies the implementation of read-only mode. In read-only mode we have to defer any persist writes, including the initial upper bump. Having only a single place that performs writes makes it easy to ensure we are doing that correctly.