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, and skip the rows that were output, see Self::skip_output_rows.
• Windows that may be closed by the watermark of the order key are closed by WatermarkTriggers.

Fields

input: Executor

inner: ExecutorInner<S>

Implementations

impl<S: StateStore> EowcOverWindowExecutor<S>

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

fn restore_intermediate_state( this: &ExecutorInner<S>, partition: &mut Partition, partition_key: &OwnedRow, row: Option<OwnedRow>, ) -> StreamExecutorResult<()>

Restore intermediate state snapshots and the output progress of the partition from its row in the intermediate state table.

fn persist_partition( this: &mut ExecutorInner<S>, partition: &mut Partition, partition_key: &OwnedRow, )

Persist intermediate state snapshots and the output progress to the state table, if changed.

async fn load_partition( this: &ExecutorInner<S>, partition_key: &OwnedRow, ) -> StreamExecutorResult<Partition>

Load the partition from the state tables. The rows that were output are skipped by Self::skip_output_rows afterwards.

fn skip_output_rows( this: &mut ExecutorInner<S>, partition: &mut Partition, partition_key: &OwnedRow, ) -> StreamExecutorResult<()>

Skip the rows of a loaded partition that were output. The rows not needed any more are evicted, as they were retained when output without the later rows.

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

async fn ensure_partitions_in_cache( this: &mut ExecutorInner<S>, cache: &mut ManagedLruCache<MemcmpEncoded, Partition>, partitions: &[(MemcmpEncoded, &OwnedRow)], ) -> StreamExecutorResult<()>

Load the partitions not in the cache concurrently, and put them into the cache.

fn reset_partition( this: &ExecutorInner<S>, partition: &mut Partition, ) -> StreamExecutorResult<()>

Rebuild a partition whose states are empty, as if it’s reloaded. The window states may hold memory for the closed windows that their estimated sizes don’t count.

fn evict( this: &mut ExecutorInner<S>, partition_key: &OwnedRow, evict_hint: StateEvictHint, ) -> StreamExecutorResult<()>

Delete the evicted rows from the state table.

fn output_ready_rows( this: &mut ExecutorInner<S>, partition: &mut Partition, partition_key: &OwnedRow, chunk_builder: &mut StreamChunkBuilder, chunks: &mut Vec<StreamChunk>, ) -> StreamExecutorResult<Readiness>

Output the rows of the partition whose windows are closed, under the current watermark of the order key if windows may be closed by watermark. Returns the readiness of the next row.

fn register_trigger( this: &mut ExecutorInner<S>, partition: &mut Partition, partition_key: &OwnedRow, readiness: Readiness, )

Register a trigger for the partition if the window of its next row to output, whose readiness is given, may be closed by watermark, unless one no later than its bound is registered.

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

fn apply_watermark<'a>( this: &'a mut ExecutorInner<S>, vars: &'a mut ExecutionVars<S>, watermark: ScalarImpl, ) -> impl Stream<Item = Result<StreamChunk, StreamExecutorError>> + 'a

Output the rows whose windows are closed by the new watermark of the order key.

fn fire_trigger( this: &mut ExecutorInner<S>, partition: &mut Partition, entry: &TriggerEntry, chunk_builder: &mut StreamChunkBuilder, chunks: &mut Vec<StreamChunk>, ) -> StreamExecutorResult<()>

Handle the partition whose trigger entry is satisfied by the watermark: output the rows whose windows are closed, and register a trigger again if needed.

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

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 boxed(self) -> Box<dyn Execute>

where Self: Sized + Send + 'static,