Yaoxuan Wu created FLINK-41506:
----------------------------------
Summary: Batch sliding group window planned as a one-phase
SortWindowAggregate reads the pane time at the input row's time index
Key: FLINK-41506
URL: https://issues.apache.org/jira/browse/FLINK-41506
Project: Flink
Issue Type: Bug
Components: Table SQL / Planner, Table SQL / Runtime
Affects Versions: 2.3.0
Reporter: Yaoxuan Wu
{code:python}
from pyflink.table import TableEnvironment, EnvironmentSettings, DataTypes
from pyflink.table import expressions as T
from pyflink.table.window import Slide
t_env = TableEnvironment.create(EnvironmentSettings.in_batch_mode())
cfg = t_env.get_config()
cfg.set('parallelism.default', '1')
cfg.set('table.local-time-zone', 'UTC')
cfg.set('table.optimizer.agg-phase-strategy', 'ONE_PHASE')
src = t_env.from_elements([(1, 7000)], DataTypes.ROW([DataTypes.FIELD('k',
DataTypes.INT()),
DataTypes.FIELD('v',
DataTypes.BIGINT())]))
t = src.add_columns(T.call_sql("TIMESTAMP '2001-09-09
01:46:40.000'").alias('rt')) # (k, v, rt)
r =
(t.window(Slide.over(T.lit(12).seconds).every(T.lit(3).seconds).on(T.col('rt')).alias('w'))
.group_by(T.col('w'), T.col('k'))
.select(T.col('k'), T.col('w').start, T.col('w').end, T.col('v').sum))
for row in sorted(r.execute().collect(), key=str):
print(row)
# expected: 4 windows starting 2001-09-09 01:46:30, :33, :36, :39
# actual: 4 windows starting 1969-12-31 23:59:57, 1970-01-01 00:00:00,
00:00:03, 00:00:06
{code}
Same on 2.0.1, 2.2.1, 2.3.0 and a 2.4-SNAPSHOT build. The row is placed at 7000
ms, the value of the {{SUM(v)}} buffer. With the input columns in the order
{{(rt, k, v)}} it is placed at 1 ms (the key). With {{(k, rt, v)}} the result
is right. With a fourth column after {{rt}}, the job fails with
{{java.lang.IndexOutOfBoundsException}} in {{WindowsGrouping.getTimeValue}}.
Tumbling windows and the two-phase plan are not affected.
The one-phase plan is not only chosen by ONE_PHASE: with the default strategy,
the same happens when the window input comes from two aggregates, e.g.
{code:python}
a2 = (s2.group_by(T.col('a')).select(T.col('a'), T.lit(1).count.alias('c'))
.group_by(T.col('c')).select(T.col('c').alias('k'), (T.col('c').sum *
7000).alias('v'))
.add_columns(T.call_sql("TIMESTAMP '2001-09-09
01:46:40.000'").alias('rt')))
# s2 = (7, 100), (8, 200): one row (k=1, v=14000); Slide 12 s every 3 s ->
windows around 1970-01-01 00:00:14
{code}
Cause (planner, master):
- {{BatchPhysicalWindowAggregateRule.transformTimeSlidingWindow}}, one-phase
branch: a sliding window with pane optimization ({{enableAssignPane = true}})
becomes a {{BatchPhysicalSortWindowAggregate}} with {{isMerge = false}} and
{{inputTimeFieldIndex}} = the time attribute's index in the input row.
- {{SortWindowCodeGenerator}}: {{enablePreAcc = choosePreAcc || isMerge}}, and
{{choosePreAcc}} is true when {{enableAssignPane}} is set. Input rows are then
pre-accumulated into pane elements laid out as {{[group keys, assignedTs$, agg
buffers]}} ({{WindowCodeGenerator.getWindowsGroupingElementInfo}}).
- {{WindowCodeGenerator.genWindowAggCodes}} passes the same input-row index to
{{genCreateWindowsGroupingCode}}, so {{HeapWindowsGrouping}} reads a pane
element's time ({{WindowsGrouping.getTimeValue}}) at that index. For a
pre-accumulated element the index should be the number of grouping keys (the
{{assignedTs$}} field). The two-phase merge branch already passes
{{newInputTimeFieldIndexFromLocal}}, which points there.
The result is only right when the input's time index happens to equal the
number of grouping keys. This is separate from FLINK-41287, which is about the
same operator. Its reproducer has no key and the time field at index 0, so it
is not affected by this issue.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
