[
https://issues.apache.org/jira/browse/FLINK-41294?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18141701#comment-18141701
]
Dian Fu commented on FLINK-41294:
---------------------------------
[~shawngarello] Thanks for working on this issue. Assigned it to you~
> PyFlink batch: pandas UDAF over a RANGE frame with CURRENT ROW or n PRECEDING
> / FOLLOWING reads the ORDER BY key as TIMESTAMP(3) and gets wrong frames for
> INT and DECIMAL keys
> -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-41294
> URL: https://issues.apache.org/jira/browse/FLINK-41294
> Project: Flink
> Issue Type: Bug
> Components: API / Python
> Affects Versions: 2.0.1, 2.1.1, 2.3.0
> Reporter: Yaoxuan Wu
> Assignee: Shawn
> 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([(-2,), (-1,), (0,),
> (3,)], D.ROW([D.FIELD('o', D.INT())])))
> t_env.create_temporary_function('pandas_count',
> udaf(lambda s: len(s), result_type=D.BIGINT(), func_type='pandas'))
> w = "OVER (ORDER BY o RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)"
> t_env.execute_sql(f"SELECT o, COUNT(o) {w} AS c FROM t").print() # c
> = 1, 2, 3, 4
> t_env.execute_sql(f"SELECT o, pandas_count(o) {w} AS c FROM t").print() # c
> = 1, 4, 4, 4
> {code}
> Same data: {{RANGE BETWEEN 1 PRECEDING AND CURRENT ROW}} gives native 1, 2,
> 2, 1 and pandas 1, 4, 4, 4. With a DECIMAL(10, 2) key, {{UNBOUNDED PRECEDING
> AND CURRENT ROW}} is correct but {{1 PRECEDING AND CURRENT ROW}} gives pandas
> 1, 1, 1, 1. With a DECIMAL(38, 10) key pandas gives 4, 4, 4, 4 for both
> frames. BIGINT keys are correct.
> Cause:
> {{BatchArrowPythonOverWindowAggregateFunctionOperator.invokeCurrentBatch}}
> computes the boundaries of every bounded RANGE window from
> {{curData.getTimestamp(inputTimeFieldIndex, 3).getMillisecond()}} and
> compares the neighbouring rows the same way ({{isInCurrentOverWindow}}).
> {{BatchExecPythonOverAggregate}} passes the index of the ORDER BY field
> ({{sortSpec.getFieldIndices()[0]}}) whatever its type, and a {{CURRENT ROW}}
> bound is a bounded bound (boundary 0 from
> {{OverAggregateUtil.getLongBoundary}}), so even {{UNBOUNDED PRECEDING AND
> CURRENT ROW}} takes this path. Reading the key as a compact TIMESTAMP(3)
> ({{BinaryRowData.getTimestamp}}:
> {{segments[0].getLong(getFieldOffset(pos))}}) reads its 8-byte fixed-length
> slot as a long: right for BIGINT; for DECIMAL(p <= 18) it is the unscaled
> value, so the frame offset is applied in units of 10^-scale (1 PRECEDING
> becomes 0.01 PRECEDING for DECIMAL(10, 2)); for a negative INT only the low 4
> bytes are set (-2 reads as 4294967294); for DECIMAL(p > 18) the slot holds an
> offset and length rather than the value. Since the boundary scan only moves
> forward, keys read out of order also give wrong frames for later rows.
> Reproduces on 2.0.1, 2.2.1, 2.3.0 and master.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
