hequn8128 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_r282853222
##########
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
Review comment:
You are right. I think we can remove it.
----------------------------------------------------------------
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:
[email protected]
With regards,
Apache Git Services
