[
https://issues.apache.org/jira/browse/FLINK-41099?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Yaoxuan Wu updated FLINK-41099:
-------------------------------
Description:
{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.
was:
{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.
> 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.3.0, 2.2.1
> Reporter: Yaoxuan Wu
> Priority: Major
>
> {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)
