flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Christopher Rost (Jira)" <j...@apache.org>
Subject [jira] [Comment Edited] (FLINK-22530) RuntimeException after subsequent windowed grouping in TableAPI
Date Mon, 03 May 2021 13:09:00 GMT

    [ https://issues.apache.org/jira/browse/FLINK-22530?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17338361#comment-17338361
] 

Christopher Rost edited comment on FLINK-22530 at 5/3/21, 1:08 PM:
-------------------------------------------------------------------

[~AHeise] and [~libenchao]] It Seems that you had a similar issue with two subsequent tumbling
windows in https://issues.apache.org/jira/browse/FLINK-15494. Do you think it is the same
problem here?


was (Author: chrizzz110):
@libenchao Seems that you had a similar issue with two subsequent tumbling windows.

> RuntimeException after subsequent windowed grouping in TableAPI
> ---------------------------------------------------------------
>
>                 Key: FLINK-22530
>                 URL: https://issues.apache.org/jira/browse/FLINK-22530
>             Project: Flink
>          Issue Type: Bug
>          Components: Table SQL / Planner
>    Affects Versions: 1.12.0
>            Reporter: Christopher Rost
>            Priority: Major
>
> After applying the following using the TableAPI v 1.12.0, an error is thrown: 
> {code:java}
> java.lang.RuntimeException: Error while applying rule StreamExecGroupWindowAggregateRule(in:LOGICAL,out:STREAM_PHYSICAL),
args [rel#505:FlinkLogicalWindowAggregate.LOGICAL.any.None: 0.[NONE].[NONE](input=RelSubset#504,group={1},window=TumblingGroupWindow('w2,
w1_rowtime, 10000),properties=EXPR$1)]{code}
> The code snippet to reproduce:
> {code:java}
>     Table table2 = table1
>       .window(Tumble.over(lit(10).seconds()).on($(EVENT_TIME)).as("w1"))
>       .groupBy($(ID), $(LABEL), $("w1"))
>       .select($(ID), $(LABEL), $("w1").rowtime().as("w1_rowtime"));
> // table2.execute().print(); --> work well
>     Table table3 = table2
>       .window(Tumble.over(lit(10).seconds()).on($("w1_rowtime")).as("w2"))
>       .groupBy($(LABEL), $("w2"))
>       .select(
>         $(LABEL).as("super_label"),
>         lit(1).count().as("super_count"),
>         $("w2").rowtime().as("w2_rowtime")
>       );
> // table3.execute().print(); //--> work well
>    table3.select($("super_label"), $("w2_rowtime"))
>       .execute().print(); // --> throws exception
> {code}
> It seems that the alias "w1_rowtime" is no longer available for further usages of table3,
since the cause of the exception is: 
> {noformat}
> Caused by: java.lang.IllegalArgumentException: field [w1_rowtime] not found; input fields
are: [vertex_id, vertex_label, EXPR$0
> {noformat}
> {{The complete trace:}}
> {code:java}
> java.lang.RuntimeException: Error while applying rule StreamExecGroupWindowAggregateRule(in:LOGICAL,out:STREAM_PHYSICAL),
args [rel#197:FlinkLogicalWindowAggregate.LOGICAL.any.None: 0.[NONE].[NONE](input=RelSubset#196,group={1},window=TumblingGroupWindow('w2,
w1_rowtime, 10000),properties=EXPR$1)]	at org.apache.calcite.plan.volcano.VolcanoRuleCall.onMatch(VolcanoRuleCall.java:256)
> 	at org.apache.calcite.plan.volcano.IterativeRuleDriver.drive(IterativeRuleDriver.java:58)
> 	at org.apache.calcite.plan.volcano.VolcanoPlanner.findBestExp(VolcanoPlanner.java:510)
> 	at org.apache.calcite.tools.Programs$RuleSetProgram.run(Programs.java:312)
> 	at org.apache.flink.table.planner.plan.optimize.program.FlinkVolcanoProgram.optimize(FlinkVolcanoProgram.scala:64)
> 	at org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram$$anonfun$optimize$1.apply(FlinkChainedProgram.scala:62)
> 	at org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram$$anonfun$optimize$1.apply(FlinkChainedProgram.scala:58)
> 	at scala.collection.TraversableOnce$$anonfun$foldLeft$1.apply(TraversableOnce.scala:157)
> 	at scala.collection.TraversableOnce$$anonfun$foldLeft$1.apply(TraversableOnce.scala:157)
> 	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.TraversableOnce$class.foldLeft(TraversableOnce.scala:157)
> 	at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:104)
> 	at org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.optimize(FlinkChainedProgram.scala:57)
> 	at org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.optimizeTree(StreamCommonSubGraphBasedOptimizer.scala:163)
> 	at org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.doOptimize(StreamCommonSubGraphBasedOptimizer.scala:79)
> 	at org.apache.flink.table.planner.plan.optimize.CommonSubGraphBasedOptimizer.optimize(CommonSubGraphBasedOptimizer.scala:77)
> 	at org.apache.flink.table.planner.delegation.PlannerBase.optimize(PlannerBase.scala:286)
> 	at org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:165)
> 	at org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1267)
> 	at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:703)
> 	at org.apache.flink.table.api.internal.TableImpl.execute(TableImpl.java:570)
> 	at edu.leipzig.impl.algorithm.GraphStreamGroupingTest.testDoubleGrouping(GraphStreamGroupingTest.java:224)
> 	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> 	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> 	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> 	at java.base/java.lang.reflect.Method.invoke(Method.java:566)
> 	at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
> 	at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
> 	at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
> 	at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
> 	at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
> 	at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
> 	at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
> 	at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
> 	at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
> 	at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
> 	at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
> 	at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
> 	at org.junit.rules.ExternalResource$1.evaluate(ExternalResource.java:48)
> 	at org.junit.rules.RunRules.evaluate(RunRules.java:20)
> 	at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
> 	at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
> 	at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:69)
> 	at com.intellij.rt.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:33)
> 	at com.intellij.rt.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:221)
> 	at com.intellij.rt.junit.JUnitStarter.main(JUnitStarter.java:54)
> Caused by: java.lang.IllegalArgumentException: field [w1_rowtime] not found; input fields
are: [vertex_id, vertex_label, EXPR$0]
> 	at org.apache.calcite.tools.RelBuilder.field(RelBuilder.java:415)
> 	at org.apache.calcite.tools.RelBuilder.field(RelBuilder.java:398)
> 	at org.apache.flink.table.planner.plan.utils.AggregateUtil$.timeFieldIndex(AggregateUtil.scala:888)
> 	at org.apache.flink.table.planner.plan.rules.physical.stream.StreamExecGroupWindowAggregateRule.convert(StreamExecGroupWindowAggregateRule.scala:81)
> 	at org.apache.calcite.rel.convert.ConverterRule.onMatch(ConverterRule.java:167)
> 	at org.apache.calcite.plan.volcano.VolcanoRuleCall.onMatch(VolcanoRuleCall.java:229)
> 	... 48 more
> {code}
> Maybe correlated to this issue: https://issues.apache.org/jira/browse/FLINK-15494



--
This message was sent by Atlassian Jira
(v8.3.4#803005)

Mime
View raw message