title: Simple TableAPI SQL query doesn't work on Flink 1.10 and Blink
body: <p>I want to define Kafka connector using TableAPI and run SQL over such described table (backed by Kafka). Unfortunately, it seems that <code>Rowtime</code> definition doesn't work as expected.</p>
  
  <p>Here's a reproducible example:</p>
  
  <pre><code>object DefineSource extends App {
  
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.table.api.scala._
  
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
  
    val config = EnvironmentSettings.newInstance().inStreamingMode().useBlinkPlanner().build()
    val tEnv = StreamTableEnvironment.create(env, config)
  
    val rowtime = new Rowtime().watermarksPeriodicBounded(5000)
    val schema = new Schema()
      .field("k", "string")
      .field("ts", "timestamp(3)").rowtime(rowtime)
  
    tEnv.connect(new Kafka()
      .topic("test")
      .version("universal"))
      .withSchema(schema)
      .withFormat(new Csv())
      .createTemporaryTable("InputTable")
  
    val output = tEnv.sqlQuery(
      """SELECT k, COUNT(*)
        |  FROM InputTable
        | GROUP BY k, TUMBLE(ts, INTERVAL '15' MINUTE)
        |""".stripMargin
    )
  
    tEnv.toAppendStream[(String, Long)](output).print()
  
    env.execute()
  }
  </code></pre>
  
  <p>which yields</p>
  
  <pre><code>org.apache.flink.table.api.TableException: Window aggregate can only be defined over a time attribute column, but TIMESTAMP(3) encountered.
      at org.apache.flink.table.planner.plan.rules.logical.StreamLogicalWindowAggregateRule.getInAggregateGroupExpression(StreamLogicalWindowAggregateRule.scala:51)
      at org.apache.flink.table.planner.plan.rules.logical.LogicalWindowAggregateRuleBase.onMatch(LogicalWindowAggregateRuleBase.scala:79)
      at org.apache.calcite.plan.AbstractRelOptPlanner.fireRule(AbstractRelOptPlanner.java:319)
      at org.apache.calcite.plan.hep.HepPlanner.applyRule(HepPlanner.java:560)
      at org.apache.calcite.plan.hep.HepPlanner.applyRules(HepPlanner.java:419)
      at org.apache.calcite.plan.hep.HepPlanner.executeInstruction(HepPlanner.java:256)
      at org.apache.calcite.plan.hep.HepInstruction$RuleInstance.execute(HepInstruction.java:127)
      at org.apache.calcite.plan.hep.HepPlanner.executeProgram(HepPlanner.java:215)
      at org.apache.calcite.plan.hep.HepPlanner.findBestExp(HepPlanner.java:202)
      at org.apache.flink.table.planner.plan.optimize.program.FlinkHepProgram.optimize(FlinkHepProgram.scala:69)
      at org.apache.flink.table.planner.plan.optimize.program.FlinkHepRuleSetProgram.optimize(FlinkHepRuleSetProgram.scala:87)
      at org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.$anonfun$optimize$1(FlinkChainedProgram.scala:62)
      at scala.collection.TraversableOnce.$anonfun$foldLeft$1(TraversableOnce.scala:160)
      at scala.collection.TraversableOnce.$anonfun$foldLeft$1$adapted(TraversableOnce.scala:160)
      at scala.collection.Iterator.foreach(Iterator.scala:941)
      at scala.collection.Iterator.foreach$(Iterator.scala:941)
      at scala.collection.AbstractIterator.foreach(Iterator.scala:1429)
      at scala.collection.IterableLike.foreach(IterableLike.scala:74)
      at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
      at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
      at scala.collection.TraversableOnce.foldLeft(TraversableOnce.scala:160)
      at scala.collection.TraversableOnce.foldLeft$(TraversableOnce.scala:158)
      at scala.collection.AbstractTraversable.foldLeft(Traversable.scala:108)
      at org.apache.flink.table.planner.plan.optimize.program.FlinkChainedProgram.optimize(FlinkChainedProgram.scala:58)
      at org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.optimizeTree(StreamCommonSubGraphBasedOptimizer.scala:170)
      at org.apache.flink.table.planner.plan.optimize.StreamCommonSubGraphBasedOptimizer.doOptimize(StreamCommonSubGraphBasedOptimizer.scala:94)
      at org.apache.flink.table.planner.plan.optimize.CommonSubGraphBasedOptimizer.optimize(CommonSubGraphBasedOptimizer.scala:77)
      at org.apache.flink.table.planner.delegation.PlannerBase.optimize(PlannerBase.scala:248)
      at org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:151)
      at org.apache.flink.table.api.scala.internal.StreamTableEnvironmentImpl.toDataStream(StreamTableEnvironmentImpl.scala:210)
      at org.apache.flink.table.api.scala.internal.StreamTableEnvironmentImpl.toAppendStream(StreamTableEnvironmentImpl.scala:107)
  </code></pre>
  
  <p>I'm on Flink <code>1.10.0</code>.</p>