Struct CreateMviewProgressReporter

pub struct CreateMviewProgressReporter {
    barrier_manager: LocalBarrierManager,
    backfill_actor_id: ActorId,
    state: Option<BackfillState>,
}

The progress held by the backfill executors to report to the local barrier manager.

Progress can be computed by total_rows_consumed / total_rows_upstream. This yields the (approximate) percentage of rows we are done backfilling.

For total_rows_consumed, the progress is tracked in the following way:

1. Fetching the row count from our state table. This number is the total number, NOT incremental. This is done per actor.
2. Refreshing this number on the meta side, on every barrier. This is done by just summing up all the row counts from the actors.

For total_rows_upstream, this is fetched from HummockVersion’s statistics (TableStats::total_key_count).

This is computed per HummockVersion, which is updated whenever a checkpoint is committed. The total_key_count figure just counts the number of storage keys. For example, if a key is inserted and then deleted, it results two storage entries in LSMt, so count=2. Only after compaction, the count will drop back to 0.

So the total count could be more pessimistic, than actual progress.

It is fine for this number not to be precise, since we don’t rely on it to update the status of a stream job internally.

TODO(kwannoel): Perhaps it is possible to get total key count of the replicated state table for arrangement backfill. We can use that to estimate the progress as well, and avoid recording row_count state for it.

Fields

barrier_manager: LocalBarrierManager

backfill_actor_id: ActorId

The id of the actor containing the backfill executors.

state: Option<BackfillState>

Implementations

impl CreateMviewProgressReporter

pub fn new( barrier_manager: LocalBarrierManager, backfill_actor_id: ActorId, ) -> Self

pub fn actor_id(&self) -> u32

fn update_inner(&mut self, epoch: EpochPair, state: BackfillState)

pub fn update( &mut self, epoch: EpochPair, consumed_epoch: u64, current_consumed_rows: u64, )

Update the progress to ConsumingUpstream(consumed_epoch, consumed_rows). The epoch must be monotonically increasing. current_epoch should be provided to locate the barrier under concurrent checkpoint. current_consumed_rows is an accumulated value.

pub fn update_for_source_backfill( &mut self, epoch: EpochPair, current_consumed_rows: u64, )

The difference from Self::update (MV backfill) is that we don’t care ConsumedEpoch here.

pub fn finish(&mut self, epoch: EpochPair, current_consumed_rows: u64)

Finish the progress. If the progress is already finished, then perform no-op. current_epoch should be provided to locate the barrier under concurrent checkpoint.

pub(crate) fn update_create_mview_log_store_progress( &mut self, epoch: EpochPair, pending_barrier_num: usize, )

pub(crate) fn finish_consuming_log_store(&mut self, epoch: EpochPair)