Struct LocalityProviderExecutor 

pub struct LocalityProviderExecutor<S: StateStore> {

    upstream: Executor,
    locality_columns: Vec<usize>,
    state_table: StateTable<S>,
    progress_table: StateTable<S>,
    sort_buffer_table: Option<StateTable<S>>,
    sort_buffer_settings: SortBufferSettings,
    input_schema: Schema,
    progress: CreateMviewProgressReporter,
    fragment_id: FragmentId,
    actor_id: ActorId,
    metrics: Arc<StreamingMetrics>,
    chunk_size: usize,
    rate_limiter: MonitoredRateLimiter,
}

The LocalityProviderExecutor provides locality for operators during backfilling. It buffers input data into a state table using locality columns as primary key prefix.

The executor implements a proper backfill process similar to arrangement backfill:

1. Backfill phase: Buffer incoming data and provide locality-ordered snapshot reads
2. Completion transition: Drain buffered input and forward new input directly until the barrier that persists backfill completion
3. Forward phase: Optionally sort amplified epochs after that completion barrier

The sort buffer intentionally does not cover the completion transition. Input drained when the snapshot reaches EOF, or received while waiting for the completion barrier, is forwarded in arrival order. Only input after that barrier is eligible for sorting.

Key improvements over the original implementation:

• Removes arbitrary barrier buffer limit
• Implements proper upstream chunk tracking during backfill
• Uses per-vnode progress tracking for better state management

Fields

upstream: Executor

Upstream input

locality_columns: Vec<usize>

Locality columns (indices in input schema)

state_table: StateTable<S>

State table for buffering input data

progress_table: StateTable<S>

Progress table for tracking backfill progress per vnode

sort_buffer_table: Option<StateTable<S>>

Ephemeral sorted log table for buffering and re-ordering amplified input within an epoch after backfill is finished. None if the feature was not enabled at plan time.

sort_buffer_settings: SortBufferSettings

Runtime settings for the sort buffer.

input_schema: Schema

progress: CreateMviewProgressReporter

Progress reporter for materialized view creation

fragment_id: FragmentId

actor_id: ActorId

metrics: Arc<StreamingMetrics>

Metrics

chunk_size: usize

Chunk size for output

rate_limiter: MonitoredRateLimiter

Implementations

impl<S: StateStore> LocalityProviderExecutor<S>

pub fn new( upstream: Executor, locality_columns: Vec<usize>, state_table: StateTable<S>, progress_table: StateTable<S>, sort_buffer_table: Option<StateTable<S>>, sort_buffer_settings: SortBufferSettings, input_schema: Schema, progress: CreateMviewProgressReporter, metrics: Arc<StreamingMetrics>, chunk_size: usize, fragment_id: FragmentId, rate_limit: RateLimit, ) -> Self

fn apply_throttle( rate_limiter: &MonitoredRateLimiter, fragment_id: FragmentId, barrier: &Barrier, ) -> Option<RateLimit>

Returns the new rate limit if it changed.

fn make_snapshot_stream<'a>( reader: FlushedStateTableReader<S>, backfill_state: LocalityBackfillState, rate_limiter: &'a MonitoredRateLimiter, ) -> impl Stream<Item = Result<(VirtualNode, OwnedRow), StreamExecutorError>> + 'a

Creates a snapshot stream that reads from state table in locality order

async fn persist_backfill_state( progress_table: &mut StateTable<S>, backfill_state: &LocalityBackfillState, ) -> StreamExecutorResult<()>

Persist backfill state to progress table

async fn load_backfill_state( progress_table: &StateTable<S>, ) -> StreamExecutorResult<LocalityBackfillState>

Load backfill state from progress table

fn mark_chunk( chunk: StreamChunk, backfill_state: &LocalityBackfillState, state_table: &StateTable<S>, ) -> StreamExecutorResult<StreamChunk>

Mark chunk for forwarding based on backfill progress

fn handle_snapshot_chunk( data_chunk: DataChunk, vnode: VirtualNode, pk_indices: &[usize], backfill_state: &mut LocalityBackfillState, cur_barrier_snapshot_processed_rows: &mut u64, ) -> StreamExecutorResult<StreamChunk>

fn write_chunk_to_sort_buffer( sort_buffer_table: &mut StateTable<S>, chunk: &StreamChunk, generation: i64, seq: &mut i64, )

Write a compacted input chunk into the sort buffer table as an append-only log. Each row is extended with gen (the current epoch generation), op (normalized op code) and seq (monotonic sequence number within the epoch), all writes using Op::Insert with unique keys.

fn make_sort_buffer_replay_stream( reader: FlushedStateTableReader<S>, generation: i64, group_key_indices: Vec<usize>, input_len: usize, data_types: Vec<DataType>, chunk_size: usize, ) -> impl Stream<Item = Result<StreamChunk, StreamExecutorError>>

Replay this epoch’s buffered changes from the sort buffer table in locality order, compacting consecutive changes of the same key (locality columns + stream key) into at most one record.

Must be called after the sort buffer table has been committed for the epoch, so that all buffered rows are visible to the flushed snapshot reader (staging imms + spilled SSTs).

impl<S: StateStore> LocalityProviderExecutor<S>

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

Trait Implementations

impl<S: StateStore> Execute for LocalityProviderExecutor<S>

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

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

where Self: Sized + Send + 'static,