You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tw...@apache.org on 2017/07/13 10:18:43 UTC
[34/44] flink git commit: [FLINK-6617] [table] Restructuring of tests
http://git-wip-us.apache.org/repos/asf/flink/blob/f1fafc0e/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/table/validation/UserDefinedTableFunctionValidationTest.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/table/validation/UserDefinedTableFunctionValidationTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/table/validation/UserDefinedTableFunctionValidationTest.scala
deleted file mode 100644
index 218cc89..0000000
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/table/validation/UserDefinedTableFunctionValidationTest.scala
+++ /dev/null
@@ -1,199 +0,0 @@
-/*
- * 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.table.validation
-
-import org.apache.flink.api.scala._
-import org.apache.flink.table.api._
-import org.apache.flink.table.api.scala._
-import org.apache.flink.table.expressions.utils._
-import org.apache.flink.table.utils._
-import org.junit.Assert.{assertTrue, fail}
-import org.junit.Test
-
-class UserDefinedTableFunctionValidationTest extends TableTestBase {
-
- @Test
- def testRegisterFunctionException(): Unit ={
- val util = streamTestUtil()
- val t = util.addTable[(Int, Long, String)]('a, 'b, 'c)
-
- // check scala object is forbidden
- expectExceptionThrown(
- util.tableEnv.registerFunction("func3", ObjectTableFunction), "Scala object")
- expectExceptionThrown(
- util.javaTableEnv.registerFunction("func3", ObjectTableFunction), "Scala object")
-
- expectExceptionThrown(t.join(ObjectTableFunction('a, 1)), "Scala object")
- }
-
- @Test
- def testInvalidTableFunctions(): Unit = {
- val util = streamTestUtil()
-
- val func1 = new TableFunc1
- util.javaTableEnv.registerFunction("func1", func1)
-
- // table function call select
- expectExceptionThrown(
- func1('c).select("f0"),
- "TableFunction can only be used in join and leftOuterJoin."
- )
-
- // table function call select
- expectExceptionThrown(
- func1('c).select('f0),
- "TableFunction can only be used in join and leftOuterJoin."
- )
-
- // table function call writeToSink
- expectExceptionThrown(
- func1('c).writeToSink(null),
- "Cannot translate a query with an unbounded table function call."
- )
-
- // table function call distinct
- expectExceptionThrown(
- func1('c).distinct(),
- "TableFunction can only be used in join and leftOuterJoin."
- )
-
- // table function call filter
- expectExceptionThrown(
- func1('c).filter('f0 === "?"),
- "TableFunction can only be used in join and leftOuterJoin."
- )
-
- // table function call filter
- expectExceptionThrown(
- func1('c).filter("f0 = '?'"),
- "TableFunction can only be used in join and leftOuterJoin."
- )
-
- // table function call limit
- expectExceptionThrown(
- func1('c).orderBy('f0).limit(3),
- "TableFunction can only be used in join and leftOuterJoin."
- )
-
- // table function call limit
- expectExceptionThrown(
- func1('c).orderBy('f0).limit(0, 3),
- "TableFunction can only be used in join and leftOuterJoin."
- )
-
- // table function call orderBy
- expectExceptionThrown(
- func1('c).orderBy("f0"),
- "TableFunction can only be used in join and leftOuterJoin."
- )
-
- // table function call orderBy
- expectExceptionThrown(
- func1('c).orderBy('f0),
- "TableFunction can only be used in join and leftOuterJoin."
- )
-
- // table function call where
- expectExceptionThrown(
- func1('c).where("f0 = '?'"),
- "TableFunction can only be used in join and leftOuterJoin."
- )
-
- // table function call where
- expectExceptionThrown(
- func1('c).where('f0 === "?"),
- "TableFunction can only be used in join and leftOuterJoin."
- )
-
- }
-
- @Test
- def testInvalidTableFunction(): Unit = {
- val util = streamTestUtil()
- val t = util.addTable[(Int, Long, String)]("MyTable", 'a, 'b, 'c)
-
- //=================== check scala object is forbidden =====================
- // Scala table environment register
- expectExceptionThrown(util.addFunction("udtf", ObjectTableFunction), "Scala object")
- // Java table environment register
- expectExceptionThrown(
- util.tableEnv.registerFunction("udtf", ObjectTableFunction), "Scala object")
- // Scala Table API directly call
- expectExceptionThrown(t.join(ObjectTableFunction('a, 1)), "Scala object")
-
-
- //============ throw exception when table function is not registered =========
- // Java Table API call
- expectExceptionThrown(
- t.join(new Table(util.tableEnv, "nonexist(a)")
- ), "Undefined function: NONEXIST")
- // SQL API call
- expectExceptionThrown(
- util.tableEnv.sql("SELECT * FROM MyTable, LATERAL TABLE(nonexist(a))"),
- "No match found for function signature nonexist(<NUMERIC>)")
-
-
- //========= throw exception when the called function is a scalar function ====
- util.tableEnv.registerFunction("func0", Func0)
-
- // Java Table API call
- expectExceptionThrown(
- t.join(new Table(util.tableEnv, "func0(a)")),
- "only accept String that define table function",
- classOf[TableException])
- // SQL API call
- // NOTE: it doesn't throw an exception but an AssertionError, maybe a Calcite bug
- expectExceptionThrown(
- util.tableEnv.sql("SELECT * FROM MyTable, LATERAL TABLE(func0(a))"),
- null,
- classOf[AssertionError])
-
- //========== throw exception when the parameters is not correct ===============
- // Java Table API call
- util.addFunction("func2", new TableFunc2)
- expectExceptionThrown(
- t.join(new Table(util.tableEnv, "func2(c, c)")),
- "Given parameters of function 'FUNC2' do not match any signature")
- // SQL API call
- expectExceptionThrown(
- util.tableEnv.sql("SELECT * FROM MyTable, LATERAL TABLE(func2(c, c))"),
- "No match found for function signature func2(<CHARACTER>, <CHARACTER>)")
- }
-
- // ----------------------------------------------------------------------------------------------
-
- private def expectExceptionThrown(
- function: => Unit,
- keywords: String,
- clazz: Class[_ <: Throwable] = classOf[ValidationException])
- : Unit = {
- try {
- function
- fail(s"Expected a $clazz, but no exception is thrown.")
- } catch {
- case e if e.getClass == clazz =>
- if (keywords != null) {
- assertTrue(
- s"The exception message '${e.getMessage}' doesn't contain keyword '$keywords'",
- e.getMessage.contains(keywords))
- }
- case e: Throwable => fail(s"Expected throw ${clazz.getSimpleName}, but is $e.")
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f1fafc0e/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/utils/StreamTestData.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/utils/StreamTestData.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/utils/StreamTestData.scala
deleted file mode 100644
index 6745039..0000000
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/utils/StreamTestData.scala
+++ /dev/null
@@ -1,83 +0,0 @@
-/*
- * 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.utils
-
-import org.apache.flink.api.scala._
-import scala.collection.mutable
-import org.apache.flink.streaming.api.scala.DataStream
-import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
-
-object StreamTestData {
-
- def getSmall3TupleDataStream(env: StreamExecutionEnvironment): DataStream[(Int, Long, String)] = {
- val data = new mutable.MutableList[(Int, Long, String)]
- data.+=((1, 1L, "Hi"))
- data.+=((2, 2L, "Hello"))
- data.+=((3, 2L, "Hello world"))
- env.fromCollection(data)
- }
-
- def get3TupleDataStream(env: StreamExecutionEnvironment): DataStream[(Int, Long, String)] = {
- val data = new mutable.MutableList[(Int, Long, String)]
- data.+=((1, 1L, "Hi"))
- data.+=((2, 2L, "Hello"))
- data.+=((3, 2L, "Hello world"))
- data.+=((4, 3L, "Hello world, how are you?"))
- data.+=((5, 3L, "I am fine."))
- data.+=((6, 3L, "Luke Skywalker"))
- data.+=((7, 4L, "Comment#1"))
- data.+=((8, 4L, "Comment#2"))
- data.+=((9, 4L, "Comment#3"))
- data.+=((10, 4L, "Comment#4"))
- data.+=((11, 5L, "Comment#5"))
- data.+=((12, 5L, "Comment#6"))
- data.+=((13, 5L, "Comment#7"))
- data.+=((14, 5L, "Comment#8"))
- data.+=((15, 5L, "Comment#9"))
- data.+=((16, 6L, "Comment#10"))
- data.+=((17, 6L, "Comment#11"))
- data.+=((18, 6L, "Comment#12"))
- data.+=((19, 6L, "Comment#13"))
- data.+=((20, 6L, "Comment#14"))
- data.+=((21, 6L, "Comment#15"))
- env.fromCollection(data)
- }
-
- def get5TupleDataStream(env: StreamExecutionEnvironment):
- DataStream[(Int, Long, Int, String, Long)] = {
-
- val data = new mutable.MutableList[(Int, Long, Int, String, Long)]
- data.+=((1, 1L, 0, "Hallo", 1L))
- data.+=((2, 2L, 1, "Hallo Welt", 2L))
- data.+=((2, 3L, 2, "Hallo Welt wie", 1L))
- data.+=((3, 4L, 3, "Hallo Welt wie gehts?", 2L))
- data.+=((3, 5L, 4, "ABC", 2L))
- data.+=((3, 6L, 5, "BCD", 3L))
- data.+=((4, 7L, 6, "CDE", 2L))
- data.+=((4, 8L, 7, "DEF", 1L))
- data.+=((4, 9L, 8, "EFG", 1L))
- data.+=((4, 10L, 9, "FGH", 2L))
- data.+=((5, 11L, 10, "GHI", 1L))
- data.+=((5, 12L, 11, "HIJ", 3L))
- data.+=((5, 13L, 12, "IJK", 3L))
- data.+=((5, 14L, 13, "JKL", 2L))
- data.+=((5, 15L, 14, "KLM", 2L))
- env.fromCollection(data)
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f1fafc0e/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/validation/TableEnvironmentValidationTest.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/validation/TableEnvironmentValidationTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/validation/TableEnvironmentValidationTest.scala
deleted file mode 100644
index af8aff0..0000000
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/stream/validation/TableEnvironmentValidationTest.scala
+++ /dev/null
@@ -1,78 +0,0 @@
-/*
- * 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.validation
-
-import org.apache.flink.api.scala._
-import org.apache.flink.table.api.TableException
-import org.apache.flink.table.api.scala._
-import org.apache.flink.table.utils.TableTestBase
-import org.junit.Test
-
-class TableEnvironmentValidationTest extends TableTestBase {
- def testInvalidTimeAttributes(): Unit = {
- val util = streamTestUtil()
- // table definition makes no sense
- util.addTable[(Long, Int, String, Int, Long)]('a.rowtime.rowtime, 'b, 'c, 'd, 'e)
- }
-
- @Test(expected = classOf[TableException])
- def testInvalidProctimeAttribute(): Unit = {
- val util = streamTestUtil()
- // cannot replace an attribute with proctime
- util.addTable[(Long, Int, String, Int, Long)]('a, 'b.proctime, 'c, 'd, 'e)
- }
-
- @Test(expected = classOf[TableException])
- def testRowtimeAttributeReplaceFieldOfInvalidType(): Unit = {
- val util = streamTestUtil()
- // cannot replace a non-time attribute with rowtime
- util.addTable[(Long, Int, String, Int, Long)]('a, 'b, 'c.rowtime, 'd, 'e)
- }
-
- @Test(expected = classOf[TableException])
- def testRowtimeAndInvalidProctimeAttribute(): Unit = {
- val util = streamTestUtil()
- util.addTable[(Long, Int, String, Int, Long)]('rt.rowtime, 'b, 'c, 'd, 'pt.proctime)
- }
-
- @Test(expected = classOf[TableException])
- def testOnlyOneRowtimeAttribute1(): Unit = {
- val util = streamTestUtil()
- util.addTable[(Long, Int, String, Int, Long)]('a.rowtime, 'b, 'c, 'd, 'e, 'rt.rowtime)
- }
-
- @Test(expected = classOf[TableException])
- def testOnlyOneProctimeAttribute1(): Unit = {
- val util = streamTestUtil()
- util.addTable[(Long, Int, String, Int, Long)]('a, 'b, 'c, 'd, 'e, 'pt1.proctime, 'pt2.proctime)
- }
-
- @Test(expected = classOf[TableException])
- def testRowtimeAttributeUsedName(): Unit = {
- val util = streamTestUtil()
- util.addTable[(Long, Int, String, Int, Long)]('a, 'b, 'c, 'd, 'e, 'a.rowtime)
- }
-
- @Test(expected = classOf[TableException])
- def testProctimeAttributeUsedName(): Unit = {
- val util = streamTestUtil()
- util.addTable[(Long, Int, String, Int, Long)]('a, 'b, 'c, 'd, 'e, 'b.proctime)
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f1fafc0e/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/validation/TableSchemaValidationTest.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/validation/TableSchemaValidationTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/validation/TableSchemaValidationTest.scala
deleted file mode 100644
index e275d75..0000000
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/scala/validation/TableSchemaValidationTest.scala
+++ /dev/null
@@ -1,35 +0,0 @@
-/*
- * 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.validation
-
-import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, TypeInformation}
-import org.apache.flink.table.api.{TableException, TableSchema}
-import org.apache.flink.table.utils.TableTestBase
-import org.junit.Test
-
-class TableSchemaValidationTest extends TableTestBase {
-
- @Test(expected = classOf[TableException])
- def testInvalidSchema() {
- val fieldNames = Array("a", "b", "c")
- val typeInfos: Array[TypeInformation[_]] = Array(
- BasicTypeInfo.INT_TYPE_INFO,
- BasicTypeInfo.STRING_TYPE_INFO)
- new TableSchema(fieldNames, typeInfos)
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/f1fafc0e/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/ExplainTest.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/ExplainTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/ExplainTest.scala
new file mode 100644
index 0000000..820de08
--- /dev/null
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/ExplainTest.scala
@@ -0,0 +1,73 @@
+/*
+ * 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.stream
+
+import org.apache.flink.api.scala._
+import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
+import org.apache.flink.streaming.util.StreamingMultipleProgramsTestBase
+import org.apache.flink.table.api.TableEnvironment
+import org.apache.flink.table.api.scala._
+import org.junit.Assert.assertEquals
+import org.junit._
+
+class ExplainTest extends StreamingMultipleProgramsTestBase {
+
+ private val testFilePath = this.getClass.getResource("/").getFile
+
+ @Test
+ def testFilter(): Unit = {
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ val tEnv = TableEnvironment.getTableEnvironment(env)
+
+ val table = env.fromElements((1, "hello"))
+ .toTable(tEnv, 'a, 'b)
+ .filter("a % 2 = 0")
+
+ val result = replaceString(tEnv.explain(table))
+
+ val source = scala.io.Source.fromFile(testFilePath +
+ "../../src/test/scala/resources/testFilterStream0.out").mkString
+ val expect = replaceString(source)
+ assertEquals(result, expect)
+ }
+
+ @Test
+ def testUnion(): Unit = {
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ val tEnv = TableEnvironment.getTableEnvironment(env)
+
+ val table1 = env.fromElements((1, "hello")).toTable(tEnv, 'count, 'word)
+ val table2 = env.fromElements((1, "hello")).toTable(tEnv, 'count, 'word)
+ val table = table1.unionAll(table2)
+
+ val result = replaceString(tEnv.explain(table))
+
+ val source = scala.io.Source.fromFile(testFilePath +
+ "../../src/test/scala/resources/testUnionStream0.out").mkString
+ val expect = replaceString(source)
+ assertEquals(result, expect)
+ }
+
+ def replaceString(s: String): String = {
+ /* Stage {id} is ignored, because id keeps incrementing in test class
+ * while StreamExecutionEnvironment is up
+ */
+ s.replaceAll("\\r\\n", "\n").replaceAll("Stage \\d+", "")
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f1fafc0e/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/StreamTableEnvironmentTest.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/StreamTableEnvironmentTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/StreamTableEnvironmentTest.scala
new file mode 100644
index 0000000..0943ea6
--- /dev/null
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/StreamTableEnvironmentTest.scala
@@ -0,0 +1,150 @@
+/*
+ * 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.stream
+
+import java.lang.{Integer => JInt, Long => JLong}
+
+import org.apache.flink.api.java.tuple.{Tuple5 => JTuple5}
+import org.apache.flink.api.java.typeutils.TupleTypeInfo
+import org.apache.flink.api.scala._
+import org.apache.flink.streaming.api.TimeCharacteristic
+import org.apache.flink.streaming.api.datastream.DataStream
+import org.apache.flink.streaming.api.environment.{StreamExecutionEnvironment => JStreamExecEnv}
+import org.apache.flink.table.api.java.{StreamTableEnvironment => JStreamTableEnv}
+import org.apache.flink.table.api.scala._
+import org.apache.flink.table.api.{TableEnvironment, Types}
+import org.apache.flink.table.utils.TableTestUtil.{binaryNode, streamTableNode, term, unaryNode}
+import org.apache.flink.table.utils.TableTestBase
+import org.junit.Test
+import org.mockito.Mockito.{mock, when}
+
+class StreamTableEnvironmentTest extends TableTestBase {
+
+ @Test
+ def testSqlWithoutRegistering(): Unit = {
+ val util = streamTestUtil()
+ val table = util.addTable[(Long, Int, String)]("tableName", 'a, 'b, 'c)
+
+ val sqlTable = util.tableEnv.sql(s"SELECT a, b, c FROM $table WHERE b > 12")
+
+ val expected = unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "a, b, c"),
+ term("where", ">(b, 12)"))
+
+ util.verifyTable(sqlTable, expected)
+
+ val table2 = util.addTable[(Long, Int, String)]('d, 'e, 'f)
+
+ val sqlTable2 = util.tableEnv.sql(s"SELECT d, e, f FROM $table2 " +
+ s"UNION ALL SELECT a, b, c FROM $table")
+
+ val expected2 = binaryNode(
+ "DataStreamUnion",
+ streamTableNode(1),
+ streamTableNode(0),
+ term("union all", "d, e, f"))
+
+ util.verifyTable(sqlTable2, expected2)
+ }
+
+ @Test
+ def testProctimeAttribute(): Unit = {
+ val util = streamTestUtil()
+ // cannot replace an attribute with proctime
+ util.addTable[(Long, Int, String, Int, Long)]('a, 'b, 'c, 'd, 'e, 'pt.proctime)
+ }
+
+ @Test
+ def testReplacedRowtimeAttribute(): Unit = {
+ val util = streamTestUtil()
+ util.addTable[(Long, Int, String, Int, Long)]('rt.rowtime, 'b, 'c, 'd, 'e)
+ }
+
+ @Test
+ def testAppendedRowtimeAttribute(): Unit = {
+ val util = streamTestUtil()
+ util.addTable[(Long, Int, String, Int, Long)]('a, 'b, 'c, 'd, 'e, 'rt.rowtime)
+ }
+
+ @Test
+ def testRowtimeAndProctimeAttribute1(): Unit = {
+ val util = streamTestUtil()
+ util.addTable[(Long, Int, String, Int, Long)]('a, 'b, 'c, 'd, 'e, 'rt.rowtime, 'pt.proctime)
+ }
+
+ @Test
+ def testRowtimeAndProctimeAttribute2(): Unit = {
+ val util = streamTestUtil()
+ util.addTable[(Long, Int, String, Int, Long)]('a, 'b, 'c, 'd, 'e, 'pt.proctime, 'rt.rowtime)
+ }
+
+ @Test
+ def testRowtimeAndProctimeAttribute3(): Unit = {
+ val util = streamTestUtil()
+ util.addTable[(Long, Int, String, Int, Long)]('rt.rowtime, 'b, 'c, 'd, 'e, 'pt.proctime)
+ }
+
+ @Test
+ def testProctimeAttributeParsed(): Unit = {
+ val (jTEnv, ds) = prepareSchemaExpressionParser
+ jTEnv.fromDataStream(ds, "a, b, c, d, e, pt.proctime")
+ }
+
+ @Test
+ def testReplacingRowtimeAttributeParsed(): Unit = {
+ val (jTEnv, ds) = prepareSchemaExpressionParser
+ jTEnv.fromDataStream(ds, "a.rowtime, b, c, d, e")
+ }
+
+ @Test
+ def testAppedingRowtimeAttributeParsed(): Unit = {
+ val (jTEnv, ds) = prepareSchemaExpressionParser
+ jTEnv.fromDataStream(ds, "a, b, c, d, e, rt.rowtime")
+ }
+
+ @Test
+ def testRowtimeAndProctimeAttributeParsed1(): Unit = {
+ val (jTEnv, ds) = prepareSchemaExpressionParser
+ jTEnv.fromDataStream(ds, "a, b, c, d, e, pt.proctime, rt.rowtime")
+ }
+
+ @Test
+ def testRowtimeAndProctimeAttributeParsed2(): Unit = {
+ val (jTEnv, ds) = prepareSchemaExpressionParser
+ jTEnv.fromDataStream(ds, "rt.rowtime, b, c, d, e, pt.proctime")
+ }
+
+ private def prepareSchemaExpressionParser:
+ (JStreamTableEnv, DataStream[JTuple5[JLong, JInt, String, JInt, JLong]]) = {
+
+ val jStreamExecEnv = mock(classOf[JStreamExecEnv])
+ when(jStreamExecEnv.getStreamTimeCharacteristic).thenReturn(TimeCharacteristic.EventTime)
+ val jTEnv = TableEnvironment.getTableEnvironment(jStreamExecEnv)
+
+ val sType = new TupleTypeInfo(Types.LONG, Types.INT, Types.STRING, Types.INT, Types.LONG)
+ .asInstanceOf[TupleTypeInfo[JTuple5[JLong, JInt, String, JInt, JLong]]]
+ val ds = mock(classOf[DataStream[JTuple5[JLong, JInt, String, JInt, JLong]]])
+ when(ds.getType).thenReturn(sType)
+
+ (jTEnv, ds)
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f1fafc0e/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/StreamTableEnvironmentValidationTest.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/StreamTableEnvironmentValidationTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/StreamTableEnvironmentValidationTest.scala
new file mode 100644
index 0000000..1fd58cc
--- /dev/null
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/StreamTableEnvironmentValidationTest.scala
@@ -0,0 +1,112 @@
+/*
+ * 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.stream
+
+import java.math.BigDecimal
+
+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.runtime.stream.TimeAttributesITCase.TimestampWithEqualWatermark
+import org.apache.flink.table.utils.TableTestBase
+import org.junit.Test
+
+class StreamTableEnvironmentValidationTest extends TableTestBase {
+
+ @Test(expected = classOf[TableException])
+ def testInvalidTimeAttributes(): Unit = {
+ val util = streamTestUtil()
+ // table definition makes no sense
+ util.addTable[(Long, Int, String, Int, Long)]('a.rowtime.rowtime, 'b, 'c, 'd, 'e)
+ }
+
+ @Test(expected = classOf[TableException])
+ def testInvalidProctimeAttribute(): Unit = {
+ val util = streamTestUtil()
+ // cannot replace an attribute with proctime
+ util.addTable[(Long, Int, String, Int, Long)]('a, 'b.proctime, 'c, 'd, 'e)
+ }
+
+ @Test(expected = classOf[TableException])
+ def testRowtimeAttributeReplaceFieldOfInvalidType(): Unit = {
+ val util = streamTestUtil()
+ // cannot replace a non-time attribute with rowtime
+ util.addTable[(Long, Int, String, Int, Long)]('a, 'b, 'c.rowtime, 'd, 'e)
+ }
+
+ @Test(expected = classOf[TableException])
+ def testRowtimeAndInvalidProctimeAttribute(): Unit = {
+ val util = streamTestUtil()
+ util.addTable[(Long, Int, String, Int, Long)]('rt.rowtime, 'b, 'c, 'd, 'pt.proctime)
+ }
+
+ @Test(expected = classOf[TableException])
+ def testOnlyOneRowtimeAttribute1(): Unit = {
+ val util = streamTestUtil()
+ util.addTable[(Long, Int, String, Int, Long)]('a.rowtime, 'b, 'c, 'd, 'e, 'rt.rowtime)
+ }
+
+ @Test(expected = classOf[TableException])
+ def testOnlyOneProctimeAttribute1(): Unit = {
+ val util = streamTestUtil()
+ util.addTable[(Long, Int, String, Int, Long)]('a, 'b, 'c, 'd, 'e, 'pt1.proctime, 'pt2.proctime)
+ }
+
+ @Test(expected = classOf[TableException])
+ def testRowtimeAttributeUsedName(): Unit = {
+ val util = streamTestUtil()
+ util.addTable[(Long, Int, String, Int, Long)]('a, 'b, 'c, 'd, 'e, 'a.rowtime)
+ }
+
+ @Test(expected = classOf[TableException])
+ def testProctimeAttributeUsedName(): Unit = {
+ val util = streamTestUtil()
+ util.addTable[(Long, Int, String, Int, Long)]('a, 'b, 'c, 'd, 'e, 'b.proctime)
+ }
+
+ @Test(expected = classOf[TableException])
+ def testAsWithToManyFields(): Unit = {
+ val util = streamTestUtil()
+ util.addTable[(Int, Long, String)]('a, 'b, 'c, 'd)
+ }
+
+ @Test(expected = classOf[TableException])
+ def testAsWithAmbiguousFields(): Unit = {
+ val util = streamTestUtil()
+ util.addTable[(Int, Long, String)]('a, 'b, 'b)
+ }
+
+ @Test(expected = classOf[TableException])
+ def testOnlyFieldRefInAs(): Unit = {
+ val util = streamTestUtil()
+ util.addTable[(Int, Long, String)]('a, 'b as 'c, 'd)
+ }
+
+ @Test(expected = classOf[TableException])
+ def testInvalidTimeCharacteristic(): Unit = {
+ val data = List((1L, 1, 1d, 1f, new BigDecimal("1"), "Hi"))
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ val tEnv = TableEnvironment.getTableEnvironment(env)
+ val stream = env
+ .fromCollection(data)
+ .assignTimestampsAndWatermarks(new TimestampWithEqualWatermark())
+ stream.toTable(tEnv, 'rowtime.rowtime, 'int, 'double, 'float, 'bigdec, 'string)
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f1fafc0e/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/AggregateTest.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/AggregateTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/AggregateTest.scala
new file mode 100644
index 0000000..70d1d21
--- /dev/null
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/AggregateTest.scala
@@ -0,0 +1,157 @@
+/*
+ * 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.stream.sql
+
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.java.typeutils.RowTypeInfo
+import org.apache.flink.api.scala._
+import org.apache.flink.api.scala.typeutils.CaseClassTypeInfo
+import org.apache.flink.table.api.Types
+import org.apache.flink.table.api.scala._
+import org.apache.flink.table.expressions.AggFunctionCall
+import org.apache.flink.table.functions.AggregateFunction
+import org.apache.flink.table.utils.TableTestUtil.{streamTableNode, term, unaryNode}
+import org.apache.flink.table.utils.{StreamTableTestUtil, TableTestBase}
+import org.apache.flink.types.Row
+import org.junit.Assert.{assertEquals, assertTrue}
+import org.junit.{Ignore, Test}
+
+class AggregateTest extends TableTestBase {
+
+ private val streamUtil: StreamTableTestUtil = streamTestUtil()
+ streamUtil.addTable[(Int, String, Long)](
+ "MyTable", 'a, 'b, 'c, 'proctime.proctime, 'rowtime.rowtime)
+
+ @Test
+ def testGroupbyWithoutWindow() = {
+ val sql = "SELECT COUNT(a) FROM MyTable GROUP BY b"
+
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+ "DataStreamGroupAggregate",
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "b", "a")
+ ),
+ term("groupBy", "b"),
+ term("select", "b", "COUNT(a) AS EXPR$0")
+ ),
+ term("select", "EXPR$0")
+ )
+ streamUtil.verifySql(sql, expected)
+ }
+
+ @Test
+ def testDistinct(): Unit = {
+ val sql = "SELECT DISTINCT a, b, c FROM MyTable"
+
+ val expected =
+ unaryNode(
+ "DataStreamGroupAggregate",
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "a, b, c")
+ ),
+ term("groupBy", "a, b, c"),
+ term("select", "a, b, c")
+ )
+ streamUtil.verifySql(sql, expected)
+ }
+
+ // TODO: this query should be optimized to only have a single DataStreamGroupAggregate
+ // TODO: reopen this until FLINK-7144 fixed
+ @Ignore
+ @Test
+ def testDistinctAfterAggregate(): Unit = {
+ val sql = "SELECT DISTINCT a FROM MyTable GROUP BY a, b, c"
+
+ val expected =
+ unaryNode(
+ "DataStreamGroupAggregate",
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "a")
+ ),
+ term("groupBy", "a"),
+ term("select", "a")
+ )
+ streamUtil.verifySql(sql, expected)
+ }
+
+
+ @Test
+ def testUserDefinedAggregateFunctionWithScalaAccumulator(): Unit = {
+ streamUtil.addFunction("udag", new MyAgg)
+ val call = streamUtil
+ .tableEnv
+ .functionCatalog
+ .lookupFunction("udag", Seq())
+ .asInstanceOf[AggFunctionCall]
+
+ val typeInfo = call.accTypeInfo
+ assertTrue(typeInfo.isInstanceOf[CaseClassTypeInfo[_]])
+ assertEquals(2, typeInfo.getTotalFields)
+ val caseTypeInfo = typeInfo.asInstanceOf[CaseClassTypeInfo[_]]
+ assertEquals(Types.LONG, caseTypeInfo.getTypeAt(0))
+ assertEquals(Types.LONG, caseTypeInfo.getTypeAt(1))
+
+ streamUtil.addFunction("udag2", new MyAgg2)
+ val call2 = streamUtil
+ .tableEnv
+ .functionCatalog
+ .lookupFunction("udag2", Seq())
+ .asInstanceOf[AggFunctionCall]
+
+ val typeInfo2 = call2.accTypeInfo
+ assertTrue(s"actual type: $typeInfo2", typeInfo2.isInstanceOf[RowTypeInfo])
+ assertEquals(2, typeInfo2.getTotalFields)
+ val rowTypeInfo = typeInfo2.asInstanceOf[RowTypeInfo]
+ assertEquals(Types.LONG, rowTypeInfo.getTypeAt(0))
+ assertEquals(Types.INT, rowTypeInfo.getTypeAt(1))
+ }
+}
+
+case class MyAccumulator(var sum: Long, var count: Long)
+
+class MyAgg extends AggregateFunction[Long, MyAccumulator] {
+
+ //Overloaded accumulate method
+ def accumulate(acc: MyAccumulator, value: Long): Unit = {
+ }
+
+ override def createAccumulator(): MyAccumulator = MyAccumulator(0, 0)
+
+ override def getValue(accumulator: MyAccumulator): Long = 1L
+}
+
+class MyAgg2 extends AggregateFunction[Long, Row] {
+
+ def accumulate(acc: Row, value: Long): Unit = {}
+
+ override def createAccumulator(): Row = new Row(2)
+
+ override def getValue(accumulator: Row): Long = 1L
+
+ def getAccumulatorType: TypeInformation[_] = new RowTypeInfo(Types.LONG, Types.INT)
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f1fafc0e/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/CorrelateTest.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/CorrelateTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/CorrelateTest.scala
new file mode 100644
index 0000000..52d1320
--- /dev/null
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/CorrelateTest.scala
@@ -0,0 +1,236 @@
+/*
+ * 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.stream.sql
+
+import org.apache.flink.api.scala._
+import org.apache.flink.table.api.scala._
+import org.apache.flink.table.utils.TableTestUtil._
+import org.apache.flink.table.utils.{HierarchyTableFunction, PojoTableFunc, TableFunc2, _}
+import org.junit.Test
+
+class CorrelateTest extends TableTestBase {
+
+ @Test
+ def testCrossJoin(): Unit = {
+ val util = streamTestUtil()
+ val func1 = new TableFunc1
+ util.addTable[(Int, Long, String)]("MyTable", 'a, 'b, 'c)
+ util.addFunction("func1", func1)
+
+ val sqlQuery = "SELECT c, s FROM MyTable, LATERAL TABLE(func1(c)) AS T(s)"
+
+ val expected = unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+ "DataStreamCorrelate",
+ streamTableNode(0),
+ term("invocation", "func1($cor0.c)"),
+ term("function", func1.getClass.getCanonicalName),
+ term("rowType",
+ "RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, VARCHAR(2147483647) f0)"),
+ term("joinType", "INNER")
+ ),
+ term("select", "c", "f0 AS s")
+ )
+
+ util.verifySql(sqlQuery, expected)
+
+ // test overloading
+
+ val sqlQuery2 = "SELECT c, s FROM MyTable, LATERAL TABLE(func1(c, '$')) AS T(s)"
+
+ val expected2 = unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+ "DataStreamCorrelate",
+ streamTableNode(0),
+ term("invocation", "func1($cor0.c, '$')"),
+ term("function", func1.getClass.getCanonicalName),
+ term("rowType",
+ "RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, VARCHAR(2147483647) f0)"),
+ term("joinType", "INNER")
+ ),
+ term("select", "c", "f0 AS s")
+ )
+
+ util.verifySql(sqlQuery2, expected2)
+ }
+
+ @Test
+ def testLeftOuterJoin(): Unit = {
+ val util = streamTestUtil()
+ val func1 = new TableFunc1
+ util.addTable[(Int, Long, String)]("MyTable", 'a, 'b, 'c)
+ util.addFunction("func1", func1)
+
+ val sqlQuery = "SELECT c, s FROM MyTable LEFT JOIN LATERAL TABLE(func1(c)) AS T(s) ON TRUE"
+
+ val expected = unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+ "DataStreamCorrelate",
+ streamTableNode(0),
+ term("invocation", "func1($cor0.c)"),
+ term("function", func1.getClass.getCanonicalName),
+ term("rowType",
+ "RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, VARCHAR(2147483647) f0)"),
+ term("joinType", "LEFT")
+ ),
+ term("select", "c", "f0 AS s")
+ )
+
+ util.verifySql(sqlQuery, expected)
+ }
+
+ @Test
+ def testCustomType(): Unit = {
+ val util = streamTestUtil()
+ val func2 = new TableFunc2
+ util.addTable[(Int, Long, String)]("MyTable", 'a, 'b, 'c)
+ util.addFunction("func2", func2)
+
+ val sqlQuery = "SELECT c, name, len FROM MyTable, LATERAL TABLE(func2(c)) AS T(name, len)"
+
+ val expected = unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+ "DataStreamCorrelate",
+ streamTableNode(0),
+ term("invocation", "func2($cor0.c)"),
+ term("function", func2.getClass.getCanonicalName),
+ term("rowType",
+ "RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, " +
+ "VARCHAR(2147483647) f0, INTEGER f1)"),
+ term("joinType", "INNER")
+ ),
+ term("select", "c", "f0 AS name", "f1 AS len")
+ )
+
+ util.verifySql(sqlQuery, expected)
+ }
+
+ @Test
+ def testHierarchyType(): Unit = {
+ val util = streamTestUtil()
+ util.addTable[(Int, Long, String)]("MyTable", 'a, 'b, 'c)
+ val function = new HierarchyTableFunction
+ util.addFunction("hierarchy", function)
+
+ val sqlQuery = "SELECT c, T.* FROM MyTable, LATERAL TABLE(hierarchy(c)) AS T(name, adult, len)"
+
+ val expected = unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+ "DataStreamCorrelate",
+ streamTableNode(0),
+ term("invocation", "hierarchy($cor0.c)"),
+ term("function", function.getClass.getCanonicalName),
+ term("rowType",
+ "RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c," +
+ " VARCHAR(2147483647) f0, BOOLEAN f1, INTEGER f2)"),
+ term("joinType", "INNER")
+ ),
+ term("select", "c", "f0 AS name", "f1 AS adult", "f2 AS len")
+ )
+
+ util.verifySql(sqlQuery, expected)
+ }
+
+ @Test
+ def testPojoType(): Unit = {
+ val util = streamTestUtil()
+ util.addTable[(Int, Long, String)]("MyTable", 'a, 'b, 'c)
+ val function = new PojoTableFunc
+ util.addFunction("pojo", function)
+
+ val sqlQuery = "SELECT c, name, age FROM MyTable, LATERAL TABLE(pojo(c))"
+
+ val expected = unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+ "DataStreamCorrelate",
+ streamTableNode(0),
+ term("invocation", "pojo($cor0.c)"),
+ term("function", function.getClass.getCanonicalName),
+ term("rowType",
+ "RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c," +
+ " INTEGER age, VARCHAR(2147483647) name)"),
+ term("joinType", "INNER")
+ ),
+ term("select", "c", "name", "age")
+ )
+
+ util.verifySql(sqlQuery, expected)
+ }
+
+ @Test
+ def testFilter(): Unit = {
+ val util = streamTestUtil()
+ val func2 = new TableFunc2
+ util.addTable[(Int, Long, String)]("MyTable", 'a, 'b, 'c)
+ util.addFunction("func2", func2)
+
+ val sqlQuery = "SELECT c, name, len FROM MyTable, LATERAL TABLE(func2(c)) AS T(name, len) " +
+ "WHERE len > 2"
+
+ val expected = unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+ "DataStreamCorrelate",
+ streamTableNode(0),
+ term("invocation", "func2($cor0.c)"),
+ term("function", func2.getClass.getCanonicalName),
+ term("rowType",
+ "RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, " +
+ "VARCHAR(2147483647) f0, INTEGER f1)"),
+ term("joinType", "INNER"),
+ term("condition", ">($1, 2)")
+ ),
+ term("select", "c", "f0 AS name", "f1 AS len")
+ )
+
+ util.verifySql(sqlQuery, expected)
+ }
+
+ @Test
+ def testScalarFunction(): Unit = {
+ val util = streamTestUtil()
+ val func1 = new TableFunc1
+ util.addTable[(Int, Long, String)]("MyTable", 'a, 'b, 'c)
+ util.addFunction("func1", func1)
+
+ val sqlQuery = "SELECT c, s FROM MyTable, LATERAL TABLE(func1(SUBSTRING(c, 2))) AS T(s)"
+
+ val expected = unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+ "DataStreamCorrelate",
+ streamTableNode(0),
+ term("invocation", "func1(SUBSTRING($cor0.c, 2))"),
+ term("function", func1.getClass.getCanonicalName),
+ term("rowType",
+ "RecordType(INTEGER a, BIGINT b, VARCHAR(2147483647) c, VARCHAR(2147483647) f0)"),
+ term("joinType", "INNER")
+ ),
+ term("select", "c", "f0 AS s")
+ )
+
+ util.verifySql(sqlQuery, expected)
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f1fafc0e/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/GroupWindowTest.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/GroupWindowTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/GroupWindowTest.scala
new file mode 100644
index 0000000..4823903
--- /dev/null
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/GroupWindowTest.scala
@@ -0,0 +1,146 @@
+/*
+ * 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.stream.sql
+
+import org.apache.flink.api.scala._
+import org.apache.flink.table.api.scala._
+import org.apache.flink.table.plan.logical._
+import org.apache.flink.table.runtime.utils.JavaUserDefinedAggFunctions.WeightedAvgWithMerge
+import org.apache.flink.table.utils.TableTestUtil._
+import org.apache.flink.table.utils.{StreamTableTestUtil, TableTestBase}
+import org.junit.Test
+
+class GroupWindowTest extends TableTestBase {
+ private val streamUtil: StreamTableTestUtil = streamTestUtil()
+ streamUtil.addTable[(Int, String, Long)](
+ "MyTable", 'a, 'b, 'c, 'proctime.proctime, 'rowtime.rowtime)
+
+ @Test
+ def testTumbleFunction() = {
+ streamUtil.tableEnv.registerFunction("weightedAvg", new WeightedAvgWithMerge)
+
+ val sql =
+ "SELECT " +
+ " COUNT(*), weightedAvg(c, a) AS wAvg, " +
+ " TUMBLE_START(rowtime, INTERVAL '15' MINUTE), " +
+ " TUMBLE_END(rowtime, INTERVAL '15' MINUTE)" +
+ "FROM MyTable " +
+ "GROUP BY TUMBLE(rowtime, INTERVAL '15' MINUTE)"
+ val expected =
+ unaryNode(
+ "DataStreamGroupWindowAggregate",
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "rowtime", "c", "a")
+ ),
+ term("window", TumblingGroupWindow('w$, 'rowtime, 900000.millis)),
+ term("select",
+ "COUNT(*) AS EXPR$0, " +
+ "weightedAvg(c, a) AS wAvg, " +
+ "start('w$) AS w$start, " +
+ "end('w$) AS w$end")
+ )
+ streamUtil.verifySql(sql, expected)
+ }
+
+ @Test
+ def testHoppingFunction() = {
+ streamUtil.tableEnv.registerFunction("weightedAvg", new WeightedAvgWithMerge)
+
+ val sql =
+ "SELECT COUNT(*), weightedAvg(c, a) AS wAvg, " +
+ " HOP_START(proctime, INTERVAL '15' MINUTE, INTERVAL '1' HOUR), " +
+ " HOP_END(proctime, INTERVAL '15' MINUTE, INTERVAL '1' HOUR) " +
+ "FROM MyTable " +
+ "GROUP BY HOP(proctime, INTERVAL '15' MINUTE, INTERVAL '1' HOUR)"
+ val expected =
+ unaryNode(
+ "DataStreamGroupWindowAggregate",
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "proctime", "c", "a")
+ ),
+ term("window", SlidingGroupWindow('w$, 'proctime, 3600000.millis, 900000.millis)),
+ term("select",
+ "COUNT(*) AS EXPR$0, " +
+ "weightedAvg(c, a) AS wAvg, " +
+ "start('w$) AS w$start, " +
+ "end('w$) AS w$end")
+ )
+ streamUtil.verifySql(sql, expected)
+ }
+
+ @Test
+ def testSessionFunction() = {
+ streamUtil.tableEnv.registerFunction("weightedAvg", new WeightedAvgWithMerge)
+
+ val sql =
+ "SELECT " +
+ " COUNT(*), weightedAvg(c, a) AS wAvg, " +
+ " SESSION_START(proctime, INTERVAL '15' MINUTE), " +
+ " SESSION_END(proctime, INTERVAL '15' MINUTE) " +
+ "FROM MyTable " +
+ "GROUP BY SESSION(proctime, INTERVAL '15' MINUTE)"
+ val expected =
+ unaryNode(
+ "DataStreamGroupWindowAggregate",
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "proctime", "c", "a")
+ ),
+ term("window", SessionGroupWindow('w$, 'proctime, 900000.millis)),
+ term("select",
+ "COUNT(*) AS EXPR$0, " +
+ "weightedAvg(c, a) AS wAvg, " +
+ "start('w$) AS w$start, " +
+ "end('w$) AS w$end")
+ )
+ streamUtil.verifySql(sql, expected)
+ }
+
+ @Test
+ def testExpressionOnWindowAuxFunction() = {
+ val sql =
+ "SELECT " +
+ " COUNT(*), " +
+ " TUMBLE_END(rowtime, INTERVAL '15' MINUTE) + INTERVAL '1' MINUTE " +
+ "FROM MyTable " +
+ "GROUP BY TUMBLE(rowtime, INTERVAL '15' MINUTE)"
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+ "DataStreamGroupWindowAggregate",
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "rowtime")
+ ),
+ term("window", TumblingGroupWindow('w$, 'rowtime, 900000.millis)),
+ term("select", "COUNT(*) AS EXPR$0", "start('w$) AS w$start", "end('w$) AS w$end")
+ ),
+ term("select", "EXPR$0", "DATETIME_PLUS(w$end, 60000) AS $f1")
+ )
+
+ streamUtil.verifySql(sql, expected)
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f1fafc0e/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/OverWindowTest.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/OverWindowTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/OverWindowTest.scala
new file mode 100644
index 0000000..eea395c
--- /dev/null
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/OverWindowTest.scala
@@ -0,0 +1,505 @@
+/*
+ * 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.stream.sql
+
+import org.apache.flink.api.scala._
+import org.apache.flink.table.api.scala._
+import org.apache.flink.table.utils.TableTestUtil._
+import org.apache.flink.table.utils.{StreamTableTestUtil, TableTestBase}
+import org.junit.Test
+
+class OverWindowTest extends TableTestBase {
+ private val streamUtil: StreamTableTestUtil = streamTestUtil()
+ streamUtil.addTable[(Int, String, Long)](
+ "MyTable",
+ 'a, 'b, 'c,
+ 'proctime.proctime, 'rowtime.rowtime)
+
+ @Test
+ def testProcTimeBoundedPartitionedRowsOver() = {
+ val sql = "SELECT " +
+ "c, " +
+ "count(a) OVER (PARTITION BY c ORDER BY proctime ROWS BETWEEN 2 preceding AND " +
+ "CURRENT ROW) as cnt1, " +
+ "sum(a) OVER (PARTITION BY c ORDER BY proctime ROWS BETWEEN 2 preceding AND " +
+ "CURRENT ROW) as sum1 " +
+ "from MyTable"
+
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+ "DataStreamOverAggregate",
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "a", "c", "proctime")
+ ),
+ term("partitionBy", "c"),
+ term("orderBy", "proctime"),
+ term("rows", "BETWEEN 2 PRECEDING AND CURRENT ROW"),
+ term("select", "a", "c", "proctime", "COUNT(a) AS w0$o0, $SUM0(a) AS w0$o1")
+ ),
+ term("select", "c", "w0$o0 AS cnt1, CASE(>(w0$o0, 0), CAST(w0$o1), null) AS sum1")
+ )
+ streamUtil.verifySql(sql, expected)
+ }
+
+ @Test
+ def testProcTimeBoundedPartitionedRangeOver() = {
+
+ val sqlQuery =
+ "SELECT a, " +
+ " AVG(c) OVER (PARTITION BY a ORDER BY proctime " +
+ " RANGE BETWEEN INTERVAL '2' HOUR PRECEDING AND CURRENT ROW) AS avgA " +
+ "FROM MyTable"
+
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+ "DataStreamOverAggregate",
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "a", "c", "proctime")
+ ),
+ term("partitionBy", "a"),
+ term("orderBy", "proctime"),
+ term("range", "BETWEEN 7200000 PRECEDING AND CURRENT ROW"),
+ term(
+ "select",
+ "a",
+ "c",
+ "proctime",
+ "COUNT(c) AS w0$o0",
+ "$SUM0(c) AS w0$o1"
+ )
+ ),
+ term("select", "a", "/(CASE(>(w0$o0, 0)", "CAST(w0$o1), null), w0$o0) AS avgA")
+ )
+
+ streamUtil.verifySql(sqlQuery, expected)
+ }
+
+ @Test
+ def testProcTimeBoundedNonPartitionedRangeOver() = {
+
+ val sqlQuery =
+ "SELECT a, " +
+ " COUNT(c) OVER (ORDER BY proctime " +
+ " RANGE BETWEEN INTERVAL '10' SECOND PRECEDING AND CURRENT ROW) " +
+ "FROM MyTable"
+
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+ "DataStreamOverAggregate",
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "a", "c", "proctime")
+ ),
+ term("orderBy", "proctime"),
+ term("range", "BETWEEN 10000 PRECEDING AND CURRENT ROW"),
+ term("select", "a", "c", "proctime", "COUNT(c) AS w0$o0")
+ ),
+ term("select", "a", "w0$o0 AS $1")
+ )
+
+ streamUtil.verifySql(sqlQuery, expected)
+ }
+
+ @Test
+ def testProcTimeBoundedNonPartitionedRowsOver() = {
+ val sql = "SELECT " +
+ "c, " +
+ "count(a) OVER (ORDER BY proctime ROWS BETWEEN 2 preceding AND " +
+ "CURRENT ROW)" +
+ "from MyTable"
+
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+ "DataStreamOverAggregate",
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "a", "c", "proctime")
+ ),
+ term("orderBy", "proctime"),
+ term("rows", "BETWEEN 2 PRECEDING AND CURRENT ROW"),
+ term("select", "a", "c", "proctime", "COUNT(a) AS w0$o0")
+ ),
+ term("select", "c", "w0$o0 AS $1")
+ )
+ streamUtil.verifySql(sql, expected)
+ }
+
+
+ @Test
+ def testProcTimeUnboundedPartitionedRangeOver() = {
+ val sql = "SELECT " +
+ "c, " +
+ "count(a) OVER (PARTITION BY c ORDER BY proctime RANGE UNBOUNDED preceding) as cnt1, " +
+ "sum(a) OVER (PARTITION BY c ORDER BY proctime RANGE UNBOUNDED preceding) as cnt2 " +
+ "from MyTable"
+
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+ "DataStreamOverAggregate",
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "a", "c", "proctime")
+ ),
+ term("partitionBy", "c"),
+ term("orderBy", "proctime"),
+ term("range", "BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW"),
+ term("select", "a", "c", "proctime", "COUNT(a) AS w0$o0", "$SUM0(a) AS w0$o1")
+ ),
+ term("select", "c", "w0$o0 AS cnt1", "CASE(>(w0$o0, 0)", "CAST(w0$o1), null) AS cnt2")
+ )
+ streamUtil.verifySql(sql, expected)
+ }
+
+ @Test
+ def testProcTimeUnboundedPartitionedRowsOver() = {
+ val sql = "SELECT " +
+ "c, " +
+ "count(a) OVER (PARTITION BY c ORDER BY proctime ROWS BETWEEN UNBOUNDED preceding AND " +
+ "CURRENT ROW) " +
+ "from MyTable"
+
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+ "DataStreamOverAggregate",
+ streamTableNode(0),
+ term("partitionBy", "c"),
+ term("orderBy", "proctime"),
+ term("rows", "BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW"),
+ term("select", "a", "b", "c", "proctime", "rowtime", "COUNT(a) AS w0$o0")
+ ),
+ term("select", "c", "w0$o0 AS $1")
+ )
+ streamUtil.verifySql(sql, expected)
+ }
+
+ @Test
+ def testProcTimeUnboundedNonPartitionedRangeOver() = {
+ val sql = "SELECT " +
+ "c, " +
+ "count(a) OVER (ORDER BY proctime RANGE UNBOUNDED preceding) as cnt1, " +
+ "sum(a) OVER (ORDER BY proctime RANGE UNBOUNDED preceding) as cnt2 " +
+ "from MyTable"
+
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+ "DataStreamOverAggregate",
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "a", "c", "proctime")
+ ),
+ term("orderBy", "proctime"),
+ term("range", "BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW"),
+ term("select", "a", "c", "proctime", "COUNT(a) AS w0$o0", "$SUM0(a) AS w0$o1")
+ ),
+ term("select", "c", "w0$o0 AS cnt1", "CASE(>(w0$o0, 0)", "CAST(w0$o1), null) AS cnt2")
+ )
+ streamUtil.verifySql(sql, expected)
+ }
+
+ @Test
+ def testProcTimeUnboundedNonPartitionedRowsOver() = {
+ val sql = "SELECT " +
+ "c, " +
+ "count(a) OVER (ORDER BY proctime ROWS BETWEEN UNBOUNDED preceding AND " +
+ "CURRENT ROW) " +
+ "from MyTable"
+
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+ "DataStreamOverAggregate",
+ streamTableNode(0),
+ term("orderBy", "proctime"),
+ term("rows", "BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW"),
+ term("select", "a", "b", "c", "proctime", "rowtime", "COUNT(a) AS w0$o0")
+ ),
+ term("select", "c", "w0$o0 AS $1")
+ )
+ streamUtil.verifySql(sql, expected)
+ }
+
+ @Test
+ def testRowTimeBoundedPartitionedRowsOver() = {
+ val sql = "SELECT " +
+ "c, " +
+ "count(a) OVER (PARTITION BY c ORDER BY rowtime ROWS BETWEEN 5 preceding AND " +
+ "CURRENT ROW) " +
+ "from MyTable"
+
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+ "DataStreamOverAggregate",
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "a", "c", "rowtime")
+ ),
+ term("partitionBy", "c"),
+ term("orderBy", "rowtime"),
+ term("rows", "BETWEEN 5 PRECEDING AND CURRENT ROW"),
+ term("select", "a", "c", "rowtime", "COUNT(a) AS w0$o0")
+ ),
+ term("select", "c", "w0$o0 AS $1")
+ )
+ streamUtil.verifySql(sql, expected)
+ }
+
+ @Test
+ def testRowTimeBoundedPartitionedRangeOver() = {
+ val sql = "SELECT " +
+ "c, " +
+ "count(a) OVER (PARTITION BY c ORDER BY rowtime " +
+ "RANGE BETWEEN INTERVAL '1' SECOND preceding AND CURRENT ROW) " +
+ "from MyTable"
+
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+ "DataStreamOverAggregate",
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "a", "c", "rowtime")
+ ),
+ term("partitionBy", "c"),
+ term("orderBy", "rowtime"),
+ term("range", "BETWEEN 1000 PRECEDING AND CURRENT ROW"),
+ term("select", "a", "c", "rowtime", "COUNT(a) AS w0$o0")
+ ),
+ term("select", "c", "w0$o0 AS $1")
+ )
+ streamUtil.verifySql(sql, expected)
+ }
+
+ @Test
+ def testRowTimeBoundedNonPartitionedRowsOver() = {
+ val sql = "SELECT " +
+ "c, " +
+ "count(a) OVER (ORDER BY rowtime ROWS BETWEEN 5 preceding AND " +
+ "CURRENT ROW) " +
+ "from MyTable"
+
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+ "DataStreamOverAggregate",
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "a", "c", "rowtime")
+ ),
+ term("orderBy", "rowtime"),
+ term("rows", "BETWEEN 5 PRECEDING AND CURRENT ROW"),
+ term("select", "a", "c", "rowtime", "COUNT(a) AS w0$o0")
+ ),
+ term("select", "c", "w0$o0 AS $1")
+ )
+ streamUtil.verifySql(sql, expected)
+ }
+
+ @Test
+ def testRowTimeBoundedNonPartitionedRangeOver() = {
+ val sql = "SELECT " +
+ "c, " +
+ "count(a) OVER (ORDER BY rowtime " +
+ "RANGE BETWEEN INTERVAL '1' SECOND preceding AND CURRENT ROW) as cnt1 " +
+ "from MyTable"
+
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+ "DataStreamOverAggregate",
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "a", "c", "rowtime")
+ ),
+ term("orderBy", "rowtime"),
+ term("range", "BETWEEN 1000 PRECEDING AND CURRENT ROW"),
+ term("select", "a", "c", "rowtime", "COUNT(a) AS w0$o0")
+ ),
+ term("select", "c", "w0$o0 AS $1")
+ )
+ streamUtil.verifySql(sql, expected)
+ }
+
+ @Test
+ def testRowTimeUnboundedPartitionedRangeOver() = {
+ val sql = "SELECT " +
+ "c, " +
+ "count(a) OVER (PARTITION BY c ORDER BY rowtime RANGE UNBOUNDED preceding) as cnt1, " +
+ "sum(a) OVER (PARTITION BY c ORDER BY rowtime RANGE UNBOUNDED preceding) as cnt2 " +
+ "from MyTable"
+
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+ "DataStreamOverAggregate",
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "a", "c", "rowtime")
+ ),
+ term("partitionBy", "c"),
+ term("orderBy", "rowtime"),
+ term("range", "BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW"),
+ term("select", "a", "c", "rowtime", "COUNT(a) AS w0$o0", "$SUM0(a) AS w0$o1")
+ ),
+ term("select", "c", "w0$o0 AS cnt1", "CASE(>(w0$o0, 0)", "CAST(w0$o1), null) AS cnt2")
+ )
+ streamUtil.verifySql(sql, expected)
+ }
+
+ @Test
+ def testRowTimeUnboundedPartitionedRowsOver() = {
+ val sql = "SELECT " +
+ "c, " +
+ "count(a) OVER (PARTITION BY c ORDER BY rowtime ROWS UNBOUNDED preceding) as cnt1, " +
+ "sum(a) OVER (PARTITION BY c ORDER BY rowtime ROWS UNBOUNDED preceding) as cnt2 " +
+ "from MyTable"
+
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+ "DataStreamOverAggregate",
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "a", "c", "rowtime")
+ ),
+ term("partitionBy", "c"),
+ term("orderBy", "rowtime"),
+ term("rows", "BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW"),
+ term(
+ "select",
+ "a",
+ "c",
+ "rowtime",
+ "COUNT(a) AS w0$o0",
+ "$SUM0(a) AS w0$o1"
+ )
+ ),
+ term(
+ "select",
+ "c",
+ "w0$o0 AS cnt1",
+ "CASE(>(w0$o0, 0)",
+ "CAST(w0$o1), null) AS cnt2"
+ )
+ )
+ streamUtil.verifySql(sql, expected)
+ }
+
+ @Test
+ def testRowTimeUnboundedNonPartitionedRangeOver() = {
+ val sql = "SELECT " +
+ "c, " +
+ "count(a) OVER (ORDER BY rowtime RANGE UNBOUNDED preceding) as cnt1, " +
+ "sum(a) OVER (ORDER BY rowtime RANGE UNBOUNDED preceding) as cnt2 " +
+ "from MyTable"
+
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+ "DataStreamOverAggregate",
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "a", "c", "rowtime")
+ ),
+ term("orderBy", "rowtime"),
+ term("range", "BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW"),
+ term("select", "a", "c", "rowtime", "COUNT(a) AS w0$o0", "$SUM0(a) AS w0$o1")
+ ),
+ term("select", "c", "w0$o0 AS cnt1", "CASE(>(w0$o0, 0)", "CAST(w0$o1), null) AS cnt2")
+ )
+ streamUtil.verifySql(sql, expected)
+ }
+
+ @Test
+ def testRowTimeUnboundedNonPartitionedRowsOver() = {
+ val sql = "SELECT " +
+ "c, " +
+ "count(a) OVER (ORDER BY rowtime ROWS UNBOUNDED preceding) as cnt1, " +
+ "sum(a) OVER (ORDER BY rowtime ROWS UNBOUNDED preceding) as cnt2 " +
+ "from MyTable"
+
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+ "DataStreamOverAggregate",
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "a", "c", "rowtime")
+ ),
+ term("orderBy", "rowtime"),
+ term("rows", "BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW"),
+ term(
+ "select",
+ "a",
+ "c",
+ "rowtime",
+ "COUNT(a) AS w0$o0",
+ "$SUM0(a) AS w0$o1"
+ )
+ ),
+ term(
+ "select",
+ "c",
+ "w0$o0 AS cnt1",
+ "CASE(>(w0$o0, 0)",
+ "CAST(w0$o1), null) AS cnt2"
+ )
+ )
+ streamUtil.verifySql(sql, expected)
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f1fafc0e/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/SortTest.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/SortTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/SortTest.scala
new file mode 100644
index 0000000..a5a1319
--- /dev/null
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/SortTest.scala
@@ -0,0 +1,64 @@
+/*
+ * 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.stream.sql
+
+import org.apache.flink.api.scala._
+import org.apache.flink.table.api.scala._
+import org.apache.flink.table.utils.TableTestUtil._
+import org.apache.flink.table.utils.{StreamTableTestUtil, TableTestBase}
+import org.junit.Test
+
+class SortTest extends TableTestBase {
+
+ private val streamUtil: StreamTableTestUtil = streamTestUtil()
+ streamUtil.addTable[(Int, String, Long)]("MyTable", 'a, 'b, 'c,
+ 'proctime.proctime, 'rowtime.rowtime)
+
+ @Test
+ def testSortProcessingTime() = {
+
+ val sqlQuery = "SELECT a FROM MyTable ORDER BY proctime, c"
+
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ unaryNode("DataStreamSort",
+ streamTableNode(0),
+ term("orderBy", "proctime ASC", "c ASC")),
+ term("select", "a", "TIME_MATERIALIZATION(proctime) AS proctime", "c"))
+
+ streamUtil.verifySql(sqlQuery, expected)
+ }
+
+ @Test
+ def testSortRowTime() = {
+
+ val sqlQuery = "SELECT a FROM MyTable ORDER BY rowtime, c"
+
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ unaryNode("DataStreamSort",
+ streamTableNode(0),
+ term("orderBy", "rowtime ASC, c ASC")),
+ term("select", "a", "TIME_MATERIALIZATION(rowtime) AS rowtime", "c"))
+
+ streamUtil.verifySql(sqlQuery, expected)
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f1fafc0e/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/validation/OverWindowValidationTest.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/validation/OverWindowValidationTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/validation/OverWindowValidationTest.scala
new file mode 100644
index 0000000..413cca7
--- /dev/null
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/validation/OverWindowValidationTest.scala
@@ -0,0 +1,71 @@
+/*
+ * 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.stream.sql.validation
+
+import org.apache.flink.api.scala._
+import org.apache.flink.table.runtime.utils.JavaUserDefinedAggFunctions.OverAgg0
+import org.apache.flink.table.api.scala._
+import org.apache.flink.table.api.{TableException, ValidationException}
+import org.apache.flink.table.utils.TableTestBase
+import org.apache.flink.types.Row
+import org.junit.Test
+
+class OverWindowValidationTest extends TableTestBase {
+
+ private val streamUtil = streamTestUtil()
+ streamUtil.addTable[(Int, String, Long)]("T1", 'a, 'b, 'c, 'proctime.proctime)
+
+ /**
+ * All aggregates must be computed on the same window.
+ */
+ @Test(expected = classOf[TableException])
+ def testMultiWindow(): Unit = {
+
+ val sqlQuery = "SELECT " +
+ "c, " +
+ "count(a) OVER (PARTITION BY c ORDER BY proctime RANGE UNBOUNDED preceding), " +
+ "sum(a) OVER (PARTITION BY b ORDER BY proctime RANGE UNBOUNDED preceding) " +
+ "from T1"
+
+ streamUtil.tableEnv.sql(sqlQuery).toAppendStream[Row]
+ }
+
+ /**
+ * OVER clause is necessary for [[OverAgg0]] window function.
+ */
+ @Test(expected = classOf[ValidationException])
+ def testInvalidOverAggregation(): Unit = {
+ streamUtil.addFunction("overAgg", new OverAgg0)
+
+ val sqlQuery = "SELECT overAgg(c, a) FROM MyTable"
+
+ streamUtil.tableEnv.sql(sqlQuery)
+ }
+
+ /**
+ * OVER clause is necessary for [[OverAgg0]] window function.
+ */
+ @Test(expected = classOf[ValidationException])
+ def testInvalidOverAggregation2(): Unit = {
+ streamUtil.addFunction("overAgg", new OverAgg0)
+
+ val sqlQuery = "SELECT overAgg(c, a) FROM MyTable"
+ streamUtil.tableEnv.sql(sqlQuery)
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f1fafc0e/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/validation/SortValidationTest.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/validation/SortValidationTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/validation/SortValidationTest.scala
new file mode 100644
index 0000000..ae7486d
--- /dev/null
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/validation/SortValidationTest.scala
@@ -0,0 +1,49 @@
+/*
+ * 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.stream.sql.validation
+
+import org.apache.flink.api.scala._
+import org.apache.flink.table.api.TableException
+import org.apache.flink.table.api.scala._
+import org.apache.flink.table.utils.{StreamTableTestUtil, TableTestBase}
+import org.junit.Test
+
+class SortValidationTest extends TableTestBase {
+
+ private val streamUtil: StreamTableTestUtil = streamTestUtil()
+ streamUtil.addTable[(Int, String, Long)]("MyTable", 'a, 'b, 'c,
+ 'proctime.proctime, 'rowtime.rowtime)
+
+ // test should fail because time order is descending
+ @Test(expected = classOf[TableException])
+ def testSortProcessingTimeDesc() = {
+
+ val sqlQuery = "SELECT a FROM MyTable ORDER BY proctime DESC, c"
+ streamUtil.verifySql(sqlQuery, "")
+ }
+
+
+ // test should fail because time is not the primary order field
+ @Test(expected = classOf[TableException])
+ def testSortProcessingTimeSecondaryField() = {
+
+ val sqlQuery = "SELECT a FROM MyTable ORDER BY c, proctime"
+ streamUtil.verifySql(sqlQuery, "")
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f1fafc0e/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/validation/WindowAggregateValidationTest.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/validation/WindowAggregateValidationTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/validation/WindowAggregateValidationTest.scala
new file mode 100644
index 0000000..e7cf597
--- /dev/null
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/validation/WindowAggregateValidationTest.scala
@@ -0,0 +1,80 @@
+/*
+ * 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.stream.sql.validation
+
+import org.apache.flink.api.scala._
+import org.apache.flink.table.api.scala._
+import org.apache.flink.table.api.{TableException, ValidationException}
+import org.apache.flink.table.runtime.utils.JavaUserDefinedAggFunctions.WeightedAvgWithMerge
+import org.apache.flink.table.utils.{StreamTableTestUtil, TableTestBase}
+import org.junit.Test
+
+class WindowAggregateValidationTest extends TableTestBase {
+ private val streamUtil: StreamTableTestUtil = streamTestUtil()
+ streamUtil.addTable[(Int, String, Long)](
+ "MyTable", 'a, 'b, 'c, 'proctime.proctime, 'rowtime.rowtime)
+
+ @Test(expected = classOf[TableException])
+ def testTumbleWindowNoOffset(): Unit = {
+ val sqlQuery =
+ "SELECT SUM(a) AS sumA, COUNT(b) AS cntB " +
+ "FROM MyTable " +
+ "GROUP BY TUMBLE(proctime, INTERVAL '2' HOUR, TIME '10:00:00')"
+
+ streamUtil.verifySql(sqlQuery, "n/a")
+ }
+
+ @Test(expected = classOf[TableException])
+ def testHopWindowNoOffset(): Unit = {
+ val sqlQuery =
+ "SELECT SUM(a) AS sumA, COUNT(b) AS cntB " +
+ "FROM MyTable " +
+ "GROUP BY HOP(proctime, INTERVAL '1' HOUR, INTERVAL '2' HOUR, TIME '10:00:00')"
+
+ streamUtil.verifySql(sqlQuery, "n/a")
+ }
+
+ @Test(expected = classOf[TableException])
+ def testSessionWindowNoOffset(): Unit = {
+ val sqlQuery =
+ "SELECT SUM(a) AS sumA, COUNT(b) AS cntB " +
+ "FROM MyTable " +
+ "GROUP BY SESSION(proctime, INTERVAL '2' HOUR, TIME '10:00:00')"
+
+ streamUtil.verifySql(sqlQuery, "n/a")
+ }
+
+ @Test(expected = classOf[TableException])
+ def testVariableWindowSize() = {
+ val sql = "SELECT COUNT(*) FROM MyTable GROUP BY TUMBLE(proctime, c * INTERVAL '1' MINUTE)"
+ streamUtil.verifySql(sql, "n/a")
+ }
+
+ @Test(expected = classOf[ValidationException])
+ def testWindowUdAggInvalidArgs(): Unit = {
+ streamUtil.tableEnv.registerFunction("weightedAvg", new WeightedAvgWithMerge)
+
+ val sqlQuery =
+ "SELECT SUM(a) AS sumA, weightedAvg(a, b) AS wAvg " +
+ "FROM MyTable " +
+ "GROUP BY TUMBLE(proctime(), INTERVAL '2' HOUR, TIME '10:00:00')"
+
+ streamUtil.verifySql(sqlQuery, "n/a")
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/f1fafc0e/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/table/AggregateTest.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/table/AggregateTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/table/AggregateTest.scala
new file mode 100644
index 0000000..533235a
--- /dev/null
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/table/AggregateTest.scala
@@ -0,0 +1,189 @@
+/*
+ * 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.stream.table
+
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo
+import org.apache.flink.api.scala._
+import org.apache.flink.table.api.scala._
+import org.apache.flink.table.utils.TableTestUtil._
+import org.apache.flink.table.utils.TableTestBase
+import org.junit.Test
+
+class AggregateTest extends TableTestBase {
+
+ @Test
+ def testGroupAggregate() = {
+ val util = streamTestUtil()
+ val table = util.addTable[(Long, Int, String)]('a, 'b, 'c)
+
+ val resultTable = table
+ .groupBy('b)
+ .select('a.count)
+
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+ "DataStreamGroupAggregate",
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "a", "b")
+ ),
+ term("groupBy", "b"),
+ term("select", "b", "COUNT(a) AS TMP_0")
+ ),
+ term("select", "TMP_0")
+ )
+ util.verifyTable(resultTable, expected)
+ }
+
+ @Test
+ def testGroupAggregateWithConstant1(): Unit = {
+ val util = streamTestUtil()
+ val table = util.addTable[(Long, Int, String)]('a, 'b, 'c)
+
+ val resultTable = table
+ .select('a, 4 as 'four, 'b)
+ .groupBy('four, 'a)
+ .select('four, 'b.sum)
+
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+ "DataStreamGroupAggregate",
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "4 AS four", "b", "a")
+ ),
+ term("groupBy", "four", "a"),
+ term("select", "four", "a", "SUM(b) AS TMP_0")
+ ),
+ term("select", "4 AS four", "TMP_0")
+ )
+ util.verifyTable(resultTable, expected)
+ }
+
+ @Test
+ def testGroupAggregateWithConstant2(): Unit = {
+ val util = streamTestUtil()
+ val table = util.addTable[(Long, Int, String)]('a, 'b, 'c)
+
+ val resultTable = table
+ .select('b, 4 as 'four, 'a)
+ .groupBy('b, 'four)
+ .select('four, 'a.sum)
+
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+ "DataStreamGroupAggregate",
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "4 AS four", "a", "b")
+ ),
+ term("groupBy", "four", "b"),
+ term("select", "four", "b", "SUM(a) AS TMP_0")
+ ),
+ term("select", "4 AS four", "TMP_0")
+ )
+ util.verifyTable(resultTable, expected)
+ }
+
+ @Test
+ def testGroupAggregateWithExpressionInSelect(): Unit = {
+ val util = streamTestUtil()
+ val table = util.addTable[(Long, Int, String)]('a, 'b, 'c)
+
+ val resultTable = table
+ .select('a as 'a, 'b % 3 as 'd, 'c as 'c)
+ .groupBy('d)
+ .select('c.min, 'a.avg)
+
+ val expected =
+ unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+ "DataStreamGroupAggregate",
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "a", "MOD(b, 3) AS d", "c")
+ ),
+ term("groupBy", "d"),
+ term("select", "d", "MIN(c) AS TMP_0", "AVG(a) AS TMP_1")
+ ),
+ term("select", "TMP_0", "TMP_1")
+ )
+ util.verifyTable(resultTable, expected)
+ }
+
+ @Test
+ def testGroupAggregateWithFilter(): Unit = {
+ val util = streamTestUtil()
+ val table = util.addTable[(Long, Int, String)]('a, 'b, 'c)
+
+ val resultTable = table
+ .groupBy('b)
+ .select('b, 'a.sum)
+ .where('b === 2)
+
+ val expected =
+ unaryNode(
+ "DataStreamGroupAggregate",
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "b", "a"),
+ term("where", "=(b, 2)")
+ ),
+ term("groupBy", "b"),
+ term("select", "b", "SUM(a) AS TMP_0")
+ )
+ util.verifyTable(resultTable, expected)
+ }
+
+ @Test
+ def testGroupAggregateWithAverage(): Unit = {
+ val util = streamTestUtil()
+ val table = util.addTable[(Long, Int, String)]('a, 'b, 'c)
+
+ val resultTable = table
+ .groupBy('b)
+ .select('b, 'a.cast(BasicTypeInfo.DOUBLE_TYPE_INFO).avg)
+
+ val expected =
+ unaryNode(
+ "DataStreamGroupAggregate",
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "b", "a", "CAST(a) AS a0")
+ ),
+ term("groupBy", "b"),
+ term("select", "b", "AVG(a0) AS TMP_0")
+ )
+
+ util.verifyTable(resultTable, expected)
+ }
+}