[
https://issues.apache.org/jira/browse/FLINK-40735?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
sepuri sai krishna updated FLINK-40735:
---------------------------------------
Release Note:
An OVER window ordered by a non-time attribute compared accumulators with
RowData#equals. On the
heap state backend that threw UnsupportedOperationException for any aggregate
whose accumulator
holds a RAW field, such as LAG and ARRAY_AGG. On RocksDB it never matched, so
every row after an
insert emitted an UPDATE_BEFORE and UPDATE_AFTER pair carrying identical values.
The window also handed accumulators straight from state to the aggregate
functions. On the heap
state backend that changed the accumulator stored for another sort key, so LAG,
ARRAY_AGG and the
BITMAP aggregates returned wrong results once a row sorted in between two
existing ones.
Both are fixed. A query of this shape now returns the same results on both
state backends, and
downstream consumers see fewer no-op updates than before.
> OVER window ordered by a non-time attribute fails for LAG and ARRAY_AGG when
> a row arrives out of order
> -------------------------------------------------------------------------------------------------------
>
> Key: FLINK-40735
> URL: https://issues.apache.org/jira/browse/FLINK-40735
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Runtime
> Affects Versions: 2.3.0, 2.2.1, 2.1.3
> Reporter: sepuri sai krishna
> Assignee: sepuri sai krishna
> Priority: Major
> Labels: pull-request-available
> Attachments: NonTimeOverRangeRepro.java, pom.xml
>
>
> In streaming mode, an {{OVER}} window ordered by a non-time attribute fails
> at runtime for
> {{LAG}}, {{ARRAY_AGG}}, {{COLLECT}} and {{PERCENTILE}} as soon as a row
> arrives that does
> not sort to the end of its partition.
> {code:sql}
> SELECT ord, LAG(v,1) OVER (PARTITION BY k ORDER BY ord)
> FROM (VALUES ('a',30,'p'),('a',20,'q'),('a',10,'r')) AS t(k,ord,v);
> {code}
> {noformat}
> java.lang.UnsupportedOperationException: Unmaterialized BinaryRawValueData
> cannot be compared.
> at
> org.apache.flink.table.data.binary.BinaryRawValueData.equals(BinaryRawValueData.java:85)
> at
> org.apache.flink.table.data.GenericRowData.equals(GenericRowData.java:229)
> ...
> {noformat}
> Two triggers, both ordinary in a stream:
> * a row arriving out of order on the ordering column
> * two rows sharing the same ordering value
> A single row, or rows arriving in ascending order, both work.
> Affected: {{LAG}}, {{ARRAY_AGG}}, {{COLLECT}}, {{PERCENTILE}}.
> Unaffected on the same data: {{COUNT}}, {{SUM}}, {{AVG}}, {{MIN}}, {{MAX}},
> {{FIRST_VALUE}},
> {{LAST_VALUE}}, {{LISTAGG}}, {{ROW_NUMBER}}, {{RANK}}, {{DENSE_RANK}},
> {{VAR_POP}},
> {{STDDEV_POP}}.
> Also reproduces with a {{datagen}} source whose ordering column is a random
> INT, so the
> rows arrive unordered naturally rather than through {{VALUES}}:
> {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.1.3, 2.2.1 and 2.3.0. On 2.0.0 the same query is rejected at
> plan time with
> {{OVER windows' ordering in stream mode must be defined on a time attribute}}.
> Reproducer attached.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
