Return-Path: X-Original-To: archive-asf-public-internal@cust-asf2.ponee.io Delivered-To: archive-asf-public-internal@cust-asf2.ponee.io Received: from cust-asf.ponee.io (cust-asf.ponee.io [163.172.22.183]) by cust-asf2.ponee.io (Postfix) with ESMTP id F1553200C2F for ; Mon, 6 Mar 2017 09:19:39 +0100 (CET) Received: by cust-asf.ponee.io (Postfix) id EFE3E160B81; Mon, 6 Mar 2017 08:19:39 +0000 (UTC) Delivered-To: archive-asf-public@cust-asf.ponee.io Received: from mail.apache.org (hermes.apache.org [140.211.11.3]) by cust-asf.ponee.io (Postfix) with SMTP id EE3BD160B73 for ; Mon, 6 Mar 2017 09:19:38 +0100 (CET) Received: (qmail 95820 invoked by uid 500); 6 Mar 2017 08:19:38 -0000 Mailing-List: contact issues-help@flink.apache.org; run by ezmlm Precedence: bulk List-Help: List-Unsubscribe: List-Post: List-Id: Reply-To: dev@flink.apache.org Delivered-To: mailing list issues@flink.apache.org Received: (qmail 95811 invoked by uid 99); 6 Mar 2017 08:19:38 -0000 Received: from pnap-us-west-generic-nat.apache.org (HELO spamd2-us-west.apache.org) (209.188.14.142) by apache.org (qpsmtpd/0.29) with ESMTP; Mon, 06 Mar 2017 08:19:38 +0000 Received: from localhost (localhost [127.0.0.1]) by spamd2-us-west.apache.org (ASF Mail Server at spamd2-us-west.apache.org) with ESMTP id B9F4B1A051C for ; Mon, 6 Mar 2017 08:19:37 +0000 (UTC) X-Virus-Scanned: Debian amavisd-new at spamd2-us-west.apache.org X-Spam-Flag: NO X-Spam-Score: -1.547 X-Spam-Level: X-Spam-Status: No, score=-1.547 tagged_above=-999 required=6.31 tests=[KAM_ASCII_DIVIDERS=0.8, RP_MATCHES_RCVD=-2.999, SPF_NEUTRAL=0.652] autolearn=disabled Received: from mx1-lw-us.apache.org ([10.40.0.8]) by localhost (spamd2-us-west.apache.org [10.40.0.9]) (amavisd-new, port 10024) with ESMTP id GHK-oF3GfJWb for ; Mon, 6 Mar 2017 08:19:35 +0000 (UTC) Received: from mailrelay1-us-west.apache.org (mailrelay1-us-west.apache.org [209.188.14.139]) by mx1-lw-us.apache.org (ASF Mail Server at mx1-lw-us.apache.org) with ESMTP id 3886B5FB6B for ; Mon, 6 Mar 2017 08:19:35 +0000 (UTC) Received: from jira-lw-us.apache.org (unknown [207.244.88.139]) by mailrelay1-us-west.apache.org (ASF Mail Server at mailrelay1-us-west.apache.org) with ESMTP id 38CC1E0A0D for ; Mon, 6 Mar 2017 08:19:34 +0000 (UTC) Received: from jira-lw-us.apache.org (localhost [127.0.0.1]) by jira-lw-us.apache.org (ASF Mail Server at jira-lw-us.apache.org) with ESMTP id 4294D2417E for ; Mon, 6 Mar 2017 08:19:33 +0000 (UTC) Date: Mon, 6 Mar 2017 08:19:33 +0000 (UTC) From: "ASF GitHub Bot (JIRA)" To: issues@flink.apache.org Message-ID: In-Reply-To: References: Subject: [jira] [Commented] (FLINK-5653) Add processing time OVER ROWS BETWEEN x PRECEDING aggregation to SQL MIME-Version: 1.0 Content-Type: text/plain; charset=utf-8 Content-Transfer-Encoding: 7bit X-JIRA-FingerPrint: 30527f35849b9dde25b450d4833f0394 archived-at: Mon, 06 Mar 2017 08:19:40 -0000 [ https://issues.apache.org/jira/browse/FLINK-5653?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15896880#comment-15896880 ] ASF GitHub Bot commented on FLINK-5653: --------------------------------------- Github user huawei-flink commented on a diff in the pull request: https://github.com/apache/flink/pull/3443#discussion_r104361048 --- Diff: flink-libraries/flink-table/src/test/java/org/apache/flink/table/api/java/stream/sql/ProcTimeRowStreamAggregationSqlITCase.java --- @@ -0,0 +1,278 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.table.api.java.stream.sql; + +import org.apache.flink.table.api.java.StreamTableEnvironment; +import org.apache.flink.api.java.tuple.Tuple5; +import org.apache.flink.types.Row; +import org.apache.flink.table.api.scala.stream.utils.StreamITCase; +import org.apache.flink.table.api.Table; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.streaming.util.StreamingMultipleProgramsTestBase; +import org.apache.flink.table.api.java.stream.utils.StreamTestData; +import org.junit.Ignore; +import org.junit.Test; + +import java.util.ArrayList; +import java.util.List; + +public class ProcTimeRowStreamAggregationSqlITCase extends StreamingMultipleProgramsTestBase { + + + @Test + public void testMaxAggregatation() throws Exception { + StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); + StreamTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env); + StreamITCase.clear(); + + env.setParallelism(1); + + DataStream> ds = StreamTestData.get5TupleDataStream(env); + Table in = tableEnv.fromDataStream(ds, "a,b,c,d,e"); + tableEnv.registerTable("MyTable", in); + + String sqlQuery = "SELECT a, MAX(c) OVER (PARTITION BY a ORDER BY procTime() ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) AS maxC FROM MyTable"; + Table result = tableEnv.sql(sqlQuery); + + DataStream resultSet = tableEnv.toDataStream(result, Row.class); + resultSet.addSink(new StreamITCase.StringSink()); + env.execute(); + + List expected = new ArrayList<>(); + expected.add("1,0"); + expected.add("2,1"); + expected.add("2,2"); + expected.add("3,3"); + expected.add("3,4"); + expected.add("3,5"); + expected.add("4,6"); + expected.add("4,7"); + expected.add("4,8"); + expected.add("4,9"); + expected.add("5,10"); + expected.add("5,11"); + expected.add("5,12"); + expected.add("5,14"); + expected.add("5,14"); + + StreamITCase.compareWithList(expected); + } + + @Test + public void testMinAggregatation() throws Exception { + StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); + StreamTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env); + StreamITCase.clear(); + + env.setParallelism(1); + + DataStream> ds = StreamTestData.get5TupleDataStream(env); + Table in = tableEnv.fromDataStream(ds, "a,b,c,d,e"); + tableEnv.registerTable("MyTable", in); + + String sqlQuery = "SELECT a, MIN(c) OVER (PARTITION BY a ORDER BY procTime() ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) AS maxC FROM MyTable"; + Table result = tableEnv.sql(sqlQuery); + + DataStream resultSet = tableEnv.toDataStream(result, Row.class); + resultSet.addSink(new StreamITCase.StringSink()); + env.execute(); + + List expected = new ArrayList<>(); + expected.add("1,0"); + expected.add("2,1"); + expected.add("2,1"); + expected.add("3,3"); + expected.add("3,3"); + expected.add("3,4"); + expected.add("4,6"); + expected.add("4,6"); + expected.add("4,7"); + expected.add("4,8"); + expected.add("5,10"); + expected.add("5,10"); + expected.add("5,11"); + expected.add("5,12"); + expected.add("5,13"); + + StreamITCase.compareWithList(expected); + } + + @Test + public void testSumAggregatation() throws Exception { + StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); + StreamTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env); + StreamITCase.clear(); + + env.setParallelism(1); + + DataStream> ds = StreamTestData.get5TupleDataStream(env); + Table in = tableEnv.fromDataStream(ds, "a,b,c,d,e"); + tableEnv.registerTable("MyTable", in); + + String sqlQuery = "SELECT a, SUM(c) OVER (PARTITION BY a ORDER BY procTime() ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) AS sumC FROM MyTable"; + Table result = tableEnv.sql(sqlQuery); + + DataStream resultSet = tableEnv.toDataStream(result, Row.class); + resultSet.addSink(new StreamITCase.StringSink()); + env.execute(); + + List expected = new ArrayList<>(); + expected.add("1,0"); + expected.add("2,1"); + expected.add("2,3"); + expected.add("3,3"); + expected.add("3,7"); + expected.add("3,9"); + expected.add("4,6"); + expected.add("4,13"); + expected.add("4,15"); + expected.add("4,17"); + expected.add("5,10"); + expected.add("5,21"); + expected.add("5,23"); + expected.add("5,26"); + expected.add("5,27"); + + StreamITCase.compareWithList(expected); + } + + @Test + public void testSumMinAggregatation() throws Exception { + StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); + StreamTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env); + StreamITCase.clear(); + + env.setParallelism(1); + + DataStream> ds = StreamTestData.get5TupleDataStream(env); + Table in = tableEnv.fromDataStream(ds, "a,b,c,d,e"); + tableEnv.registerTable("MyTable", in); + + String sqlQuery = "SELECT a, " + + "SUM(c) OVER (PARTITION BY a ORDER BY procTime() ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) AS sumC," + + "MIN(c) OVER (PARTITION BY a ORDER BY procTime() ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) AS sumC" + + " FROM MyTable"; + Table result = tableEnv.sql(sqlQuery); + + DataStream resultSet = tableEnv.toDataStream(result, Row.class); + resultSet.addSink(new StreamITCase.StringSink()); + env.execute(); + + List expected = new ArrayList<>(); + expected.add("1,0,0"); + expected.add("2,1,1"); + expected.add("2,3,1"); + expected.add("3,3,3"); + expected.add("3,7,3"); + expected.add("3,9,4"); + expected.add("4,6,6"); + expected.add("4,13,6"); + expected.add("4,15,7"); + expected.add("4,17,8"); + expected.add("5,10,10"); + expected.add("5,21,10"); + expected.add("5,23,11"); + expected.add("5,26,12"); + expected.add("5,27,13"); + + StreamITCase.compareWithList(expected); + } + + @Test + public void testGlobalSumAggregatation() throws Exception { + StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); + StreamTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env); + StreamITCase.clear(); + + env.setParallelism(1); + + DataStream> ds = StreamTestData.get5TupleDataStream(env); + Table in = tableEnv.fromDataStream(ds, "a,b,c,d,e"); + tableEnv.registerTable("MyTable", in); + + String sqlQuery = "SELECT SUM(c) OVER (ORDER BY procTime() ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) AS sumC FROM MyTable"; + Table result = tableEnv.sql(sqlQuery); + + DataStream resultSet = tableEnv.toDataStream(result, Row.class); + resultSet.addSink(new StreamITCase.StringSink()); + env.execute(); + + List expected = new ArrayList<>(); + expected.add("0"); + expected.add("1"); + expected.add("3"); + expected.add("5"); + expected.add("7"); + expected.add("9"); + expected.add("11"); + expected.add("13"); + expected.add("15"); + expected.add("17"); + expected.add("19"); + expected.add("21"); + expected.add("23"); + expected.add("26"); + expected.add("27"); + + StreamITCase.compareWithList(expected); + } + + @Test + public void testAvgAggregatation() throws Exception { + StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); --- End diff -- I have added a test, of course did not modify this one. > Add processing time OVER ROWS BETWEEN x PRECEDING aggregation to SQL > -------------------------------------------------------------------- > > Key: FLINK-5653 > URL: https://issues.apache.org/jira/browse/FLINK-5653 > Project: Flink > Issue Type: Sub-task > Components: Table API & SQL > Reporter: Fabian Hueske > Assignee: Stefano Bortoli > > The goal of this issue is to add support for OVER ROWS aggregations on processing time streams to the SQL interface. > Queries similar to the following should be supported: > {code} > SELECT > a, > SUM(b) OVER (PARTITION BY c ORDER BY procTime() ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) AS sumB, > MIN(b) OVER (PARTITION BY c ORDER BY procTime() ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) AS minB > FROM myStream > {code} > The following restrictions should initially apply: > - All OVER clauses in the same SELECT clause must be exactly the same. > - The PARTITION BY clause is optional (no partitioning results in single threaded execution). > - The ORDER BY clause may only have procTime() as parameter. procTime() is a parameterless scalar function that just indicates processing time mode. > - UNBOUNDED PRECEDING is not supported (see FLINK-5656) > - FOLLOWING is not supported. > The restrictions will be resolved in follow up issues. If we find that some of the restrictions are trivial to address, we can add the functionality in this issue as well. > This issue includes: > - Design of the DataStream operator to compute OVER ROW aggregates > - Translation from Calcite's RelNode representation (LogicalProject with RexOver expression). -- This message was sent by Atlassian JIRA (v6.3.15#6346)