Github user fhueske commented on a diff in the pull request: https://github.com/apache/flink/pull/5241#discussion_r162645683 --- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/dataset/DataSetAggregate.scala --- @@ -157,6 +158,7 @@ class DataSetAggregate( } else { inputDS .reduceGroup(finalAgg) + .mapPartition(emptyProcessMapPartition.get) --- End diff -- We should implement the pre-aggregated non-grouped agg also with `mapPartition` then, IMO.
---