Windowed stream aggregation

Deep dive

How it works

Parts:

• assigner — ts to slice start floor(ts/slide)*slide; window [w, w+size) = size/slide consecutive slices.
• slice store — ordered map (key, slice start) -> accumulator with an associative merge.
• watermark — max_ts - bound, never decreasing: the claim that older events are done (watermarks).
• timers — ordered sets fire (window end) and expire (slice end + lateness), popped as the watermark advances.
• side output — events past allowed lateness, e.g. to a dead-letter queue.

1. Compute the slice; if slice + size + lateness <= watermark, divert the event and stop.
2. Fold into the slice accumulator; a new slice arms fire timers for unfired windows and one expire timer.
3. Windows containing it that already fired, within lateness, emit a refined pane (accumulating mode).
4. Raise max_ts; if max_ts - bound exceeds the watermark, advance it.
5. Pop fire timers with end <= watermark: merge the window's slices, emit the on-time pane.
6. Pop expire timers: delete slices no remaining window uses.
7. Checkpoint slices, timers, watermark and input offsets together (below).

events (unordered) -> slice(key, ts/slide) -> [acc][acc][acc][acc] ... per key
                                                \_________/ window = merge of size/slide slices
 watermark = max_ts - bound ---> pop fire timers (end<=wm) -> emit pane
                             ---> pop expire timers (end+lateness<=wm) -> free slices

Session windows: each event opens [ts, ts+gap), overlapping ones merge (Flink needs a merging function). Exactly once: snapshot slices, timers, watermark and source offsets at a consistent cut (Chandy-Lamport style; Flink's asynchronous barrier snapshots), restore all after failure and replay from the offsets; panes may repeat unless the sink is transactional or idempotent (exactly-once processing, consumer offsets).

Code

use std::collections::{BTreeMap, BTreeSet};

#[derive(Clone, Copy, Default, PartialEq, Debug)]
pub struct Acc { pub sum: i64, pub count: u64 }
impl Acc {
    fn merge(&mut self, o: Acc) { self.sum += o.sum; self.count += o.count; }
}

#[derive(Clone, Copy, Debug)]
pub struct Event { pub key: u64, pub ts: i64, pub val: i64 }

#[derive(Debug, PartialEq)]
pub enum Pane { OnTime, Late }
#[derive(Debug, PartialEq)]
pub struct Out { pub key: u64, pub start: i64, pub end: i64, pub acc: Acc, pub pane: Pane }

/// Hopping window [w, w+size), w a multiple of `slide`; tumbling when size == slide.
/// State: one accumulator per (key, slice of width `slide`); a window merges size/slide slices.
pub struct Windower {
    size: i64, slide: i64, bound: i64, lateness: i64,
    slices: BTreeMap<(u64, i64), Acc>,
    fire: BTreeSet<(i64, u64, i64)>,   // timers (window end, key, window start)
    expire: BTreeSet<(i64, u64, i64)>, // (slice start + size + lateness, key, slice start)
    max_ts: i64, wm: i64,
    pub dropped: Vec<Event>,           // later than allowed lateness
}

impl Windower {
    pub fn new(size: i64, slide: i64, bound: i64, lateness: i64) -> Self {
        assert!(slide > 0 && size % slide == 0);
        Windower { size, slide, bound, lateness, slices: BTreeMap::new(),
                   fire: BTreeSet::new(), expire: BTreeSet::new(),
                   max_ts: i64::MIN / 2, wm: i64::MIN / 2, dropped: vec![] }
    }
    pub fn watermark(&self) -> i64 { self.wm }
    pub fn slice_count(&self) -> usize { self.slices.len() }

    fn pane(&self, key: u64, w: i64, pane: Pane) -> Out {
        let mut acc = Acc::default();
        for (_, a) in self.slices.range((key, w)..(key, w + self.size)) { acc.merge(*a); }
        Out { key, start: w, end: w + self.size, acc, pane }
    }

    pub fn ingest(&mut self, e: Event, out: &mut Vec<Out>) {
        let s = e.ts.div_euclid(self.slide) * self.slide; // slice start
        if s + self.size + self.lateness <= self.wm { self.dropped.push(e); return; }
        let a = self.slices.entry((e.key, s)).or_insert_with(|| {
            // new slice: arm timers for its windows that have not fired
            for j in 0..self.size / self.slide {
                let w = s - j * self.slide;
                if w + self.size > self.wm { self.fire.insert((w + self.size, e.key, w)); }
            }
            self.expire.insert((s + self.size + self.lateness, e.key, s));
            Acc::default()
        });
        a.sum += e.val; a.count += 1;
        // already-fired windows inside allowed lateness: refined pane
        for j in 0..self.size / self.slide {
            let end = s - j * self.slide + self.size;
            if end <= self.wm && end + self.lateness > self.wm {
                out.push(self.pane(e.key, end - self.size, Pane::Late));
            }
        }
        self.max_ts = self.max_ts.max(e.ts);
        if self.max_ts - self.bound > self.wm { self.advance(self.max_ts - self.bound, out); }
    }

    /// End of input: watermark to infinity.
    pub fn flush(&mut self, out: &mut Vec<Out>) { self.advance(i64::MAX / 2, out); }

    fn advance(&mut self, wm: i64, out: &mut Vec<Out>) {
        self.wm = wm;
        while let Some(&(end, key, w)) = self.fire.first() {
            if end > wm { break; }
            self.fire.pop_first();
            out.push(self.pane(key, w, Pane::OnTime));
        }
        while let Some(&(t, key, s)) = self.expire.first() {
            if t > wm { break; }
            self.expire.pop_first();
            self.slices.remove(&(key, s));
        }
    }
}

Run here: for size/slide 10/10, 60/10, 60/1, 7/7, 20,000 out-of-order events matched a brute-force reference exactly, in end order, none dropped, all slices freed; a late event inside lateness gave a refined pane, one past it landed in dropped.

Complexity and constants

• Update: one BTreeMap entry, O(log slices). Emit: size/slide lookups per window; on sparse keys output is size/slide panes per event, so keep slide coarse.
• State: 32 B payload per slice entry here (16 B key, 16 B accumulator) plus map overhead; buffering costs 24 B per event times size/slide copies.
• Measured (Apple M4 Pro, one thread, rustc 1.99.1, 3M events, bound 200; a repeat matched), slices vs one accumulator per window:
  • 6000/6000 tumbling, 10 keys: 55 vs 93 M ev/s (slicing loses).
  • 6000/100 (60 slices), 10 keys, 10 events/key/slice: 5.3 vs 1.8 M ev/s, state 620 vs 630 entries.
  • 60000/100 (600 slices), same load: 0.5 vs 0.2 M ev/s.
  • 6000/100, 1000 keys (0.1 events/key/slice): 0.2 vs 0.4 M ev/s, but state 5,923 vs 53,267 entries.

Slicing wins on dense keys (about 3x), saves state only on sparse keys, and loses at about one event per slice or in tumbling shapes.

Efficiency

• Cost: one ordered-map update per event, order of 20-200 ns here on an M4 Pro (5-55 M ev/s dense); bound is memory latency (real engines add serialization and checkpoint I/O).
• Levers:
  • incremental combiner, not buffering — O(1) state per window instead of per event (Flink docs) — any associative aggregate
  • slices/panes — 1 update instead of size/slide, about 3x measured at 60 slices (this page; Li 2005, Cutty, Scotty) — dense keys, overlapping windows
  • pre-aggregate before the shuffle — one partial per key per slice on the wire (data shuffle) — high-volume keys
  • bounded lateness — slices live only until end + lateness (Beam, Flink docs) — set from measured delay percentiles
  • incremental checkpoints — only changed state written (Flink docs) — large RocksDB state; interval trades replay time for overhead
• Naive → standard → state of the art: buffer events, aggregate at emit · state and work grow with events x overlap; one accumulator per window (Flink Reduce/AggregateFunction, Kafka Streams) · size/slide updates per event; shared slices (this page, Scotty) · one update per event, merge at emit, also for out-of-order input and user-defined windows.

Implementation notes

• Slice width gcd(size, slide) (the pane idea); queries sharing a stream can share one store.
• Heap state (Flink HashMapStateBackend) is fastest, memory-bound; RocksDB state is serialized, disk-bound, lower throughput (Flink docs). Key slices as (key, slice_start) so a window is one range scan.
• Write the sink once per checkpoint, not per pane. With several inputs the watermark is their minimum; mark idle ones idle.

Pitfalls

• bound too small drops events; too large delays every result by bound. One idle partition pins the minimum watermark.
• Late panes update an earlier result (Flink docs): downstream must upsert by (key, window) or consume retractions (Dataflow model).
• Client clocks skew event times; clamp them, or see hybrid logical clocks.
• Dropped late events skew totals silently: alert on dropped.

Variants and alternatives

In production

Apache Flink, Apache Beam (including Google Cloud Dataflow), Kafka Streams and RisingWave materialized views over TUMBLE/HOP.

Implementations

Sources

• Tyler Akidau et al., The Dataflow Model, PVLDB 8(12) 2015. paper
• Jin Li, David Maier, Kristin Tufte, Vassilis Papadimos, Peter Tucker, No Pane, No Gain, SIGMOD Record 34(1) 2005. doi
• Paris Carbone et al., Cutty: Aggregate Sharing for User-Defined Windows, CIKM 2016. doi
• Jonas Traub et al., Scotty: Efficient Window Aggregation for Out-of-Order Stream Processing, ICDE 2018. doi · code
• Kanat Tangwongsan, Martin Hirzel, Scott Schneider, Kun-Lung Wu, General Incremental Sliding-Window Aggregation, PVLDB 8(7) 2015. doi
• Paris Carbone et al., Lightweight Asynchronous Snapshots for Distributed Dataflows, 2015. arXiv
• Docs: Flink windows · Flink state backends · Kafka Streams DSL · Beam windowing