Struct SimpleAggExecutor

pub struct SimpleAggExecutor<S: StateStore> {
    input: Executor,
    inner: ExecutorInner<S>,
}

SimpleAggExecutor is the aggregation operator for streaming system. To create an aggregation operator, states and expressions should be passed along the constructor.

SimpleAggExecutor maintain multiple states together. If there are n states and n expressions, there will be n columns as output.

As the engine processes data in chunks, it is possible that multiple update messages could consolidate to a single row update. For example, our source emits 1000 inserts in one chunk, and we aggregates count function on that. Current SimpleAggExecutor will only emit one row for a whole chunk. Therefore, we “automatically” implemented a window function inside SimpleAggExecutor.

Fields

input: Executor

inner: ExecutorInner<S>

Implementations

impl<S: StateStore> SimpleAggExecutor<S>

pub fn new( args: AggExecutorArgs<S, SimpleAggExecutorExtraArgs>, ) -> StreamResult<Self>

async fn apply_chunk( this: &mut ExecutorInner<S>, vars: &mut ExecutionVars<S>, chunk: StreamChunk, ) -> StreamExecutorResult<()>

async fn flush_data( this: &mut ExecutorInner<S>, vars: &mut ExecutionVars<S>, epoch: EpochPair, ) -> StreamExecutorResult<Option<StreamChunk>>

async fn try_flush_data(this: &mut ExecutorInner<S>) -> StreamExecutorResult<()>

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

Trait Implementations

impl<S: StateStore> Execute for SimpleAggExecutor<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,