[FLINK-5047] [table] Add sliding group-windows for batch tables #3364

twalthr · 2017-02-20T15:41:44Z

This PR implements sliding group-windows. It covers the following cases:

Grouped Slide count-windows with incremental aggregates
Grouped and All Slide event-time time-windows with incremental and non-incremental aggregates

All windows support the overlapping and non-overlapping case. Slide windows are pre-tumbled if possible. This PR also fixes some bugs.

If the general design is ok, I will also implement the missing non-incremental aggregates for count-windows. And add a bit of documentation.

fhueske

Thanks for the PR @twalthr.
I think it looks pretty good. Can we split the PR into two time and count windows? There are lots for different cases to consider when reviewing the code. The PR also needs to be rebased to the current master and the new interface for AggregationFunctions.

After rebasing, the time windows should be pretty much good to merge.
The missing parts for the count windows could be addressed in a second PR.

What do you think?
Fabian

fhueske · 2017-03-06T19:57:36Z

...ries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala

+            asLong(slide),
+            returnType)
+        } else {
+          // for non-incremental aggregations


Do we need this case? Without partial aggregation, we cannot use tumbling windows to compute sliding windows. So this method is only called if all aggregates support partial aggregation, right?

You are right. I will remove this case distinction.

fhueske · 2017-03-06T20:17:37Z

...cala/org/apache/flink/table/runtime/aggregate/DataSetSlideTimeWindowAggFlatMapFunction.scala

+    if (start > windowStart - windowSize) {
+
+      // prepare output
+      for (i <- aggregates.indices) {


Isn't this just copying data from record to the aggregateBuffer?
Doesn't the input record have the same schema as the output record? Isn't is sufficient emit the input record multiple times with adapted outWindowStartIndex?

You are right. Will simplify that.

fhueske · 2017-03-06T20:31:42Z

...org/apache/flink/table/runtime/aggregate/DataSetSlideCountWindowAggReduceGroupFunction.scala

+  * @param windowSlide window slide of the sliding window
+  * @param returnType return type of this function
+  */
+class DataSetSlideCountWindowAggReduceGroupFunction(


I think it makes sense to add forward field annotations for the internal operators.

I will create an issue for this.

fhueske · 2017-03-06T20:47:26Z

...la/org/apache/flink/table/runtime/aggregate/DataSetSlideWindowAggReduceCombineFunction.scala

+    }
+
+    // this code path should never be reached as we return before the loop finishes
+    throw new IllegalArgumentException("Group is empty. This should never happen.")


Can be removed. Combine is only called if there is at least one record.

I cannot remove it. The compiler complains otherwise.

OK, but how about we make that more clear with a comment like:

// This will never happen because the iterator is never null but we have to satisfy the compiler.

fhueske · 2017-03-06T21:00:38Z

...org/apache/flink/table/runtime/aggregate/DataSetSlideCountWindowAggReduceGroupFunction.scala

+  private var outWindowStartIndex: Int = _
+
+  override def open(config: Configuration) {
+    Preconditions.checkNotNull(aggregates)


I think we can move everything into the constructor and remove the open() method.

fhueske · 2017-03-06T21:13:04Z

...rg/apache/flink/table/runtime/aggregate/DataSetSlideTimeWindowAggReduceCombineFunction.scala

+    }
+
+    // this code path should never be reached as we return before the loop finishes
+    throw new IllegalArgumentException("Group is empty. This should never happen.")


can be removed. It's the responsibility of the DataSet API to call the user functions correctly. And even if it would not, this function would behave correctly.

fhueske · 2017-03-06T21:13:54Z

.../org/apache/flink/table/runtime/aggregate/DataSetSlideTimeWindowAggReduceGroupFunction.scala

+  private var outWindowStartIndex: Int = _
+
+  override def open(config: Configuration) {
+    Preconditions.checkNotNull(aggregates)


Move checks and initialization to constructor and remove open()?

fhueske · 2017-03-06T21:16:54Z

...cala/org/apache/flink/table/runtime/aggregate/DataSetSlideWindowAggReduceGroupFunction.scala

+  private var output: Row = _
+
+  override def open(config: Configuration) {
+    Preconditions.checkNotNull(aggregates)


Except for the initialization of TimeWindowPropertyCollector everything can be moved into the constructor.

fhueske · 2017-03-06T21:19:53Z

...link-table/src/test/scala/org/apache/flink/table/api/scala/batch/table/GroupWindowTest.scala

@@ -360,6 +360,19 @@ class GroupWindowTest extends TableTestBase {
      .window(Session withGap 7.milli as 'w) // require on a time attribute
      .groupBy('string, 'w)
      .select('string, 'int.count)
+
+    val expected = unaryNode(


Why do you add the expected result to a test which is expected to fail?

Seems that I haven't seen the exception. Thanks.

fhueske · 2017-03-06T21:20:44Z

...-table/src/test/scala/org/apache/flink/table/api/scala/stream/table/AggregationsITCase.scala

@@ -146,42 +146,9 @@ class AggregationsITCase extends StreamingMultipleProgramsTestBase {
      "Hi,1,1,1970-01-01 00:00:00.0,1970-01-01 00:00:00.005")
    assertEquals(expected.sorted, StreamITCase.testResults.sorted)
  }
-
-  @Test
-  def testEventTimeSlidingWindow(): Unit = {


Is this test subsumed by DataStreamAggregateITCase?

twalthr · 2017-03-07T08:36:27Z

@fhueske thanks for your comments. I will create a time window PR.

twalthr · 2017-03-08T12:52:39Z

Thanks for the feedback @fhueske. I updated the PR.

fhueske

Hi @twalthr,
thanks for the update. I had a few minor comments.
Feel free to merge once these are resolved.

Cheers, Fabian

fhueske · 2017-03-08T13:48:32Z

...-table/src/main/scala/org/apache/flink/table/plan/nodes/dataset/DataSetWindowAggregate.scala

+      grouping.length)
+
+    // only pre-tumble if it is worth it
+    val littleTumblingSize = determineLargestTumblingSize(size, slide) <= 1


isLittleTumblingSize

fhueske · 2017-03-08T14:02:06Z

...cala/org/apache/flink/table/runtime/aggregate/DataSetSlideTimeWindowAggFlatMapFunction.scala

+
+  Preconditions.checkNotNull(aggregatesLength)
+
+  private var intermediateRow: Row = _


can be removed because we are replicating the input record.

fhueske · 2017-03-08T14:02:25Z

...cala/org/apache/flink/table/runtime/aggregate/DataSetSlideTimeWindowAggFlatMapFunction.scala

+  // add one field to store window start
+  private val intermediateRowArity: Int = groupingKeysLength + aggregatesLength + 1
+
+  override def open(config: Configuration) {


open() can be removed

fhueske · 2017-03-08T14:02:41Z

...cala/org/apache/flink/table/runtime/aggregate/DataSetSlideTimeWindowAggFlatMapFunction.scala

+
+  private var intermediateRow: Row = _
+  // add one field to store window start
+  private val intermediateRowArity: Int = groupingKeysLength + aggregatesLength + 1


can be removed

fhueske · 2017-03-08T14:17:51Z

...cala/org/apache/flink/table/runtime/aggregate/DataSetSlideTimeWindowAggFlatMapFunction.scala

+  * @param returnType return type of this function
+  */
+class DataSetSlideTimeWindowAggFlatMapFunction(
+    private val aggregatesLength: Int,


aggregatesLength and groupingKeysLength can be removed

fhueske · 2017-03-08T14:21:04Z

...ries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala

+      isParserCaseSensitive: Boolean)
+    : FlatMapFunction[Row, Row] = {
+
+    val aggregates = transformToAggregateFunctions(


not needed because input type = output type

fhueske · 2017-03-08T14:21:30Z

...ries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala

+    */
+  def createDataSetSlideWindowPrepareFlatMapFunction(
+      window: LogicalWindow,
+      namedAggregates: Seq[CalcitePair[AggregateCall, String]],


remove unnecessary parameters namedAggregates and groupings

fhueske · 2017-03-08T14:24:31Z

...rg/apache/flink/table/runtime/aggregate/DataSetSlideTimeWindowAggReduceCombineFunction.scala

+  * @param windowSlide window slide of the sliding window
+  * @param returnType return type of this function
+  */
+class DataSetSlideTimeWindowAggReduceCombineFunction(


Can be merged with DataSetSlideTimeWindowAggReduceGroupFunction because both can only be used if input is combinable.

fhueske · 2017-03-08T14:26:30Z

...org/apache/flink/table/runtime/aggregate/DataSetTumbleTimeWindowAggReduceGroupFunction.scala

@@ -112,12 +112,10 @@ class DataSetTumbleTimeWindowAggReduceGroupFunction(
    }

    // get final aggregate value and set to output.
-    aggregateMapping.foreach {
-      case (after, previous) => {
+    aggregateMapping.foreach { case (after, previous) =>


revert this change? Will be fixed with #3489

fhueske · 2017-03-08T14:28:25Z

...n/scala/org/apache/flink/table/runtime/aggregate/IncrementalAggregateAllWindowFunction.scala

-      out.collect(record)
+      var i = 0
+      while (i < record.getArity) {
+        output.setField(i, record.getField(0))


record.getField(0) -> record.getField(i)?

Good point!

This closes apache#3364.

fhueske reviewed Mar 6, 2017

View reviewed changes

twalthr force-pushed the FLINK-5047 branch from adcedd9 to 8da47a4 Compare March 8, 2017 10:39

[FLINK-5047] [table] Add sliding group-windows for batch tables

db363d7

twalthr force-pushed the FLINK-5047 branch from 8da47a4 to db363d7 Compare March 8, 2017 12:52

fhueske reviewed Mar 8, 2017

View reviewed changes

asfgit closed this in 31a57c5 Mar 8, 2017

p16i pushed a commit to p16i/flink that referenced this pull request Apr 16, 2017

[FLINK-5047] [table] Add sliding group-windows for batch tables

d5c0bf6

This closes apache#3364.

hequn8128 pushed a commit to hequn8128/flink that referenced this pull request Jun 22, 2017

[FLINK-5047] [table] Add sliding group-windows for batch tables

a96ba15

This closes apache#3364.

rmetzger added the component=API/TableSQL label Mar 14, 2019

flinkbot added component=TableSQL/API and removed component=API/TableSQL labels Mar 17, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-5047] [table] Add sliding group-windows for batch tables #3364

[FLINK-5047] [table] Add sliding group-windows for batch tables #3364

twalthr commented Feb 20, 2017 •

edited

fhueske left a comment •

edited

fhueske Mar 6, 2017

twalthr Mar 7, 2017

fhueske Mar 6, 2017

twalthr Mar 7, 2017

fhueske Mar 6, 2017

twalthr Mar 7, 2017

fhueske Mar 6, 2017

twalthr Mar 7, 2017

fhueske Mar 7, 2017

fhueske Mar 6, 2017

fhueske Mar 6, 2017

fhueske Mar 6, 2017

fhueske Mar 6, 2017

fhueske Mar 6, 2017

twalthr Mar 7, 2017

fhueske Mar 6, 2017

twalthr Mar 7, 2017

twalthr commented Mar 7, 2017

twalthr commented Mar 8, 2017

fhueske left a comment

fhueske Mar 8, 2017

fhueske Mar 8, 2017

fhueske Mar 8, 2017

fhueske Mar 8, 2017

fhueske Mar 8, 2017

fhueske Mar 8, 2017

fhueske Mar 8, 2017

fhueske Mar 8, 2017

fhueske Mar 8, 2017

fhueske Mar 8, 2017

twalthr Mar 8, 2017


		Preconditions.checkNotNull(aggregatesLength)

		private var intermediateRow: Row = _

[FLINK-5047] [table] Add sliding group-windows for batch tables #3364

[FLINK-5047] [table] Add sliding group-windows for batch tables #3364

Conversation

twalthr commented Feb 20, 2017 • edited

fhueske left a comment • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

twalthr commented Mar 7, 2017

twalthr commented Mar 8, 2017

fhueske left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

twalthr commented Feb 20, 2017 •

edited

fhueske left a comment •

edited