Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3423#discussion_r103434035
--- Diff:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamAggregate.scala
---
@@ -119,110 +119,54 @@ class DataStreamAggregate(
s"select: ($aggString)"
val nonKeyedAggOpName = s"window: ($window), select: ($aggString)"
- val mapFunction = AggregateUtil.createPrepareMapFunction(
- namedAggregates,
- grouping,
- inputType)
-
- val mappedInput = inputDS.map(mapFunction).name(prepareOpName)
-
-
- // check whether all aggregates support partial aggregate
- if (AggregateUtil.doAllSupportPartialAggregation(
- namedAggregates.map(_.getKey),
- inputType,
- grouping.length)) {
- // do Incremental Aggregation
- val reduceFunction =
AggregateUtil.createIncrementalAggregateReduceFunction(
- namedAggregates,
- inputType,
- getRowType,
- grouping)
- // grouped / keyed aggregation
- if (groupingKeys.length > 0) {
- val windowFunction =
AggregateUtil.createWindowIncrementalAggregationFunction(
- window,
- namedAggregates,
- inputType,
- rowRelDataType,
- grouping,
- namedProperties)
+ // grouped / keyed aggregation
+ if (groupingKeys.length > 0) {
+ val windowFunction =
AggregateUtil.createWindowIncrementalAggregationFunction(
+ window,
+ rowRelDataType.getFieldCount,
+ namedProperties)
- val keyedStream = mappedInput.keyBy(groupingKeys: _*)
- val windowedStream =
- createKeyedWindowedStream(window, keyedStream)
+ val keyedStream = inputDS.keyBy(groupingKeys: _*)
+ val windowedStream =
+ createKeyedWindowedStream(window, keyedStream)
.asInstanceOf[WindowedStream[Row, Tuple, DataStreamWindow]]
- windowedStream
- .reduce(reduceFunction, windowFunction)
- .returns(rowTypeInfo)
- .name(keyedAggOpName)
- }
- // global / non-keyed aggregation
- else {
- val windowFunction =
AggregateUtil.createAllWindowIncrementalAggregationFunction(
- window,
+ val (aggFunction, accumulatorRowType) =
+ AggregateUtil.createDataStreamAggregateFunction(
namedAggregates,
inputType,
rowRelDataType,
- grouping,
- namedProperties)
-
- val windowedStream =
- createNonKeyedWindowedStream(window, mappedInput)
- .asInstanceOf[AllWindowedStream[Row, DataStreamWindow]]
+ grouping)
- windowedStream
- .reduce(reduceFunction, windowFunction)
- .returns(rowTypeInfo)
- .name(nonKeyedAggOpName)
- }
+ windowedStream
+ .aggregate(aggFunction, windowFunction, accumulatorRowType,
rowTypeInfo, rowTypeInfo)
+ .name(keyedAggOpName)
}
+ // global / non-keyed aggregation
else {
- // do non-Incremental Aggregation
- // grouped / keyed aggregation
- if (groupingKeys.length > 0) {
-
- val windowFunction = AggregateUtil.createWindowAggregationFunction(
- window,
- namedAggregates,
- inputType,
- rowRelDataType,
- grouping,
- namedProperties)
+ val windowFunction =
AggregateUtil.createAllWindowIncrementalAggregationFunction(
--- End diff --
AggregationFunction -> WindowFunction
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
