[
https://issues.apache.org/jira/browse/FLINK-40888?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-40888:
-----------------------------------
Labels: pull-request-available (was: )
> 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
> Priority: Minor
> Labels: pull-request-available
>
> 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)
