Struct UpstreamBuffer

struct UpstreamBuffer<'a, S> {
    upstream: &'a mut MergeExecutorInput,
    max_pending_checkpoint_barrier_num: usize,
    pending_non_checkpoint_barriers: Vec<DispatcherBarrier>,
    upstream_pending_barriers: VecDeque<Vec<DispatcherBarrier>>,
    is_polling_epoch_data: bool,
    consume_upstream_row_count: LabelGuardedIntCounter<3>,
    _phase: S,
}

Fields

upstream: &'a mut MergeExecutorInput

max_pending_checkpoint_barrier_num: usize

pending_non_checkpoint_barriers: Vec<DispatcherBarrier>

upstream_pending_barriers: VecDeque<Vec<DispatcherBarrier>>

Barriers received from upstream but not yet received the barrier from local barrier worker.

In the outer VecDeque, newer barriers at the front. In the inner Vec, newer barrier at the back, with the last barrier as checkpoint barrier, and others as non-checkpoint barrier

is_polling_epoch_data: bool

Whether we have started polling any upstream data before the next barrier. When true, we should continue polling until the next barrier, because some data in this epoch have been discarded and data in this epoch must be read from log store

consume_upstream_row_count: LabelGuardedIntCounter<3>

_phase: S

Implementations

impl<'a> UpstreamBuffer<'a, ConsumingSnapshot>

fn new( upstream: &'a mut MergeExecutorInput, consume_upstream_row_count: LabelGuardedIntCounter<3>, ) -> Self

fn start_consuming_log_store(self) -> UpstreamBuffer<'a, ConsumingLogStore>

impl<'a, S> UpstreamBuffer<'a, S>

async fn concurrently_consume_upstream(&mut self) -> StreamExecutorError

async fn consume_until_next_checkpoint_barrier( &mut self, ) -> StreamExecutorResult<Vec<DispatcherBarrier>>

Consume the upstream until seeing the next barrier. pending_barriers must be non-empty after this method returns.

impl<'a> UpstreamBuffer<'a, ConsumingLogStore>

async fn next_checkpoint_barrier( &mut self, ) -> StreamExecutorResult<Option<Vec<DispatcherBarrier>>>

impl<'a, S> UpstreamBuffer<'a, S>

async fn run_future<T, E: Into<StreamExecutorError>>( &mut self, future: impl Future<Output = Result<T, E>>, ) -> StreamExecutorResult<T>

Run a future while concurrently polling the upstream so that the upstream won’t be back-pressured.

fn barrier_count(&self) -> usize