Yaoxuan Wu created FLINK-41099:
----------------------------------
Summary: Streaming RANK() / DENSE_RANK() OVER a non-time ORDER BY
ignore ties (return the row count instead of the rank)
Key: FLINK-41099
URL: https://issues.apache.org/jira/browse/FLINK-41099
Project: Flink
Issue Type: Bug
Components: Table SQL / Planner
Affects Versions: 2.2.1, 2.3.0
Reporter: Yaoxuan Wu
{code:python} from pyflink.table import EnvironmentSettings, TableEnvironment t
= TableEnvironment.create(EnvironmentSettings.in_streaming_mode())
t.execute_sql("CREATE TEMPORARY VIEW v AS SELECT * FROM (VALUES (1, 'a'), (1,
'b'), (2, 'c')) AS t(k, s)") t.execute_sql("SELECT k, s, RANK() OVER (PARTITION
BY k ORDER BY k) AS r FROM v").print() # expected (and batch mode): (1, a, 1),
(1, b, 1), (2, c, 1) # actual (final changelog): (1, a, 2), (1, b, 2), (2, c,
1) {code} The two rows with k = 1 are peers and must both get rank 1.
DENSE_RANK() gives the same wrong values. Streaming returns the number of rows
up to and including the peer group, i.e. COUNT(*) over the same window. Cause:
the streaming OVER aggregate creates its aggregate functions with
{{AggregateUtil.transformToStreamAggregateInfoList}}, which passes
{{orderKeyIndexes = null}} to {{AggFunctionFactory}}. {{RankAggFunction}} /
{{DenseRankAggFunction}} are therefore built without order keys, and
{{RankLikeAggFunctionBase.orderKeyEqualsExpression()}} with no keys returns
{{literal(false)}}: every accumulated row counts as a new rank. Since
{{NonTimeRangeUnboundedPrecedingFunction}} accumulates all peers before
emitting one value for them, the peers get the count. The batch path passes the
order key indexes and is correct. Reproduces on 2.2.1, 2.3.0 and master.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
