Yaoxuan Wu created FLINK-41287:
----------------------------------
Summary: Batch sliding (HOP) group window loses the windows of the
last pane when an empty window precedes it (SortWindowAggregate)
Key: FLINK-41287
URL: https://issues.apache.org/jira/browse/FLINK-41287
Project: Flink
Issue Type: Bug
Affects Versions: 2.2.1, 2.3.0, 2.0.0, 2.4.1
Reporter: Yaoxuan Wu
{code:python}
import datetime as dt
from pyflink.table import EnvironmentSettings, TableEnvironment, DataTypes as D
t = TableEnvironment.create(EnvironmentSettings.in_batch_mode())
t.get_config().set('parallelism.default', '1')
t.get_config().set('table.optimizer.agg-phase-strategy', 'ONE_PHASE')
t0 = dt.datetime(2023, 1, 1)
t.create_temporary_view('src', t.from_elements([(t0,), (t0 +
dt.timedelta(seconds=6),)],
D.ROW([D.FIELD('ts', D.TIMESTAMP(3))])))
t.execute_sql("""SELECT HOP_START(ts, INTERVAL '2' SECOND, INTERVAL '4' SECOND)
AS ws, COUNT(*) AS c
FROM src GROUP BY HOP(ts, INTERVAL '2' SECOND, INTERVAL '4'
SECOND)""").print()
# expected: windows starting at 23:59:58, 00:00:00, 00:00:04, 00:00:06 (c = 1
each)
# actual: only 23:59:58 and 00:00:00; the windows [00:04, 00:08) and [00:06,
00:10) holding the second row are missing
{code}
The plan is {{SortWindowAggregate(window=[SlidingGroupWindow(...)])}}. The same
happens with the default configuration whenever the aggregate cannot be
hash-based, e.g. {{MAX}} over a STRING column, and with the Table API {{Slide}}
window. With two-phase hash aggregation all 4 windows are returned; window TVFs
({{TABLE(HOP(...))}}) are not affected.
Cause ({{WindowsGrouping}}, flink-table-runtime): after the last pane of a
group, {{advanceWatermarkToTriggerAllWindows()}} sets {{watermark}} to the last
timestamp + window size and triggers the remaining windows. When a triggered
window is empty (here [00:02, 00:06)), the next {{hasTriggerWindow()}} calls
{{skipEmptyWindow()}}, which moves {{nextWindow}} to the first window
containing {{watermark}} ({{advanceNextWindowByWatermark(watermark)}}). That
treats {{watermark}} as the latest element's timestamp, which no longer holds
in this leftover phase, so {{nextWindow}} jumps past the windows of the last
pane and they are never triggered. Without an empty window in between (rows at
0 s and 4 s) the result is correct.
Reproduces on 2.0.1, 2.2.1, 2.3.0 and master.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
