Struct SourceBackfillExecutorInner

pub struct SourceBackfillExecutorInner<S: StateStore> {
    actor_ctx: ActorContextRef,
    info: ExecutorInfo,
    stream_source_core: StreamSourceCore<S>,
    backfill_state_store: BackfillStateTableHandler<S>,
    metrics: Arc<StreamingMetrics>,
    source_split_change_count: LabelGuardedIntCounter<4>,
    system_params: SystemParamsReaderRef,
    rate_limit_rps: Option<u32>,
    progress: CreateMviewProgressReporter,
}

Fields

actor_ctx: ActorContextRef

info: ExecutorInfo

stream_source_core: StreamSourceCore<S>

Streaming source for external

backfill_state_store: BackfillStateTableHandler<S>

metrics: Arc<StreamingMetrics>

Metrics for monitor.

source_split_change_count: LabelGuardedIntCounter<4>

system_params: SystemParamsReaderRef

System parameter reader to read barrier interval

rate_limit_rps: Option<u32>

Rate limit in rows/s.

progress: CreateMviewProgressReporter

Implementations

impl<S: StateStore> SourceBackfillExecutorInner<S>

pub fn new( actor_ctx: ActorContextRef, info: ExecutorInfo, stream_source_core: StreamSourceCore<S>, metrics: Arc<StreamingMetrics>, system_params: SystemParamsReaderRef, backfill_state_store: BackfillStateTableHandler<S>, rate_limit_rps: Option<u32>, progress: CreateMviewProgressReporter, ) -> Self

async fn build_stream_source_reader( &self, source_desc: &SourceDesc, splits: Vec<SplitImpl>, ) -> StreamExecutorResult<(BoxChunkSourceStream, HashMap<SplitId, BackfillInfo>)>

fn execute( self, input: Executor, ) -> impl Stream<Item = Result<Message, StreamExecutorError>>

fn should_report_finished(&self, states: &BackfillStates) -> bool

When we should call progress.finish() to let blocking DDL return. We report as soon as SourceCachingUp. Otherwise the DDL might be blocked forever until upstream messages come.

Note: split migration (online scaling) is related with progress tracking.

• For foreground DDL, scaling is not allowed before progress is finished.
• For background DDL, scaling is skipped when progress is not finished, and can be triggered by recreating actors during recovery.

See https://github.com/risingwavelabs/risingwave/issues/18300 for more details.

async fn backfill_finished( &self, states: &BackfillStates, ) -> StreamExecutorResult<bool>

All splits entered Finished state.

We check all splits for the source, including other actors’ splits here, before going to the forward stage. Otherwise if we break early, but after rescheduling, an unfinished split is migrated to this actor, we still need to backfill it.

Note: at the beginning, the actor will only read the state written by itself. It needs to wait until it can read all actors’ written data. i.e., wait for the first checkpoint has been available.

See https://github.com/risingwavelabs/risingwave/issues/18300 for more details.

async fn apply_split_change( &mut self, split_assignment: &HashMap<ActorId, Vec<SplitImpl>>, stage: &mut BackfillStage, should_trim_state: bool, ) -> StreamExecutorResult<bool>

For newly added splits, we do not need to backfill and can directly forward from upstream.

async fn update_state_if_changed( &mut self, target_splits: Vec<SplitImpl>, stage: &mut BackfillStage, should_trim_state: bool, ) -> StreamExecutorResult<bool>

Returns true if split changed. Otherwise false.

async fn apply_split_change_forward_stage( &mut self, split_assignment: &HashMap<ActorId, Vec<SplitImpl>>, states: &mut BackfillStates, should_trim_state: bool, ) -> StreamExecutorResult<()>

For split change during forward stage, all newly added splits should be already finished.

async fn update_state_if_changed_forward_stage( &mut self, target_splits: Vec<SplitImpl>, states: &mut BackfillStates, should_trim_state: bool, ) -> StreamExecutorResult<()>

Trait Implementations

impl<S: StateStore> Debug for SourceBackfillExecutorInner<S>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.