Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/5241#discussion_r160267929
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamOverAggregate.scala
---
@@ -145,6 +145,14 @@ class DataStreamOverAggregate(
inputSchema.typeInfo,
Some(constants))
+ val constantsTypeInfo =
+
Some(constants).map(_.map(generator.generateExpression(_))).getOrElse(Seq()).map(_.resultType)
+ val aggInputTypeInfo = constantsTypeInfo.++:(inputSchema.fieldTypeInfos)
+
+ val aggregateInputType =
+ cluster.getTypeFactory.asInstanceOf[FlinkTypeFactory]
--- End diff --
Something like:
```
val constantTypes = constants.map(_.getType)
val fieldTypes = input.getRowType.getFieldList.asScala.map(_.getType)
val aggInTypes = constantTypes ++ fieldTypes
val aggInNames = aggInTypes.indices.map("f" + _)
val aggInRowType =
getCluster.getTypeFactory.createStructType(aggInTypes.asJava, aggInNames.asJava)
```
---
