[
https://issues.apache.org/jira/browse/FLINK-21893?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17305682#comment-17305682
]
Jark Wu commented on FLINK-21893:
---------------------------------
Would you like to fix this [~qingru zhang]?
> A ValidationException will be thrown out if partition key of Rank is an
> expression result of input Node.
> --------------------------------------------------------------------------------------------------------
>
> Key: FLINK-21893
> URL: https://issues.apache.org/jira/browse/FLINK-21893
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
> Reporter: Andy
> Priority: Major
> Attachments: image-2021-03-21-21-07-43-230.png
>
>
> A ValidationException will be thrown out when deriveRowType of Rank if
> partition key is an expression result of Input Node. e.g If run the following
> sql, A validationException will be thrown out.
>
> {code:java}
> //代码占位符
> @Test
> def test(): Unit = {
> val data = List(
> (2001L, 2L),
> (2002L, 3L)
> )
> val ds = failingDataSource(data).toTable(tEnv, 'video_id, 'cnt,
> 'proctime.proctime)
> tEnv.registerTable("T", ds)
> val sql =
> """
> |SELECT
> | video_id,
> | cnt,
> | rownum_2
> |FROM
> |(
> | SELECT
> | video_id,
> | cnt,
> | ROW_NUMBER() OVER (
> | ORDER BY cnt DESC
> | ) AS rownum_2
> | FROM
> | (
> | SELECT
> | video_id,
> | cnt,
> | ROW_NUMBER() OVER (
> | PARTITION BY bucket_id
> | ORDER BY cnt DESC
> | ) AS rownum_1
> | FROM
> | (
> | SELECT
> | video_id,
> | cnt,
> | MOD(video_id, 64) as bucket_id
> | FROM T
> | )
> | )
> | WHERE rownum_1 <= 1000
> |)
> |WHERE rownum_2 <= 1000
> |""".stripMargin
> val sink = new TestingRetractSink
> tEnv.sqlQuery(sql).toRetractStream[Row].addSink(sink).setParallelism(1)
> env.execute()
> }
> {code}
> Exception detail
> {code:java}
> //代码占位符
> org.apache.flink.table.api.ValidationException: Field names must be unique.
> Found duplicates: [$2]org.apache.flink.table.api.ValidationException: Field
> names must be unique. Found duplicates: [$2]
> at
> org.apache.flink.table.types.logical.RowType.validateFields(RowType.java:277)
> at org.apache.flink.table.types.logical.RowType.<init>(RowType.java:158) at
> org.apache.flink.table.types.logical.RowType.<init>(RowType.java:162) at
> org.apache.flink.table.types.logical.RowType.of(RowType.java:294) at
> org.apache.flink.table.planner.calcite.FlinkTypeFactory$.toLogicalRowType(FlinkTypeFactory.scala:503)
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecRank.translateToPlanInternal(StreamExecRank.scala:212)
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecRank.translateToPlanInternal(StreamExecRank.scala:53)
> at
> org.apache.flink.table.planner.plan.nodes.exec.ExecNode$class.translateToPlan(ExecNode.scala:58)
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecRank.translateToPlan(StreamExecRank.scala:53)
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecCalc.translateToPlanInternal(StreamExecCalc.scala:54)
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecCalc.translateToPlanInternal(StreamExecCalc.scala:39)
> at
> org.apache.flink.table.planner.plan.nodes.exec.ExecNode$class.translateToPlan(ExecNode.scala:58)
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecCalcBase.translateToPlan(StreamExecCalcBase.scala:38)
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecExchange.translateToPlanInternal(StreamExecExchange.scala:84)
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecExchange.translateToPlanInternal(StreamExecExchange.scala:44)
> at
> org.apache.flink.table.planner.plan.nodes.exec.ExecNode$class.translateToPlan(ExecNode.scala:58)
> {code}
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
