sunjincheng121 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_r246240354
##########
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:
Hmm, also make sense to me due to `TableAggFunctionCallAliasable` is inner
type. but I'm not sure, because table can be `as` many times. e.g.:
`tab.as(..).as(..)`, I think @twalthr and @fhueske can give us more useful
suggestions.
----------------------------------------------------------------
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:
[email protected]
With regards,
Apache Git Services
