flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "ASF GitHub Bot (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (FLINK-5768) Apply new aggregation functions for datastream and dataset tables
Date Tue, 28 Feb 2017 15:32:47 GMT

    [ https://issues.apache.org/jira/browse/FLINK-5768?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15888220#comment-15888220
] 

ASF GitHub Bot commented on FLINK-5768:
---------------------------------------

Github user fhueske commented on a diff in the pull request:

    https://github.com/apache/flink/pull/3423#discussion_r103433985
  
    --- 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(
    --- End diff --
    
    AggregationFunction -> WindowFunction?


> Apply new aggregation functions for datastream and dataset tables
> -----------------------------------------------------------------
>
>                 Key: FLINK-5768
>                 URL: https://issues.apache.org/jira/browse/FLINK-5768
>             Project: Flink
>          Issue Type: Sub-task
>          Components: Table API & SQL
>            Reporter: Shaoxuan Wang
>            Assignee: Shaoxuan Wang
>
> Apply new aggregation functions for datastream and dataset tables
> This includes:
> 1. Change the implementation of the DataStream aggregation runtime code to use new aggregation
functions and aggregate dataStream API.
> 2. DataStream will be always running in incremental mode, as explained in 06/Feb/2017
in FLINK5564.
> 2. Change the implementation of the Dataset aggregation runtime code to use new aggregation
functions.
> 3. Clean up unused class and method.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)

Mime
View raw message