Struct State

struct State {
    sink_id: GlobalId,
    worker_id: usize,
    persist_writer: WriteHandle<SourceData, (), Timestamp, StorageDiff>,
    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, StorageDiff>

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.

Note that this is not the same as the write frontier of the output persist shard! It usually is, but during snapshot processing, these frontiers will start at the shard’s read frontier, so they can be beyond its write frontier. This is important as it means we must not discard batch descriptions based on these persist frontiers: A batch description might still be valid even if its lower is before the persist frontiers we observe.

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

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

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, StorageDiff>, metrics: SinkMetrics, as_of: Antichain<Timestamp>, worker_config: &ConfigSet, ) -> 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<(BatchDescription, ProtoBatch, Capability<Timestamp>)>