Yaoxuan Wu created FLINK-41294:
----------------------------------
Summary: 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.3.0, 2.1.1, 2.0.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([(-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)
