You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by huawei-flink <gi...@git.apache.org> on 2017/03/15 13:09:25 UTC
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
GitHub user huawei-flink opened a pull request:
https://github.com/apache/flink/pull/3547
[FLINK-5653] Add processing time OVER ROWS BETWEEN x PRECEDING aggregation to SQL
Thanks for contributing to Apache Flink. Before you open your pull request, please take the following check list into consideration.
If your changes take all of the items into account, feel free to open your pull request. For more information and/or questions please refer to the [How To Contribute guide](http://flink.apache.org/how-to-contribute.html).
In addition to going through the list, please provide a meaningful description of your changes.
- [ x] General
- The pull request references the related JIRA issue ("[FLINK-XXX] Jira title text")
- The pull request addresses only one issue
- Each commit in the PR has a meaningful commit message (including the JIRA id)
- [ x ] Documentation
- Documentation has been added for new functionality
- Old documentation affected by the pull request has been updated
- JavaDoc for public methods has been added
- [ x] Tests & Build
- Functionality added by the pull request is covered by tests
- `mvn clean verify` has been executed successfully locally or a Travis build has passed
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/huawei-flink/flink FLINK-5653b
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/3547.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #3547
----
commit 0e33399e145154b2dd68e0539b65a1baaba512bd
Author: Stefano Bortoli <s....@gmail.com>
Date: 2017-03-15T12:25:54Z
First implementation of aggregation over procTime row bounded window
commit 77a7eff006a1a4aaec2f785994a716bcd9c84133
Author: Stefano Bortoli <s....@gmail.com>
Date: 2017-03-15T12:25:54Z
First implementation of aggregation over procTime row bounded window
commit ecec5527b454ae9c1c3b037103b98660729d8958
Author: Stefano Bortoli <s....@gmail.com>
Date: 2017-03-15T13:03:13Z
Merge branch 'master' of https://github.com/huawei-flink/flink into
FLINK-5653b
# Conflicts:
# flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamOverAggregate.scala
# flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala
----
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by rtudoran <gi...@git.apache.org>.
Github user rtudoran commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106615434
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/BoundedProcessingOverAllWindowFunction.scala ---
@@ -0,0 +1,97 @@
+/*
+ * 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.runtime.aggregate
+
+import org.apache.flink.api.java.tuple.Tuple
+import org.apache.flink.types.Row
+import org.apache.flink.configuration.Configuration
+import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction
+import org.apache.flink.streaming.api.windowing.windows.Window
+import org.apache.flink.util.Collector
+import org.apache.flink.table.functions.AggregateFunction
+import org.apache.flink.api.java.typeutils.RowTypeInfo
+import org.apache.flink.util.Preconditions
+import org.apache.flink.table.functions.Accumulator
+import java.lang.Iterable
+import org.apache.flink.streaming.api.functions.windowing.RichAllWindowFunction
+
+class BoundedProcessingOverAllWindowFunction[W <: Window](
+ private val aggregates: Array[AggregateFunction[_]],
+ private val aggFields: Array[Int],
+ private val forwardedFieldCount: Int)
+ extends RichAllWindowFunction[Row, Row, W] {
+
+ Preconditions.checkNotNull(aggregates)
+ Preconditions.checkNotNull(aggFields)
+ Preconditions.checkArgument(aggregates.length == aggFields.length)
+
+ private var output: Row = _
+ private var accumulators: Row = _
+ private var reuse: Row = _
+
+ output = new Row(forwardedFieldCount + aggregates.length)
+ if (null == accumulators) {
+ accumulators = new Row(aggregates.length)
+ }
+
+ override def apply(
+ window: W,
+ records: Iterable[Row],
+ out: Collector[Row]): Unit = {
+
+ var i = 0
+ // setting the accumulators for each aggregation
+ while (i < aggregates.length) {
+ accumulators.setField(i, aggregates(i).createAccumulator())
--- End diff --
What is the meaning of the acc?
What should be passed to the accumulator for reset? logically I would expect that this is a non parameterized method
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by sunjincheng121 <gi...@git.apache.org>.
Github user sunjincheng121 commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106193738
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamOverAggregate.scala ---
@@ -17,34 +17,38 @@
*/
package org.apache.flink.table.plan.nodes.datastream
-import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
+import org.apache.calcite.plan.{ RelOptCluster, RelTraitSet }
import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rel.core.AggregateCall
-import org.apache.calcite.rel.{RelNode, RelWriter, SingleRel}
+import org.apache.calcite.rel.{ RelNode, RelWriter, SingleRel }
import org.apache.flink.api.java.typeutils.RowTypeInfo
import org.apache.flink.streaming.api.datastream.DataStream
-import org.apache.flink.table.api.{StreamTableEnvironment, TableException}
+import org.apache.flink.table.api.{ StreamTableEnvironment, TableException }
import org.apache.flink.table.calcite.FlinkTypeFactory
import org.apache.flink.table.runtime.aggregate._
import org.apache.flink.table.plan.nodes.OverAggregate
import org.apache.flink.types.Row
import org.apache.calcite.rel.core.Window
import org.apache.calcite.rel.core.Window.Group
-import java.util.{List => JList}
+import java.util.{ List => JList }
import org.apache.flink.table.functions.{ProcTimeType, RowTimeType}
import org.apache.flink.table.runtime.aggregate.AggregateUtil.CalcitePair
+import org.apache.calcite.sql.`type`.BasicSqlType
+import org.apache.flink.streaming.api.functions.windowing.WindowFunction
+import org.apache.flink.streaming.api.windowing.windows.GlobalWindow
+import org.apache.flink.api.java.tuple.Tuple
class DataStreamOverAggregate(
- logicWindow: Window,
- cluster: RelOptCluster,
- traitSet: RelTraitSet,
- inputNode: RelNode,
- rowRelDataType: RelDataType,
- inputType: RelDataType)
- extends SingleRel(cluster, traitSet, inputNode)
- with OverAggregate
- with DataStreamRel {
+ logicWindow: Window,
+ cluster: RelOptCluster,
+ traitSet: RelTraitSet,
+ inputNode: RelNode,
+ rowRelDataType: RelDataType,
+ inputType: RelDataType)
+ extends SingleRel(cluster, traitSet, inputNode)
--- End diff --
Indent 2 space.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106520755
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala ---
@@ -89,6 +116,58 @@ object AggregateUtil {
aggregationStateType)
}
}
+
+ private[flink] def CreateBoundedProcessingOverWindowFunction(
+ namedAggregates: Seq[CalcitePair[AggregateCall, String]],
+ inputType: RelDataType): WindowFunction[Row, Row, Tuple, GlobalWindow] = {
+
+ val (aggFields, aggregates) =
+ transformToAggregateFunctions(
+ namedAggregates.map(_.getKey),
+ inputType,
+ needRetraction = false)
+
+ val aggregationStateType: RowTypeInfo =
--- End diff --
this is not used
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by huawei-flink <gi...@git.apache.org>.
Github user huawei-flink commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106616178
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamOverAggregate.scala ---
@@ -106,9 +113,14 @@ class DataStreamOverAggregate(
if (overWindow.lowerBound.isUnbounded &&
overWindow.upperBound.isCurrentRow) {
createUnboundedAndCurrentRowProcessingTimeOverWindow(inputDS)
+ } // lowerbound is a BasicType and upperbound is PRECEEDING or CURRENT ROW
+ else if (overWindow.lowerBound.getOffset.getType.isInstanceOf[BasicSqlType]
--- End diff --
I guess that is also a way to do it. The check allows to distinguish between time bounded and row bounded. I have no particular affection for my solution, it just worked. I will apply and test yours as well.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by sunjincheng121 <gi...@git.apache.org>.
Github user sunjincheng121 commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106195116
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamOverAggregate.scala ---
@@ -159,6 +168,42 @@ class DataStreamOverAggregate(
result
}
+def createBoundedAndCurrentRowProcessingTimeOverWindow(
+ inputDS: DataStream[Row]): DataStream[Row] = {
+
+ val overWindow: Group = logicWindow.groups.get(0)
+ val partitionKeys: Array[Int] = overWindow.keys.toArray
+ val namedAggregates: Seq[CalcitePair[AggregateCall, String]] = generateNamedAggregates
+
+ // get the output types
+ val rowTypeInfo = FlinkTypeFactory.toInternalRowTypeInfo(getRowType).asInstanceOf[RowTypeInfo]
+
+ val result: DataStream[Row] =
+ // partitioned aggregation
+ if (partitionKeys.nonEmpty) {
+ val windowFunction = AggregateUtil.CreateBoundedProcessingOverWindowFunction(
+ namedAggregates,
+ inputType)
+
+ val lowerbound: Int = AggregateUtil.getLowerBoundary(
+ logicWindow.constants,
+ overWindow.lowerBound,
+ getInput())
+
+ inputDS
+ .keyBy(partitionKeys: _*)
+ .countWindow(lowerbound, 1).apply(windowFunction)
+ .returns(rowTypeInfo)
+ .name(aggOpName)
+ .asInstanceOf[DataStream[Row]]
+ } // global non-partitioned aggregation
+ else {
+ throw TableException(
--- End diff --
Can you supported not-partitioned case in this JIRA.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106532760
--- Diff: flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/sql/ProcTimeRowBoundedAggregationTest.scala ---
@@ -0,0 +1,311 @@
+/*
+ * 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.scala.stream.sql
+
+import org.apache.flink.api.scala._
+import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
+import org.apache.flink.table.api.{ TableEnvironment, TableException }
+import org.apache.flink.table.api.scala._
+import org.apache.flink.table.api.scala.stream.utils.{
+ StreamingWithStateTestBase,
+ StreamITCase,
+ StreamTestData
+}
+import org.junit.Assert._
+import org.junit._
+import scala.collection.mutable
+import org.apache.flink.types.Row
+
+class ProcTimeRowBoundedAggregationTest extends StreamingWithStateTestBase {
--- End diff --
This class is an integration tests because it starts a Flink minicluster. Hence its name should end with `ITCase` and not `Test` which would indicate a lightweight unit test.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by sunjincheng121 <gi...@git.apache.org>.
Github user sunjincheng121 commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106344697
--- Diff: flink-libraries/flink-table/src/test/java/org/apache/flink/table/api/java/stream/sql/ProcTimeRowStreamAggregationSqlITCase.java ---
@@ -0,0 +1,317 @@
+/*
+ * 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 java.util.ArrayList;
+import java.util.List;
+
+import org.apache.flink.api.java.tuple.Tuple5;
+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.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.java.StreamTableEnvironment;
+import org.apache.flink.table.api.java.stream.utils.StreamTestData;
+import org.apache.flink.table.api.scala.stream.utils.StreamITCase;
+import org.apache.flink.types.Row;
+import org.junit.Ignore;
+import org.junit.Test;
+
+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<Tuple5<Integer, Long, Integer, String, Long>> 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<Row> resultSet = tableEnv.toDataStream(result, Row.class);
+ resultSet.addSink(new StreamITCase.StringSink());
+ env.execute();
+
+ List<String> 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<Tuple5<Integer, Long, Integer, String, Long>> 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<Row> resultSet = tableEnv.toDataStream(result, Row.class);
+ resultSet.addSink(new StreamITCase.StringSink());
+ env.execute();
+
+ List<String> 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<Tuple5<Integer, Long, Integer, String, Long>> 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<Row> resultSet = tableEnv.toDataStream(result, Row.class);
+ resultSet.addSink(new StreamITCase.StringSink());
+ env.execute();
+
+ List<String> 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<Tuple5<Integer, Long, Integer, String, Long>> 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 minC"
+ + " FROM MyTable";
+ Table result = tableEnv.sql(sqlQuery);
+
+ DataStream<Row> resultSet = tableEnv.toDataStream(result, Row.class);
+ resultSet.addSink(new StreamITCase.StringSink());
+ env.execute();
+
+ List<String> 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");
--- End diff --
This result incorrect.The reason for the error I left the comments in `DataStreamOverAggregate.scala`
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106517060
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala ---
@@ -89,6 +116,58 @@ object AggregateUtil {
aggregationStateType)
}
}
+
+ private[flink] def CreateBoundedProcessingOverWindowFunction(
+ namedAggregates: Seq[CalcitePair[AggregateCall, String]],
--- End diff --
Please align the arguments as the other functions do.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by huawei-flink <gi...@git.apache.org>.
Github user huawei-flink closed the pull request at:
https://github.com/apache/flink/pull/3547
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by sunjincheng121 <gi...@git.apache.org>.
Github user sunjincheng121 commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106335062
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/logical/rel/LogicalWindowAggregate.scala ---
@@ -18,8 +18,6 @@
package org.apache.flink.table.plan.logical.rel
-import java.util
--- End diff --
Your current package is `org.apache.flink.table.plan.logical.rel`, in this package there is a` org.apache.flink.table.plan.logical.rel.util`, this and `util .List` package conflicts. So you can try the following 2 way solution:
solution1:
```
import java.util.List
groupSets: List[ImmutableBitSet],
aggCalls: List[AggregateCall])
```
solution2:
```
import java.{util => JUtil}
groupSets: JUtil.List[ImmutableBitSet],
aggCalls: JUtil.List[AggregateCall])
```
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by huawei-flink <gi...@git.apache.org>.
Github user huawei-flink commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106219814
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala ---
@@ -19,33 +19,124 @@ package org.apache.flink.table.runtime.aggregate
import java.util
-import org.apache.calcite.rel.`type`._
+import scala.collection.JavaConversions.asScalaBuffer
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rel.core.AggregateCall
+import org.apache.calcite.rex.RexInputRef
+import org.apache.calcite.rex.RexLiteral
+import org.apache.calcite.rex.RexWindowBound
+import org.apache.calcite.sql.SqlAggFunction
+import org.apache.calcite.sql.SqlKind
import org.apache.calcite.sql.`type`.SqlTypeName
-import org.apache.calcite.sql.`type`.SqlTypeName._
-import org.apache.calcite.sql.fun._
-import org.apache.calcite.sql.{SqlAggFunction, SqlKind}
-import org.apache.flink.streaming.api.functions.ProcessFunction
+import org.apache.calcite.sql.`type`.SqlTypeName.BIGINT
+import org.apache.calcite.sql.`type`.SqlTypeName.BOOLEAN
+import org.apache.calcite.sql.`type`.SqlTypeName.DECIMAL
+import org.apache.calcite.sql.`type`.SqlTypeName.DOUBLE
+import org.apache.calcite.sql.`type`.SqlTypeName.FLOAT
+import org.apache.calcite.sql.`type`.SqlTypeName.INTEGER
+import org.apache.calcite.sql.`type`.SqlTypeName.SMALLINT
+import org.apache.calcite.sql.`type`.SqlTypeName.TINYINT
--- End diff --
sure
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by huawei-flink <gi...@git.apache.org>.
Github user huawei-flink commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106222135
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/logical/rel/LogicalWindowAggregate.scala ---
@@ -18,8 +18,6 @@
package org.apache.flink.table.plan.logical.rel
-import java.util
--- End diff --
In principle I agree, but I it caused some building problem. Is there a practical reason for not importing the List class directly rather than creating "half references"?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by sunjincheng121 <gi...@git.apache.org>.
Github user sunjincheng121 commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106336720
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala ---
@@ -19,33 +19,124 @@ package org.apache.flink.table.runtime.aggregate
import java.util
-import org.apache.calcite.rel.`type`._
+import scala.collection.JavaConversions.asScalaBuffer
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rel.core.AggregateCall
+import org.apache.calcite.rex.RexInputRef
+import org.apache.calcite.rex.RexLiteral
+import org.apache.calcite.rex.RexWindowBound
+import org.apache.calcite.sql.SqlAggFunction
+import org.apache.calcite.sql.SqlKind
import org.apache.calcite.sql.`type`.SqlTypeName
-import org.apache.calcite.sql.`type`.SqlTypeName._
-import org.apache.calcite.sql.fun._
-import org.apache.calcite.sql.{SqlAggFunction, SqlKind}
-import org.apache.flink.streaming.api.functions.ProcessFunction
+import org.apache.calcite.sql.`type`.SqlTypeName.BIGINT
+import org.apache.calcite.sql.`type`.SqlTypeName.BOOLEAN
+import org.apache.calcite.sql.`type`.SqlTypeName.DECIMAL
+import org.apache.calcite.sql.`type`.SqlTypeName.DOUBLE
+import org.apache.calcite.sql.`type`.SqlTypeName.FLOAT
+import org.apache.calcite.sql.`type`.SqlTypeName.INTEGER
+import org.apache.calcite.sql.`type`.SqlTypeName.SMALLINT
+import org.apache.calcite.sql.`type`.SqlTypeName.TINYINT
--- End diff --
Yes, I had clone your code.May be in this case,we can use `._` ,
e.g.: `import org.apache.flink.table.functions.aggfunctions._` In this way, we can reduce 53 line to 1 line. What do you think?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by huawei-flink <gi...@git.apache.org>.
Github user huawei-flink commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106868683
--- Diff: flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/sql/ProcTimeRowBoundedAggregationTest.scala ---
@@ -0,0 +1,311 @@
+/*
+ * 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.scala.stream.sql
+
+import org.apache.flink.api.scala._
+import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
+import org.apache.flink.table.api.{ TableEnvironment, TableException }
+import org.apache.flink.table.api.scala._
+import org.apache.flink.table.api.scala.stream.utils.{
+ StreamingWithStateTestBase,
+ StreamITCase,
+ StreamTestData
+}
+import org.junit.Assert._
+import org.junit._
+import scala.collection.mutable
+import org.apache.flink.types.Row
+
+class ProcTimeRowBoundedAggregationTest extends StreamingWithStateTestBase {
--- End diff --
@fhueske Should I add all the test unit under the SqlITCase class?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by sunjincheng121 <gi...@git.apache.org>.
Github user sunjincheng121 commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106344891
--- Diff: flink-libraries/flink-table/src/test/java/org/apache/flink/table/api/java/stream/sql/ProcTimeRowStreamAggregationSqlITCase.java ---
@@ -0,0 +1,317 @@
+/*
+ * 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 java.util.ArrayList;
+import java.util.List;
+
+import org.apache.flink.api.java.tuple.Tuple5;
+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.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.java.StreamTableEnvironment;
+import org.apache.flink.table.api.java.stream.utils.StreamTestData;
+import org.apache.flink.table.api.scala.stream.utils.StreamITCase;
+import org.apache.flink.types.Row;
+import org.junit.Ignore;
+import org.junit.Test;
+
+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<Tuple5<Integer, Long, Integer, String, Long>> 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<Row> resultSet = tableEnv.toDataStream(result, Row.class);
+ resultSet.addSink(new StreamITCase.StringSink());
+ env.execute();
+
+ List<String> 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<Tuple5<Integer, Long, Integer, String, Long>> 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<Row> resultSet = tableEnv.toDataStream(result, Row.class);
+ resultSet.addSink(new StreamITCase.StringSink());
+ env.execute();
+
+ List<String> 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");
--- End diff --
This result incorrect.The reason for the error I left the comments in `DataStreamOverAggregate.scala`
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by huawei-flink <gi...@git.apache.org>.
Github user huawei-flink commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106219673
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/BoundedProcessingOverWindowFunction.scala ---
@@ -0,0 +1,97 @@
+/*
+ * 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.runtime.aggregate
+
+import org.apache.flink.api.java.tuple.Tuple
+import org.apache.flink.types.Row
+import org.apache.flink.configuration.Configuration
+import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction
+import org.apache.flink.streaming.api.windowing.windows.Window
+import org.apache.flink.util.Collector
+import org.apache.flink.table.functions.AggregateFunction
+import org.apache.flink.api.java.typeutils.RowTypeInfo
+import org.apache.flink.util.Preconditions
+import org.apache.flink.table.functions.Accumulator
+import java.lang.Iterable
+
+class BoundedProcessingOverWindowFunction[W <: Window](
--- End diff --
@sunjincheng121 thanks for the suggestion. I decided to use Window because it is convenient in the row bounded case. Within the window I apply the incremental aggregation in the same way.
It is not clear to me what are the flexibility advantages in this specific case. Can you be more explicit?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106517261
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala ---
@@ -89,6 +116,58 @@ object AggregateUtil {
aggregationStateType)
}
}
+
+ private[flink] def CreateBoundedProcessingOverWindowFunction(
+ namedAggregates: Seq[CalcitePair[AggregateCall, String]],
+ inputType: RelDataType): WindowFunction[Row, Row, Tuple, GlobalWindow] = {
+
+ val (aggFields, aggregates) =
+ transformToAggregateFunctions(
+ namedAggregates.map(_.getKey),
+ inputType,
+ needRetraction = false)
+
+ val aggregationStateType: RowTypeInfo =
+ createDataSetAggregateBufferDataType(Array(), aggregates, inputType)
+
+ new BoundedProcessingOverWindowFunction[GlobalWindow](
+ aggregates,
+ aggFields,
+ inputType.getFieldCount)
+
+ }
+
+ private[flink] def CreateBoundedProcessingOverGlobalWindowFunction(
--- End diff --
indent -2
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by huawei-flink <gi...@git.apache.org>.
Github user huawei-flink commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106917566
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala ---
@@ -17,35 +17,62 @@
*/
package org.apache.flink.table.runtime.aggregate
-import java.util
+import scala.collection.JavaConversions.asScalaBuffer
--- End diff --
I need that import, without it does not build.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by sunjincheng121 <gi...@git.apache.org>.
Github user sunjincheng121 commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106343147
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamOverAggregate.scala ---
@@ -130,32 +142,76 @@ class DataStreamOverAggregate(
val rowTypeInfo = FlinkTypeFactory.toInternalRowTypeInfo(getRowType).asInstanceOf[RowTypeInfo]
val result: DataStream[Row] =
- // partitioned aggregation
- if (partitionKeys.nonEmpty) {
- val processFunction = AggregateUtil.CreateUnboundedProcessingOverProcessFunction(
- namedAggregates,
- inputType)
+ // partitioned aggregation
+ if (partitionKeys.nonEmpty) {
+ val processFunction = AggregateUtil.CreateUnboundedProcessingOverProcessFunction(
+ namedAggregates,
+ inputType)
- inputDS
+ inputDS
.keyBy(partitionKeys: _*)
.process(processFunction)
.returns(rowTypeInfo)
.name(aggOpName)
.asInstanceOf[DataStream[Row]]
- }
- // non-partitioned aggregation
- else {
- val processFunction = AggregateUtil.CreateUnboundedProcessingOverProcessFunction(
- namedAggregates,
- inputType,
- false)
-
- inputDS
- .process(processFunction).setParallelism(1).setMaxParallelism(1)
- .returns(rowTypeInfo)
- .name(aggOpName)
- .asInstanceOf[DataStream[Row]]
- }
+ } // non-partitioned aggregation
+ else {
+ val processFunction = AggregateUtil.CreateUnboundedProcessingOverProcessFunction(
+ namedAggregates,
+ inputType,
+ false)
+
+ inputDS
+ .process(processFunction).setParallelism(1).setMaxParallelism(1)
+ .returns(rowTypeInfo)
+ .name(aggOpName)
+ .asInstanceOf[DataStream[Row]]
+ }
+ result
+ }
+
+ def createBoundedAndCurrentRowProcessingTimeOverWindow(
+ inputDS: DataStream[Row]): DataStream[Row] = {
+
+ val overWindow: Group = logicWindow.groups.get(0)
+ val partitionKeys: Array[Int] = overWindow.keys.toArray
+ val namedAggregates: Seq[CalcitePair[AggregateCall, String]] = generateNamedAggregates
+
+ // get the output types
+ val rowTypeInfo = FlinkTypeFactory.toInternalRowTypeInfo(getRowType).asInstanceOf[RowTypeInfo]
+
+ val lowerbound: Int = AggregateUtil.getLowerBoundary(
+ logicWindow.constants,
+ overWindow.lowerBound,
+ getInput())
+
+ val result: DataStream[Row] =
+ // partitioned aggregation
+ if (partitionKeys.nonEmpty) {
+ val windowFunction = AggregateUtil.CreateBoundedProcessingOverWindowFunction(
+ namedAggregates,
+ inputType)
+ inputDS
+ .keyBy(partitionKeys: _*)
+ .countWindow(lowerbound,1)
+ .apply(windowFunction)
+ .returns(rowTypeInfo)
+ .name(aggOpName)
+ .asInstanceOf[DataStream[Row]]
+ } // global non-partitioned aggregation
+ else {
+ val windowFunction = AggregateUtil.CreateBoundedProcessingOverGlobalWindowFunction(
+ namedAggregates,
+ inputType)
+
+ inputDS
+ .countWindowAll(lowerbound,1)
--- End diff --
`lowerbound -> (lowerbound+1)`
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by sunjincheng121 <gi...@git.apache.org>.
Github user sunjincheng121 commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106344846
--- Diff: flink-libraries/flink-table/src/test/java/org/apache/flink/table/api/java/stream/sql/ProcTimeRowStreamAggregationSqlITCase.java ---
@@ -0,0 +1,317 @@
+/*
+ * 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 java.util.ArrayList;
+import java.util.List;
+
+import org.apache.flink.api.java.tuple.Tuple5;
+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.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.java.StreamTableEnvironment;
+import org.apache.flink.table.api.java.stream.utils.StreamTestData;
+import org.apache.flink.table.api.scala.stream.utils.StreamITCase;
+import org.apache.flink.types.Row;
+import org.junit.Ignore;
+import org.junit.Test;
+
+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<Tuple5<Integer, Long, Integer, String, Long>> 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<Row> resultSet = tableEnv.toDataStream(result, Row.class);
+ resultSet.addSink(new StreamITCase.StringSink());
+ env.execute();
+
+ List<String> 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<Tuple5<Integer, Long, Integer, String, Long>> 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<Row> resultSet = tableEnv.toDataStream(result, Row.class);
+ resultSet.addSink(new StreamITCase.StringSink());
+ env.execute();
+
+ List<String> 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<Tuple5<Integer, Long, Integer, String, Long>> 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<Row> resultSet = tableEnv.toDataStream(result, Row.class);
+ resultSet.addSink(new StreamITCase.StringSink());
+ env.execute();
+
+ List<String> 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<Tuple5<Integer, Long, Integer, String, Long>> 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 minC"
+ + " FROM MyTable";
+ Table result = tableEnv.sql(sqlQuery);
+
+ DataStream<Row> resultSet = tableEnv.toDataStream(result, Row.class);
+ resultSet.addSink(new StreamITCase.StringSink());
+ env.execute();
+
+ List<String> 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<Tuple5<Integer, Long, Integer, String, Long>> 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<Row> resultSet = tableEnv.toDataStream(result, Row.class);
+ resultSet.addSink(new StreamITCase.StringSink());
+ env.execute();
+
+ List<String> 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");
--- End diff --
This result incorrect.The reason for the error I left the comments in `DataStreamOverAggregate.scala`
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106538072
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamOverAggregate.scala ---
@@ -106,9 +113,14 @@ class DataStreamOverAggregate(
if (overWindow.lowerBound.isUnbounded &&
overWindow.upperBound.isCurrentRow) {
createUnboundedAndCurrentRowProcessingTimeOverWindow(inputDS)
+ } // lowerbound is a BasicType and upperbound is PRECEEDING or CURRENT ROW
+ else if (overWindow.lowerBound.getOffset.getType.isInstanceOf[BasicSqlType]
--- End diff --
What is the check `overWindow.lowerBound.getOffset.getType.isInstanceOf[BasicSqlType]` for and why do you check if `overWindow.upperBound.isPreceding`?
Should the condition rather be like this:
```
else if (overWindow.lowerBound.isPreceding() && !overWindow.lowerBound.isUnbounded() && // bounded preceding
overWindow.upperBound.isCurrentRow() && // until current row
overWindow.isRows) // is rows window
```
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106525346
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala ---
@@ -89,6 +116,58 @@ object AggregateUtil {
aggregationStateType)
}
}
+
+ private[flink] def CreateBoundedProcessingOverWindowFunction(
+ namedAggregates: Seq[CalcitePair[AggregateCall, String]],
+ inputType: RelDataType): WindowFunction[Row, Row, Tuple, GlobalWindow] = {
+
+ val (aggFields, aggregates) =
+ transformToAggregateFunctions(
+ namedAggregates.map(_.getKey),
+ inputType,
+ needRetraction = false)
+
+ val aggregationStateType: RowTypeInfo =
+ createDataSetAggregateBufferDataType(Array(), aggregates, inputType)
+
+ new BoundedProcessingOverWindowFunction[GlobalWindow](
+ aggregates,
+ aggFields,
+ inputType.getFieldCount)
+
+ }
+
+ private[flink] def CreateBoundedProcessingOverGlobalWindowFunction(
+ namedAggregates: Seq[CalcitePair[AggregateCall, String]],
+ inputType: RelDataType): AllWindowFunction[Row,Row,GlobalWindow] = {
+
+ val (aggFields, aggregates) =
+ transformToAggregateFunctions(
+ namedAggregates.map(_.getKey),
+ inputType,
+ needRetraction = false)
+
+ val aggregationStateType: RowTypeInfo =
+ createDataSetAggregateBufferDataType(Array(), aggregates, inputType)
+
+ new BoundedProcessingOverAllWindowFunction[GlobalWindow](
+ aggregates,
+ aggFields,
+ inputType.getFieldCount)
+
+ }
+
+ def getLowerBoundary(
+ constants: ImmutableList[RexLiteral],
+ lowerBound: RexWindowBound,
+ input: RelNode):Int = {
--- End diff --
`input: ReoNode): Int = {` +space
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by sunjincheng121 <gi...@git.apache.org>.
Github user sunjincheng121 commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106193639
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamOverAggregate.scala ---
@@ -17,34 +17,38 @@
*/
package org.apache.flink.table.plan.nodes.datastream
-import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
+import org.apache.calcite.plan.{ RelOptCluster, RelTraitSet }
import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rel.core.AggregateCall
-import org.apache.calcite.rel.{RelNode, RelWriter, SingleRel}
+import org.apache.calcite.rel.{ RelNode, RelWriter, SingleRel }
import org.apache.flink.api.java.typeutils.RowTypeInfo
import org.apache.flink.streaming.api.datastream.DataStream
-import org.apache.flink.table.api.{StreamTableEnvironment, TableException}
+import org.apache.flink.table.api.{ StreamTableEnvironment, TableException }
import org.apache.flink.table.calcite.FlinkTypeFactory
import org.apache.flink.table.runtime.aggregate._
import org.apache.flink.table.plan.nodes.OverAggregate
import org.apache.flink.types.Row
import org.apache.calcite.rel.core.Window
import org.apache.calcite.rel.core.Window.Group
-import java.util.{List => JList}
+import java.util.{ List => JList }
import org.apache.flink.table.functions.{ProcTimeType, RowTimeType}
import org.apache.flink.table.runtime.aggregate.AggregateUtil.CalcitePair
+import org.apache.calcite.sql.`type`.BasicSqlType
+import org.apache.flink.streaming.api.functions.windowing.WindowFunction
+import org.apache.flink.streaming.api.windowing.windows.GlobalWindow
+import org.apache.flink.api.java.tuple.Tuple
class DataStreamOverAggregate(
- logicWindow: Window,
- cluster: RelOptCluster,
- traitSet: RelTraitSet,
- inputNode: RelNode,
- rowRelDataType: RelDataType,
- inputType: RelDataType)
- extends SingleRel(cluster, traitSet, inputNode)
- with OverAggregate
- with DataStreamRel {
+ logicWindow: Window,
--- End diff --
Indent 4 space.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106488388
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/BoundedProcessingOverWindowFunction.scala ---
@@ -0,0 +1,97 @@
+/*
+ * 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.runtime.aggregate
+
+import org.apache.flink.api.java.tuple.Tuple
+import org.apache.flink.types.Row
+import org.apache.flink.configuration.Configuration
+import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction
+import org.apache.flink.streaming.api.windowing.windows.Window
+import org.apache.flink.util.Collector
+import org.apache.flink.table.functions.AggregateFunction
+import org.apache.flink.api.java.typeutils.RowTypeInfo
+import org.apache.flink.util.Preconditions
+import org.apache.flink.table.functions.Accumulator
+import java.lang.Iterable
+
+class BoundedProcessingOverWindowFunction[W <: Window](
--- End diff --
As @huawei-flink pointed out, I agreed to use a WindowFunction for the bounded OVER ROW case in our previous discussions. Although it would not be consistent with the other OVER windows, this would be an easier solution, IMO.
However, I think we have to use a `ProcessFunction` for correctness reasons. An OVER aggregation must emit one row for each input row. If we have an `OVER (PARTITION BY a ORDER BY procTime() ROWS BETWEEN 3 PRECEDING AND CURRENT ROW)` and implement this with a `SlidingCountWindow(3, 1)` we will lose the two rows because the window function is called the first time, when three elements have arrived.
Regarding the aggregation of COUNT DISTINCT, we could implement a custom aggregation function, similar to the retractable min or max aggregation function.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by sunjincheng121 <gi...@git.apache.org>.
Github user sunjincheng121 commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106195520
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala ---
@@ -19,33 +19,124 @@ package org.apache.flink.table.runtime.aggregate
import java.util
-import org.apache.calcite.rel.`type`._
+import scala.collection.JavaConversions.asScalaBuffer
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rel.core.AggregateCall
+import org.apache.calcite.rex.RexInputRef
+import org.apache.calcite.rex.RexLiteral
+import org.apache.calcite.rex.RexWindowBound
+import org.apache.calcite.sql.SqlAggFunction
+import org.apache.calcite.sql.SqlKind
import org.apache.calcite.sql.`type`.SqlTypeName
-import org.apache.calcite.sql.`type`.SqlTypeName._
-import org.apache.calcite.sql.fun._
-import org.apache.calcite.sql.{SqlAggFunction, SqlKind}
-import org.apache.flink.streaming.api.functions.ProcessFunction
+import org.apache.calcite.sql.`type`.SqlTypeName.BIGINT
+import org.apache.calcite.sql.`type`.SqlTypeName.BOOLEAN
+import org.apache.calcite.sql.`type`.SqlTypeName.DECIMAL
+import org.apache.calcite.sql.`type`.SqlTypeName.DOUBLE
+import org.apache.calcite.sql.`type`.SqlTypeName.FLOAT
+import org.apache.calcite.sql.`type`.SqlTypeName.INTEGER
+import org.apache.calcite.sql.`type`.SqlTypeName.SMALLINT
+import org.apache.calcite.sql.`type`.SqlTypeName.TINYINT
--- End diff --
Can you remove those unused import ?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by huawei-flink <gi...@git.apache.org>.
Github user huawei-flink commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106615708
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamOverAggregate.scala ---
@@ -17,34 +17,41 @@
*/
package org.apache.flink.table.plan.nodes.datastream
-import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
+import org.apache.calcite.plan.{ RelOptCluster, RelTraitSet }
--- End diff --
@fhueske sorry about that. Will be more careful in the next one.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by huawei-flink <gi...@git.apache.org>.
Github user huawei-flink commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106361152
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala ---
@@ -19,33 +19,124 @@ package org.apache.flink.table.runtime.aggregate
import java.util
-import org.apache.calcite.rel.`type`._
+import scala.collection.JavaConversions.asScalaBuffer
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rel.core.AggregateCall
+import org.apache.calcite.rex.RexInputRef
+import org.apache.calcite.rex.RexLiteral
+import org.apache.calcite.rex.RexWindowBound
+import org.apache.calcite.sql.SqlAggFunction
+import org.apache.calcite.sql.SqlKind
import org.apache.calcite.sql.`type`.SqlTypeName
-import org.apache.calcite.sql.`type`.SqlTypeName._
-import org.apache.calcite.sql.fun._
-import org.apache.calcite.sql.{SqlAggFunction, SqlKind}
-import org.apache.flink.streaming.api.functions.ProcessFunction
+import org.apache.calcite.sql.`type`.SqlTypeName.BIGINT
+import org.apache.calcite.sql.`type`.SqlTypeName.BOOLEAN
+import org.apache.calcite.sql.`type`.SqlTypeName.DECIMAL
+import org.apache.calcite.sql.`type`.SqlTypeName.DOUBLE
+import org.apache.calcite.sql.`type`.SqlTypeName.FLOAT
+import org.apache.calcite.sql.`type`.SqlTypeName.INTEGER
+import org.apache.calcite.sql.`type`.SqlTypeName.SMALLINT
+import org.apache.calcite.sql.`type`.SqlTypeName.TINYINT
--- End diff --
I see your point, although I thought that wildcard import was not a best practice. It seems that the java and scala implementation are following different conventions. I have no problems with it in principle.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by huawei-flink <gi...@git.apache.org>.
Github user huawei-flink commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106366744
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/BoundedProcessingOverWindowFunction.scala ---
@@ -0,0 +1,97 @@
+/*
+ * 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.runtime.aggregate
+
+import org.apache.flink.api.java.tuple.Tuple
+import org.apache.flink.types.Row
+import org.apache.flink.configuration.Configuration
+import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction
+import org.apache.flink.streaming.api.windowing.windows.Window
+import org.apache.flink.util.Collector
+import org.apache.flink.table.functions.AggregateFunction
+import org.apache.flink.api.java.typeutils.RowTypeInfo
+import org.apache.flink.util.Preconditions
+import org.apache.flink.table.functions.Accumulator
+import java.lang.Iterable
+
+class BoundedProcessingOverWindowFunction[W <: Window](
--- End diff --
Thanks a lot for the clarification. I am really willing to do it right, but at the same time I need to understand. So, please be patient. :-)
When we started discussing the issue with @fhueske (https://issues.apache.org/jira/browse/FLINK-5654?filter=-1) there was a decision to use window, not process function.
Code consistency is pretty much the same, just extening a different interface. I understand that ProcessFunction can manage its state, but window checkpointing should replay all events in case of failure, so we would have consistent processing even without managing this level of granularity in the state. With procTime semantic, we can neglect retraction, and window can anyway customize triggering function.
I don't understand the third point.
The main argument I see for this specific case is that ProcessFunction supports granular state management. Besides the alleged code consistency.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by huawei-flink <gi...@git.apache.org>.
Github user huawei-flink commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106220189
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamOverAggregate.scala ---
@@ -159,6 +168,42 @@ class DataStreamOverAggregate(
result
}
+def createBoundedAndCurrentRowProcessingTimeOverWindow(
+ inputDS: DataStream[Row]): DataStream[Row] = {
+
+ val overWindow: Group = logicWindow.groups.get(0)
+ val partitionKeys: Array[Int] = overWindow.keys.toArray
+ val namedAggregates: Seq[CalcitePair[AggregateCall, String]] = generateNamedAggregates
+
+ // get the output types
+ val rowTypeInfo = FlinkTypeFactory.toInternalRowTypeInfo(getRowType).asInstanceOf[RowTypeInfo]
+
+ val result: DataStream[Row] =
+ // partitioned aggregation
+ if (partitionKeys.nonEmpty) {
+ val windowFunction = AggregateUtil.CreateBoundedProcessingOverWindowFunction(
+ namedAggregates,
+ inputType)
+
+ val lowerbound: Int = AggregateUtil.getLowerBoundary(
+ logicWindow.constants,
+ overWindow.lowerBound,
+ getInput())
+
+ inputDS
+ .keyBy(partitionKeys: _*)
+ .countWindow(lowerbound, 1).apply(windowFunction)
+ .returns(rowTypeInfo)
+ .name(aggOpName)
+ .asInstanceOf[DataStream[Row]]
+ } // global non-partitioned aggregation
+ else {
+ throw TableException(
--- End diff --
If needed, I can do it.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106627390
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamOverAggregate.scala ---
@@ -17,34 +17,41 @@
*/
package org.apache.flink.table.plan.nodes.datastream
-import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
+import org.apache.calcite.plan.{ RelOptCluster, RelTraitSet }
--- End diff --
No worries. I know that IDEs tend to reformat code but it really makes reviews harder.
Thanks!
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by rtudoran <gi...@git.apache.org>.
Github user rtudoran commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106615587
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/BoundedProcessingOverAllWindowFunction.scala ---
@@ -0,0 +1,97 @@
+/*
+ * 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.runtime.aggregate
+
+import org.apache.flink.api.java.tuple.Tuple
+import org.apache.flink.types.Row
+import org.apache.flink.configuration.Configuration
+import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction
+import org.apache.flink.streaming.api.windowing.windows.Window
+import org.apache.flink.util.Collector
+import org.apache.flink.table.functions.AggregateFunction
+import org.apache.flink.api.java.typeutils.RowTypeInfo
+import org.apache.flink.util.Preconditions
+import org.apache.flink.table.functions.Accumulator
+import java.lang.Iterable
+import org.apache.flink.streaming.api.functions.windowing.RichAllWindowFunction
+
+class BoundedProcessingOverAllWindowFunction[W <: Window](
+ private val aggregates: Array[AggregateFunction[_]],
+ private val aggFields: Array[Int],
+ private val forwardedFieldCount: Int)
+ extends RichAllWindowFunction[Row, Row, W] {
+
+ Preconditions.checkNotNull(aggregates)
+ Preconditions.checkNotNull(aggFields)
+ Preconditions.checkArgument(aggregates.length == aggFields.length)
+
+ private var output: Row = _
+ private var accumulators: Row = _
+ private var reuse: Row = _
+
+ output = new Row(forwardedFieldCount + aggregates.length)
+ if (null == accumulators) {
+ accumulators = new Row(aggregates.length)
+ }
+
+ override def apply(
+ window: W,
+ records: Iterable[Row],
+ out: Collector[Row]): Unit = {
+
+ var i = 0
+ // setting the accumulators for each aggregation
+ while (i < aggregates.length) {
+ accumulators.setField(i, aggregates(i).createAccumulator())
--- End diff --
What is the meaning of the acc?
What should be passed to the accumulator for reset? logically I would expect that this is a non parameterized method
should it be aggregates(i).resetAccumulator(aggregates(i))?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106519205
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamOverAggregate.scala ---
@@ -130,32 +142,76 @@ class DataStreamOverAggregate(
val rowTypeInfo = FlinkTypeFactory.toInternalRowTypeInfo(getRowType).asInstanceOf[RowTypeInfo]
val result: DataStream[Row] =
- // partitioned aggregation
- if (partitionKeys.nonEmpty) {
- val processFunction = AggregateUtil.CreateUnboundedProcessingOverProcessFunction(
- namedAggregates,
- inputType)
+ // partitioned aggregation
+ if (partitionKeys.nonEmpty) {
+ val processFunction = AggregateUtil.CreateUnboundedProcessingOverProcessFunction(
+ namedAggregates,
+ inputType)
- inputDS
+ inputDS
.keyBy(partitionKeys: _*)
.process(processFunction)
.returns(rowTypeInfo)
.name(aggOpName)
.asInstanceOf[DataStream[Row]]
- }
- // non-partitioned aggregation
- else {
- val processFunction = AggregateUtil.CreateUnboundedProcessingOverProcessFunction(
- namedAggregates,
- inputType,
- false)
-
- inputDS
- .process(processFunction).setParallelism(1).setMaxParallelism(1)
- .returns(rowTypeInfo)
- .name(aggOpName)
- .asInstanceOf[DataStream[Row]]
- }
+ } // non-partitioned aggregation
+ else {
+ val processFunction = AggregateUtil.CreateUnboundedProcessingOverProcessFunction(
+ namedAggregates,
+ inputType,
+ false)
+
+ inputDS
+ .process(processFunction).setParallelism(1).setMaxParallelism(1)
+ .returns(rowTypeInfo)
+ .name(aggOpName)
+ .asInstanceOf[DataStream[Row]]
+ }
+ result
+ }
+
+ def createBoundedAndCurrentRowProcessingTimeOverWindow(
+ inputDS: DataStream[Row]): DataStream[Row] = {
+
+ val overWindow: Group = logicWindow.groups.get(0)
+ val partitionKeys: Array[Int] = overWindow.keys.toArray
+ val namedAggregates: Seq[CalcitePair[AggregateCall, String]] = generateNamedAggregates
+
+ // get the output types
+ val rowTypeInfo = FlinkTypeFactory.toInternalRowTypeInfo(getRowType).asInstanceOf[RowTypeInfo]
+
+ val lowerbound: Int = AggregateUtil.getLowerBoundary(
+ logicWindow.constants,
+ overWindow.lowerBound,
+ getInput())
+
+ val result: DataStream[Row] =
+ // partitioned aggregation
+ if (partitionKeys.nonEmpty) {
+ val windowFunction = AggregateUtil.CreateBoundedProcessingOverWindowFunction(
+ namedAggregates,
+ inputType)
+ inputDS
+ .keyBy(partitionKeys: _*)
+ .countWindow(lowerbound,1)
+ .apply(windowFunction)
+ .returns(rowTypeInfo)
+ .name(aggOpName)
+ .asInstanceOf[DataStream[Row]]
+ } // global non-partitioned aggregation
+ else {
+ val windowFunction = AggregateUtil.CreateBoundedProcessingOverGlobalWindowFunction(
+ namedAggregates,
+ inputType)
+
+ inputDS
+ .countWindowAll(lowerbound,1)
--- End diff --
From the [MS SQL Server documentation](https://msdn.microsoft.com/en-us/library/ms189461.aspx):
> For example, ROWS BETWEEN 2 PRECEDING AND CURRENT ROW means that the window of rows that the function operates on is three rows in size, starting with 2 rows preceding until and including the current row.
So, since `lowerBound` is already `AggregateUtil.getLowerBoundary(...) + 1`, we should be good.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106528174
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/BoundedProcessingOverAllWindowFunction.scala ---
@@ -0,0 +1,97 @@
+/*
+ * 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.runtime.aggregate
+
+import org.apache.flink.api.java.tuple.Tuple
+import org.apache.flink.types.Row
+import org.apache.flink.configuration.Configuration
+import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction
+import org.apache.flink.streaming.api.windowing.windows.Window
+import org.apache.flink.util.Collector
+import org.apache.flink.table.functions.AggregateFunction
+import org.apache.flink.api.java.typeutils.RowTypeInfo
+import org.apache.flink.util.Preconditions
+import org.apache.flink.table.functions.Accumulator
+import java.lang.Iterable
+import org.apache.flink.streaming.api.functions.windowing.RichAllWindowFunction
+
+class BoundedProcessingOverAllWindowFunction[W <: Window](
+ private val aggregates: Array[AggregateFunction[_]],
+ private val aggFields: Array[Int],
+ private val forwardedFieldCount: Int)
+ extends RichAllWindowFunction[Row, Row, W] {
+
+ Preconditions.checkNotNull(aggregates)
+ Preconditions.checkNotNull(aggFields)
+ Preconditions.checkArgument(aggregates.length == aggFields.length)
+
+ private var output: Row = _
+ private var accumulators: Row = _
+ private var reuse: Row = _
+
+ output = new Row(forwardedFieldCount + aggregates.length)
+ if (null == accumulators) {
+ accumulators = new Row(aggregates.length)
+ }
+
+ override def apply(
+ window: W,
+ records: Iterable[Row],
+ out: Collector[Row]): Unit = {
+
+ var i = 0
+ // setting the accumulators for each aggregation
+ while (i < aggregates.length) {
+ accumulators.setField(i, aggregates(i).createAccumulator())
--- End diff --
accumulators can be reset with `aggregate.resetAccumulator(acc)`. So we can initialize once and reuse them
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106523715
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala ---
@@ -89,6 +116,58 @@ object AggregateUtil {
aggregationStateType)
}
}
+
+ private[flink] def CreateBoundedProcessingOverWindowFunction(
+ namedAggregates: Seq[CalcitePair[AggregateCall, String]],
+ inputType: RelDataType): WindowFunction[Row, Row, Tuple, GlobalWindow] = {
+
+ val (aggFields, aggregates) =
+ transformToAggregateFunctions(
+ namedAggregates.map(_.getKey),
+ inputType,
+ needRetraction = false)
+
+ val aggregationStateType: RowTypeInfo =
+ createDataSetAggregateBufferDataType(Array(), aggregates, inputType)
+
+ new BoundedProcessingOverWindowFunction[GlobalWindow](
+ aggregates,
+ aggFields,
+ inputType.getFieldCount)
+
+ }
+
+ private[flink] def CreateBoundedProcessingOverGlobalWindowFunction(
+ namedAggregates: Seq[CalcitePair[AggregateCall, String]],
+ inputType: RelDataType): AllWindowFunction[Row,Row,GlobalWindow] = {
+
+ val (aggFields, aggregates) =
+ transformToAggregateFunctions(
+ namedAggregates.map(_.getKey),
+ inputType,
+ needRetraction = false)
+
+ val aggregationStateType: RowTypeInfo =
+ createDataSetAggregateBufferDataType(Array(), aggregates, inputType)
+
+ new BoundedProcessingOverAllWindowFunction[GlobalWindow](
+ aggregates,
+ aggFields,
+ inputType.getFieldCount)
+
+ }
+
+ def getLowerBoundary(
+ constants: ImmutableList[RexLiteral],
+ lowerBound: RexWindowBound,
+ input: RelNode):Int = {
+ val ref: RexInputRef = lowerBound.getOffset.asInstanceOf[RexInputRef]
+ val index:Int = ref.getIndex
+ val count: Int = input.getRowType.getFieldCount
+ val lowerBoundIndex = count - index;
--- End diff --
Please remove `;`
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN x PREC...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on the issue:
https://github.com/apache/flink/pull/3547
No worries @sunjincheng121!
Everybody is welcome and encouraged to review pull requests.
Thanks for doing this!
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by sunjincheng121 <gi...@git.apache.org>.
Github user sunjincheng121 commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106192954
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/logical/rel/LogicalWindowAggregate.scala ---
@@ -37,8 +35,8 @@ class LogicalWindowAggregate(
child: RelNode,
indicator: Boolean,
groupSet: ImmutableBitSet,
- groupSets: util.List[ImmutableBitSet],
- aggCalls: util.List[AggregateCall])
+ groupSets: java.util.List[ImmutableBitSet],
--- End diff --
Can we keep the original way? `util.List[ImmutableBitSet]`
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by huawei-flink <gi...@git.apache.org>.
Github user huawei-flink commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106367506
--- Diff: flink-libraries/flink-table/src/test/java/org/apache/flink/table/api/java/stream/sql/ProcTimeRowStreamAggregationSqlITCase.java ---
@@ -0,0 +1,317 @@
+/*
+ * 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 java.util.ArrayList;
+import java.util.List;
+
+import org.apache.flink.api.java.tuple.Tuple5;
+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.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.java.StreamTableEnvironment;
+import org.apache.flink.table.api.java.stream.utils.StreamTestData;
+import org.apache.flink.table.api.scala.stream.utils.StreamITCase;
+import org.apache.flink.types.Row;
+import org.junit.Ignore;
+import org.junit.Test;
+
+public class ProcTimeRowStreamAggregationSqlITCase extends StreamingMultipleProgramsTestBase {
--- End diff --
Why should also the test be implemented in scala?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106516596
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala ---
@@ -17,35 +17,62 @@
*/
package org.apache.flink.table.runtime.aggregate
-import java.util
+import scala.collection.JavaConversions.asScalaBuffer
--- End diff --
Can we keep the imports as they are?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by huawei-flink <gi...@git.apache.org>.
Github user huawei-flink commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106238200
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala ---
@@ -19,33 +19,124 @@ package org.apache.flink.table.runtime.aggregate
import java.util
-import org.apache.calcite.rel.`type`._
+import scala.collection.JavaConversions.asScalaBuffer
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rel.core.AggregateCall
+import org.apache.calcite.rex.RexInputRef
+import org.apache.calcite.rex.RexLiteral
+import org.apache.calcite.rex.RexWindowBound
+import org.apache.calcite.sql.SqlAggFunction
+import org.apache.calcite.sql.SqlKind
import org.apache.calcite.sql.`type`.SqlTypeName
-import org.apache.calcite.sql.`type`.SqlTypeName._
-import org.apache.calcite.sql.fun._
-import org.apache.calcite.sql.{SqlAggFunction, SqlKind}
-import org.apache.flink.streaming.api.functions.ProcessFunction
+import org.apache.calcite.sql.`type`.SqlTypeName.BIGINT
+import org.apache.calcite.sql.`type`.SqlTypeName.BOOLEAN
+import org.apache.calcite.sql.`type`.SqlTypeName.DECIMAL
+import org.apache.calcite.sql.`type`.SqlTypeName.DOUBLE
+import org.apache.calcite.sql.`type`.SqlTypeName.FLOAT
+import org.apache.calcite.sql.`type`.SqlTypeName.INTEGER
+import org.apache.calcite.sql.`type`.SqlTypeName.SMALLINT
+import org.apache.calcite.sql.`type`.SqlTypeName.TINYINT
--- End diff --
after an inspection, I realized that the imports you mentioned are used. I think there is no unused import at this moment. Am I missing something?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by huawei-flink <gi...@git.apache.org>.
Github user huawei-flink commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106369028
--- Diff: flink-libraries/flink-table/src/test/java/org/apache/flink/table/api/java/stream/sql/ProcTimeRowStreamAggregationSqlITCase.java ---
@@ -0,0 +1,317 @@
+/*
+ * 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 java.util.ArrayList;
+import java.util.List;
+
+import org.apache.flink.api.java.tuple.Tuple5;
+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.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.java.StreamTableEnvironment;
+import org.apache.flink.table.api.java.stream.utils.StreamTestData;
+import org.apache.flink.table.api.scala.stream.utils.StreamITCase;
+import org.apache.flink.types.Row;
+import org.junit.Ignore;
+import org.junit.Test;
+
+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<Tuple5<Integer, Long, Integer, String, Long>> 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<Row> resultSet = tableEnv.toDataStream(result, Row.class);
+ resultSet.addSink(new StreamITCase.StringSink());
+ env.execute();
+
+ List<String> 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<Tuple5<Integer, Long, Integer, String, Long>> 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<Row> resultSet = tableEnv.toDataStream(result, Row.class);
+ resultSet.addSink(new StreamITCase.StringSink());
+ env.execute();
+
+ List<String> 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<Tuple5<Integer, Long, Integer, String, Long>> 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<Row> resultSet = tableEnv.toDataStream(result, Row.class);
+ resultSet.addSink(new StreamITCase.StringSink());
+ env.execute();
+
+ List<String> 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");
--- End diff --
Besides the fix you suggested, which is trivial (and thanks! :-)), it sounds a little odd to define a range with excluding boundaries. could you please point out the document in which this is discussed?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by sunjincheng121 <gi...@git.apache.org>.
Github user sunjincheng121 commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106343179
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamOverAggregate.scala ---
@@ -130,32 +142,76 @@ class DataStreamOverAggregate(
val rowTypeInfo = FlinkTypeFactory.toInternalRowTypeInfo(getRowType).asInstanceOf[RowTypeInfo]
val result: DataStream[Row] =
- // partitioned aggregation
- if (partitionKeys.nonEmpty) {
- val processFunction = AggregateUtil.CreateUnboundedProcessingOverProcessFunction(
- namedAggregates,
- inputType)
+ // partitioned aggregation
+ if (partitionKeys.nonEmpty) {
+ val processFunction = AggregateUtil.CreateUnboundedProcessingOverProcessFunction(
+ namedAggregates,
+ inputType)
- inputDS
+ inputDS
.keyBy(partitionKeys: _*)
.process(processFunction)
.returns(rowTypeInfo)
.name(aggOpName)
.asInstanceOf[DataStream[Row]]
- }
- // non-partitioned aggregation
- else {
- val processFunction = AggregateUtil.CreateUnboundedProcessingOverProcessFunction(
- namedAggregates,
- inputType,
- false)
-
- inputDS
- .process(processFunction).setParallelism(1).setMaxParallelism(1)
- .returns(rowTypeInfo)
- .name(aggOpName)
- .asInstanceOf[DataStream[Row]]
- }
+ } // non-partitioned aggregation
+ else {
+ val processFunction = AggregateUtil.CreateUnboundedProcessingOverProcessFunction(
+ namedAggregates,
+ inputType,
+ false)
+
+ inputDS
+ .process(processFunction).setParallelism(1).setMaxParallelism(1)
+ .returns(rowTypeInfo)
+ .name(aggOpName)
+ .asInstanceOf[DataStream[Row]]
+ }
+ result
+ }
+
+ def createBoundedAndCurrentRowProcessingTimeOverWindow(
+ inputDS: DataStream[Row]): DataStream[Row] = {
+
+ val overWindow: Group = logicWindow.groups.get(0)
+ val partitionKeys: Array[Int] = overWindow.keys.toArray
+ val namedAggregates: Seq[CalcitePair[AggregateCall, String]] = generateNamedAggregates
+
+ // get the output types
+ val rowTypeInfo = FlinkTypeFactory.toInternalRowTypeInfo(getRowType).asInstanceOf[RowTypeInfo]
+
+ val lowerbound: Int = AggregateUtil.getLowerBoundary(
+ logicWindow.constants,
+ overWindow.lowerBound,
+ getInput())
+
+ val result: DataStream[Row] =
+ // partitioned aggregation
+ if (partitionKeys.nonEmpty) {
+ val windowFunction = AggregateUtil.CreateBoundedProcessingOverWindowFunction(
+ namedAggregates,
+ inputType)
+ inputDS
+ .keyBy(partitionKeys: _*)
+ .countWindow(lowerbound,1)
--- End diff --
`lowerbound -> (lowerbound+1)`
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by huawei-flink <gi...@git.apache.org>.
Github user huawei-flink commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106465657
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/BoundedProcessingOverWindowFunction.scala ---
@@ -0,0 +1,97 @@
+/*
+ * 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.runtime.aggregate
+
+import org.apache.flink.api.java.tuple.Tuple
+import org.apache.flink.types.Row
+import org.apache.flink.configuration.Configuration
+import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction
+import org.apache.flink.streaming.api.windowing.windows.Window
+import org.apache.flink.util.Collector
+import org.apache.flink.table.functions.AggregateFunction
+import org.apache.flink.api.java.typeutils.RowTypeInfo
+import org.apache.flink.util.Preconditions
+import org.apache.flink.table.functions.Accumulator
+import java.lang.Iterable
+
+class BoundedProcessingOverWindowFunction[W <: Window](
--- End diff --
I just thought about a case where one wants to do a COUNT DISTINCT type of aggregation. How does a processFunction work for that?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by sunjincheng121 <gi...@git.apache.org>.
Github user sunjincheng121 commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106566112
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/BoundedProcessingOverWindowFunction.scala ---
@@ -0,0 +1,97 @@
+/*
+ * 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.runtime.aggregate
+
+import org.apache.flink.api.java.tuple.Tuple
+import org.apache.flink.types.Row
+import org.apache.flink.configuration.Configuration
+import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction
+import org.apache.flink.streaming.api.windowing.windows.Window
+import org.apache.flink.util.Collector
+import org.apache.flink.table.functions.AggregateFunction
+import org.apache.flink.api.java.typeutils.RowTypeInfo
+import org.apache.flink.util.Preconditions
+import org.apache.flink.table.functions.Accumulator
+import java.lang.Iterable
+
+class BoundedProcessingOverWindowFunction[W <: Window](
--- End diff --
Hi @huawei-flink @fhueske in bounded OVER ROW case, I still recommend using processFunction. But we can just agree to disagree at current time. Because in the continuous progress of flip11, we may change our view.
Best,
SunJIncheng
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by huawei-flink <gi...@git.apache.org>.
Github user huawei-flink commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106221708
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/logical/rel/LogicalWindowAggregate.scala ---
@@ -37,8 +35,8 @@ class LogicalWindowAggregate(
child: RelNode,
indicator: Boolean,
groupSet: ImmutableBitSet,
- groupSets: util.List[ImmutableBitSet],
- aggCalls: util.List[AggregateCall])
+ groupSets: java.util.List[ImmutableBitSet],
--- End diff --
for some reason it does not build in eclipse in the original way, and honestly I struggle to understand "the half package name" usage in Scala. Is there any practical reason for that? and does it break some convention to use the complete package name? I am asking with honest curiosity, and no polemical intention. Thanks for clarifying.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN x PREC...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on the issue:
https://github.com/apache/flink/pull/3547
One more thing. The PR includes a merge commit which makes squashing the commits and merging the PR a lot more difficult (see also the [contribution guidelines](http://flink.apache.org/contribute-code.html#coding-guidelines)).
You can avoid merge commits if you branch from the master and just put commits on top. If you want to include somebody else's work, you can use `git cherry-pick` to pull over a commit. Use `git rebase` to rebase your commits on the latest master.
Best, Fabian
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106517628
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamOverAggregate.scala ---
@@ -130,32 +142,77 @@ class DataStreamOverAggregate(
val rowTypeInfo = FlinkTypeFactory.toInternalRowTypeInfo(getRowType).asInstanceOf[RowTypeInfo]
val result: DataStream[Row] =
- // partitioned aggregation
- if (partitionKeys.nonEmpty) {
- val processFunction = AggregateUtil.CreateUnboundedProcessingOverProcessFunction(
- namedAggregates,
- inputType)
+ // partitioned aggregation
+ if (partitionKeys.nonEmpty) {
+ val processFunction = AggregateUtil.CreateUnboundedProcessingOverProcessFunction(
+ namedAggregates,
+ inputType)
- inputDS
+ inputDS
.keyBy(partitionKeys: _*)
.process(processFunction)
.returns(rowTypeInfo)
.name(aggOpName)
.asInstanceOf[DataStream[Row]]
- }
- // non-partitioned aggregation
- else {
- val processFunction = AggregateUtil.CreateUnboundedProcessingOverProcessFunction(
- namedAggregates,
- inputType,
- false)
-
- inputDS
- .process(processFunction).setParallelism(1).setMaxParallelism(1)
- .returns(rowTypeInfo)
- .name(aggOpName)
- .asInstanceOf[DataStream[Row]]
- }
+ } // non-partitioned aggregation
+ else {
+ val processFunction = AggregateUtil.CreateUnboundedProcessingOverProcessFunction(
+ namedAggregates,
+ inputType,
+ false)
+
+ inputDS
+ .process(processFunction).setParallelism(1).setMaxParallelism(1)
+ .returns(rowTypeInfo)
+ .name(aggOpName)
+ .asInstanceOf[DataStream[Row]]
+ }
+ result
+ }
+
+ def createBoundedAndCurrentRowProcessingTimeOverWindow(
+ inputDS: DataStream[Row]): DataStream[Row] = {
+
+ val overWindow: Group = logicWindow.groups.get(0)
+ val partitionKeys: Array[Int] = overWindow.keys.toArray
+ val namedAggregates: Seq[CalcitePair[AggregateCall, String]] = generateNamedAggregates
+
+ // get the output types
+ val rowTypeInfo = FlinkTypeFactory.toInternalRowTypeInfo(getRowType).asInstanceOf[RowTypeInfo]
+
+ // window size is lowerbound +1 to comply with over semantics
+ val lowerbound: Int = AggregateUtil.getLowerBoundary(
+ logicWindow.constants,
+ overWindow.lowerBound,
+ getInput()) + 1
+
+ val result: DataStream[Row] =
+ // partitioned aggregation
+ if (partitionKeys.nonEmpty) {
+ val windowFunction = AggregateUtil.CreateBoundedProcessingOverWindowFunction(
+ namedAggregates,
+ inputType)
+ inputDS
+ .keyBy(partitionKeys: _*)
+ .countWindow(lowerbound,1)
--- End diff --
`.countWindow(lowerbound, 1)` +space
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN x PREC...
Posted by sunjincheng121 <gi...@git.apache.org>.
Github user sunjincheng121 commented on the issue:
https://github.com/apache/flink/pull/3547
Hi @fhueske I'am sorry, I have not seen your comments until I passed over this PR. If I have any mistakes, please correct it.
Best,
SunJIncheng
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by huawei-flink <gi...@git.apache.org>.
Github user huawei-flink commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106452550
--- Diff: flink-libraries/flink-table/src/test/java/org/apache/flink/table/api/java/stream/sql/ProcTimeRowStreamAggregationSqlITCase.java ---
@@ -0,0 +1,317 @@
+/*
+ * 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 java.util.ArrayList;
+import java.util.List;
+
+import org.apache.flink.api.java.tuple.Tuple5;
+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.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.java.StreamTableEnvironment;
+import org.apache.flink.table.api.java.stream.utils.StreamTestData;
+import org.apache.flink.table.api.scala.stream.utils.StreamITCase;
+import org.apache.flink.types.Row;
+import org.junit.Ignore;
+import org.junit.Test;
+
+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<Tuple5<Integer, Long, Integer, String, Long>> 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<Row> resultSet = tableEnv.toDataStream(result, Row.class);
+ resultSet.addSink(new StreamITCase.StringSink());
+ env.execute();
+
+ List<String> 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<Tuple5<Integer, Long, Integer, String, Long>> 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<Row> resultSet = tableEnv.toDataStream(result, Row.class);
+ resultSet.addSink(new StreamITCase.StringSink());
+ env.execute();
+
+ List<String> 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<Tuple5<Integer, Long, Integer, String, Long>> 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<Row> resultSet = tableEnv.toDataStream(result, Row.class);
+ resultSet.addSink(new StreamITCase.StringSink());
+ env.execute();
+
+ List<String> 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");
--- End diff --
It makes sense anyway, I fixed according to your suggestion.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by sunjincheng121 <gi...@git.apache.org>.
Github user sunjincheng121 commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106193220
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/logical/rel/LogicalWindowAggregate.scala ---
@@ -18,8 +18,6 @@
package org.apache.flink.table.plan.logical.rel
-import java.util
--- End diff --
I think import package name is good way. How do you think?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by sunjincheng121 <gi...@git.apache.org>.
Github user sunjincheng121 commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106344788
--- Diff: flink-libraries/flink-table/src/test/java/org/apache/flink/table/api/java/stream/sql/ProcTimeRowStreamAggregationSqlITCase.java ---
@@ -0,0 +1,317 @@
+/*
+ * 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 java.util.ArrayList;
+import java.util.List;
+
+import org.apache.flink.api.java.tuple.Tuple5;
+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.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.java.StreamTableEnvironment;
+import org.apache.flink.table.api.java.stream.utils.StreamTestData;
+import org.apache.flink.table.api.scala.stream.utils.StreamITCase;
+import org.apache.flink.types.Row;
+import org.junit.Ignore;
+import org.junit.Test;
+
+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<Tuple5<Integer, Long, Integer, String, Long>> 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<Row> resultSet = tableEnv.toDataStream(result, Row.class);
+ resultSet.addSink(new StreamITCase.StringSink());
+ env.execute();
+
+ List<String> 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<Tuple5<Integer, Long, Integer, String, Long>> 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<Row> resultSet = tableEnv.toDataStream(result, Row.class);
+ resultSet.addSink(new StreamITCase.StringSink());
+ env.execute();
+
+ List<String> 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<Tuple5<Integer, Long, Integer, String, Long>> 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<Row> resultSet = tableEnv.toDataStream(result, Row.class);
+ resultSet.addSink(new StreamITCase.StringSink());
+ env.execute();
+
+ List<String> 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<Tuple5<Integer, Long, Integer, String, Long>> 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 minC"
+ + " FROM MyTable";
+ Table result = tableEnv.sql(sqlQuery);
+
+ DataStream<Row> resultSet = tableEnv.toDataStream(result, Row.class);
+ resultSet.addSink(new StreamITCase.StringSink());
+ env.execute();
+
+ List<String> 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<Tuple5<Integer, Long, Integer, String, Long>> 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<Row> resultSet = tableEnv.toDataStream(result, Row.class);
+ resultSet.addSink(new StreamITCase.StringSink());
+ env.execute();
+
+ List<String> 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();
+ StreamTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);
+ StreamITCase.clear();
+
+ env.setParallelism(1);
+
+ DataStream<Tuple5<Integer, Long, Integer, String, Long>> ds = StreamTestData.get5TupleDataStream(env);
+ Table in = tableEnv.fromDataStream(ds, "a,b,c,d,e");
+ tableEnv.registerTable("MyTable", in);
+
+ String sqlQuery = "SELECT a, AVG(c) OVER (PARTITION BY a ORDER BY procTime() ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) AS avgC, e FROM MyTable";
+ Table result = tableEnv.sql(sqlQuery);
+
+ DataStream<Row> resultSet = tableEnv.toDataStream(result, Row.class);
+ resultSet.addSink(new StreamITCase.StringSink());
+ env.execute();
+
+ List<String> expected = new ArrayList<>();
+ expected.add("1,0,1");
+ expected.add("2,1,1");
+ expected.add("2,1,2");
+ expected.add("3,3,2");
+ expected.add("3,3,2");
+ expected.add("3,4,3");
+ expected.add("4,6,1");
+ expected.add("4,6,2");
+ expected.add("4,7,1");
+ expected.add("4,8,2");
+ expected.add("5,10,1");
+ expected.add("5,10,3");
+ expected.add("5,11,3");
+ expected.add("5,13,2");
+ expected.add("5,13,2");
--- End diff --
This result incorrect.The reason for the error I left the comments in `DataStreamOverAggregate.scala`
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by huawei-flink <gi...@git.apache.org>.
Github user huawei-flink commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106367130
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamOverAggregate.scala ---
@@ -130,32 +142,76 @@ class DataStreamOverAggregate(
val rowTypeInfo = FlinkTypeFactory.toInternalRowTypeInfo(getRowType).asInstanceOf[RowTypeInfo]
val result: DataStream[Row] =
- // partitioned aggregation
- if (partitionKeys.nonEmpty) {
- val processFunction = AggregateUtil.CreateUnboundedProcessingOverProcessFunction(
- namedAggregates,
- inputType)
+ // partitioned aggregation
+ if (partitionKeys.nonEmpty) {
+ val processFunction = AggregateUtil.CreateUnboundedProcessingOverProcessFunction(
+ namedAggregates,
+ inputType)
- inputDS
+ inputDS
.keyBy(partitionKeys: _*)
.process(processFunction)
.returns(rowTypeInfo)
.name(aggOpName)
.asInstanceOf[DataStream[Row]]
- }
- // non-partitioned aggregation
- else {
- val processFunction = AggregateUtil.CreateUnboundedProcessingOverProcessFunction(
- namedAggregates,
- inputType,
- false)
-
- inputDS
- .process(processFunction).setParallelism(1).setMaxParallelism(1)
- .returns(rowTypeInfo)
- .name(aggOpName)
- .asInstanceOf[DataStream[Row]]
- }
+ } // non-partitioned aggregation
+ else {
+ val processFunction = AggregateUtil.CreateUnboundedProcessingOverProcessFunction(
+ namedAggregates,
+ inputType,
+ false)
+
+ inputDS
+ .process(processFunction).setParallelism(1).setMaxParallelism(1)
+ .returns(rowTypeInfo)
+ .name(aggOpName)
+ .asInstanceOf[DataStream[Row]]
+ }
+ result
+ }
+
+ def createBoundedAndCurrentRowProcessingTimeOverWindow(
+ inputDS: DataStream[Row]): DataStream[Row] = {
+
+ val overWindow: Group = logicWindow.groups.get(0)
+ val partitionKeys: Array[Int] = overWindow.keys.toArray
+ val namedAggregates: Seq[CalcitePair[AggregateCall, String]] = generateNamedAggregates
+
+ // get the output types
+ val rowTypeInfo = FlinkTypeFactory.toInternalRowTypeInfo(getRowType).asInstanceOf[RowTypeInfo]
+
+ val lowerbound: Int = AggregateUtil.getLowerBoundary(
+ logicWindow.constants,
+ overWindow.lowerBound,
+ getInput())
+
+ val result: DataStream[Row] =
+ // partitioned aggregation
+ if (partitionKeys.nonEmpty) {
+ val windowFunction = AggregateUtil.CreateBoundedProcessingOverWindowFunction(
+ namedAggregates,
+ inputType)
+ inputDS
+ .keyBy(partitionKeys: _*)
+ .countWindow(lowerbound,1)
+ .apply(windowFunction)
+ .returns(rowTypeInfo)
+ .name(aggOpName)
+ .asInstanceOf[DataStream[Row]]
+ } // global non-partitioned aggregation
+ else {
+ val windowFunction = AggregateUtil.CreateBoundedProcessingOverGlobalWindowFunction(
+ namedAggregates,
+ inputType)
+
+ inputDS
+ .countWindowAll(lowerbound,1)
--- End diff --
So, the semantic of between 2 rows and current row does not include the current row and I should count 3 elements?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by sunjincheng121 <gi...@git.apache.org>.
Github user sunjincheng121 commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106368103
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/BoundedProcessingOverWindowFunction.scala ---
@@ -0,0 +1,97 @@
+/*
+ * 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.runtime.aggregate
+
+import org.apache.flink.api.java.tuple.Tuple
+import org.apache.flink.types.Row
+import org.apache.flink.configuration.Configuration
+import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction
+import org.apache.flink.streaming.api.windowing.windows.Window
+import org.apache.flink.util.Collector
+import org.apache.flink.table.functions.AggregateFunction
+import org.apache.flink.api.java.typeutils.RowTypeInfo
+import org.apache.flink.util.Preconditions
+import org.apache.flink.table.functions.Accumulator
+import java.lang.Iterable
+
+class BoundedProcessingOverWindowFunction[W <: Window](
--- End diff --
Hi,@huawei-flink about the discussing I think we need @fhueske 's opinion.
Best,
SunJincheng
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106513895
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamOverAggregate.scala ---
@@ -17,34 +17,41 @@
*/
package org.apache.flink.table.plan.nodes.datastream
-import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
+import org.apache.calcite.plan.{ RelOptCluster, RelTraitSet }
--- End diff --
This PR has several changes that reformat the code but do not change the logic.
In general, we try to limit PRs to the scope of the issue and avoid reformatting.
1. It makes the PR harder to review because every reformatting change needs to be checked (was something changed or not). This is especially tiresome when indention is changed.
2. It makes it more difficult to track changes in the history.
3. Other users do not necessarily agree with your (or your IDE's) formatting and will format it back. So it might get back and forth.
There is nothing wrong with adding a space between arguments here and there, but please keep reformatting changes to a minimum.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by sunjincheng121 <gi...@git.apache.org>.
Github user sunjincheng121 commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106344518
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/BoundedProcessingOverWindowFunction.scala ---
@@ -0,0 +1,97 @@
+/*
+ * 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.runtime.aggregate
+
+import org.apache.flink.api.java.tuple.Tuple
+import org.apache.flink.types.Row
+import org.apache.flink.configuration.Configuration
+import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction
+import org.apache.flink.streaming.api.windowing.windows.Window
+import org.apache.flink.util.Collector
+import org.apache.flink.table.functions.AggregateFunction
+import org.apache.flink.api.java.typeutils.RowTypeInfo
+import org.apache.flink.util.Preconditions
+import org.apache.flink.table.functions.Accumulator
+import java.lang.Iterable
+
+class BoundedProcessingOverWindowFunction[W <: Window](
--- End diff --
I glad to explain this. IMO. There are several advantages to using `ProcessFunction`:
1. OverWindow implementation code consistent, such as: FLINK-5803, FLINK-5804, FLINK-5658 (See @fhueske propose the design in #3386 )
2. Because using `ProcessFunction` can manage their own state, the management window of the data collection, you can customize the trigger window, so in the future when the `FOLLOWING` supported we can be very convenient support.
3. For incremental aggregation, window mechanism must increase the implementation of `reduceFunction`, `ProcessFunction` in the management of the state at the same time will be very natural support.
What do you think?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106627282
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamOverAggregate.scala ---
@@ -106,9 +113,14 @@ class DataStreamOverAggregate(
if (overWindow.lowerBound.isUnbounded &&
overWindow.upperBound.isCurrentRow) {
createUnboundedAndCurrentRowProcessingTimeOverWindow(inputDS)
+ } // lowerbound is a BasicType and upperbound is PRECEEDING or CURRENT ROW
+ else if (overWindow.lowerBound.getOffset.getType.isInstanceOf[BasicSqlType]
--- End diff --
Yes, I realized that when looking at PR #3550 that `isInstanceOf[BasicSqlType]` and `.isInstanceOf[IntervalSqlType]` distinguishes ROW from RANGE windows. I think using `.isRows()` is more clear and might also be safer because it appears to be a more public API than the type of the offset.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN x PREC...
Posted by huawei-flink <gi...@git.apache.org>.
Github user huawei-flink commented on the issue:
https://github.com/apache/flink/pull/3547
sorry about the mess. I don't understand the mess the rebase does with eclipse... I will close this PR and open another one including all the changes and comments
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN x PREC...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on the issue:
https://github.com/apache/flink/pull/3547
Thanks for the new PR @huawei-flink. I'll have a look soon.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106519460
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamOverAggregate.scala ---
@@ -130,32 +142,77 @@ class DataStreamOverAggregate(
val rowTypeInfo = FlinkTypeFactory.toInternalRowTypeInfo(getRowType).asInstanceOf[RowTypeInfo]
val result: DataStream[Row] =
- // partitioned aggregation
- if (partitionKeys.nonEmpty) {
- val processFunction = AggregateUtil.CreateUnboundedProcessingOverProcessFunction(
- namedAggregates,
- inputType)
+ // partitioned aggregation
+ if (partitionKeys.nonEmpty) {
+ val processFunction = AggregateUtil.CreateUnboundedProcessingOverProcessFunction(
+ namedAggregates,
+ inputType)
- inputDS
+ inputDS
.keyBy(partitionKeys: _*)
.process(processFunction)
.returns(rowTypeInfo)
.name(aggOpName)
.asInstanceOf[DataStream[Row]]
- }
- // non-partitioned aggregation
- else {
- val processFunction = AggregateUtil.CreateUnboundedProcessingOverProcessFunction(
- namedAggregates,
- inputType,
- false)
-
- inputDS
- .process(processFunction).setParallelism(1).setMaxParallelism(1)
- .returns(rowTypeInfo)
- .name(aggOpName)
- .asInstanceOf[DataStream[Row]]
- }
+ } // non-partitioned aggregation
+ else {
+ val processFunction = AggregateUtil.CreateUnboundedProcessingOverProcessFunction(
+ namedAggregates,
+ inputType,
+ false)
+
+ inputDS
+ .process(processFunction).setParallelism(1).setMaxParallelism(1)
+ .returns(rowTypeInfo)
+ .name(aggOpName)
+ .asInstanceOf[DataStream[Row]]
+ }
+ result
+ }
+
+ def createBoundedAndCurrentRowProcessingTimeOverWindow(
+ inputDS: DataStream[Row]): DataStream[Row] = {
+
+ val overWindow: Group = logicWindow.groups.get(0)
+ val partitionKeys: Array[Int] = overWindow.keys.toArray
+ val namedAggregates: Seq[CalcitePair[AggregateCall, String]] = generateNamedAggregates
+
+ // get the output types
+ val rowTypeInfo = FlinkTypeFactory.toInternalRowTypeInfo(getRowType).asInstanceOf[RowTypeInfo]
+
+ // window size is lowerbound +1 to comply with over semantics
+ val lowerbound: Int = AggregateUtil.getLowerBoundary(
+ logicWindow.constants,
+ overWindow.lowerBound,
+ getInput()) + 1
+
+ val result: DataStream[Row] =
+ // partitioned aggregation
+ if (partitionKeys.nonEmpty) {
+ val windowFunction = AggregateUtil.CreateBoundedProcessingOverWindowFunction(
+ namedAggregates,
+ inputType)
+ inputDS
+ .keyBy(partitionKeys: _*)
+ .countWindow(lowerbound,1)
+ .apply(windowFunction)
+ .returns(rowTypeInfo)
+ .name(aggOpName)
+ .asInstanceOf[DataStream[Row]]
+ } // global non-partitioned aggregation
+ else {
+ val windowFunction = AggregateUtil.CreateBoundedProcessingOverGlobalWindowFunction(
+ namedAggregates,
+ inputType)
+
+ inputDS
+ .countWindowAll(lowerbound,1)
+ .apply(windowFunction)
+ .setParallelism(1).setMaxParallelism(1)
--- End diff --
Parallelism 1 is already implied by `countWindowAll`
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106526547
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala ---
@@ -89,6 +116,58 @@ object AggregateUtil {
aggregationStateType)
}
}
+
+ private[flink] def CreateBoundedProcessingOverWindowFunction(
+ namedAggregates: Seq[CalcitePair[AggregateCall, String]],
+ inputType: RelDataType): WindowFunction[Row, Row, Tuple, GlobalWindow] = {
+
+ val (aggFields, aggregates) =
+ transformToAggregateFunctions(
+ namedAggregates.map(_.getKey),
+ inputType,
+ needRetraction = false)
+
+ val aggregationStateType: RowTypeInfo =
+ createDataSetAggregateBufferDataType(Array(), aggregates, inputType)
+
+ new BoundedProcessingOverWindowFunction[GlobalWindow](
+ aggregates,
+ aggFields,
+ inputType.getFieldCount)
+
+ }
+
+ private[flink] def CreateBoundedProcessingOverGlobalWindowFunction(
+ namedAggregates: Seq[CalcitePair[AggregateCall, String]],
+ inputType: RelDataType): AllWindowFunction[Row,Row,GlobalWindow] = {
+
+ val (aggFields, aggregates) =
+ transformToAggregateFunctions(
+ namedAggregates.map(_.getKey),
+ inputType,
+ needRetraction = false)
+
+ val aggregationStateType: RowTypeInfo =
--- End diff --
not used
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by sunjincheng121 <gi...@git.apache.org>.
Github user sunjincheng121 commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106197313
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/BoundedProcessingOverWindowFunction.scala ---
@@ -0,0 +1,97 @@
+/*
+ * 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.runtime.aggregate
+
+import org.apache.flink.api.java.tuple.Tuple
+import org.apache.flink.types.Row
+import org.apache.flink.configuration.Configuration
+import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction
+import org.apache.flink.streaming.api.windowing.windows.Window
+import org.apache.flink.util.Collector
+import org.apache.flink.table.functions.AggregateFunction
+import org.apache.flink.api.java.typeutils.RowTypeInfo
+import org.apache.flink.util.Preconditions
+import org.apache.flink.table.functions.Accumulator
+import java.lang.Iterable
+
+class BoundedProcessingOverWindowFunction[W <: Window](
--- End diff --
I suggest use `ProcessFunction` to implement this feature. So that you can manage data flexibly and can use incremental calculations.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by sunjincheng121 <gi...@git.apache.org>.
Github user sunjincheng121 commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106342776
--- Diff: flink-libraries/flink-table/src/test/java/org/apache/flink/table/api/java/stream/sql/ProcTimeRowStreamAggregationSqlITCase.java ---
@@ -0,0 +1,317 @@
+/*
+ * 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 java.util.ArrayList;
+import java.util.List;
+
+import org.apache.flink.api.java.tuple.Tuple5;
+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.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.java.StreamTableEnvironment;
+import org.apache.flink.table.api.java.stream.utils.StreamTestData;
+import org.apache.flink.table.api.scala.stream.utils.StreamITCase;
+import org.apache.flink.types.Row;
+import org.junit.Ignore;
+import org.junit.Test;
+
+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<Tuple5<Integer, Long, Integer, String, Long>> 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<Row> resultSet = tableEnv.toDataStream(result, Row.class);
+ resultSet.addSink(new StreamITCase.StringSink());
+ env.execute();
+
+ List<String> 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<Tuple5<Integer, Long, Integer, String, Long>> 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<Row> resultSet = tableEnv.toDataStream(result, Row.class);
+ resultSet.addSink(new StreamITCase.StringSink());
+ env.execute();
+
+ List<String> 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<Tuple5<Integer, Long, Integer, String, Long>> 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<Row> resultSet = tableEnv.toDataStream(result, Row.class);
+ resultSet.addSink(new StreamITCase.StringSink());
+ env.execute();
+
+ List<String> 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");
--- End diff --
This result does not conform to the semantics of over. that is, in this sql,we must be got the result as follow:
```
1,0
2,1
2,3
3,3
3,7
3,12
4,6
4,13
4,21
4,24
5,10
5,21
5,33
5,37
5,39
```
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106534604
--- Diff: flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/sql/ProcTimeRowBoundedAggregationTest.scala ---
@@ -0,0 +1,311 @@
+/*
+ * 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.scala.stream.sql
+
+import org.apache.flink.api.scala._
+import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
+import org.apache.flink.table.api.{ TableEnvironment, TableException }
+import org.apache.flink.table.api.scala._
+import org.apache.flink.table.api.scala.stream.utils.{
+ StreamingWithStateTestBase,
+ StreamITCase,
+ StreamTestData
+}
+import org.junit.Assert._
+import org.junit._
+import scala.collection.mutable
+import org.apache.flink.types.Row
+
+class ProcTimeRowBoundedAggregationTest extends StreamingWithStateTestBase {
--- End diff --
Each ITCase class adds significantly to the build time of the project because it starts a mini cluster which takes some time. Therefore, we try to keep the number of ITCase classes low and aim to test as much as possible with faster unit tests.
Please add unit test methods that test the translation of queries to `org.apache.flink.table.api.scala.stream.sql.WindowAggregateTest`.
Please add the following integration tests to `org.apache.flink.table.api.scala.stream.sql.SqlITCase`
- multiple aggregation functions with partitioning
- multiple aggregation functions without partitioning
These tests classes contain already a few test methods that you can use as a starting point.
Thank you
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by sunjincheng121 <gi...@git.apache.org>.
Github user sunjincheng121 commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106344831
--- Diff: flink-libraries/flink-table/src/test/java/org/apache/flink/table/api/java/stream/sql/ProcTimeRowStreamAggregationSqlITCase.java ---
@@ -0,0 +1,317 @@
+/*
+ * 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 java.util.ArrayList;
+import java.util.List;
+
+import org.apache.flink.api.java.tuple.Tuple5;
+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.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.java.StreamTableEnvironment;
+import org.apache.flink.table.api.java.stream.utils.StreamTestData;
+import org.apache.flink.table.api.scala.stream.utils.StreamITCase;
+import org.apache.flink.types.Row;
+import org.junit.Ignore;
+import org.junit.Test;
+
+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<Tuple5<Integer, Long, Integer, String, Long>> 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<Row> resultSet = tableEnv.toDataStream(result, Row.class);
+ resultSet.addSink(new StreamITCase.StringSink());
+ env.execute();
+
+ List<String> 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<Tuple5<Integer, Long, Integer, String, Long>> 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<Row> resultSet = tableEnv.toDataStream(result, Row.class);
+ resultSet.addSink(new StreamITCase.StringSink());
+ env.execute();
+
+ List<String> 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<Tuple5<Integer, Long, Integer, String, Long>> 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<Row> resultSet = tableEnv.toDataStream(result, Row.class);
+ resultSet.addSink(new StreamITCase.StringSink());
+ env.execute();
+
+ List<String> 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<Tuple5<Integer, Long, Integer, String, Long>> 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 minC"
+ + " FROM MyTable";
+ Table result = tableEnv.sql(sqlQuery);
+
+ DataStream<Row> resultSet = tableEnv.toDataStream(result, Row.class);
+ resultSet.addSink(new StreamITCase.StringSink());
+ env.execute();
+
+ List<String> 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<Tuple5<Integer, Long, Integer, String, Long>> 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<Row> resultSet = tableEnv.toDataStream(result, Row.class);
+ resultSet.addSink(new StreamITCase.StringSink());
+ env.execute();
+
+ List<String> 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();
+ StreamTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);
+ StreamITCase.clear();
+
+ env.setParallelism(1);
+
+ DataStream<Tuple5<Integer, Long, Integer, String, Long>> ds = StreamTestData.get5TupleDataStream(env);
+ Table in = tableEnv.fromDataStream(ds, "a,b,c,d,e");
+ tableEnv.registerTable("MyTable", in);
+
+ String sqlQuery = "SELECT a, AVG(c) OVER (PARTITION BY a ORDER BY procTime() ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) AS avgC, e FROM MyTable";
+ Table result = tableEnv.sql(sqlQuery);
+
+ DataStream<Row> resultSet = tableEnv.toDataStream(result, Row.class);
+ resultSet.addSink(new StreamITCase.StringSink());
+ env.execute();
+
+ List<String> expected = new ArrayList<>();
+ expected.add("1,0,1");
+ expected.add("2,1,1");
+ expected.add("2,1,2");
+ expected.add("3,3,2");
+ expected.add("3,3,2");
+ expected.add("3,4,3");
+ expected.add("4,6,1");
+ expected.add("4,6,2");
+ expected.add("4,7,1");
+ expected.add("4,8,2");
+ expected.add("5,10,1");
+ expected.add("5,10,3");
+ expected.add("5,11,3");
+ expected.add("5,13,2");
+ expected.add("5,13,2");
+
+ StreamITCase.compareWithList(expected);
+ }
+
+ @Test
+ public void testAvgAggregatation2() throws Exception {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ StreamTableEnvironment tableEnv = TableEnvironment.getTableEnvironment(env);
+ StreamITCase.clear();
+
+ env.setParallelism(1);
+
+ DataStream<Tuple5<Integer, Long, Integer, String, Long>> ds = StreamTestData.get5TupleDataStream(env);
+ Table in = tableEnv.fromDataStream(ds, "a,b,c,d,e");
+ tableEnv.registerTable("MyTable", in);
+
+ String sqlQuery = "SELECT a, AVG(c) OVER (PARTITION BY a ORDER BY procTime() ROWS BETWEEN 2 PRECEDING AND CURRENT ROW) AS avgC, d FROM MyTable";
+ Table result = tableEnv.sql(sqlQuery);
+
+ DataStream<Row> resultSet = tableEnv.toDataStream(result, Row.class);
+ resultSet.addSink(new StreamITCase.StringSink());
+ env.execute();
+
+ List<String> expected = new ArrayList<>();
+ expected.add("1,0,Hallo");
+ expected.add("2,1,Hallo Welt");
+ expected.add("2,1,Hallo Welt wie");
+ expected.add("3,3,ABC");
+ expected.add("3,3,Hallo Welt wie gehts?");
+ expected.add("3,4,BCD");
+ expected.add("4,6,CDE");
+ expected.add("4,6,DEF");
+ expected.add("4,7,EFG");
+ expected.add("4,8,FGH");
+ expected.add("5,10,GHI");
+ expected.add("5,10,HIJ");
+ expected.add("5,11,IJK");
+ expected.add("5,13,JKL");
--- End diff --
This result incorrect.The reason for the error I left the comments in `DataStreamOverAggregate.scala`
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #3547: [FLINK-5653] Add processing time OVER ROWS BETWEEN...
Posted by huawei-flink <gi...@git.apache.org>.
Github user huawei-flink commented on a diff in the pull request:
https://github.com/apache/flink/pull/3547#discussion_r106360688
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/logical/rel/LogicalWindowAggregate.scala ---
@@ -18,8 +18,6 @@
package org.apache.flink.table.plan.logical.rel
-import java.util
--- End diff --
Thanks, both solutions look better than the half package import. I will apply one of those.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---