Struct dataflow::source::SourceReaderPersistence

struct SourceReaderPersistence {
    source_name: String,
    config: PersistentTimestampBindingsConfig<SourceTimestamp, AssignedTimestamp>,
}

Util for restoring persisted timestamps and rendering persistence operators.

Fields

source_name: String

config: PersistentTimestampBindingsConfig<SourceTimestamp, AssignedTimestamp>

Implementations

impl SourceReaderPersistence

fn new(
    source_name: String,
    config: PersistentTimestampBindingsConfig<SourceTimestamp, AssignedTimestamp>
) -> Self

Creates a new SourceReaderPersistence from the given configuration. The configuration determines the persistent collection that will be used by restore and render_persistence_operators, respectively.

fn restore(
    &self
) -> Result<(Vec<(SourceTimestamp, AssignedTimestamp)>, Vec<((SourceTimestamp, AssignedTimestamp), Diff)>), Error>

Restores timestamp bindings from the given StreamReadHandle by reading the differential timestamp updates and materializing/consolidating them into a Vec. This basically sums up the diff of the bindings and collects those bindings whose diff is 1.

This returns two Vecs. The first contains the valid bindings, the second one contains retractions for the bindings that are beyond the upper_seal_ts which must be applied/emitted before emitting any new bindings updates.

NOTE: Consolidated bindings with a diff other than 0 or 1 indicate a bug, and this method panics if that case happens.

fn get_starting_offsets<'a>(
    &self,
    bindings: impl Iterator<Item = &'a (SourceTimestamp, AssignedTimestamp)>
) -> HashMap<PartitionId, MzOffset>

fn render_persistence_operators<G>(
    &self,
    source_name: String,
    ts_bindings_stream: Stream<G, ((SourceTimestamp, AssignedTimestamp), Timestamp, Diff)>
) -> (Stream<G, ((SourceTimestamp, AssignedTimestamp), Timestamp, Diff)>, Stream<G, SourceError>) where
    G: Scope<Timestamp = Timestamp>, 

Renders operators that persist the given ts_bindings_stream to the configured persistent collection.

This does not seal the persistent collection, calling code must ensure that this happens eventually.