Mailing-List: contact issues-help@flink.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@flink.apache.org
Date: Tue, 28 Feb 2017 15:32:47 +0000 (UTC)
From: "ASF GitHub Bot (JIRA)" <jira@apache.org>
To: issues@flink.apache.org
Message-ID: <JIRA.13041934.1486698791000.19172.1488295967005@Atlassian.JIRA>
In-Reply-To: <JIRA.13041934.1486698791000@Atlassian.JIRA>
References: <JIRA.13041934.1486698791000@Atlassian.JIRA> <JIRA.13041934.1486698791207@jira-lw-us.apache.org>
Subject: [jira] [Commented] (FLINK-5768) Apply new aggregation functions for
 datastream and dataset tables
MIME-Version: 1.0
Content-Type: text/plain; charset=utf-8
Content-Transfer-Encoding: 7bit
archived-at: Tue, 28 Feb 2017 15:32:54 -0000


    [ 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)