[jira] [Created] (FLINK-20660) Time window operator with computed column triggers an exception in batch mode

[Shang Yuanchun (Jira)]

sujun created FLINK-20660:
-----------------------------

             Summary: Time window operator with computed column triggers an exception in batch mode
                 Key: FLINK-20660
                 URL: https://issues.apache.org/jira/browse/FLINK-20660
             Project: Flink
          Issue Type: Bug
          Components: Table SQL / Planner
    Affects Versions: 1.11.0
            Reporter: sujun


{{Time window operator with computed column triggers an exception in batch mode.}}

{{my test code:}}
{code:java}
public class WindowAggWithBigintTest {
      public static void main(String[] args) throws Exception {
            EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build();
            TableEnvironment tEnv = TableEnvironment.create(settings);            tEnv.registerFunction("longToTimestamp",new LongToTimestamp());            String ddl = "CREATE TABLE source(occur_time bigint,rowtime AS longToTimestamp(occur_time)) WITH ('connector' = 'filesystem', 'format' = 'orc', 'path' = '/path/to/orc')";
            tEnv.executeSql(ddl);            Table table = tEnv.sqlQuery("select TUMBLE_START(rowtime, INTERVAL '1' HOUR) as ts,count(1) as ct from source group by TUMBLE(rowtime, INTERVAL '1' HOUR)");            DiscardingOutputFormat<String> outputFormat = new DiscardingOutputFormat();
            TableResultSink tableResultSink = new TableResultSink(table.getSchema(), outputFormat);
            tEnv.registerTableSink("sink",tableResultSink);
            table.insertInto("sink");
            tEnv.execute("test");
      }      private static class TableResultSink implements StreamTableSink<String> {
            private final TableSchema schema;
            private final DataType rowType;
            private final OutputFormat<String> outputFormat;            TableResultSink(TableSchema schema, OutputFormat<String> outputFormat) {
                  this.schema = schema;
                  this.rowType = schema.toRowDataType();
                  this.outputFormat = outputFormat;
            }            @Override
            public DataType getConsumedDataType() {
                  return rowType;
            }            @Override
            public TableSchema getTableSchema() {
                  return schema;
            }            @Override
            public TableSink<String> configure(String[] fieldNames, TypeInformation<?>[] fieldTypes) {
                  throw new UnsupportedOperationException(
                              "This sink is configured by passing a static schema when initiating");
            }            @Override
            public DataStreamSink<?> consumeDataStream(DataStream<String> dataStream) {
                  return dataStream.writeUsingOutputFormat(outputFormat).setParallelism(1).name("tableResult");
            }
      }
}
{code}
 

Exception:

 
{code:java}
Exception in thread "main" java.lang.RuntimeException: Error while applying rule BatchExecWindowAggregateRule, args [rel#264:FlinkLogicalWindowAggregate.LOGICAL.any.[](input=RelSubset#263,group={},ct=COUNT(),window=TumblingGroupWindow('w$, $f0, 3600000),properties=w$start, w$end, w$rowtime), rel#250:FlinkLogicalLegacyTableSourceScan.LOGICAL.any.[](table=[default_catalog, default_database, source, source: [FileSystemTableSource(occur_time, rowtime)]],fields=occur_time, rowtime)]
      at org.apache.calcite.plan.volcano.VolcanoRuleCall.onMatch(VolcanoRuleCall.java:244)
      at org.apache.calcite.plan.volcano.VolcanoPlanner.findBestExp(VolcanoPlanner.java:636)
      at org.apache.calcite.tools.Programs$RuleSetProgram.run(Programs.java:327)
      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:893)
      at scala.collection.AbstractIterator.foreach(Iterator.scala:1336)
      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.BatchCommonSubGraphBasedOptimizer.optimizeTree(BatchCommonSubGraphBasedOptimizer.scala:86)
      at org.apache.flink.table.planner.plan.optimize.BatchCommonSubGraphBasedOptimizer.org$apache$flink$table$planner$plan$optimize$BatchCommonSubGraphBasedOptimizer$$optimizeBlock(BatchCommonSubGraphBasedOptimizer.scala:57)
      at org.apache.flink.table.planner.plan.optimize.BatchCommonSubGraphBasedOptimizer$$anonfun$doOptimize$1.apply(BatchCommonSubGraphBasedOptimizer.scala:45)
      at org.apache.flink.table.planner.plan.optimize.BatchCommonSubGraphBasedOptimizer$$anonfun$doOptimize$1.apply(BatchCommonSubGraphBasedOptimizer.scala:45)
      at scala.collection.immutable.List.foreach(List.scala:381)
      at org.apache.flink.table.planner.plan.optimize.BatchCommonSubGraphBasedOptimizer.doOptimize(BatchCommonSubGraphBasedOptimizer.scala:45)
      at org.apache.flink.table.planner.plan.optimize.CommonSubGraphBasedOptimizer.optimize(CommonSubGraphBasedOptimizer.scala:77)
      at org.apache.flink.table.planner.delegation.PlannerBase.optimize(PlannerBase.scala:279)
      at org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:164)
      at org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1248)
      at org.apache.flink.table.api.internal.TableEnvironmentImpl.translateAndClearBuffer(TableEnvironmentImpl.java:1240)
      at org.apache.flink.table.api.internal.TableEnvironmentImpl.execute(TableEnvironmentImpl.java:1197)
      at example.WindowAggWithBigintTest.main(WindowAggWithBigintTest.java:34)
Caused by: java.lang.IllegalArgumentException: field [$f0] not found; input fields are: [occur_time]
      at org.apache.calcite.tools.RelBuilder.field(RelBuilder.java:402)
      at org.apache.calcite.tools.RelBuilder.field(RelBuilder.java:385)
      at org.apache.flink.table.planner.plan.utils.AggregateUtil$.timeFieldIndex(AggregateUtil.scala:720)
      at org.apache.flink.table.planner.plan.rules.physical.batch.BatchExecWindowAggregateRule.transformTimeSlidingWindow(BatchExecWindowAggregateRule.scala:161)
      at org.apache.flink.table.planner.plan.rules.physical.batch.BatchExecWindowAggregateRule.onMatch(BatchExecWindowAggregateRule.scala:111)
      at org.apache.calcite.plan.volcano.VolcanoRuleCall.onMatch(VolcanoRuleCall.java:217)
      ... 27 more
{code}



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