[
https://issues.apache.org/jira/browse/FLINK-22530?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17341717#comment-17341717
]
Christopher Rost commented on FLINK-22530:
------------------------------------------
Can no one help? It would help if someone can agree that this might be a bug.
I'm not sure if consecutive windowing is the best practice for my case or
should I do it in some other way. Maybe someone knows a workaround? The first
windowed grouping is just a workaround to distinct the elements because in my
use case it is possible that the same entities arrive at slightly different
times. Because I want to aggregate in the second window (e.g., sum, count, avg)
just distinct entities, I have to apply the first window.
> 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)
