[flink] branch master updated: [FLINK-9935] [table] Fix incorrect group field access in batch window combiner.

[fh...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

fhueske pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/master by this push:
     new 63e84ad  [FLINK-9935] [table] Fix incorrect group field access in batch window combiner.
63e84ad is described below

commit 63e84ad6532aff561d3abbd0dbadd420f836b28f
Author: Fabian Hueske <fh...@apache.org>
AuthorDate: Wed Jul 25 17:04:30 2018 +0200

    [FLINK-9935] [table] Fix incorrect group field access in batch window combiner.
---
 .../scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala  | 2 +-
 .../org/apache/flink/table/runtime/batch/table/GroupWindowITCase.scala  | 2 ++
 2 files changed, 3 insertions(+), 1 deletion(-)

diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala
index 7ce44a6..a3c9c1e 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/aggregate/AggregateUtil.scala
@@ -586,7 +586,7 @@ object AggregateUtil {
       isDistinctAggs,
       isStateBackedDataViews = false,
       partialResults = true,
-      groupings,
+      groupings.indices.toArray,
       Some(aggregates.indices.map(_ + groupings.length).toArray),
       outputType.getFieldCount,
       needRetract,
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/batch/table/GroupWindowITCase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/batch/table/GroupWindowITCase.scala
index 3d9223e..2c984c1 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/batch/table/GroupWindowITCase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/batch/table/GroupWindowITCase.scala
@@ -97,6 +97,7 @@ class GroupWindowITCase(
     val table = env
       .fromCollection(data)
       .toTable(tEnv, 'long, 'int, 'double, 'float, 'bigdec, 'string)
+      .select('int, 'long, 'string) // keep this select to enforce that the 'string key comes last
 
     val windowedTable = table
       .window(Tumble over 5.milli on 'long as 'w)
@@ -271,6 +272,7 @@ class GroupWindowITCase(
     val table = env
       .fromCollection(data)
       .toTable(tEnv, 'long, 'int, 'double, 'float, 'bigdec, 'string)
+      .select('int, 'long, 'string) // keep this select to enforce that the 'string key comes last
 
     val windowedTable = table
       .window(Slide over 10.milli every 5.milli on 'long as 'w)