Struct BackfillExecutor 

pub struct BackfillExecutor<S: StateStore> {
    upstream_table: BatchTable<S>,
    upstream: Executor,
    state_table: Option<StateTable<S>>,
    output_indices: Vec<usize>,
    progress: CreateMviewProgressReporter,
    actor_id: ActorId,
    metrics: Arc<StreamingMetrics>,
    chunk_size: usize,
    rate_limiter: MonitoredRateLimiter,
    fragment_id: FragmentId,
}

An implementation of the RFC: Use Backfill To Let Mv On Mv Stream Again. BackfillExecutor is used to create a materialized view on another materialized view.

It can only buffer chunks between two barriers instead of unbundled memory usage of RearrangedChainExecutor.

It uses the latest epoch to read the snapshot of the upstream mv during two barriers and all the StreamChunk of the snapshot read will forward to the downstream.

It uses current_pos to record the progress of the backfill (the pk of the upstream mv) and current_pos is initiated as an empty Row.

All upstream messages during the two barriers interval will be buffered and decide to forward or ignore based on the current_pos at the end of the later barrier. Once current_pos reaches the end of the upstream mv pk, the backfill would finish.

Notice: The pk we are talking about here refers to the storage primary key. We rely on the scheduler to schedule the BackfillExecutor together with the upstream mv/table in the same worker, so that we can read uncommitted data from the upstream table without waiting.

Fields

upstream_table: BatchTable<S>

Upstream table

upstream: Executor

Upstream with the same schema with the upstream table.

state_table: Option<StateTable<S>>

Internal state table for persisting state of backfill state.

output_indices: Vec<usize>

The column indices need to be forwarded to the downstream from the upstream and table scan.

progress: CreateMviewProgressReporter

PTAL at the docstring for CreateMviewProgress to understand how we compute it.

actor_id: ActorId

metrics: Arc<StreamingMetrics>

chunk_size: usize

rate_limiter: MonitoredRateLimiter

fragment_id: FragmentId

Fragment id of the fragment this backfill node belongs to.

Implementations

impl<S> BackfillExecutor<S>

where S: StateStore,

pub fn new( upstream_table: BatchTable<S>, upstream: Executor, state_table: Option<StateTable<S>>, output_indices: Vec<usize>, progress: CreateMviewProgressReporter, metrics: Arc<StreamingMetrics>, chunk_size: usize, rate_limit: RateLimit, fragment_id: FragmentId, ) -> Self

fn execute_inner( self, ) -> impl Stream<Item = Result<Message, StreamExecutorError>>

async fn recover_backfill_state( state_table: Option<&StateTable<S>>, pk_len: usize, ) -> StreamExecutorResult<BackfillState>

fn deserialize_backfill_state( row: Option<OwnedRow>, pk_len: usize, ) -> BackfillState

fn make_snapshot_stream<'a>( upstream_table: &'a BatchTable<S>, epoch: u64, current_pos: Option<OwnedRow>, paused: bool, rate_limiter: &'a MonitoredRateLimiter, ) -> impl Stream<Item = Result<Option<OwnedRow>, StreamExecutorError>> + 'a

pub fn snapshot_read<'_async0>( upstream_table: &'_async0 BatchTable<S>, epoch: HummockReadEpoch, current_pos: Option<OwnedRow>, ) -> impl Stream<Item = Result<OwnedRow, StreamExecutorError>> + '_async0

Snapshot read the upstream mv. The rows from upstream snapshot read will be buffered inside the builder. If snapshot is dropped before its rows are consumed, remaining data in builder must be flushed manually. Otherwise when we scan a new snapshot, it is possible the rows in the builder would be present, Then when we flush we contain duplicate rows.

async fn persist_state( epoch: EpochPair, table: &mut Option<StateTable<S>>, is_finished: bool, current_pos: &Option<OwnedRow>, row_count: u64, old_state: &mut Option<Vec<Datum>>, current_state: &mut [Datum], ) -> StreamExecutorResult<()>

fn handle_snapshot_chunk( data_chunk: DataChunk, current_pos: &mut Option<OwnedRow>, cur_barrier_snapshot_processed_rows: &mut u64, total_snapshot_processed_rows: &mut u64, pk_indices: &[usize], output_indices: &[usize], ) -> StreamChunk

1. Converts from data chunk to stream chunk.
2. Update the current position.
3. Update Metrics
4. Map the chunk according to output indices, return the stream chunk and do wrapping outside.

Trait Implementations

impl<S> Execute for BackfillExecutor<S>

where S: StateStore,

fn execute(self: Box<Self>) -> BoxedMessageStream

fn execute_with_epoch(self: Box<Self>, _epoch: u64) -> BoxedMessageStream

fn boxed(self) -> Box<dyn Execute>

where Self: Sized + Send + 'static,