hequn8128 commented on a change in pull request #7209: [FLINK-10977][table] Add
UnBounded FlatAggregate operator to streaming Table API
URL: https://github.com/apache/flink/pull/7209#discussion_r245671247
##########
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/expressions/aggregations.scala
##########
@@ -414,3 +414,98 @@ case class AggFunctionCall(
relBuilder.call(this.getSqlAggFunction(), args.map(_.toRexNode): _*)
}
}
+
+case class TableAggFunctionCall(
+ aggregateFunction: TableAggregateFunction[_, _],
+ resultTypeInfo: TypeInformation[_],
+ accTypeInfo: TypeInformation[_],
+ args: Seq[Expression],
+ alias: Option[Seq[Expression]],
+ isDistinct: Boolean)
+ extends Aggregation {
+
+ override private[flink] def children: Seq[Expression] = args
+
+ override def resultType: TypeInformation[_] = resultTypeInfo
+
+ override def validateInput(): ValidationResult = {
+ val signature = children.map(_.resultType)
+ // look for a signature that matches the input types
+ val foundSignature = getAccumulateMethodSignature(aggregateFunction,
signature)
+ if (foundSignature.isEmpty) {
+ ValidationFailure(s"Given parameters do not match any signature. \n" +
+ s"Actual: ${signatureToString(signature)} \n" +
+ s"Expected: ${
+ getMethodSignatures(aggregateFunction, "accumulate")
+ .map(_.drop(1))
+ .map(signatureToString)
+ .mkString(", ")}")
+ } else {
+ ValidationSuccess
+ }
+ }
+
+ override def toString: String =
s"${aggregateFunction.getClass.getSimpleName}($args)"
+
+ override def toAggCall(
+ name: String, isDistinct: Boolean = false)(implicit relBuilder:
RelBuilder): AggCall = {
+ relBuilder.aggregateCall(
+ this.getSqlAggFunction(),
+ isDistinct,
+ false,
+ null,
+ name,
+ args.map(_.toRexNode): _*)
+ }
+
+ override private[flink] def getSqlAggFunction()(implicit relBuilder:
RelBuilder) = {
+ val typeFactory = relBuilder.getTypeFactory.asInstanceOf[FlinkTypeFactory]
+
+ TableAggSqlFunction(
+ aggregateFunction.functionIdentifier,
+ aggregateFunction.toString,
+ aggregateFunction,
+ alias.map(_.map(_.asInstanceOf[UnresolvedFieldReference].name)),
+ resultType,
+ accTypeInfo,
+ typeFactory)
+ }
+
+ override private[flink] def toRexNode(implicit relBuilder: RelBuilder):
RexNode = {
+ relBuilder.call(this.getSqlAggFunction(), args.map(_.toRexNode): _*)
+ }
+}
+
+class TableAggFunctionCallAliasable(
Review comment:
`TableAggFunctionCallAliasable.as()` returns `TableAggFunctionCall`, so that
users can't write multi as, such as `tab.flatAggregate(func('a, 'b) as ('a, 'b,
c) as ('d, 'e, 'f))`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
