Junfan Zhang created FLINK-41508:
------------------------------------
Summary: Support multiple bounded RANGE OVER windows with shared
state in streaming SQL
Key: FLINK-41508
URL: https://issues.apache.org/jira/browse/FLINK-41508
Project: Flink
Issue Type: New Feature
Components: Table SQL / Planner
Reporter: Junfan Zhang
Streaming SQL rejects a query that computes aggregates over several OVER
windows with different ranges, which is a common pattern in real-time ML
features engineering:
{code:java}
SELECT
entity_id,
event_time,
SUM(v) OVER w10,
SUM(v) OVER w20,
SUM(v) OVER w30
FROM events WINDOW
w10 AS (PARTITION BY entity_id ORDER BY event_time RANGE BETWEEN INTERVAL
'10' MINUTE PRECEDING AND CURRENT ROW),
w20 AS (PARTITION BY entity_id ORDER BY event_time RANGE BETWEEN INTERVAL
'20' MINUTE PRECEDING AND CURRENT ROW),
w30 AS (PARTITION BY entity_id ORDER BY event_time RANGE BETWEEN INTERVAL
'30' MINUTE PRECEDING AND CURRENT ROW);{code}
It fails with "Over Agg: Unsupported use of OVER windows. All aggregates must
be computed on the same window." Batch mode supports it.
The workaround is one nested subquery per window. Each level is a separate
operator that stores its own copy of the input rows, so state grows with the
sum of all ranges (10 + 20 + 30 = 60 minutes of rows instead of 30).
Proposal: when all windows of one OVER node are bounded event-time RANGE
windows with the same PARTITION BY and ORDER BY, evaluate them in one operator
that:
- stores the input rows once, kept for the largest range;
- keeps one accumulator per window and retracts a row from a window when the
row leaves that window's range.
Row state becomes max(range_i) instead of sum(range_i). Single-window queries
keep their current operator and state layout; other combinations (ROWS,
proctime, unbounded, Python UDAFs) keep failing with a clearer error message.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
