[
https://issues.apache.org/jira/browse/FLINK-41294?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18141521#comment-18141521
]
Shawn commented on FLINK-41294:
-------------------------------
[~dianfu] I'd like to take this one too. I'd make the operator compare the
ORDER BY key by its actual type, like the native over aggregate does. Could you
assign it to me?
> 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
> 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)
