Struct EowcOverWindowExecutor

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

EowcOverWindowExecutor consumes ordered input (on order key column with watermark in ascending order) and outputs window function results. One EowcOverWindowExecutor can handle one combination of partition key and order key.

The reason not to use SortBuffer is that the table schemas of EowcOverWindowExecutor and SortBuffer are different, since we don’t have something like a grouped sort buffer.

Basic idea:

──────────────┬────────────────────────────────────────────────────── curr evict row
              │ROWS BETWEEN 5 PRECEDING AND 1 PRECEDING
       (1)    │ ─┬─
              │  │RANGE BETWEEN '1hr' PRECEDING AND '1hr' FOLLOWING
       ─┬─    │  │ ─┬─
  LAG(1)│        │  │
────────┴──┬─────┼──┼──────────────────────────────────────────────── curr output row
    LEAD(1)│     │  │GROUPS 1 PRECEDING AND 1 FOLLOWING
                 │
                 │ (2)
─────────────────┴─────────────────────────────────────────────────── curr input row
(1): additional buffered input (unneeded) for some window
(2): additional delay (already able to output) for some window

• State table schema = input schema, state table pk = partition key | order key | input pk.
• Output schema = input schema + window function results.
• Rows in range (curr evict row, curr input row] are in state table.
• curr evict row <= min(last evict rows of all WindowStates).
• WindowState should output agg result for curr output row.
• Recover: iterate through state table, push rows to WindowState, ignore ready windows.

Fields

input: Executor

inner: ExecutorInner<S>

Implementations

impl<S: StateStore> EowcOverWindowExecutor<S>

pub fn new(args: EowcOverWindowExecutorArgs<S>) -> Self

async fn ensure_key_in_cache( this: &ExecutorInner<S>, cache: &mut ManagedLruCache<MemcmpEncoded, Partition>, partition_key: impl Row, encoded_partition_key: &MemcmpEncoded, ) -> StreamExecutorResult<()>

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

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

Trait Implementations

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