From 064b827127b15a1397c216aae6611d575a75556b Mon Sep 17 00:00:00 2001
From: Haohui Mai <wheat9@apache.org>
Date: Thu, 9 Mar 2017 13:57:49 -0800
Subject: [PATCH] [FLINK-5954] Always assign names to the window in the Stream
 SQL API.

---
 .../rules/datastream/LogicalWindowAggregateRule.scala     | 2 +-
 .../table/api/scala/stream/sql/WindowAggregateTest.scala  | 8 ++++----
 2 files changed, 5 insertions(+), 5 deletions(-)

diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/LogicalWindowAggregateRule.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/LogicalWindowAggregateRule.scala
index 37a1b7d0c0604..98839576c23a8 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/LogicalWindowAggregateRule.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/plan/rules/datastream/LogicalWindowAggregateRule.scala
@@ -107,7 +107,7 @@ class LogicalWindowAggregateRule
     val windowExpr = groupExpr
       .map(g => (g._2, identifyWindow(g._1)) )
       .filter(_._2.isDefined)
-      .map(g => (g._1, g._2.get) )
+      .map(g => (g._1, g._2.get.as("w$")) )
 
     windowExpr.size match {
       case 0 => None
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/sql/WindowAggregateTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/sql/WindowAggregateTest.scala
index 2781fb8fa3b0f..a25e59cb88395 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/sql/WindowAggregateTest.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/sql/WindowAggregateTest.scala
@@ -42,7 +42,7 @@ class WindowAggregateTest extends TableTestBase {
             streamTableNode(0),
             term("select", "1970-01-01 00:00:00 AS $f0")
           ),
-          term("window", EventTimeTumblingGroupWindow(None, 'rowtime, 3600000.millis)),
+          term("window", EventTimeTumblingGroupWindow(Some('w$), 'rowtime, 3600000.millis)),
           term("select", "COUNT(*) AS EXPR$0")
         ),
         term("select", "EXPR$0")
@@ -64,7 +64,7 @@ class WindowAggregateTest extends TableTestBase {
             term("select", "a", "1970-01-01 00:00:00 AS $f1")
           ),
           term("groupBy", "a"),
-          term("window", EventTimeTumblingGroupWindow(None, 'rowtime, 60000.millis)),
+          term("window", EventTimeTumblingGroupWindow(Some('w$), 'rowtime, 60000.millis)),
           term("select", "a", "COUNT(*) AS EXPR$1")
         ),
         term("select", "a", "EXPR$1")
@@ -86,7 +86,7 @@ class WindowAggregateTest extends TableTestBase {
             term("select", "a", "1970-01-01 00:00:00 AS $f1, b, c")
           ),
           term("groupBy", "a, b"),
-          term("window", EventTimeTumblingGroupWindow(None, 'rowtime, 1000.millis)),
+          term("window", EventTimeTumblingGroupWindow(Some('w$), 'rowtime, 1000.millis)),
           term("select", "a", "b", "SUM(c) AS EXPR$1")
         ),
         term("select", "a", "EXPR$1", "b")
@@ -107,7 +107,7 @@ class WindowAggregateTest extends TableTestBase {
             streamTableNode(0),
             term("select", "1970-01-01 00:00:00 AS $f0")
           ),
-          term("window", ProcessingTimeTumblingGroupWindow(None, 3600000.millis)),
+          term("window", ProcessingTimeTumblingGroupWindow(Some('w$), 3600000.millis)),
           term("select", "COUNT(*) AS EXPR$0")
         ),
         term("select", "EXPR$0")