DongShengHe created FLINK-40888:
-----------------------------------
Summary: Improve MATCH_RECOGNIZE performance: lazily evaluate
DEFINE aggregations and avoid extracting all matched pattern events
Key: FLINK-40888
URL: https://issues.apache.org/jira/browse/FLINK-40888
Project: Flink
Issue Type: Improvement
Components: Table SQL / Planner
Reporter: DongShengHe
h3. Motivation：
For MATCH_RECOGNIZE queries with DEFINE conditions, the generated
IterativeCondition eagerly performs the following work for every incoming event:
1. All matched pattern events are extracted. For each pattern variable
referenced in the condition, all events matched so far for that variable are
extracted from the NFA shared state into a java.util.List, even when the
condition only needs the current (last) event. For quantified patterns
accumulating many events (e.g. A+ over hundreds of rows), this is {{O(N)}} work
per event, i.e. {{O(N^2)}} overall.
{{2. All aggregations are computed unconditionally. Every aggregate referenced
in the condition is computed per record before the condition body is evaluated.
For a condition like DEFINE C AS C.flag > 0 AND SUM(C.amount) > 100,
SUM(C.amount) (which itself iterates over all events matched for C) is computed
even when C.flag > 0 is false and the AND is already decided.}}
In production jobs with long matched sequences this leads to significant CPU
overhead and backpressure. We observed repeated checkpoint failures and job
restarts for patterns matching more than 500 events on an internal Flink 1.18
based deployment.
h3. Proposed changes:
1. {*}Current-event shortcut{*}: when a DEFINE condition references the last
event (offset 0, LAST) of the pattern variable it defines, generate a
single-element list wrapping the incoming event instead of extracting all
matched events from the NFA state. This extends the existing fast path that is
already used for the star pattern variable (*)
2. {*}Lazy aggregation via memoized suppliers{*}: the per-record construction
of pattern event lists and the aggregate computation are wrapped in
java.util.function.Supplier instances that are created per record but evaluated
on first access only. The && short-circuit in the condition body therefore
skips the work entirely when it is not needed. One memoized aggregate-row
supplier is shared by all aggregates of a pattern variable, so the aggregate
function still runs at most once per record even when the condition reads
multiple aggregates of the same variable.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
