[
https://issues.apache.org/jira/browse/FLINK-26051?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17489467#comment-17489467
]
chuncheng wu commented on FLINK-26051:
--------------------------------------
the same problem in last comments:
https://stackoverflow.com/questions/65141369/flink-sql-how-to-get-the-first-record-and-the-last-record-by-eventtime-in-kafka/71047250#71047250
> row_number =1 and Subsequent SQL has "case when" and "where" statement : The
> window can only be ordered in ASCENDING mode
> -------------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-26051
> URL: https://issues.apache.org/jira/browse/FLINK-26051
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
> Affects Versions: 1.12.2
> Reporter: chuncheng wu
> Priority: Major
>
> hello,
> i have 2 sqls. One sql has rn=1 and the Subsequent SQL has "case when"
> and "where".it results the exception as follow. It happen in the occasion
> when logical plan turn into physical plan :
>
> {code:java}
> org.apache.flink.table.api.TableException: The window can only be ordered in
> ASCENDING mode.
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecOverAggregate.translateToPlanInternal(StreamExecOverAggregate.scala:98)
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecOverAggregate.translateToPlanInternal(StreamExecOverAggregate.scala:52)
> at
> org.apache.flink.table.planner.plan.nodes.exec.ExecNode$class.translateToPlan(ExecNode.scala:59)
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecOverAggregateBase.translateToPlan(StreamExecOverAggregateBase.scala:42)
> 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:59)
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecCalcBase.translateToPlan(StreamExecCalcBase.scala:38)
> at
> org.apache.flink.table.planner.delegation.StreamPlanner$$anonfun$translateToPlan$1.apply(StreamPlanner.scala:66)
> at
> org.apache.flink.table.planner.delegation.StreamPlanner$$anonfun$translateToPlan$1.apply(StreamPlanner.scala:65)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
> at scala.collection.Iterator$class.foreach(Iterator.scala:891)
> at scala.collection.AbstractIterator.foreach(Iterator.scala:1334)
> at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
> at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
> at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
> at scala.collection.AbstractTraversable.map(Traversable.scala:104)
> at
> org.apache.flink.table.planner.delegation.StreamPlanner.translateToPlan(StreamPlanner.scala:65)
> at
> org.apache.flink.table.planner.delegation.StreamPlanner.explain(StreamPlanner.scala:103)
> at
> org.apache.flink.table.planner.delegation.StreamPlanner.explain(StreamPlanner.scala:42)
> at
> org.apache.flink.table.api.internal.TableEnvironmentImpl.explainInternal(TableEnvironmentImpl.java:630)
> at
> org.apache.flink.table.api.internal.TableImpl.explain(TableImpl.java:582)
> at
> com.meituan.grocery.data.flink.test.BugTest.testRowNumber(BugTest.java:69)
> at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native
> Method)
> at
> java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
> at
> java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> at java.base/java.lang.reflect.Method.invoke(Method.java:568)
> at
> org.junit.platform.commons.util.ReflectionUtils.invokeMethod(ReflectionUtils.java:725)
> at
> org.junit.jupiter.engine.execution.MethodInvocation.proceed(MethodInvocation.java:60){code}
>
>
> example code :
> {code:java}
> import org.apache.flink.api.java.tuple.Tuple12;
> import org.apache.flink.api.java.tuple.Tuple2;
> import org.apache.flink.streaming.api.datastream.DataStream;
> import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
> import org.apache.flink.streaming.api.functions.source.SourceFunction;
> import org.apache.flink.table.api.EnvironmentSettings;
> import org.apache.flink.table.api.ExplainDetail;
> import org.apache.flink.table.api.Table;
> import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
> import org.apache.flink.types.Row;
> import org.junit.jupiter.api.Test;
> import java.sql.Timestamp;
> import static org.apache.flink.table.api.Expressions.$;
> public class BugTest {
> @Test public void testRowNumber() throws Exception {
> StreamExecutionEnvironment env =
> StreamExecutionEnvironment.getExecutionEnvironment();
> EnvironmentSettings
> mySetting=EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
> StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env,
> mySetting);
> env.setParallelism(1);
> //source table ,name is wosOutSowingTaskDetail
> DataStream<Tuple12<String, Integer, Integer, String, Integer, Integer,
> Integer, Integer, Integer, String, String, String>> oriStream =
> env.addSource(new CustomSourceRowNumber());
> Table testTable = tableEnv.fromDataStream(oriStream, $("biz_bill_no"),
> $("task_type"), $("task_mode"), $("parent_task_no"), $("total_stage_num"),
> $("current_stage_index"), $("use_pre_task_owner"), $("poi_type"),
> $("biz_origin_bill_type"), $("sowing_task_no"), $("dt"),
> $("sowing_task_detail_id"));
> tableEnv.createTemporaryView("wosOutSowingTaskDetail", testTable);
> //sql 0
> Table wosOutSowingTaskDetailLatest = tableEnv.sqlQuery( "SELECT
> `biz_bill_no`\n"
> + ",task_type\n" + ",task_mode\n"
> + ",parent_task_no\n"
> + ",total_stage_num\n"
> + ",current_stage_index\n"
> + ",use_pre_task_owner\n"
> + ",poi_type\n"
> + ",biz_origin_bill_type\n"
> + ",sowing_task_no\n"
> + " FROM (\n"
> + " SELECT *,\n"
> + " ROW_NUMBER() OVER(PARTITION BY dt,sowing_task_detail_id ORDER
> BY task_type desc) AS rn\n"
> + " FROM wosOutSowingTaskDetail\n"
> + " ) tmp\n"
> + " WHERE rn = 1");
> System.out.println("SQL 0 Plan: ");
>
> System.out.println(wosOutSowingTaskDetailLatest.explain(ExplainDetail.ESTIMATED_COST));
>
>
> System.out.println(wosOutSowingTaskDetailLatest.explain(ExplainDetail.CHANGELOG_MODE));
>
> tableEnv.createTemporaryView("wosOutSowingTaskDetailLatest",
> wosOutSowingTaskDetailLatest);
>
> //sql 1:select from sql0'result table ,field has a CASE WHEN
> Table resultTable = tableEnv.sqlQuery("SELECT\n"
> + "biz_bill_no\n"
> + ", CASE WHEN task_type = 21 AND task_mode = 51 THEN
> parent_task_no\n"
> + " WHEN task_type = 21 AND task_mode = 40 AND total_stage_num
> >= 2 AND current_stage_index >= 2 AND use_pre_task_owner = 1 THEN
> parent_task_no\n"
> + " ELSE sowing_task_no END AS parent_task_no_cw\n"
> + ",parent_task_no, sowing_task_no, task_type, task_mode,
> total_stage_num, current_stage_index,use_pre_task_owner \n"
> + "FROM wosOutSowingTaskDetailLatest\n"
> + "WHERE task_type = 21\n" + "AND task_mode IN (51, 40)\n" +
> "AND poi_type = 2\n");
> System.out.println("SQL 1 Plan: ");
> System.out.println(resultTable.explain(ExplainDetail.ESTIMATED_COST));
> System.out.println(resultTable.explain(ExplainDetail.CHANGELOG_MODE));
> DataStream<Tuple2<Boolean, Row>> resultStream =
> tableEnv.toRetractStream(resultTable, Row.class);
> env.execute();
> }
> //user define source
> static class CustomSourceRowNumber implements SourceFunction<Tuple12<String,
> Integer, Integer, String, Integer, Integer, Integer, Integer, Integer,
> String, String, String>> {
> private boolean isRuning = true;
> @Override public void run(SourceContext<Tuple12<String, Integer, Integer,
> String, Integer, Integer, Integer, Integer, Integer, String, String, String>>
> sourceContext) throws Exception {
> while (isRuning) {
>
> sourceContext.collect(Tuple12.of("xxx",21,51,"yyy",1,1,0,2,110,"zzz","aaa","bbb"));
>
>
> sourceContext.collect(Tuple12.of("xxx",21,40,"yyy",2,2,1,2,110,"zzz","aaa","bbb"));
>
> Thread.sleep(Integer.MAX_VALUE); } } @Override public void cancel()
> { isRuning = false;
> }
> }
> }
> {code}
> Oddly,if you remove the "case when" statement in sql1 or remove the "where"
> statement in sql1 ,the program will work well, the exception disappear. So I
> check the physical plan of the sql in the situation. why rownumber() 's
> physical rel is StreamExecOverAggregate ? In nomal is StreamExecRank.
> actually, if you remove the "case when" statement in sql1 or remove the
> "where" statement in sql1, rownumber() 's physical rel came back to
> StreamExecRank.
> {code:java}
> SQL 0 Plan:
> == Abstract Syntax Tree ==
> LogicalProject(biz_bill_no=[$0], task_type=[$1], task_mode=[$2],
> parent_task_no=[$3], total_stage_num=[$4], current_stage_index=[$5],
> use_pre_task_owner=[$6], poi_type=[$7], biz_origin_bill_type=[$8],
> sowing_task_no=[$9])
> +- LogicalFilter(condition=[=($12, 1)])
> +- LogicalProject(biz_bill_no=[$0], task_type=[$1], task_mode=[$2],
> parent_task_no=[$3], total_stage_num=[$4], current_stage_index=[$5],
> use_pre_task_owner=[$6], poi_type=[$7], biz_origin_bill_type=[$8],
> sowing_task_no=[$9], dt=[$10], sowing_task_detail_id=[$11], rn=[ROW_NUMBER()
> OVER (PARTITION BY $10, $11 ORDER BY $1 DESC NULLS LAST)])
> +- LogicalTableScan(table=[[default_catalog, default_database,
> wosOutSowingTaskDetail]])
> == Optimized Logical Plan ==
> Calc(select=[biz_bill_no, task_type, task_mode, parent_task_no,
> total_stage_num, current_stage_index, use_pre_task_owner, poi_type,
> biz_origin_bill_type, sowing_task_no], changelogMode=[I,UA,D])
> +- Rank(strategy=[AppendFastStrategy], rankType=[ROW_NUMBER],
> rankRange=[rankStart=1, rankEnd=1], partitionBy=[dt, sowing_task_detail_id],
> orderBy=[task_type DESC], select=[biz_bill_no, task_type, task_mode,
> parent_task_no, total_stage_num, current_stage_index, use_pre_task_owner,
> poi_type, biz_origin_bill_type, sowing_task_no, dt, sowing_task_detail_id],
> changelogMode=[I,UA,D])
> +- Exchange(distribution=[hash[dt, sowing_task_detail_id]],
> changelogMode=[I])
> +- DataStreamScan(table=[[default_catalog, default_database,
> wosOutSowingTaskDetail]], fields=[biz_bill_no, task_type, task_mode,
> parent_task_no, total_stage_num, current_stage_index, use_pre_task_owner,
> poi_type, biz_origin_bill_type, sowing_task_no, dt, sowing_task_detail_id],
> changelogMode=[I])
> == Physical Execution Plan ==
> Stage 1 : Data Source
> content : Source: Custom Source
> Stage 7 : Operator
> content :
> SourceConversion(table=[default_catalog.default_database.wosOutSowingTaskDetail],
> fields=[biz_bill_no, task_type, task_mode, parent_task_no, total_stage_num,
> current_stage_index, use_pre_task_owner, poi_type, biz_origin_bill_type,
> sowing_task_no, dt, sowing_task_detail_id])
> ship_strategy : FORWARD
> Stage 9 : Operator
> content : Rank(strategy=[AppendFastStrategy],
> rankType=[ROW_NUMBER], rankRange=[rankStart=1, rankEnd=1], partitionBy=[dt,
> sowing_task_detail_id], orderBy=[task_type DESC], select=[biz_bill_no,
> task_type, task_mode, parent_task_no, total_stage_num, current_stage_index,
> use_pre_task_owner, poi_type, biz_origin_bill_type, sowing_task_no, dt,
> sowing_task_detail_id])
> ship_strategy : HASH
> Stage 10 : Operator
> content : Calc(select=[biz_bill_no, task_type, task_mode,
> parent_task_no, total_stage_num, current_stage_index, use_pre_task_owner,
> poi_type, biz_origin_bill_type, sowing_task_no])
> ship_strategy : FORWARD
> SQL 1 Plan:
> org.apache.flink.table.api.TableException: The window can only be ordered in
> ASCENDING mode. {code}
--
This message was sent by Atlassian Jira
(v8.20.1#820001)
