sunjincheng121 commented on a change in pull request #8359:
[FLINK-11051][table] Add streaming window FlatAggregate to Table API
URL: https://github.com/apache/flink/pull/8359#discussion_r282720318
##########
File path:
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/tableImpl.scala
##########
@@ -641,9 +641,57 @@ class WindowGroupedTableImpl(
))
}
- override def flatAggregate(tableAggregateFunction: String):
FlatAggregateTable = ???
+ override def flatAggregate(tableAggregateFunction: String):
FlatAggregateTable = {
+ flatAggregate(ExpressionParser.parseExpression(tableAggregateFunction))
+ }
+
+ override def flatAggregate(tableAggregateFunction: Expression):
FlatAggregateTable = {
+ new WindowFlatAggregateTableImpl(table, groupKeys, tableAggregateFunction,
window)
+ }
+}
+
+/**
+ * The implementation of a [[WindowGroupedTable]] that has been windowed and
grouped on
+ * [[GroupWindow]]s for table aggregate.
+ */
+class WindowFlatAggregateTableImpl(
+ private[flink] val table: Table,
+ private[flink] val groupKeys: Seq[Expression],
+ private[flink] val tableAggFunction: Expression,
+ private[flink] val window: GroupWindow)
+ extends FlatAggregateTable {
+
+ private val tableImpl = table.asInstanceOf[TableImpl]
+
+ override def select(fields: String): Table = {
+ select(ExpressionParser.parseExpressionList(fields).asScala: _*)
+ }
+
+ override def select(fields: Expression*): Table = {
+ val expressionsWithResolvedCalls =
fields.map(_.accept(tableImpl.callResolver)).asJava
+ val extracted = extractAggregationsAndProperties(
+ expressionsWithResolvedCalls,
+ tableImpl.getUniqueAttributeSupplier)
- override def flatAggregate(tableAggregateFunction: Expression):
FlatAggregateTable = ???
+ if (!extracted.getAggregations.isEmpty) {
Review comment:
`extracted.getAggregations.nonEmpty` is better than
`!extracted.getAggregations.isEmpty`, right?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to 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
