Pattern Matching (MATCH_RECOGNIZE / CEP)

Pattern matching (CEP, Complex Event Processing) recognizes event sequences that appear in a specific order on an event stream — e.g. "3 consecutive threshold crossings", "rise then drop", "start → run → stop". It uses the SQL:2016 standard MATCH_RECOGNIZE clause, aligned with Flink SQL (not the FlinkCEP Java API).

When to use pattern matching vs analytic functions vs windows

• Multi-event sequential/ordered patterns ("N consecutive threshold crossings", "A followed by B", "V-shaped reversal", "workflow Start→End") → pattern matching.
• Comparing adjacent events ("did it change vs last time", "what was the previous value") → analytic functions (lag/had_changed).
• Statistics over a time range ("min over the last 10 seconds", "per-minute average") → windowed aggregation + HAVING.
• Pure "device idle for N seconds" offline alerts belong to windows + idle detection, not forced into MATCH_RECOGNIZE.

Dimension	Pattern matching (CEP)	Analytic functions	Windowed aggregation
What it cares about	Event order / sequence	Adjacent-event relation	Time-range statistics
State	Per-partition NFA simulation, across many events	Cross-event state machine	Aggregated in-window, cleared on trigger
Output	A batch per pattern match	One row per event	A batch when the window closes
Trigger	When a pattern match completes	Every event	Window boundary
API	Emit + AddSink	EmitSync or Emit	Emit + AddSink

A minimal example

Temperature crosses 50 three times in a row (debounce, to avoid single-point jitter false alarms):

SELECT * FROM stream
MATCH_RECOGNIZE (
    ORDER BY ts
    MEASURES MATCH_NUMBER() AS mn, LAST(A.temp) AS peak
    ONE ROW PER MATCH
    PATTERN (A{3})
    WITHIN '1h'
    DEFINE A AS temp > 50
)

Input:

{"ts": 1, "temp": 10}
{"ts": 2, "temp": 60}
{"ts": 3, "temp": 70}
{"ts": 4, "temp": 80}
{"ts": 5, "temp": 5}

Output (rows 2–4 form one A{3} match):

{"mn": 1, "peak": 80}

• PATTERN (A{3}): A repeats exactly 3 times.
• DEFINE A AS temp > 50: symbol A matches rows where temp > 50.
• MEASURES: which columns a match emits — MATCH_NUMBER() is the match index, LAST(A.temp) is A's last temperature.
• WITHIN '1h': the whole match must fall within 1 hour (required for boundedness, see WITHIN).

Syntax overview

SELECT * FROM stream
MATCH_RECOGNIZE (
    [PARTITION BY <expr> [, ...]]               -- partition; each partition matches independently
    ORDER BY <expr> [ASC] [, ...]               -- required: event ordering (usually a timestamp)
    [MEASURES <expr> AS <alias> [, ...]]        -- output columns
    [ONE ROW PER MATCH | ALL ROWS PER MATCH]    -- 1 row per match / one row per input row
    [AFTER MATCH SKIP ...]                      -- where to resume after a match
    PATTERN ( <pattern> )                       -- the event-sequence pattern
    [SUBSET <name> = (<sym> [, ...]) [, ...]]   -- group symbols
    [WITHIN '<interval>']                       -- time window
    DEFINE <sym> AS <cond> [, ...]              -- per-symbol conditions
)

Clause	Required	Notes
PARTITION BY	no	Match independently per partition (e.g. per device). High cardinality → see pitfall 5
ORDER BY	yes	Events are processed ascending by this field (usually event timestamp)
MEASURES	no	Match output columns; if omitted, emits the match's last-row fields
ONE / ALL ROWS PER MATCH	no	Defaults to ONE ROW (one row per match)
AFTER MATCH SKIP	no	Defaults to PAST LAST ROW
PATTERN	yes	The pattern expression
SUBSET	no	Group multiple symbols into a set
WITHIN	no (recommended)	Time window; idle partitions' partial matches are actively expired
DEFINE	no	Symbol conditions; an undefined symbol is always true (standard)

SELECT does not project in CEP mode

MATCH_RECOGNIZE output is decided by MEASURES; the outer SELECT only exists as SELECT * FROM stream MATCH_RECOGNIZE(...). Put the columns you want into MEASURES.

PATTERN — defining the event sequence

Syntax	Meaning
A B C	Sequence: A then B then C
A?	A occurs 0 or 1 time
A*	A occurs 0 or more times (greedy, as many as possible)
A+	A occurs 1 or more times
A{3}	A exactly 3 times
A{2,}	A at least 2 times
A{2,4}	A 2 to 4 times
A \| B	Alternation: A or B
(A B)	Grouping
PERMUTE(A, B, C)	A/B/C each once, in any order

A ? suffix on a quantifier makes it reluctant (match as few as possible): A*? picks the shortest A run.

Greedy vs reluctant

• A* (greedy): match as many as possible. A* B on A A A B picks A A A B (longest).
• A*? (reluctant): match as few as possible. A*? B picks B (shortest — 0 A's then B).

When to use reluctant

Greedy (the default) covers most "take the longest run" needs. Reluctant is for "end as early as possible, trigger the next match sooner". Pure-greedy and pure-reluctant patterns are fully accurate; mixed greedy/reluctant quantifiers do not guarantee per-quantifier priority.

DEFINE — symbol conditions

Each pattern variable (symbol) gets a boolean condition; the condition may use fields, navigation functions, and aggregates:

PATTERN (A B)
DEFINE
    A AS temp > 50,
    B AS temp < PREV(temp)      -- B is colder than "the previous row"

• An undefined symbol is always true (SQL standard) — PATTERN (A B) with only A defined means B matches any row.
• Use PREV/NEXT/FIRST/LAST for cross-row references in DEFINE (not LAG/LEAD, those are window functions).

MEASURES — output columns

MEASURES defines which columns each match emits. You can use fields, navigation, aggregates, and two special functions:

Expression	Meaning
A.temp	temp of A's last occurrence (symbol-qualified field)
LAST(temp) / FIRST(temp)	temp of the match's last / first row
SUM(temp) / AVG(temp) / COUNT(*) / MIN / MAX	Aggregate over the match
CLASSIFIER()	The symbol the current row matched (varies per row under ALL ROWS)
MATCH_NUMBER()	The match index (per-partition counter)

MEASURES
    MATCH_NUMBER() AS mn,
    FIRST(A.temp)  AS start_temp,
    LAST(A.temp)   AS end_temp,
    MAX(A.temp)    AS peak,
    CLASSIFIER()   AS sym

Navigation functions

Function	Meaning
PREV(field [, n])	The field of the row n rows before the current one (default n=1)
NEXT(field [, n])	The field n rows after
FIRST(field [, n])	The n-th occurrence of the field in the match (default 1)
LAST(field [, n])	The n-th from the end

In DEFINE, "current row" is the candidate being tested; in MEASURES it advances with ALL ROWS.

Aggregates (symbol-scoped)

SUM/AVG/COUNT/MIN/MAX act over the whole match's rows; add a symbol qualifier to aggregate only that symbol's rows:

MEASURES SUM(A.v) AS a_total, SUM(v) AS all_total
-- rows: A.v=1, B.v=2, A.v=3 → a_total=4 (A only), all_total=6 (all)

FINAL / RUNNING semantics

Aggregates and FIRST/LAST may take a FINAL / RUNNING prefix:

Prefix	Meaning
RUNNING (default)	Only up to the current row (changes as it advances under ALL ROWS)
FINAL	Over the entire match (independent of current row)

MEASURES
    RUNNING SUM(v) AS run_total,   -- cumulative up to the current row
    FINAL   SUM(v) AS match_total  -- total over the whole match

Under ONE ROW PER MATCH there is only the last-row view, so FINAL == RUNNING; the two differ only under ALL ROWS PER MATCH.

SUBSET — grouping symbols

Group several symbols into one set, usable as a whole in PATTERN and MEASURES/DEFINE:

PATTERN (S)
SUBSET S = (A, B)
MEASURES SUM(S.v) AS total    -- aggregate over both A and B rows

CLASSIFIER() returns the actual member symbol (A or B), not S, under a SUBSET. Nesting S2 = (S1, C) is supported.

WITHIN — time window

WITHIN '<interval>' requires the whole match to fall within the interval. Common units: '200ms' / '1s' / '5m' / '1h'.

PATTERN (A B) WITHIN '5s'     -- A and B must arrive within 5 seconds of each other

• The ORDER BY timestamp field is unit-normalized automatically (ns/μs/ms/s epoch).
• Partial matches past the window are actively reaped: a background sweeper periodically scans idle partitions with wall-clock and clears expired partial matches — no need to wait for the next event to trigger passive cleanup.

WITHIN is recommended

Edge memory is limited; prefer an explicit WITHIN so partial matches in idle partitions are actively expired rather than lingering. Without WITHIN, the partition LRU + partial-match caps act as a fallback (see Bounded memory).

PARTITION BY / ORDER BY

• PARTITION BY: partition independently (e.g. "each device's own fault sequence"). Required when many devices share a stream, otherwise they cross-contaminate.
• ORDER BY (required): events are processed ascending by this field, usually the event timestamp.

PARTITION BY deviceId ORDER BY ts

ONE ROW vs ALL ROWS PER MATCH

Mode	Output
ONE ROW PER MATCH (default)	1 row per match (the last-row view + MEASURES)
ALL ROWS PER MATCH	N rows per match (one per input row in the match; CLASSIFIER()/RUNNING advance per row)

ALL ROWS output is large; pair it with WITHIN and output restraint.

AFTER MATCH SKIP

Where to resume after a match completes:

Strategy	Meaning
PAST LAST ROW (default)	After the last row of this match
TO NEXT ROW	From the row after the last
TO FIRST <sym>	After the first occurrence of a symbol
TO LAST <sym>	After the last occurrence of a symbol

PAST LAST ROW keeps matches non-overlapping; TO NEXT ROW allows overlap.

API — how to run it

CEP cannot use EmitSync

Pattern matching is a many-in/many-out batch semantic; EmitSync (synchronous single-row return) does not support it and returns synchronous mode does not support MATCH_RECOGNIZE. You must use Emit (async) + AddSink to receive matches.

package main

import (
    "fmt"
    "time"

    "github.com/rulego/streamsql"
)

func main() {
    ssql := streamsql.New()
    defer ssql.Stop()

    sql := `SELECT * FROM stream
        MATCH_RECOGNIZE (
            PARTITION BY deviceId ORDER BY ts
            MEASURES MATCH_NUMBER() AS mn, LAST(A.temp) AS peak
            ONE ROW PER MATCH
            PATTERN (A{3})
            WITHIN '1h'
            DEFINE A AS temp > 50
        )`
    if err := ssql.Execute(sql); err != nil {
        panic(err)
    }

    // CEP receives matches via AddSink (in arrival order; AddSyncSink preserves order)
    ssql.AddSyncSink(func(batch []map[string]any) {
        for _, row := range batch {
            fmt.Printf("%+v\n", row)
        }
    })

    ssql.Emit(map[string]any{"deviceId": "d1", "ts": 1, "temp": 60.0})
    ssql.Emit(map[string]any{"deviceId": "d1", "ts": 2, "temp": 70.0})
    ssql.Emit(map[string]any{"deviceId": "d1", "ts": 3, "temp": 80.0})

    time.Sleep(200 * time.Millisecond) // let the async sink drain
}

• AddSyncSink: invoked in order inside the data-processing goroutine, preserving match order; AddSink goes through an async worker pool and is not ordered with multiple workers.
• Unclosed matches at end-of-stream (e.g. an unbounded A+) are flushed by Stop.

Bounded memory

Edge memory is limited; pattern matching has four guards against pathological patterns / high-rate streams blowing up memory:

Guard	Default	Purpose
WITHIN	—	Time window; actively expires out-of-window partial matches
Partial-match row cap	10000	Prevents an A+ match from growing unbounded
Active partial matches per partition	10000	Prevents A*-style state explosion
Partition count	10000	Beyond this, LRU evicts the least-recently-used partition

Once a partition is LRU-evicted its pattern state is lost (treated as new when seen again). Use a low-cardinality key for PARTITION BY (e.g. device id), not a near-unique key (timestamp, event id).

Common pitfalls

Pitfall 1: EmitSync errors on CEP

EmitSync supports only non-aggregation simple queries / analytic functions, not CEP (nor windowed aggregation). CEP uses Emit + AddSink/AddSyncSink.

Pitfall 2: SELECT column names have no effect

CEP output is decided by MEASURES; the outer SELECT does not project. To get a column, write it into MEASURES ... AS alias; the outer is always SELECT * FROM stream MATCH_RECOGNIZE(...).

Pitfall 3: using LAG / LEAD in DEFINE

Row-pattern navigation uses PREV/NEXT/FIRST/LAST. LAG/LEAD are window functions and are not available inside MATCH_RECOGNIZE.

Pitfall 4: forgetting ORDER BY

ORDER BY is required inside MATCH_RECOGNIZE — pattern matching depends on event order. Omitting it is an error.

Pitfall 5: high-cardinality PARTITION BY

PARTITION BY partition count grows with distinct keys. Use a low-cardinality key; at massive device counts the default LRU cap (10000) kicks in and evicted partitions lose state (matches restart).

Pitfall 6: assuming A* picks the shortest

A* is greedy by default (picks the longest run). For the shortest, use A*? (reluctant). See Greedy vs reluctant.

Cases

Full business-scenario cases are in Device Fault Pattern Recognition, covering:

• Consecutive threshold debounce (A{3})
• Rise then drop (A B, failure precursor)
• V-shaped reversal (A+ B+ C)
• Start/stop workflow (Start Running Stop)
• Out-of-order events with PERMUTE
• Time-constrained sequences with WITHIN

Performance

Single-core (AMD Ryzen 4800U) baseline (cep package BenchmarkCEP_*):

Pattern	Throughput	Allocs
Sequence A B	~410k ops/s	20 allocs/op
Greedy star A* B	~270k ops/s	29 allocs/op
Partitioned sequence A B + PARTITION BY	~370k ops/s	23 allocs/op

Same order of magnitude as analytic + partition (~480k ops/s) — suitable for edge-node CEP workloads.

Known limitations

• Exclusion {- A -} (absence / negation) is not yet implemented (rejected at compile time).
• ALL ROWS PER MATCH WITH UNMATCHED ROWS is not implemented.
• SUBSET as the left-hand side of DEFINE (DEFINE S AS ...) is not implemented; right-hand references are supported.
• Outer ORDER BY / LIMIT do not apply to CEP output.
• Multi-stream CEP (matching across joined streams) is not implemented.
• Mixed greedy/reluctant quantifiers do not guarantee per-quantifier priority (pure-greedy / pure-reluctant are accurate).