[
https://issues.apache.org/jira/browse/FLINK-41287?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Yaoxuan Wu updated FLINK-41287:
-------------------------------
Component/s: Table SQL / Runtime
Labels: correctness (was: )
> 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
> Components: Table SQL / Runtime
> Affects Versions: 2.0.0, 2.3.0, 2.2.1, 2.4.1
> Reporter: Yaoxuan Wu
> Priority: Major
> Labels: correctness
>
> {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)
