Yaoxuan Wu created FLINK-41293:
----------------------------------
Summary: PyFlink batch: pandas UDAF over ROWS BETWEEN n PRECEDING
AND k PRECEDING (k >= 2) aggregates rows outside the frame for the first rows
of a partition
Key: FLINK-41293
URL: https://issues.apache.org/jira/browse/FLINK-41293
Project: Flink
Issue Type: Bug
Components: API / Python
Affects Versions: 2.3.0, 2.1.1
Reporter: Yaoxuan Wu
{code:python}
from pyflink.table import EnvironmentSettings, TableEnvironment, DataTypes as D
from pyflink.table.udf import udaf
t_env = TableEnvironment.create(EnvironmentSettings.in_batch_mode())
t_env.get_config().set('parallelism.default', '1')
t_env.create_temporary_view('t', t_env.from_elements([(i,) for i in range(1,
6)], D.ROW([D.FIELD('o', D.INT())])))
t_env.create_temporary_function('pandas_list',
udaf(lambda s: str(list(s)), result_type=D.STRING(), func_type='pandas'))
t_env.execute_sql("""SELECT o, pandas_list(o) OVER (ORDER BY o ROWS BETWEEN 2
PRECEDING AND 2 PRECEDING) AS w
FROM t""").print()
# expected: o=1 [], o=2 [], o=3 [1], o=4 [2], o=5 [3]
# actual: o=1 [1, 2, 3, 4], o=2 [], o=3 [1], o=4 [2], o=5 [3]
{code}
Native {{COUNT(o)}} over the same frame gives 0, 0, 1, 1, 1; the pandas UDAF
sees 4 rows for o=1. {{ROWS BETWEEN 2 PRECEDING AND 1 PRECEDING}} and {{ROWS
BETWEEN 1 FOLLOWING AND 2 FOLLOWING}} are correct.
Cause:
{{PandasBatchOverWindowAggregateFunctionOperation.wrapped_over_window_function}}
(pyflink/fn_execution/table/operations.py), row sliding window:
{code:python}
start = max(j + window_start, 0)
end = min(j + window_end + 1, input_cnt)
series_slices = [s.iloc[start: end] for s in input_series]
{code}
{{end}} is clamped from above but not from below. With an upper bound of k
PRECEDING ({{window_end = -k}}), {{end = j - k + 1}} is negative for the first
k - 1 rows, and pandas reads a negative stop from the end of the series
({{iloc[0:-1]}} = all rows but the last). For k = 1, {{end = 0}} and the frame
is correctly empty. The {{ROW_UNBOUNDED_PRECEDING}} branch ({{s.iloc[: end]}}
with the same {{end}}) has the same unclamped end for {{UNBOUNDED PRECEDING AND
k PRECEDING}}.
Reproduces on 2.0.1, 2.2.1, 2.3.0 and master.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
