flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From fhueske <...@git.apache.org>
Subject [GitHub] flink pull request #5241: [FLINK-8325][table] Add COUNT(*),COUNT(1) supporte...
Date Mon, 08 Jan 2018 22:26:15 GMT
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)
    ```



---

Mime
View raw message