Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/5241#discussion_r160260174
--- Diff:
flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/stream/sql/GroupWindowITCase.scala
---
@@ -0,0 +1,122 @@
+/*
+ * 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.stream.sql
+
+import org.apache.flink.api.scala._
+import org.apache.flink.streaming.api.TimeCharacteristic
+import
org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks
+import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
+import org.apache.flink.streaming.api.watermark.Watermark
+import org.apache.flink.table.api.TableEnvironment
+import org.apache.flink.table.api.scala._
+import
org.apache.flink.table.runtime.stream.sql.GroupWindowITCase.TimestampAndWatermarkWithOffset
+import org.apache.flink.table.runtime.utils.{StreamITCase,
StreamingWithStateTestBase}
+import org.apache.flink.types.Row
+import org.junit.Assert._
+import org.junit._
+
+import scala.collection.mutable
+
+class GroupWindowITCase extends StreamingWithStateTestBase {
+
+ val data = List(
+ (1000L, "1", "Hello"),
+ (2000L, "2", "Hello"),
+ (3000L, null.asInstanceOf[String], "Hello"),
+ (4000L, "4", "Hello"),
+ (5000L, null.asInstanceOf[String], "Hello"),
+ (6000L, "6", "Hello"),
+ (7000L, "7", "Hello World"),
+ (8000L, "8", "Hello World"),
+ (20000L, "20", "Hello World"))
+
+ @Test
+ def testRowTimeTumbleWindow(): Unit = {
+
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
+ val tEnv = TableEnvironment.getTableEnvironment(env)
+ StreamITCase.testResults = mutable.MutableList()
+ StreamITCase.clear
+ env.setParallelism(1)
+
+ val stream = env
+ .fromCollection(data)
+ .assignTimestampsAndWatermarks(
+ new TimestampAndWatermarkWithOffset[(Long, String, String)](0L))
+ val table = stream.toTable(tEnv, 'a, 'b, 'c, 'rowtime.rowtime)
+
+ tEnv.registerTable("T1", table)
+
+ val sqlQuery = "SELECT c, COUNT(*), COUNT(1), COUNT(b) FROM T1 " +
+ "GROUP BY TUMBLE(rowtime, interval '5' SECOND), c"
+
+ val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
+ result.addSink(new StreamITCase.StringSink[Row])
+ env.execute()
+
+ val expected = List("Hello World,2,2,2", "Hello World,1,1,1",
"Hello,4,4,3", "Hello,2,2,1")
+ assertEquals(expected.sorted, StreamITCase.testResults.sorted)
+ }
+
+ @Test
+ def testUnboundedGroupWindow(): Unit = {
--- End diff --
Rename to `testNonWindowedCount()`
---
