Struct ArrangementBackfillExecutor

pub struct ArrangementBackfillExecutor<S: StateStore, SD: ValueRowSerde> {
    upstream_table: ReplicatedStateTable<S, SD>,
    upstream: Executor,
    state_table: StateTable<S>,
    output_indices: Vec<usize>,
    progress: CreateMviewProgressReporter,
    actor_id: ActorId,
    metrics: Arc<StreamingMetrics>,
    chunk_size: usize,
    rate_limit: Option<usize>,
}

Similar to super::no_shuffle_backfill::BackfillExecutor. Main differences:

• ArrangementBackfillExecutor can reside on a different CN, so it can be scaled independently.
• To synchronize upstream shared buffer, it is initialized with a ReplicatedStateTable.

Fields

upstream_table: ReplicatedStateTable<S, SD>

Upstream table

upstream: Executor

Upstream with the same schema with the upstream table.

state_table: 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

actor_id: ActorId

metrics: Arc<StreamingMetrics>

chunk_size: usize

rate_limit: Option<usize>

Implementations

impl<S, SD> ArrangementBackfillExecutor<S, SD>

where S: StateStore, SD: ValueRowSerde,

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

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

fn make_snapshot_stream<'a>( upstream_table: &'a ReplicatedStateTable<S, SD>, backfill_state: BackfillState, paused: bool, rate_limiter: &'a Option<RateLimiter<NotKeyed, InMemoryState, MonotonicClock, NoOpMiddleware<Instant>>>, ) -> impl Stream<Item = Result<Option<(VirtualNode, OwnedRow)>, StreamExecutorError>> + 'a

fn handle_snapshot_chunk( chunk: DataChunk, vnode: VirtualNode, pk_in_output_indices: &[usize], backfill_state: &mut BackfillState, cur_barrier_snapshot_processed_rows: &mut u64, total_snapshot_processed_rows: &mut u64, output_indices: &[usize], ) -> StreamExecutorResult<StreamChunk>

fn snapshot_read_per_vnode<'_async0>( upstream_table: &'_async0 ReplicatedStateTable<S, SD>, backfill_state: BackfillState, ) -> impl Stream<Item = Result<Option<(VirtualNode, OwnedRow)>, StreamExecutorError>> + '_async0

Read snapshot per vnode. These streams should be sorted in storage layer.

1. Get row iterator / vnode.
2. Merge it with select_all.
3. Change it into a chunk iterator with iter_chunks. This means it should fetch a row from each iterator to form a chunk.

We interleave at chunk per vnode level rather than rows. This is so that we can compute current_pos once per chunk, since they correspond to 1 vnode.

The stream contains pairs of (VirtualNode, StreamChunk). The VirtualNode is the vnode that the chunk belongs to. The StreamChunk is the chunk that contains the rows from the vnode. If it’s None, it means the vnode has no more rows for this snapshot read.

The snapshot_read_epoch is supplied as a parameter for state_table. It is required to ensure we read a fully-checkpointed snapshot the first time.

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.

Trait Implementations

impl<S, SD> Execute for ArrangementBackfillExecutor<S, SD>

where S: StateStore, SD: ValueRowSerde,

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,