Struct mz_storage::statistics::AggregatedStatistics

pub struct AggregatedStatistics {
    worker_id: usize,
    worker_count: usize,
    local_source_statistics: BTreeMap<GlobalId, (usize, SourceStatistics)>,
    local_sink_statistics: BTreeMap<GlobalId, (usize, SinkStatistics)>,
    global_source_statistics: BTreeMap<GlobalId, (usize, Vec<Option<SourceStatisticsUpdate>>)>,
    global_sink_statistics: BTreeMap<GlobalId, (usize, Vec<Option<SinkStatisticsUpdate>>)>,
}

A structure that keeps track of local statistics, as well as aggregating statistics into a single worker (currently worker 0).

This is because we ONLY want to emit statistics for gauge-style-statistics when ALL workers have caught up to the currently running instance of a source/sink and have reported an accurate gauge. This prevents issues like 1 worker reporting snapshot_committed=true before the others workers have caught up to report that they haven’t finished snapshotting.

The API flow is:

• Initialize sources/sinks with initialize_source or initialize_sink. This advances the local epoch.
• Advance the global epoch with advance_global_epoch. This should always be called strictly before reinitializing objects.
• Emit a snapshot of local data with emit_local, which can be broadcasted to other workers.
• Ingest data from other workers with ingest.
• Emit a snapshot of global data (i.e. skipping objects that don’t have all workers caught up) with snapshot.

All functions can and should be called from ALL workers.

Note that we hand-roll a statistics epoch here, but in the future, when the storage controller has top-level support for dataflow epochs, this logic can be entirely moved there.

Fields

worker_id: usize

worker_count: usize

local_source_statistics: BTreeMap<GlobalId, (usize, SourceStatistics)>

local_sink_statistics: BTreeMap<GlobalId, (usize, SinkStatistics)>

global_source_statistics: BTreeMap<GlobalId, (usize, Vec<Option<SourceStatisticsUpdate>>)>

global_sink_statistics: BTreeMap<GlobalId, (usize, Vec<Option<SinkStatisticsUpdate>>)>

Implementations

impl AggregatedStatistics

pub fn new(worker_id: usize, worker_count: usize) -> Self

Initialize a new AggregatedStatistics.

pub fn get_source(&self, id: &GlobalId) -> Option<&SourceStatistics>

Get a SourceStatistics for an id, if it exists.

pub fn get_sink(&self, id: &GlobalId) -> Option<&SinkStatistics>

Get a SinkStatistics for an id, if it exists.

pub fn deinitialize(&mut self, id: GlobalId)

Deinitialize an object. Other methods other than initialize_source and initialize_sink will never overwrite this.

pub fn advance_global_epoch(&mut self, id: GlobalId)

Advance the global epoch for statistics.

Gauge values from previous epochs will be ignored. Counter values from previous epochs will still be applied as usual.

pub fn initialize_source<F: FnOnce() -> SourceStatistics>( &mut self, id: GlobalId, resume_upper: Antichain<Timestamp>, stats: F, )

Re-initialize a source. If it already exists, then its local epoch is advanced.

pub fn initialize_sink<F: FnOnce() -> SinkStatistics>( &mut self, id: GlobalId, stats: F, )

Re-initialize a sink. If it already exists, then its local epoch is advanced.

pub fn ingest( &mut self, source_statistics: Vec<(usize, SourceStatisticsRecord)>, sink_statistics: Vec<(usize, SinkStatisticsRecord)>, )

Ingest data from other workers.

fn _emit_local( &mut self, ) -> (Vec<(usize, SourceStatisticsRecord)>, Vec<(usize, SinkStatisticsRecord)>)

pub fn emit_local( &mut self, ) -> (Vec<(usize, SourceStatisticsRecord)>, Vec<(usize, SinkStatisticsRecord)>)

Emit a snapshot of this workers local data.

pub fn snapshot( &mut self, ) -> (Vec<SourceStatisticsUpdate>, Vec<SinkStatisticsUpdate>)

Emit a global snapshot of data. This does not include objects whose workers have not initialized gauges for the current epoch.