sepuri sai krishna created FLINK-40736:
------------------------------------------
Summary: Non-time ROWS OVER window emits a malformed row for
FIRST_VALUE, LAST_VALUE, LISTAGG, COUNT(DISTINCT), ARRAY_AGG, COLLECT and
PERCENTILE
Key: FLINK-40736
URL: https://issues.apache.org/jira/browse/FLINK-40736
Project: Flink
Issue Type: Bug
Components: Table SQL / Runtime
Affects Versions: 2.2.1, 2.3.0
Reporter: sepuri sai krishna
Attachments: NonTimeOverRowsRepro.java, pom.xml
In streaming mode, an {{OVER ... ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT
ROW}} window
ordered by a non-time attribute emits a row of the wrong shape when a row
arrives out of
order.
{code:sql}
SELECT ord, LISTAGG(v)
OVER (PARTITION BY k ORDER BY ord ROWS BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW)
FROM (VALUES ('a',30,'p'),('a',20,'q'),('a',10,'r')) AS t(k,ord,v);
{code}
{noformat}
java.lang.IllegalArgumentException: Row arity: 6, but serializer arity: 4
at
org.apache.flink.table.runtime.typeutils.RowDataSerializer.copy(RowDataSerializer.java:124)
...
{noformat}
{{FIRST_VALUE}}, {{LAST_VALUE}}, {{ARRAY_AGG}}, {{COLLECT}} and {{PERCENTILE}}
fail on the
same query shape with:
{noformat}
java.lang.ClassCastException: class org.apache.flink.table.data.GenericRowData
cannot be cast
to class org.apache.flink.table.data.ArrayData
at
org.apache.flink.table.data.GenericRowData.getArray(GenericRowData.java:197)
at
org.apache.flink.table.data.utils.JoinedRowData.getArray(JoinedRowData.java:232)
...
{noformat}
Measured on 2.3.0 with rows arriving 30, 20, 10:
{noformat}
COUNT(*), COUNT(v), SUM, AVG, MIN, MAX, VAR_POP, STDDEV_POP ok
FIRST_VALUE, LAST_VALUE, ARRAY_AGG, COLLECT, PERCENTILE
ClassCastException
LISTAGG(v), LISTAGG(v,'-'), COUNT(DISTINCT v)
IllegalArgumentException
{noformat}
Ascending input, tied ordering values and a single row all work. Only
out-of-order arrival
fails. The aggregates that do not throw return correct values, so this looks
like a crash
rather than a wrong result.
Also reproduces with a {{datagen}} source whose ordering column is a random INT:
{code:sql}
CREATE TABLE src (k STRING, ord INT, v STRING) WITH (
'connector'='datagen','number-of-rows'='20',
'fields.k.length'='1','fields.v.length'='2',
'fields.ord.min'='1','fields.ord.max'='1000');
{code}
Reproduced on 2.2.1 and 2.3.0. On 2.1.3 the same query is rejected at plan time
with
{{OVER windows with UNBOUNDED PRECEDING are not supported when sorting on a
non-time
attribute column}}.
Reproducer attached.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
