Struct mz_compute::sink::materialized_view_v2::write::State

struct State {
    sink_id: GlobalId,
    worker_id: usize,
    persist_writer: WriteHandle<SourceData, (), Timestamp, Diff>,
    corrections: OkErr<Correction<Row>, Correction<DataflowError>>,
    desired_frontiers: OkErr<Antichain<Timestamp>, Antichain<Timestamp>>,
    persist_frontiers: OkErr<Antichain<Timestamp>, Antichain<Timestamp>>,
    batch_description: Option<(BatchDescription, Capability<Timestamp>)>,
    force_consolidation_after: Option<Antichain<Timestamp>>,
}

State maintained by the write operator.

Fields

sink_id: GlobalId

worker_id: usize

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

corrections: OkErr<Correction<Row>, Correction<DataflowError>>

Contains desired - persist, reflecting the updates we would like to commit to persist in order to “correct” it to track desired. This collection is only modified by updates received from either the desired or persist inputs.

desired_frontiers: OkErr<Antichain<Timestamp>, Antichain<Timestamp>>

The frontiers of the desired inputs.

persist_frontiers: OkErr<Antichain<Timestamp>, Antichain<Timestamp>>

The frontiers of the persist inputs.

batch_description: Option<(BatchDescription, Capability<Timestamp>)>

The current valid batch description and associated output capability, if any.

Note that “valid” here implies that if a batch description is set, it must be true that its lower is >= the persist_frontier. Otherwise the described batch couldn’t be appended anymore, rendering the batch description invalid.

force_consolidation_after: Option<Antichain<Timestamp>>

A request to force a consolidation of corrections once both desired_frontiers and persist_frontiers become greater than the given frontier.

Normally we force a consolidation whenever we write a batch, but there are periods (like read-only mode) when that doesn’t happen, and we need to manually force consolidation instead. Currently this is only used to ensure we quickly get rid of the snapshot updates.

Implementations

impl State

fn new( sink_id: GlobalId, worker_id: usize, persist_writer: WriteHandle<SourceData, (), Timestamp, Diff>, metrics: SinkMetrics, as_of: Antichain<Timestamp>, ) -> Self

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

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

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

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

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

fn apply_desired_frontier_advancement(&mut self)

Apply the effects of a previous desired frontier advancement.

fn apply_persist_frontier_advancement(&mut self)

Apply the effects of a previous persist frontier advancement.

fn maybe_force_consolidation(&mut self)

If the current consolidation request has become applicable, apply it.

fn absorb_batch_description( &mut self, desc: BatchDescription, cap: Capability<Timestamp>, )

async fn maybe_write_batch( &mut self, ) -> Option<(ProtoBatch, Capability<Timestamp>)>