[
https://issues.apache.org/jira/browse/FLINK-41293?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18139485#comment-18139485
]
Yahya Kisana commented on FLINK-41293:
--------------------------------------
Hi, can I take this?
> 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.1.1, 2.3.0
> Reporter: Yaoxuan Wu
> Priority: Major
> Labels: correctness
>
> {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)
