sunjincheng121 commented on a change in pull request #8311:
[FLINK-10976][table] Add Aggregate operator to Table API
URL: https://github.com/apache/flink/pull/8311#discussion_r281126735
##########
File path:
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/api/tableImpl.scala
##########
@@ -512,6 +528,33 @@ class GroupedTableImpl(
}
}
+/**
+ * The implementation of an [[AggregatedTable]] that has been performed on an
aggregate function.
+ */
+class AggregatedTableImpl(
+ private[flink] val table: Table,
+ private[flink] val groupKeys: Seq[Expression],
+ private[flink] val aggregateFunction: Expression)
+ extends AggregatedTable {
+
+ private val tableImpl = table.asInstanceOf[TableImpl]
+
+ override def select(fields: String): Table = {
+ select(ExpressionParser.parseExpressionList(fields).asScala: _*)
+ }
+
+ override def select(fields: Expression*): Table = {
+ new TableImpl(tableImpl.tableEnv,
+ tableImpl.operationTreeBuilder.project(fields,
Review comment:
I think we can unify aggregate and flatAggregate implementations. i.e. both
`AggregatedTableImpl#select` and `FlatAggregateTableImpl#select` can
using`operationTreeBuilder.project` or `xxTable.select(fields: _*)`. What do
you think?
----------------------------------------------------------------
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
