Complex event processing

Deep dive

How it works

Parts:

• Pattern compiler — pattern to NFA (states = steps, edges = predicates plus window); a regex engine over typed events with a clock.
• Partitioner — routes by key; all state is per key, so keys shard across cores and nodes.
• Partial-match store — per key, the live runs (stage, start time, captured fields).
• Timer queue — deadlines for window expiry and for negation; heap or timing wheel.
• Event-time buffer — orders events, releases at the watermark.
• Level index — ordered map (or interval tree) for threshold queries.
• Emitter — once per match, re-arm band (alert grouping).

1. Register patterns once; compile to NFAs; threshold queries go into the key's level index.
2. Order the event by event time (buffer until the watermark passes).
3. Fire timers due before it: drop expired partials; finalize trailing negations whose quiet period passed.
4. For its key, test each live partial against its next predicate: advance, kill (a negated event arrived) or ignore (relaxed contiguity).
5. Test the start predicate and spawn a partial. Any-match: every start spawns. Next-match with consumption: one event completes all waiting partials and they are discarded.
6. A partial in its final state emits; "not followed by" emits only when its timer fires.
7. Fire once, then disarm until the value retreats by a band (hysteresis), so a price hovering at X does not alert every tick.

event -> [reorder by watermark] -> timers(<= ts) -> partials[key] -> match -> emitter
RISE --(SPIKE within 60 s)--> CONFIRM --(10 s quiet)--> ALERT
                                  +--(REVERSAL)--> dead

Contiguity: strict (adjacent), relaxed / skip-till-next-match (ignore non-matching events, take the first match), non-deterministic relaxed / skip-till-any-match (all combinations). Flink CEP offers all three; its AfterMatchSkipStrategy (NO_SKIP, SKIP_TO_NEXT, SKIP_PAST_LAST_EVENT, SKIP_TO_FIRST, SKIP_TO_LAST) limits how many matches an event joins. The SQL form is MATCH_RECOGNIZE (row pattern recognition, SQL:2016): PATTERN, DEFINE, MEASURES, AFTER MATCH SKIP.

Code

Per key: rise >= 2 %, volume spike within 60 s, no 1 % reversal within 10 s. Plus a level index for "price crosses above X" with hysteresis. Std-only, compiled and run with asserts.

use std::cmp::Reverse;
use std::collections::{BTreeMap, BinaryHeap, HashMap};

#[derive(Clone, Copy)]
pub struct Tick { pub ts: u64, pub key: u32, pub price: i64, pub vol: u64 } // ms, cents

// RISE (price >= +2 % vs previous tick), then SPIKE (vol >= 1000) within 60 s,
// NOT followed by REVERSAL (price <= -1 % from the spike price) within 10 s.
const WITHIN: u64 = 60_000;
const QUIET: u64 = 10_000;
enum Partial { WaitSpike { expires: u64 }, Confirm { spike_price: i64, until: u64 } }
#[derive(Default)]
struct KeyState { last: Option<i64>, partials: Vec<Partial> }
#[derive(Default)]
pub struct Engine {
    keys: HashMap<u32, KeyState>,
    timers: BinaryHeap<Reverse<(u64, u32)>>, // (deadline, key): expiry without scanning all keys
    pub alerts: Vec<(u32, u64)>,             // (key, deadline at which the match became final)
}
impl Engine {
    /// Time moves to `now`: drop or finalize partials past their deadline.
    pub fn advance(&mut self, now: u64) {
        while let Some(&Reverse((d, k))) = self.timers.peek() {
            if d >= now { break; }
            self.timers.pop();
            let (Some(st), alerts) = (self.keys.get_mut(&k), &mut self.alerts) else { continue };
            st.partials.retain(|p| match *p {
                Partial::WaitSpike { expires } => expires >= now,
                Partial::Confirm { until, .. } if until < now => { alerts.push((k, until)); false }
                _ => true,
            });
        }
    }

    pub fn on_tick(&mut self, t: Tick) {
        self.advance(t.ts);
        let st = self.keys.entry(t.key).or_default();
        // existing partials first, so a tick never completes a partial it creates itself
        st.partials.retain(|p| match *p {
            Partial::Confirm { spike_price, .. } => t.price * 100 > spike_price * 99, // reversal kills it
            _ => true,
        });
        // skip-till-next-match with consumption: a spike completes all waiting partials at once
        if t.vol >= 1000 && st.partials.iter().any(|p| matches!(p, Partial::WaitSpike { .. })) {
            st.partials.retain(|p| !matches!(p, Partial::WaitSpike { .. }));
            st.partials.push(Partial::Confirm { spike_price: t.price, until: t.ts + QUIET });
            self.timers.push(Reverse((t.ts + QUIET, t.key)));
        }
        if matches!(st.last, Some(prev) if t.price * 100 >= prev * 102) {
            st.partials.push(Partial::WaitSpike { expires: t.ts + WITHIN });
            self.timers.push(Reverse((t.ts + WITHIN, t.key)));
        }
        st.last = Some(t.price);
    }
}

/// "Price crosses above X" standing queries for one key, with hysteresis.
#[derive(Default)]
pub struct CrossIndex {
    armed: BTreeMap<i64, Vec<u32>>,    // level -> query ids
    disarmed: BTreeMap<i64, Vec<u32>>, // re-arm level (level - band) -> query ids
    last: Option<i64>,
}
impl CrossIndex {
    pub fn add(&mut self, level: i64, qid: u32) { self.armed.entry(level).or_default().push(qid); }

    /// Fire queries with level in (last, price]; re-arm those whose price fell back by `band`.
    pub fn on_price(&mut self, price: i64, band: i64, fired: &mut Vec<u32>) {
        let prev = self.last.replace(price).unwrap_or(price);
        if price > prev {
            let hit: Vec<i64> = self.armed.range(prev + 1..=price).map(|(l, _)| *l).collect();
            for l in hit {
                let ids = self.armed.remove(&l).unwrap();
                fired.extend(&ids);
                self.disarmed.entry(l - band).or_default().extend(ids);
            }
        } else if price < prev {
            let back: Vec<i64> = self.disarmed.range(price..).map(|(r, _)| *r).collect();
            for r in back {
                for id in self.disarmed.remove(&r).unwrap() { self.armed.entry(r + band).or_default().push(id); }
            }
        }
    }
}
fn main() {
    let tk = |ts, price, vol| Tick { ts, key: 1, price, vol };
    let run = |ticks: &[Tick]| { let mut e = Engine::default(); ticks.iter().for_each(|t| e.on_tick(*t)); e.alerts };
    // match: +2.5 %, spike 20 s later, quiet 10 s -> final at 31 s
    assert_eq!(run(&[tk(0, 10_000, 10), tk(1_000, 10_250, 10), tk(21_000, 10_260, 5_000), tk(40_000, 10_270, 10)]), [(1, 31_000)]);
    // no match: -1.5 % reversal 4 s after the spike
    assert!(run(&[tk(0, 10_000, 10), tk(1_000, 10_250, 10), tk(21_000, 10_260, 5_000), tk(25_000, 10_100, 10), tk(90_000, 10_100, 10)]).is_empty());
    // no match: spike 69 s after the rise
    assert!(run(&[tk(0, 10_000, 10), tk(1_000, 10_250, 10), tk(70_000, 10_250, 9_000), tk(200_000, 10_250, 10)]).is_empty());
    // hysteresis: fire once, re-arm only after a 5-cent retreat below the level
    let (mut ix, mut f) = (CrossIndex::default(), vec![]);
    ix.add(10_100, 7);
    for (p, n) in [(10_000, 0), (10_120, 1), (10_130, 0), (10_098, 0), (10_110, 0), (10_090, 0), (10_105, 1)] {
        f.clear(); ix.on_price(p, 5, &mut f); assert_eq!(f.len(), n, "price {p}");
    }
    // throughput: 10k keys x 100 cross queries, 3M random-walk ticks, rare +3 % jumps
    let (keys, qpk, n) = (10_000u32, 100u32, 3_000_000usize);
    let mut s = 0x9E3779B97F4A7C15u64;
    let mut rnd = move || { s ^= s << 13; s ^= s >> 7; s ^= s << 17; s };
    let mut price = vec![10_000i64; keys as usize];
    let mut idx: Vec<CrossIndex> = (0..keys).map(|_| CrossIndex::default()).collect();
    idx.iter_mut().for_each(|ix| (0..qpk).for_each(|q| ix.add(9_000 + (rnd() % 2_000) as i64, q)));
    let ev: Vec<Tick> = (0..n).map(|i| {
        let k = (rnd() % keys as u64) as u32;
        let p = &mut price[k as usize];
        *p = (*p + (rnd() % 41) as i64 - 20 + if rnd() % 400 == 0 { *p / 30 } else { 0 }).max(1);
        Tick { ts: i as u64 * 2, key: k, price: *p, vol: if rnd() % 50 == 0 { 2000 } else { 10 } }
    }).collect();
    let (mut e, t0) = (Engine::default(), std::time::Instant::now());
    ev.iter().for_each(|t| e.on_tick(*t));
    println!("pattern: {} alerts, {:.0} M events/s", e.alerts.len(), n as f64 / t0.elapsed().as_secs_f64() / 1e6);
    let (mut fired, t0) = (vec![], std::time::Instant::now());
    ev.iter().for_each(|t| idx[t.key as usize].on_price(t.price, 5, &mut fired));
    println!("cross index: {} fires, {:.1} M events/s", fired.len(), n as f64 / t0.elapsed().as_secs_f64() / 1e6);
}

Measured on this Mac (Apple M4 Pro), release build, one thread, 3M synthetic ticks over 10,000 keys, three runs: pattern engine about 100-107 M events/s (450 alerts); level index with 1,000,000 standing queries (100 per key) about 5.4-6.1 M events/s (757k fires). Real rates depend on live partials.

Complexity and constants

• Per event: O(p) for the key's p live partials, plus O(log t) per timer push. Level index: O(log q + f) for q queries on the key, f fired.
• State: O(keys x p x bytes per partial). Without a window state never drains; Flink CEP docs advise until() or within() to enable state clearing.
• Repetition (B+) with any-match semantics can multiply partials exponentially in the run length; skip strategies and greedy matching cut it.

Efficiency

• Cost: a hash lookup, a scan of the key's live partials, a few heap pushes: about 10 ns per event (measured here, cache-resident state, one M4 Pro core); level index with 100 queries per key about 170-190 ns. Bound: memory access to per-key state.
• Levers:
  • Partition by key — independent state scales across cores and stays cache-sized — always.
  • within plus a skip strategy — bounds partials per key (Flink CEP docs: NO_SKIP emits every possible match) — repetition or any-match.
  • Level index instead of testing every query — cost follows fired, not registered, queries — many threshold alerts per key.
  • Cheap start predicate first — no partial allocated for non-candidates — selective first step.
  • Share sub-expressions across similar rules — common predicate evaluated once per event (no figure quoted) — many similar rules.
• Naive → standard → state of the art:
  • Re-run every rule over a stored window per event · rules x window size per event
  • Per-key NFA, within, timers · live partials per key
  • Same plus level/predicate indexes, shared sub-expressions, watermark ordering · events plus fired rules

Implementation notes

• Trailing negation is decided only at the deadline; the code emits when the next event or an explicit advance passes it. Drive advance from a clock too, or an idle stream never finalizes.
• Event time: Flink CEP buffers events and releases them in timestamp order when a watermark arrives; late events need a policy (drop or side output).
• Keep partials small (start time, fields later predicates read); index prices as integer cents.
• Hysteresis band in the unit of the noise (tick size, spread), or alerts flap.
• Emit with an id (key, start time) so retries and failover do not duplicate alerts.

Pitfalls

• No window: state grows forever.
• Any-match on a repeating pattern: a burst of 30 qualifying events can mean thousands of matches.
• Out-of-order events in arrival order: "B then A" looks like "A then B".
• Hot key: per-key serial processing makes one symbol the bottleneck.
• Oscillation around a level re-fires without hysteresis.

Variants and alternatives

In production

• Price alerts, screeners and surveillance over ticks; card-fraud sequence rules; sensor alarms; SIEM correlation.
• Engines: Flink CEP and Flink SQL, Esper, Siddhi; Trino and Oracle MATCH_RECOGNIZE on stored rows.

Implementations

Sources

• Wu, Diao, Rizvi, High-performance complex event processing over streams, SIGMOD 2006. paper
• Agrawal, Diao, Gyllstrom, Immerman, Efficient pattern matching over event streams, SIGMOD 2008. paper
• Cugola, Margara, Processing flows of information: from data stream to complex event processing, ACM Computing Surveys 44(3), 2012. paper
• Luckham, The Power of Events, Addison-Wesley 2002 (book).
• Apache Flink CEP documentation · Trino MATCH_RECOGNIZE · SQL:2016 (row pattern recognition)