Struct LookupExecutor

pub struct LookupExecutor<S: StateStore> {
    ctx: ActorContextRef,
    chunk_data_types: Vec<DataType>,
    arrangement: ArrangeJoinSide<S>,
    stream: StreamJoinSide,
    arrangement_executor: Option<Executor>,
    stream_executor: Option<Executor>,
    last_barrier: Option<Barrier>,
    column_mapping: Vec<usize>,
    key_indices_mapping: Vec<usize>,
    lookup_cache: LookupCache,
    chunk_size: usize,
}

LookupExecutor takes one input stream and one arrangement. It joins the input stream with the arrangement. Currently, it only supports inner join. See LookupExecutorParams for more information.

The output schema is | stream columns | arrangement columns |. The input is required to be first stream and then arrangement.

Fields

ctx: ActorContextRef

chunk_data_types: Vec<DataType>

the data types of the produced data chunk inside lookup (before reordering)

arrangement: ArrangeJoinSide<S>

The join side of the arrangement

stream: StreamJoinSide

The join side of the stream

arrangement_executor: Option<Executor>

The executor for arrangement.

stream_executor: Option<Executor>

The executor for stream.

last_barrier: Option<Barrier>

The last received barrier.

column_mapping: Vec<usize>

Information of column reordering

key_indices_mapping: Vec<usize>

When we receive a row from the stream side, we will first convert it to join key, and then map it to arrange side. For example, if we receive [a, b, c] from the stream side, where: stream_join_key = [1, 2], arrange_join_key = [3, 2] with order rules [2 ascending, 3 ascending].

• We will first extract join key [b, c],
• then map it to the order of arrangement join key [c, b].

This vector records such mapping.

lookup_cache: LookupCache

The cache for arrangement side.

chunk_size: usize

The maximum size of the chunk produced by executor at a time.

Implementations

impl<S: StateStore> LookupExecutor<S>

pub fn new(params: LookupExecutorParams<S>) -> Self

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

Try produce one stream message from LookupExecutor. If there’s no message to produce, it will return None, and the next function of LookupExecutor will continuously polling messages until there’s one.

If we can use async_stream to write this part, things could be easier.

fn process_barrier(&mut self, barrier: &Barrier)

Process the barrier and apply changes if necessary.

async fn lookup_one_row( &mut self, stream_row: &RowRef<'_>, epoch_pair: EpochPair, metrics: &LookupExecutorMetrics, ) -> StreamExecutorResult<Vec<OwnedRow>>

Lookup all rows corresponding to a join key in shared buffer.

Trait Implementations

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

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,