You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ku...@apache.org on 2019/04/18 08:55:15 UTC
[flink] branch master updated: [FLINK-12211][table-planner-blink]
Add more it cases to blink batch (#8184)
This is an automated email from the ASF dual-hosted git repository.
kurt 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 96bb0d4 [FLINK-12211][table-planner-blink] Add more it cases to blink batch (#8184)
96bb0d4 is described below
commit 96bb0d47b36f869e9b007fc3312e004a90779807
Author: Jingsong Lee <lz...@aliyun.com>
AuthorDate: Thu Apr 18 16:54:55 2019 +0800
[FLINK-12211][table-planner-blink] Add more it cases to blink batch (#8184)
---
.../flink/table/codegen/calls/DivCallGen.scala | 2 +-
.../flink/table/sinks/CollectTableSink.scala | 3 +-
.../table/expressions/ScalarFunctionsTest.scala | 60 +-
.../table/expressions/SqlExpressionTest.scala | 7 +-
.../flink/table/runtime/batch/sql/CalcITCase.scala | 447 ++++++++++++++-
.../flink/table/runtime/batch/sql/MiscITCase.scala | 612 +++++++++++++++++++++
.../runtime/batch/sql/join/InnerJoinITCase.scala | 3 +-
.../join/JoinConditionTypeCoerceRuleITCase.scala | 4 +-
.../table/runtime/batch/sql/join/JoinITCase.scala | 4 +-
...JoinITCaseBase.scala => JoinITCaseHelper.scala} | 3 +-
.../runtime/batch/sql/join/OuterJoinITCase.scala | 4 +-
.../table/runtime/functions/SqlFunctionUtils.java | 4 +-
12 files changed, 1134 insertions(+), 19 deletions(-)
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/calls/DivCallGen.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/calls/DivCallGen.scala
index 6f2f1d0..41d3786 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/calls/DivCallGen.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/codegen/calls/DivCallGen.scala
@@ -36,7 +36,7 @@ class DivCallGen extends CallGenerator {
operands: Seq[GeneratedExpression],
returnType: InternalType): GeneratedExpression = {
- val (arg1, type1) = (operands(0).resultTerm, operands(0).resultType)
+ val (arg1, type1) = (operands.head.resultTerm, operands.head.resultType)
val (arg2, type2) = (operands(1).resultTerm, operands(1).resultType)
def toDec(arg: String) = s"$DECIMAL.castFrom($arg, 19, 0)"
def decDiv(arg1: String, arg2: String) = {
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/sinks/CollectTableSink.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/sinks/CollectTableSink.scala
index 27a7d49..3cc3903 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/sinks/CollectTableSink.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/sinks/CollectTableSink.scala
@@ -42,7 +42,8 @@ class CollectTableSink[T](produceOutputType: (Array[TypeInformation[_]] => TypeI
tableConfig: TableConfig,
executionConfig: ExecutionConfig): DataStreamSink[T] = {
boundedStream.writeUsingOutputFormat(collectOutputFormat)
- .name("collect")
+ .setParallelism(1)
+ .name("collect")
}
override protected def copy: TableSinkBase[T] = {
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/expressions/ScalarFunctionsTest.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/expressions/ScalarFunctionsTest.scala
index bf8128a..f05314b 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/expressions/ScalarFunctionsTest.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/expressions/ScalarFunctionsTest.scala
@@ -229,6 +229,11 @@ class ScalarFunctionsTest extends ScalarTypesTestBase {
"ascii(cast (null AS VARCHAR))",
"null"
)
+
+ testSqlApi(
+ "ascii('val_238') = ascii('val_239')",
+ "true"
+ )
}
@Test
@@ -270,6 +275,16 @@ class ScalarFunctionsTest extends ScalarTypesTestBase {
"instr('e', cast (null AS VARCHAR))",
"null"
)
+
+ testSqlApi(
+ "instr('val_238', '_') = instr('val_239', '_')",
+ "true"
+ )
+
+ testSqlApi(
+ "instr('val_239', '_')",
+ "4"
+ )
}
@Test
@@ -446,6 +461,10 @@ class ScalarFunctionsTest extends ScalarTypesTestBase {
testSqlApi(
"LENGTH(f0)",
"22")
+
+ testSqlApi(
+ "length(uuid())",
+ "36")
}
@Test
@@ -873,6 +892,7 @@ class ScalarFunctionsTest extends ScalarTypesTestBase {
testSqlApi("lpad('äääääääää',10,'??')", "?äääääääää")
testSqlApi("lpad('Hello', -1, 'x') IS NULL", "true")
testSqlApi("lpad('Hello', -1, 'x') IS NOT NULL", "false")
+ testSqlApi("lpad('ab', 5, '')", "null")
testSqlApi(
"lpad('äää',13,'12345')",
@@ -897,7 +917,7 @@ class ScalarFunctionsTest extends ScalarTypesTestBase {
testSqlApi("rpad('111',-2,'??')", "null")
testSqlApi("rpad('\u0061\u0062',1,'??')", "a") // the unicode of ab is \u0061\u0062
testSqlApi("rpad('üö',1,'??')", "ü")
-
+ testSqlApi("rpad('abcd', 5, '')", "null")
testSqlApi(
"rpad('äää',13,'12345')",
"äää1234512345")
@@ -1110,6 +1130,14 @@ class ScalarFunctionsTest extends ScalarTypesTestBase {
"0")
testSqlApi(
+ "BIN(-7)",
+ "1111111111111111111111111111111111111111111111111111111111111001")
+
+ testSqlApi(
+ "BIN(-1)",
+ "1111111111111111111111111111111111111111111111111111111111111111")
+
+ testSqlApi(
"BIN(f32)",
"1111111111111111111111111111111111111111111111111111111111111111")
}
@@ -1174,6 +1202,11 @@ class ScalarFunctionsTest extends ScalarTypesTestBase {
"1514356320000 / 60000",
"2.5239272E7")
+ // DIV return decimal
+ testSqlApi(
+ "DIV(1514356320000, 60000)",
+ "25239272")
+
testSqlApi(
"f7 / 2",
"1.5")
@@ -2804,6 +2837,11 @@ class ScalarFunctionsTest extends ScalarTypesTestBase {
testSqlApi(
"decode(encode('中国', 'UTF-16LE'), 'UTF-16LE')",
"中国")
+
+ testSqlApi(
+ "decode(encode('val_238', 'US-ASCII'), 'US-ASCII') =" +
+ " decode(encode('val_238', 'utf-8'), 'utf-8')",
+ "true")
}
@@ -3091,6 +3129,10 @@ class ScalarFunctionsTest extends ScalarTypesTestBase {
s"${2 & 2}")
testSqlApi(
+ "BITAND(-3, -2)",
+ "-4")
+
+ testSqlApi(
"BITOR(1, 2)",
s"${1 | 2}")
@@ -3099,6 +3141,14 @@ class ScalarFunctionsTest extends ScalarTypesTestBase {
s"${3 | 0}")
testSqlApi(
+ "BITOR(3, 4)",
+ "7")
+
+ testSqlApi(
+ "BITOR(-3, 2)",
+ "-1")
+
+ testSqlApi(
"BITXOR(1, 1)",
s"${1 ^ 1}")
@@ -3107,6 +3157,10 @@ class ScalarFunctionsTest extends ScalarTypesTestBase {
s"${3 ^ 1}")
testSqlApi(
+ "BITXOR(-3, 2)",
+ "-1")
+
+ testSqlApi(
"BITNOT(1)",
s"${~1}")
@@ -3117,6 +3171,10 @@ class ScalarFunctionsTest extends ScalarTypesTestBase {
testSqlApi(
"BITNOT(3)",
s"${~3}")
+
+ testSqlApi(
+ "BITNOT(-3)",
+ "2")
}
}
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/expressions/SqlExpressionTest.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/expressions/SqlExpressionTest.scala
index f70db66..098a51f 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/expressions/SqlExpressionTest.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/expressions/SqlExpressionTest.scala
@@ -21,7 +21,8 @@ package org.apache.flink.table.expressions
import org.apache.flink.api.java.typeutils.RowTypeInfo
import org.apache.flink.table.expressions.utils.ExpressionTestBase
import org.apache.flink.types.Row
-import org.junit.{Ignore, Test}
+
+import org.junit.Test
/**
* Tests all SQL expressions that are currently supported according to the documentation.
@@ -55,9 +56,9 @@ class SqlExpressionTest extends ExpressionTestBase {
testSqlApi("12 NOT BETWEEN ASYMMETRIC 13 AND 11", "true")
testSqlApi("12 NOT BETWEEN SYMMETRIC 13 AND 11", "false")
testSqlApi("'TEST' LIKE '%EST'", "true")
- //testSqlApi("'%EST' LIKE '.%EST' ESCAPE '.'", "true") // TODO
+ testSqlApi("'%EST' LIKE '.%EST' ESCAPE '.'", "true")
testSqlApi("'TEST' NOT LIKE '%EST'", "false")
- //testSqlApi("'%EST' NOT LIKE '.%EST' ESCAPE '.'", "false") // TODO
+ testSqlApi("'%EST' NOT LIKE '.%EST' ESCAPE '.'", "false")
testSqlApi("'TEST' SIMILAR TO '.EST'", "true")
//testSqlApi("'TEST' SIMILAR TO ':.EST' ESCAPE ':'", "true") // TODO
testSqlApi("'TEST' NOT SIMILAR TO '.EST'", "false")
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/CalcITCase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/CalcITCase.scala
index 2cc265f..ea17a51 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/CalcITCase.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/CalcITCase.scala
@@ -18,14 +18,15 @@
package org.apache.flink.table.runtime.batch.sql
-import org.apache.flink.api.common.typeinfo.BasicTypeInfo.{INT_TYPE_INFO, STRING_TYPE_INFO}
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo.{INT_TYPE_INFO, LONG_TYPE_INFO, STRING_TYPE_INFO}
import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO
import org.apache.flink.api.common.typeinfo.SqlTimeTypeInfo.{DATE, TIME, TIMESTAMP}
import org.apache.flink.api.common.typeinfo.{SqlTimeTypeInfo, TypeInformation}
import org.apache.flink.api.java.typeutils.{ListTypeInfo, PojoField, PojoTypeInfo, RowTypeInfo, TypeExtractor}
import org.apache.flink.api.scala._
import org.apache.flink.api.scala.typeutils.Types
-import org.apache.flink.table.api.ValidationException
+import org.apache.flink.table.api.{TableConfigOptions, ValidationException}
+import org.apache.flink.table.dataformat.DataFormatConverters.{DateConverter, TimestampConverter}
import org.apache.flink.table.dataformat.{BaseRow, BinaryString, Decimal}
import org.apache.flink.table.expressions.utils.{RichFunc1, RichFunc2, RichFunc3, SplitUDF}
import org.apache.flink.table.functions.ScalarFunction
@@ -38,7 +39,7 @@ import org.apache.flink.types.Row
import org.junit.Assert.assertEquals
import org.junit._
-import java.sql.Timestamp
+import java.sql.{Date, Timestamp}
import java.util
import scala.collection.JavaConversions._
@@ -49,7 +50,9 @@ class CalcITCase extends BatchTestBase {
@Before
def before(): Unit = {
registerCollection("Table3", data3, type3, nullablesOfData3, "a, b, c")
+ registerCollection("NullTable3", nullData3, type3, nullablesOfData3, "a, b, c")
registerCollection("SmallTable3", smallData3, type3, nullablesOfData3, "a, b, c")
+ registerCollection("testTable", buildInData, buildInType, "a,b,c,d,e,f,g,h,i,j")
}
@Test
@@ -779,6 +782,440 @@ class CalcITCase extends BatchTestBase {
row("x", "x", "z", "z", "z", "z"))
)
}
+
+ @Test
+ def testCast(): Unit = {
+ checkResult(
+ "SELECT CAST(a AS VARCHAR(10)) FROM Table3 WHERE CAST(a AS VARCHAR(10)) = '1'",
+ Seq(row(1)))
+ }
+
+ @Test
+ def testLike(): Unit = {
+ checkResult(
+ "SELECT a FROM NullTable3 WHERE c LIKE '%llo%'",
+ Seq(row(2), row(3), row(4)))
+
+ checkResult(
+ "SELECT a FROM NullTable3 WHERE CAST(a as VARCHAR(10)) LIKE CAST(b as VARCHAR(10))",
+ Seq(row(1), row(2)))
+
+ checkResult(
+ "SELECT a FROM NullTable3 WHERE c NOT LIKE '%Comment%' AND c NOT LIKE '%Hello%'",
+ Seq(row(1), row(5), row(6), row(null), row(null)))
+
+ checkResult(
+ "SELECT a FROM NullTable3 WHERE c LIKE 'Comment#%' and c LIKE '%2'",
+ Seq(row(8), row(18)))
+
+ checkResult(
+ "SELECT a FROM NullTable3 WHERE c LIKE 'Comment#12'",
+ Seq(row(18)))
+
+ checkResult(
+ "SELECT a FROM NullTable3 WHERE c LIKE '%omm%nt#12'",
+ Seq(row(18)))
+ }
+
+ @Test
+ def testLikeWithEscape(): Unit = {
+
+ val rows = Seq(
+ (1, "ha_ha"),
+ (2, "ffhaha_hahaff"),
+ (3, "aaffhaha_hahaffaa"),
+ (4, "aaffhaaa_aahaffaa"),
+ (5, "a%_ha")
+ )
+
+ BatchScalaTableEnvUtil.registerCollection(tEnv, "MyT", rows, "a, b")
+
+ checkResult(
+ "SELECT a FROM MyT WHERE b LIKE '%ha?_ha%' ESCAPE '?'",
+ Seq(row(1), row(2), row(3)))
+
+ checkResult(
+ "SELECT a FROM MyT WHERE b LIKE '%ha?_ha' ESCAPE '?'",
+ Seq(row(1)))
+
+ checkResult(
+ "SELECT a FROM MyT WHERE b LIKE 'ha?_ha%' ESCAPE '?'",
+ Seq(row(1)))
+
+ checkResult(
+ "SELECT a FROM MyT WHERE b LIKE 'ha?_ha' ESCAPE '?'",
+ Seq(row(1)))
+
+ checkResult(
+ "SELECT a FROM MyT WHERE b LIKE '%affh%ha?_ha%' ESCAPE '?'",
+ Seq(row(3)))
+
+ checkResult(
+ "SELECT a FROM MyT WHERE b LIKE 'a?%?_ha' ESCAPE '?'",
+ Seq(row(5)))
+
+ checkResult(
+ "SELECT a FROM MyT WHERE b LIKE 'h_?_ha' ESCAPE '?'",
+ Seq(row(1)))
+ }
+
+ @Test
+ def testChainLike(): Unit = {
+ // special case to test CHAIN_PATTERN.
+ checkResult(
+ "SELECT a FROM NullTable3 WHERE c LIKE '% /sys/kvengine/KVServerRole/kvengine/kv_server%'",
+ Seq())
+
+ // special case to test CHAIN_PATTERN.
+ checkResult(
+ "SELECT a FROM NullTable3 WHERE c LIKE '%Tuple%%'",
+ Seq(row(null), row(null)))
+
+ // special case to test CHAIN_PATTERN.
+ checkResult(
+ "SELECT a FROM NullTable3 WHERE c LIKE '%/order/inter/touch/backwayprice.do%%'",
+ Seq())
+ }
+
+ @Test
+ def testEqual(): Unit = {
+ checkResult(
+ "SELECT a FROM Table3 WHERE c = 'Hi'",
+ Seq(row(1)))
+
+ checkResult(
+ "SELECT c FROM Table3 WHERE c <> 'Hello' AND b = 2",
+ Seq(row("Hello world")))
+ }
+
+ @Test
+ def testSubString(): Unit = {
+ checkResult(
+ "SELECT SUBSTRING(c, 6, 13) FROM Table3 WHERE a = 6",
+ Seq(row("Skywalker")))
+ }
+
+ @Test
+ def testConcat(): Unit = {
+ checkResult(
+ "SELECT CONCAT(c, '-haha') FROM Table3 WHERE a = 1",
+ Seq(row("Hi-haha")))
+
+ checkResult(
+ "SELECT CONCAT_WS('-x-', c, 'haha') FROM Table3 WHERE a = 1",
+ Seq(row("Hi-x-haha")))
+ }
+
+ @Test
+ def testStringAgg(): Unit = {
+ checkResult(
+ "SELECT MIN(c) FROM NullTable3",
+ Seq(row("Comment#1")))
+
+ checkResult(
+ "SELECT SUM(b) FROM NullTable3 WHERE c = 'NullTuple' OR c LIKE '%Hello world%' GROUP BY c",
+ Seq(row(1998), row(2), row(3)))
+ }
+
+ @Test
+ def testStringUdf(): Unit = {
+ tEnv.registerFunction("myFunc", MyStringFunc)
+ checkResult(
+ "SELECT myFunc(c) FROM Table3 WHERE a = 1",
+ Seq(row("Hihaha")))
+ }
+
+ @Test
+ def testNestUdf(): Unit = {
+ tEnv.registerFunction("func", MyStringFunc)
+ checkResult(
+ "SELECT func(func(func(c))) FROM SmallTable3",
+ Seq(row("Hello worldhahahahahaha"), row("Hellohahahahahaha"), row("Hihahahahahaha")))
+ }
+
+
+ @Test
+ def testCurrentDate(): Unit = {
+ // Execution in on Query should return the same value
+ checkResult("SELECT CURRENT_DATE = CURRENT_DATE FROM testTable WHERE a = TRUE",
+ Seq(row(true)))
+
+ val d0 = DateConverter.INSTANCE.toInternal(new Date(System.currentTimeMillis()))
+
+ val table = parseQuery("SELECT CURRENT_DATE FROM testTable WHERE a = TRUE")
+ val result = executeQuery(table)
+ val d1 = DateConverter.INSTANCE.toInternal(
+ result.toList.head.getField(0).asInstanceOf[java.sql.Date])
+
+ Assert.assertTrue(d0 <= d1 && d1 - d0 <= 1)
+ }
+
+ @Test
+ def testCurrentTimestamp(): Unit = {
+ // Execution in on Query should return the same value
+ checkResult("SELECT CURRENT_TIMESTAMP = CURRENT_TIMESTAMP FROM testTable WHERE a = TRUE",
+ Seq(row(true)))
+
+ // CURRENT_TIMESTAMP should return the current timestamp
+ val ts0 = System.currentTimeMillis()
+
+ val table = parseQuery("SELECT CURRENT_TIMESTAMP FROM testTable WHERE a = TRUE")
+ val result = executeQuery(table)
+ val ts1 = TimestampConverter.INSTANCE.toInternal(
+ result.toList.head.getField(0).asInstanceOf[java.sql.Timestamp])
+
+ val ts2 = System.currentTimeMillis()
+
+ Assert.assertTrue(ts0 <= ts1 && ts1 <= ts2)
+ }
+
+ @Test
+ def testCurrentTime(): Unit = {
+ // Execution in on Query should return the same value
+ checkResult("SELECT CURRENT_TIME = CURRENT_TIME FROM testTable WHERE a = TRUE",
+ Seq(row(true)))
+ }
+
+ def testTimestampCompareWithDate(): Unit = {
+ checkResult("SELECT j FROM testTable WHERE j < DATE '2017-11-11'",
+ Seq(row(true)))
+ }
+
+ /**
+ * TODO Support below string timestamp format to cast to timestamp:
+ * yyyy
+ * yyyy-[m]m
+ * yyyy-[m]m-[d]d
+ * yyyy-[m]m-[d]d
+ * yyyy-[m]m-[d]d [h]h:[m]m:[s]s.[ms][ms][ms][us][us][us]
+ * yyyy-[m]m-[d]d [h]h:[m]m:[s]s.[ms][ms][ms][us][us][us]Z
+ * yyyy-[m]m-[d]d [h]h:[m]m:[s]s.[ms][ms][ms][us][us][us]-[h]h:[m]m
+ * yyyy-[m]m-[d]d [h]h:[m]m:[s]s.[ms][ms][ms][us][us][us]+[h]h:[m]m
+ * yyyy-[m]m-[d]dT[h]h:[m]m:[s]s.[ms][ms][ms][us][us][us]
+ * yyyy-[m]m-[d]dT[h]h:[m]m:[s]s.[ms][ms][ms][us][us][us]Z
+ * yyyy-[m]m-[d]dT[h]h:[m]m:[s]s.[ms][ms][ms][us][us][us]-[h]h:[m]m
+ * yyyy-[m]m-[d]dT[h]h:[m]m:[s]s.[ms][ms][ms][us][us][us]+[h]h:[m]m
+ * [h]h:[m]m:[s]s.[ms][ms][ms][us][us][us]
+ * [h]h:[m]m:[s]s.[ms][ms][ms][us][us][us]Z
+ * [h]h:[m]m:[s]s.[ms][ms][ms][us][us][us]-[h]h:[m]m
+ * [h]h:[m]m:[s]s.[ms][ms][ms][us][us][us]+[h]h:[m]m
+ * T[h]h:[m]m:[s]s.[ms][ms][ms][us][us][us]
+ * T[h]h:[m]m:[s]s.[ms][ms][ms][us][us][us]Z
+ * T[h]h:[m]m:[s]s.[ms][ms][ms][us][us][us]-[h]h:[m]m
+ * T[h]h:[m]m:[s]s.[ms][ms][ms][us][us][us]+[h]h:[m]m
+ */
+ @Ignore
+ @Test
+ def testTimestampCompareWithDateString(): Unit = {
+ //j 2015-05-20 10:00:00.887
+ checkResult("SELECT j FROM testTable WHERE j < '2017-11-11'",
+ Seq(row(UTCTimestamp("2015-05-20 10:00:00.887"))))
+ }
+
+ @Test
+ def testDateCompareWithDateString(): Unit = {
+ checkResult("SELECT h FROM testTable WHERE h <= '2017-12-12'",
+ Seq(
+ row(UTCDate("2017-12-12")),
+ row(UTCDate("2017-12-12"))
+ ))
+ }
+
+ @Test
+ def testDateEqualsWithDateString(): Unit = {
+ checkResult("SELECT h FROM testTable WHERE h = '2017-12-12'",
+ Seq(
+ row(UTCDate("2017-12-12")),
+ row(UTCDate("2017-12-12"))
+ ))
+ }
+
+ @Test
+ def testDateFormat(): Unit = {
+ //j 2015-05-20 10:00:00.887
+ checkResult("SELECT j, " +
+ " DATE_FORMAT(j, 'yyyy/MM/dd HH:mm:ss')," +
+ " DATE_FORMAT('2015-05-20 10:00:00.887', 'yyyy/MM/dd HH:mm:ss')," +
+ " DATE_FORMAT('2015-05-20 10:00:00.887', 'yyyy-MM-dd HH:mm:ss', 'yyyy/MM/dd HH:mm:ss')" +
+ " FROM testTable WHERE a = TRUE",
+ Seq(
+ row(UTCTimestamp("2015-05-20 10:00:00.887"),
+ "2015/05/20 10:00:00",
+ "2015/05/20 10:00:00",
+ "2015/05/20 10:00:00")
+ ))
+ }
+
+ @Test
+ def testYear(): Unit = {
+ checkResult("SELECT j, YEAR(j) FROM testTable WHERE a = TRUE",
+ Seq(row(UTCTimestamp("2015-05-20 10:00:00.887"), "2015")))
+ }
+
+ @Test
+ def testQuarter(): Unit = {
+ checkResult("SELECT j, QUARTER(j) FROM testTable WHERE a = TRUE",
+ Seq(row(UTCTimestamp("2015-05-20 10:00:00.887"), "2")))
+ }
+
+ @Test
+ def testMonth(): Unit = {
+ checkResult("SELECT j, MONTH(j) FROM testTable WHERE a = TRUE",
+ Seq(row(UTCTimestamp("2015-05-20 10:00:00.887"), "5")))
+ }
+
+ @Test
+ def testWeek(): Unit = {
+ checkResult("SELECT j, WEEK(j) FROM testTable WHERE a = TRUE",
+ Seq(row(UTCTimestamp("2015-05-20 10:00:00.887"), "21")))
+ }
+
+ @Test
+ def testDayOfYear(): Unit = {
+ checkResult("SELECT j, DAYOFYEAR(j) FROM testTable WHERE a = TRUE",
+ Seq(row(UTCTimestamp("2015-05-20 10:00:00.887"), "140")))
+ }
+
+ @Test
+ def testDayOfMonth(): Unit = {
+ checkResult("SELECT j, DAYOFMONTH(j) FROM testTable WHERE a = TRUE",
+ Seq(row(UTCTimestamp("2015-05-20 10:00:00.887"), "20")))
+ }
+
+ @Test
+ def testDayOfWeek(): Unit = {
+ checkResult("SELECT j, DAYOFWEEK(j) FROM testTable WHERE a = TRUE",
+ Seq(row(UTCTimestamp("2015-05-20 10:00:00.887"), "4")))
+ }
+
+ @Test
+ def testHour(): Unit = {
+ checkResult("SELECT j, HOUR(j) FROM testTable WHERE a = TRUE",
+ Seq(row(UTCTimestamp("2015-05-20 10:00:00.887"), "10")))
+ }
+
+ @Test
+ def testMinute(): Unit = {
+ checkResult("SELECT j, MINUTE(j) FROM testTable WHERE a = TRUE",
+ Seq(row(UTCTimestamp("2015-05-20 10:00:00.887"), "0")))
+ }
+
+ @Test
+ def testSecond(): Unit = {
+ checkResult("SELECT j, SECOND(j) FROM testTable WHERE a = TRUE",
+ Seq(row(UTCTimestamp("2015-05-20 10:00:00.887"), "0")))
+ }
+
+ @Test
+ def testUnixTimestamp(): Unit = {
+ checkResult("SELECT" +
+ " UNIX_TIMESTAMP('2017-12-13 19:25:30')," +
+ " UNIX_TIMESTAMP('2017-12-13 19:25:30', 'yyyy-MM-dd HH:mm:ss')" +
+ " FROM testTable WHERE a = TRUE",
+ Seq(row(1513193130, 1513193130)))
+ }
+
+ @Test
+ def testFromUnixTime(): Unit = {
+ checkResult("SELECT" +
+ " FROM_UNIXTIME(1513193130), FROM_UNIXTIME(1513193130, 'MM/dd/yyyy HH:mm:ss')" +
+ " FROM testTable WHERE a = TRUE",
+ Seq(row("2017-12-13 19:25:30", "12/13/2017 19:25:30")))
+ }
+
+ @Test
+ def testDateDiff(): Unit = {
+ checkResult("SELECT" +
+ " DATEDIFF('2017-12-14 01:00:34', '2016-12-14 12:00:00')," +
+ " DATEDIFF(TIMESTAMP '2017-12-14 01:00:23', '2016-08-14 12:00:00')," +
+ " DATEDIFF('2017-12-14 09:00:23', TIMESTAMP '2013-08-19 11:00:00')," +
+ " DATEDIFF(TIMESTAMP '2017-12-14 09:00:23', TIMESTAMP '2018-08-19 11:00:00')" +
+ " FROM testTable WHERE a = TRUE",
+ Seq(row(365, 487, 1578, -248)))
+ }
+
+ @Test
+ def testDateSub(): Unit = {
+ checkResult("SELECT" +
+ " DATE_SUB(TIMESTAMP '2017-12-14 09:00:23', 3)," +
+ " DATE_SUB('2017-12-14 09:00:23', -3)" +
+ " FROM testTable WHERE a = TRUE",
+ Seq(row("2017-12-11", "2017-12-17")))
+ }
+
+ @Test
+ def testDateAdd(): Unit = {
+ checkResult("SELECT" +
+ " DATE_ADD('2017-12-14', 4)," +
+ " DATE_ADD(TIMESTAMP '2017-12-14 09:10:20',-4)" +
+ " FROM testTable WHERE a = TRUE",
+ Seq(row("2017-12-18", "2017-12-10")))
+ }
+
+ @Test
+ def testToDate(): Unit = {
+ checkResult("SELECT" +
+ " TO_DATE(CAST(null AS VARCHAR))," +
+ " TO_DATE('2016-12-31')," +
+ " TO_DATE('2016-12-31', 'yyyy-MM-dd')",
+ Seq(row(null, UTCDate("2016-12-31"), UTCDate("2016-12-31"))))
+ }
+
+ @Test
+ def testToTimestamp(): Unit = {
+ checkResult("SELECT" +
+ " TO_TIMESTAMP(CAST(null AS VARCHAR))," +
+ " TO_TIMESTAMP('2016-12-31 00:12:00')," +
+ " TO_TIMESTAMP('2016-12-31', 'yyyy-MM-dd')",
+ Seq(row(null, UTCTimestamp("2016-12-31 00:12:00"), UTCTimestamp("2016-12-31 00:00:00"))))
+ }
+
+ @Test
+ def testCalcBinary(): Unit = {
+ registerCollection(
+ "BinaryT",
+ nullData3.map((r) => row(r.getField(0), r.getField(1), r.getField(2).toString.getBytes)),
+ new RowTypeInfo(INT_TYPE_INFO, LONG_TYPE_INFO, BYTE_PRIMITIVE_ARRAY_TYPE_INFO),
+ "a, b, c",
+ nullablesOfNullData3)
+ checkResult(
+ "select a, b, c from BinaryT where b < 1000",
+ nullData3.map((r) => row(r.getField(0), r.getField(1), r.getField(2).toString.getBytes))
+ )
+ }
+
+ @Test(expected = classOf[UnsupportedOperationException])
+ def testOrderByBinary(): Unit = {
+ registerCollection(
+ "BinaryT",
+ nullData3.map((r) => row(r.getField(0), r.getField(1), r.getField(2).toString.getBytes)),
+ new RowTypeInfo(INT_TYPE_INFO, LONG_TYPE_INFO, BYTE_PRIMITIVE_ARRAY_TYPE_INFO),
+ "a, b, c",
+ nullablesOfNullData3)
+ conf.getConf.setInteger(TableConfigOptions.SQL_RESOURCE_DEFAULT_PARALLELISM, 1)
+ conf.getConf.setBoolean(TableConfigOptions.SQL_EXEC_SORT_RANGE_ENABLED, true)
+ checkResult(
+ "select * from BinaryT order by c",
+ nullData3.sortBy((x : Row) =>
+ x.getField(2).asInstanceOf[String]).map((r) =>
+ row(r.getField(0), r.getField(1), r.getField(2).toString.getBytes)),
+ isSorted = true
+ )
+ }
+
+ @Test
+ def testGroupByBinary(): Unit = {
+ registerCollection(
+ "BinaryT2",
+ nullData3.map((r) => row(r.getField(0), r.getField(1).toString.getBytes, r.getField(2))),
+ new RowTypeInfo(INT_TYPE_INFO, BYTE_PRIMITIVE_ARRAY_TYPE_INFO, STRING_TYPE_INFO),
+ "a, b, c",
+ nullablesOfNullData3)
+ checkResult(
+ "select sum(sumA) from (select sum(a) as sumA, b, c from BinaryT2 group by c, b) group by b",
+ Seq(row(1), row(111), row(15), row(34), row(5), row(65), row(null))
+ )
+ }
}
object MyHashCode extends ScalarFunction {
@@ -873,6 +1310,10 @@ object MyToPojoFunc extends ScalarFunction {
}
}
+object MyStringFunc extends ScalarFunction {
+ def eval(s: String): String = s + "haha"
+}
+
// TODO support getResultType from literal args.
//object LiteralHashCode extends ScalarFunction {
// def eval(s: String, t: String): Any = {
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/MiscITCase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/MiscITCase.scala
new file mode 100644
index 0000000..ba5e72c
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/MiscITCase.scala
@@ -0,0 +1,612 @@
+/*
+ * 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.batch.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.table.api.{TableConfigOptions, TableException}
+import org.apache.flink.table.runtime.batch.sql.join.JoinITCaseHelper
+import org.apache.flink.table.runtime.batch.sql.join.JoinType.SortMergeJoin
+import org.apache.flink.table.runtime.utils.BatchTestBase
+import org.apache.flink.table.runtime.utils.BatchTestBase.row
+import org.apache.flink.table.runtime.utils.TestData.{buildInData, buildInType}
+import org.apache.flink.types.Row
+
+import org.junit.{Before, Ignore, Test}
+
+import scala.collection.Seq
+
+/**
+ * Misc tests.
+ */
+class MiscITCase extends BatchTestBase {
+
+ // helper methods
+
+ private var newTableId = 0
+
+ @Before
+ def before(): Unit = {
+ registerCollection("testTable", buildInData, buildInType, "a,b,c,d,e,f,g,h,i,j")
+ }
+
+ private val toRow = (p: Product) =>
+ Row.of(p.productIterator.map(_.asInstanceOf[AnyRef]).toArray: _*)
+
+ private def addTable[T <: Product : TypeInformation](tableData: Seq[T]): String = {
+ val tableRows = tableData.map(toRow)
+
+ val tupleTypeInfo = implicitly[TypeInformation[T]]
+ val fieldInfos = tupleTypeInfo.getGenericParameters.values()
+ import scala.collection.JavaConverters._
+ val rowTypeInfo = new RowTypeInfo(fieldInfos.asScala.toArray: _*)
+
+ newTableId += 1
+ val tableName = "TestTableX" + newTableId
+
+ val fields = rowTypeInfo.getFieldNames.mkString(",")
+
+ registerCollection(tableName, tableRows, rowTypeInfo, fields)
+ tableName
+ }
+
+ private def checkQuery[T <: Product : TypeInformation](
+ table1Data: Seq[T],
+ sqlQuery: String,
+ expected: Seq[_ <: Product],
+ isSorted: Boolean = false)
+ : Unit = {
+ val table2Data: Seq[Tuple1[String]] = null
+ checkQuery2(table1Data, table2Data, sqlQuery, expected, isSorted)
+ }
+
+ private def checkQuery2[T1 <: Product : TypeInformation, T2 <: Product : TypeInformation](
+ table1Data: Seq[T1],
+ table2Data: Seq[T2],
+ sqlQuery: String,
+ expected: Seq[_ <: Product],
+ isSorted: Boolean = false)
+ : Unit = {
+
+ var sqlQueryX: String = sqlQuery
+ if (table1Data!=null) {
+ val table1Name = addTable(table1Data)
+ sqlQueryX = sqlQueryX.replace("Table1", table1Name)
+ }
+ if (table2Data!=null) {
+ val table2Name = addTable(table2Data)
+ sqlQueryX = sqlQueryX.replace("Table2", table2Name)
+ }
+
+ checkResult(sqlQueryX, expected.map(toRow), isSorted)
+ }
+
+ /// tests
+
+ @Test
+ def testBasicSelect(): Unit = {
+ val testData = (1 to 100).map(i=>(i, i.toString))
+ checkQuery(
+ testData,
+ "select * from Table1",
+ testData
+ )
+ checkQuery(
+ testData,
+ "select f1 from Table1 where f0=1",
+ Seq(Tuple1("1"))
+ )
+ checkQuery(
+ testData,
+ "select sum(f0), avg(f0), count(1) from Table1",
+ Seq((5050, 50.5, 100L))
+ )
+ val testData2 = Seq((1, 1), (1, 2), (2, 1), (2, 2), (3, 1), (3, 2))
+ checkQuery(
+ testData2,
+ "select f0+f1, f0<f1 from Table1",
+ Seq((2, false), (3, true), (3, false), (4, false), (4, false), (5, false))
+ )
+ }
+
+ @Test(expected = classOf[TableException])
+ def testBasicSelect2(): Unit = {
+ val testData2 = Seq((1, 1), (1, 2), (2, 1), (2, 2), (3, 1), (3, 2))
+ checkQuery(
+ testData2,
+ "select sum(distinct f0) from Table1",
+ Seq(Tuple1(6))
+ )
+ }
+
+ @Test
+ def testNullOrderingOnMultiColumn(): Unit = {
+ env.setParallelism(1)
+ def seq(x: (Integer, String)*): Seq[(Integer, String)] = x
+ val testData = seq(Tuple2(1, "a"), Tuple2(1, null), Tuple2(null, null), Tuple2(null, "b"))
+ checkQuery(
+ testData,
+ "select * from Table1 order by f0, f1 asc",
+ seq(Tuple2(null, null), Tuple2(null, "b"), Tuple2(1, null), Tuple2(1, "a")),
+ isSorted = true
+ )
+ checkQuery(
+ testData,
+ "select * from Table1 order by f0 asc nulls first, f1 asc nulls first",
+ seq(Tuple2(null, null), Tuple2(null, "b"), Tuple2(1, null), Tuple2(1, "a")),
+ isSorted = true
+ )
+ checkQuery(
+ testData,
+ "select * from Table1 order by f0 asc nulls last, f1 asc nulls first",
+ seq(Tuple2(1, null), Tuple2(1, "a"), Tuple2(null, null), Tuple2(null, "b")),
+ isSorted = true
+ )
+
+ checkQuery(
+ testData,
+ "select * from Table1 order by f0 asc nulls first, f1 asc nulls last",
+ seq(Tuple2(null, "b"), Tuple2(null, null), Tuple2(1, "a"), Tuple2(1, null)),
+ isSorted = true
+ )
+ checkQuery(
+ testData,
+ "select * from Table1 order by f0 asc nulls last, f1 asc nulls last",
+ seq(Tuple2(1, "a"), Tuple2(1, null), Tuple2(null, "b"), Tuple2(null, null)),
+ isSorted = true
+ )
+
+ checkQuery(
+ testData,
+ "select * from Table1 order by f0 desc, f1 asc",
+ seq(Tuple2(1, null), Tuple2(1, "a"), Tuple2(null, null), Tuple2(null, "b")),
+ isSorted = true
+ )
+
+ checkQuery(
+ testData,
+ "select * from Table1 order by f0 desc nulls last, f1 desc nulls last",
+ seq(Tuple2(1, "a"), Tuple2(1, null), Tuple2(null, "b"), Tuple2(null, null)),
+ isSorted = true
+ )
+ checkQuery(
+ testData,
+ "select * from Table1 order by f0 desc nulls first, f1 desc nulls last",
+ seq(Tuple2(null, "b"), Tuple2(null, null), Tuple2(1, "a"), Tuple2(1, null)),
+ isSorted = true
+ )
+
+ checkQuery(
+ testData,
+ "select * from Table1 order by f0 desc nulls last, f1 desc nulls first",
+ seq(Tuple2(1, null), Tuple2(1, "a"), Tuple2(null, null), Tuple2(null, "b")),
+ isSorted = true
+ )
+ checkQuery(
+ testData,
+ "select * from Table1 order by f0 desc nulls first, f1 desc nulls first",
+ seq(Tuple2(null, null), Tuple2(null, "b"), Tuple2(1, null), Tuple2(1, "a")),
+ isSorted = true
+ )
+
+ checkQuery(
+ testData,
+ "select * from Table1 order by f0 desc nulls last, f1 asc nulls last",
+ seq(Tuple2(1, "a"), Tuple2(1, null), Tuple2(null, "b"), Tuple2(null, null)),
+ isSorted = true
+ )
+ checkQuery(
+ testData,
+ "select * from Table1 order by f0 desc nulls first, f1 asc nulls last",
+ seq(Tuple2(null, "b"), Tuple2(null, null), Tuple2(1, "a"), Tuple2(1, null)),
+ isSorted = true
+ )
+
+ checkQuery(
+ testData,
+ "select * from Table1 order by f0 desc nulls last, f1 asc nulls first",
+ seq(Tuple2(1, null), Tuple2(1, "a"), Tuple2(null, null), Tuple2(null, "b")),
+ isSorted = true
+ )
+ checkQuery(
+ testData,
+ "select * from Table1 order by f0 desc nulls first, f1 asc nulls first",
+ seq(Tuple2(null, null), Tuple2(null, "b"), Tuple2(1, null), Tuple2(1, "a")),
+ isSorted = true
+ )
+ }
+
+ @Test
+ def testNullOrdering(): Unit = {
+ env.setParallelism(1)
+ def seq(x: Integer*): Seq[Tuple1[Integer]] = x.map(Tuple1(_))
+ val testData = seq(2, 1, null)
+ checkQuery(
+ testData,
+ "select * from Table1 order by f0 asc",
+ seq(null, 1, 2),
+ isSorted = true
+ )
+ checkQuery(
+ testData,
+ "select * from Table1 order by f0 asc nulls first",
+ seq(null, 1, 2),
+ isSorted = true
+ )
+ checkQuery(
+ testData,
+ "select * from Table1 order by f0 asc nulls last",
+ seq(1, 2, null),
+ isSorted = true
+ )
+ checkQuery(
+ testData,
+ "select * from Table1 order by f0 desc",
+ seq(2, 1, null),
+ isSorted = true
+ )
+ checkQuery(
+ testData,
+ "select * from Table1 order by f0 desc nulls last",
+ seq(2, 1, null),
+ isSorted = true
+ )
+ checkQuery(
+ testData,
+ "select * from Table1 order by f0 desc nulls first",
+ seq(null, 2, 1),
+ isSorted = true
+ )
+ }
+
+ @Test
+ def testGlobalSorting(): Unit = {
+ env.setParallelism(1)
+ val testData2 = Seq((1, 1), (1, 2), (2, 1), (2, 2), (3, 1), (3, 2))
+ checkQuery(
+ testData2,
+ "select * from Table1 order by f0 asc, f1 asc",
+ Seq((1, 1), (1, 2), (2, 1), (2, 2), (3, 1), (3, 2)),
+ isSorted = true
+ )
+ checkQuery(
+ testData2,
+ "select * from Table1 order by f0 asc, f1 desc",
+ Seq((1, 2), (1, 1), (2, 2), (2, 1), (3, 2), (3, 1)),
+ isSorted = true
+ )
+ checkQuery(
+ testData2,
+ "select * from Table1 order by f0 desc, f1 desc",
+ Seq((3, 2), (3, 1), (2, 2), (2, 1), (1, 2), (1, 1)),
+ isSorted = true
+ )
+ checkQuery(
+ testData2,
+ "select * from Table1 order by f0 desc, f1 asc",
+ Seq((3, 1), (3, 2), (2, 1), (2, 2), (1, 1), (1, 2)),
+ isSorted = true
+ )
+ }
+
+ @Test
+ def testLimit(): Unit = {
+ // a huge limit
+ checkQuery(
+ Seq((1, "a"), (2, "b")),
+ "select * from Table1 limit 2147483638",
+ Seq((1, "a"), (2, "b"))
+ )
+ }
+
+ @Test(expected = classOf[TableException])
+ def testExcept(): Unit = {
+ checkQuery2(
+ Seq((1, "a"), (2, "b"), (3, "c"), (4, "d")),
+ Seq((1, "A"), (2, "B"), (3, "C"), (4, "D"), (5, "E")),
+ "select * from Table1 except select * from Table2",
+ Seq((1, "a"), (2, "b"), (3, "c"), (4, "d"))
+ )
+ checkQuery2(
+ Seq((1, "a"), (2, "b"), (3, "c"), (4, "d")),
+ Seq((1, "a"), (2, "b"), (3, "c"), (4, "d")),
+ "select * from Table1 except select * from Table2",
+ Seq[(Integer, String)]()
+ )
+ // check null equality
+ val nullInts = Seq[Tuple1[Integer]](Tuple1(1), Tuple1(2), Tuple1(null:Integer))
+ val allNulls = Seq[Tuple1[Integer]](Tuple1(null:Integer), Tuple1(null:Integer))
+ checkQuery(
+ nullInts,
+ "select * from Table1 except (select * from Table1 where 1=0)",
+ nullInts
+ )
+ checkQuery(
+ nullInts,
+ "select * from Table1 except select * from Table1",
+ Seq[Tuple1[Integer]]()
+ )
+ // check if values are de-duplicated
+ checkQuery(
+ allNulls,
+ "select * from Table1 except (select * from Table1 where f0=0)",
+ Seq[Tuple1[Integer]](Tuple1(null:Integer))
+ )
+ checkQuery(
+ allNulls,
+ "select * from Table1 except select * from Table1",
+ Seq[Tuple1[Integer]]()
+ )
+ // check if values are de-duplicated
+ checkQuery(
+ Seq(("a", 1), ("a", 1), ("b", 1), ("a", 2)),
+ "select * from Table1 except select * from Table1 where f1<0",
+ Seq(("a", 1), ("b", 1), ("a", 2))
+ )
+ // check if the empty set on the left side works
+ checkQuery(
+ Seq(("a", 1), ("a", 1), ("b", 1), ("a", 2)),
+ "select * from Table1 where 1=0 except select * from Table1",
+ Seq[(String, Integer)]()
+ )
+ checkQuery2(
+ Seq((1, 1), (2, 2), (3, 3)),
+ Seq((2, 2), (3, 3), (4, 4)),
+ "select * from Table1 except select * from Table2",
+ Seq((1, 1))
+ )
+ }
+
+ @Test(expected = classOf[TableException])
+ def testIntersect(): Unit = {
+ checkQuery(
+ Seq((1, "a"), (2, "b"), (3, "c"), (4, "d")),
+ "select * from Table1 intersect select * from Table1",
+ Seq((1, "a"), (2, "b"), (3, "c"), (4, "d"))
+ )
+ checkQuery2(
+ Seq((1, "a"), (2, "b"), (3, "c"), (4, "d")),
+ Seq((1, "A"), (2, "B"), (3, "C"), (4, "D"), (5, "E")),
+ "select * from Table1 intersect select * from Table2",
+ Seq[(Integer, String)]()
+ )
+ // check null equality
+ checkQuery(
+ Seq[Tuple1[Integer]](Tuple1(1), Tuple1(2), Tuple1(null:Integer)),
+ "select * from Table1 intersect select * from Table1",
+ Seq[Tuple1[Integer]](Tuple1(1), Tuple1(2), Tuple1(null))
+ )
+ checkQuery(
+ Seq[Tuple1[Integer]](Tuple1(null:Integer), Tuple1(null:Integer)),
+ "select * from Table1 intersect select * from Table1",
+ Seq[Tuple1[Integer]](Tuple1(null))
+ )
+ // check if values are de-duplicated
+ checkQuery(
+ Seq(("a", 1), ("a", 1), ("b", 1), ("a", 2)),
+ "select * from Table1 intersect select * from Table1",
+ Seq(("a", 1), ("b", 1), ("a", 2))
+ )
+ checkQuery2(
+ Seq((1, 1), (2, 2), (3, 3)),
+ Seq((2, 2), (3, 3), (4, 4)),
+ "select * from Table1 intersect select * from Table2",
+ Seq((2, 2), (3, 3))
+ )
+ }
+
+ @Test
+ def testOrderByAgg(): Unit = {
+ tEnv.getConfig.getConf.setInteger(TableConfigOptions.SQL_RESOURCE_DEFAULT_PARALLELISM, 1)
+ env.setParallelism(1)
+ checkQuery(
+ Seq((1, 10), (1, 20), (10, 1), (10, 2)),
+ "select f0, max(f1) from Table1 group by f0 order by sum(f1)",
+ Seq((10, 2), (1, 20)),
+ isSorted = true
+ )
+ }
+
+ @Test
+ def testCastInFilter(): Unit = {
+ checkQuery(
+ Seq((1,"a")),
+ "select * from Table1 where cast(f0 as varchar(9))='1'",
+ Seq((1,"a"))
+ )
+ }
+
+ @Test
+ def testOuterJoinWithIsNullFilter(): Unit = {
+ JoinITCaseHelper.disableOtherJoinOpForJoin(tEnv, SortMergeJoin)
+ checkQuery2(
+ Seq(Tuple1("x")),
+ Seq(("y", true)),
+ "select * from Table1 left outer join Table2 on Table1.f0=Table2.f0",
+ Seq(("x", null, null))
+ )
+ checkQuery2(
+ Seq(Tuple1("x")),
+ Seq(("y", true)),
+ "select * from Table1 left outer join Table2 on Table1.f0=Table2.f0 where f1 IS NULL",
+ Seq(("x", null, null))
+ )
+ }
+
+ @Test
+ def testNotNotNull(): Unit = {
+ checkQuery(
+ Seq[(Integer, Integer)]((1, 1), (2, null)),
+ "select * from Table1 where NOT(f1 IS NOT NULL)",
+ Seq((2, null))
+ )
+ checkQuery(
+ Seq[(Integer, Integer)]((1, 1), (2, null)),
+ "select * from Table1 where NOT(-f1 IS NOT NULL)",
+ Seq((2, null))
+ )
+ }
+
+ @Test
+ def testOuterJoin(): Unit = {
+ JoinITCaseHelper.disableOtherJoinOpForJoin(tEnv, SortMergeJoin)
+ checkQuery2(
+ Seq(("a", "a!"), ("b", "b!"), ("c", "c!")),
+ Seq(("a", 1), ("b", 2)),
+ "select count(*) from Table1 left outer join Table2 on Table1.f0=Table2.f0 " +
+ "where Table2.f0 IS NOT NULL OR Table1.f1 not in ('a!')",
+ Seq(Tuple1(3))
+ )
+ }
+
+ @Ignore // TODO: allows 123L="123"
+ @Test
+ def testCompareLongAndString(): Unit = {
+ checkQuery(
+ Seq((123L, "123"), (19157170390056973L, "19157170390056971")),
+ "select f0=f1 from Table1",
+ Seq(Tuple1(true), Tuple1(false))
+ )
+ }
+
+ @Test
+ def testOrderByOrdinal(): Unit = {
+ env.setParallelism(1)
+ checkQuery(
+ Seq((1, 1), (1, 2), (2, 1), (2, 2), (3, 1), (3, 2)),
+ "select 7, f0, f1 from Table1 order by 1, 2, 3",
+ Seq((7, 1, 1), (7, 1, 2), (7, 2, 1), (7, 2, 2), (7, 3, 1), (7, 3, 2)),
+ isSorted = true
+ )
+ }
+
+ @Test // lots of when..then clauses
+ def testLargeCaseWhen(): Unit = {
+ // when f0=0 then 0 when f0=1 then 1 ...
+ val w1 = (0 to 10).map(i=>s"when f0=$i then $i").mkString(" ")
+ val w2 = (0 to 10).map(i=>s"when f0=$i then ${i + 10}").mkString(" ") + " else 0"
+ checkQuery(
+ Seq(Tuple1(5)),
+ s"select case $w1 end, case $w2 end from Table1",
+ Seq((5, 15))
+ )
+ }
+
+ @Test(expected = classOf[ClassCastException])
+ def testCompareFunctionWithSubquery(): Unit = {
+ checkResult("SELECT " +
+ "b IN (3, 4, 5)," +
+ "b NOT IN (3, 4, 5)," +
+ "EXISTS (SELECT c FROM testTable WHERE c > 2)," +
+ "NOT EXISTS (SELECT c FROM testTable WHERE c = 2)" +
+ " FROM testTable WHERE a = TRUE",
+ Seq(row(true, false, false, false)))
+
+ checkResult("SELECT a, b FROM testTable WHERE CAST(b AS INTEGER) IN " +
+ "(SELECT ABS(c) FROM testTable)",
+ Seq(row(null, 2), row(true, 3)))
+
+ checkResult("SELECT a, b FROM testTable WHERE CAST(b AS INTEGER) NOT IN " +
+ "(SELECT ABS(c) FROM testTable)",
+ Seq(row(false, 1)))
+
+ checkResult("SELECT a, b FROM testTable WHERE EXISTS (SELECT c FROM testTable WHERE c > b) " +
+ "AND b = 2",
+ Seq(row(null, 2)))
+
+ checkResult("SELECT a, b FROM testTable WHERE NOT EXISTS " +
+ "(SELECT c FROM testTable WHERE c > b) OR b <> 2",
+ Seq(row(false, 1), row(true, 3)))
+ }
+
+ @Test(expected = classOf[org.apache.flink.table.api.ValidationException])
+ def testTableGenerateFunction(): Unit = {
+ checkResult("SELECT f, g, v FROM testTable," +
+ "LATERAL TABLE(STRING_SPLIT(f, ' ')) AS T(v)",
+ Seq(
+ row("abcd", "f%g", "abcd"),
+ row("e fg", null, "e"),
+ row("e fg", null, "fg")))
+
+ // BuildInFunctions in SQL is case insensitive
+ checkResult("SELECT f, g, v FROM testTable," +
+ "LATERAL TABLE(sTRING_sPLIT(f, ' ')) AS T(v)",
+ Seq(
+ row("abcd", "f%g", "abcd"),
+ row("e fg", null, "e"),
+ row("e fg", null, "fg")))
+
+ checkResult("SELECT f, g, v FROM testTable," +
+ "LATERAL TABLE(GENERATE_SERIES(0, CAST(b AS INTEGER))) AS T(v)",
+ Seq(
+ row("abcd", "f%g", 0),
+ row(null, "hij_k", 0),
+ row(null, "hij_k", 1),
+ row("e fg", null, 0),
+ row("e fg", null, 1),
+ row("e fg", null, 2)))
+
+ checkResult("SELECT f, g, v FROM testTable," +
+ "LATERAL TABLE(JSON_TUPLE('{\"a1\": \"b1\", \"a2\": \"b2\", \"e fg\": \"b3\"}'," +
+ "'a1', f)) AS T(v)",
+ Seq(
+ row("abcd", "f%g", "b1"),
+ row("abcd", "f%g", null),
+ row(null, "hij_k", "b1"),
+ row(null, "hij_k", null),
+ row("e fg", null, "b1"),
+ row("e fg", null, "b3")))
+
+ checkResult("SELECT f, g, v FROM " +
+ "testTable JOIN LATERAL TABLE(JSON_TUPLE" +
+ "('{\"a1\": \"b1\", \"a2\": \"b2\", \"e fg\": \"b3\"}', 'a1', f)) AS T(v) " +
+ "ON CHAR_LENGTH(f) = CHAR_LENGTH(v) + 2 OR CHAR_LENGTH(g) = CHAR_LENGTH(v) + 3",
+ Seq(
+ row("abcd", "f%g", "b1"),
+ row(null, "hij_k", "b1"),
+ row("e fg", null, "b1"),
+ row("e fg", null, "b3")))
+
+ checkResult("SELECT f, g, v FROM " +
+ "testTable JOIN LATERAL TABLE(JSON_TUPLE" +
+ "('{\"a1\": \"b1\", \"a2\": \"b2\", \"e fg\": \"b3\"}', 'a1', f)) AS T(v) " +
+ "ON CHAR_LENGTH(f) = CHAR_LENGTH(v) + 2 OR CHAR_LENGTH(g) = CHAR_LENGTH(v) + 3",
+ Seq(
+ row("abcd", "f%g", "b1"),
+ row(null, "hij_k", "b1"),
+ row("e fg", null, "b1"),
+ row("e fg", null, "b3")))
+ }
+
+ /**
+ * Due to the improper translation of TableFunction left outer join (see CALCITE-2004), the
+ * join predicate can only be empty or literal true (the restriction should be removed in
+ * FLINK-7865).
+ */
+ @Test(expected = classOf[org.apache.flink.table.api.ValidationException])
+ def testTableGenerateFunctionLeftJoin(): Unit = {
+ checkResult("SELECT f, g, v FROM " +
+ "testTable LEFT OUTER JOIN LATERAL TABLE(GENERATE_SERIES(0, CAST(b AS INTEGER))) AS T(v) " +
+ "ON LENGTH(f) = v + 2 OR LENGTH(g) = v + 4",
+ Seq(
+ row(null, "hij_k", 1),
+ row("e fg", null, 2)))
+ }
+}
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/join/InnerJoinITCase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/join/InnerJoinITCase.scala
index 0fed097..d8f22fa 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/join/InnerJoinITCase.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/join/InnerJoinITCase.scala
@@ -21,6 +21,7 @@ package org.apache.flink.table.runtime.batch.sql.join
import org.apache.flink.api.common.typeinfo.BasicTypeInfo.INT_TYPE_INFO
import org.apache.flink.api.java.typeutils.RowTypeInfo
import org.apache.flink.table.api.TableConfigOptions
+import org.apache.flink.table.runtime.batch.sql.join.JoinITCaseHelper.disableOtherJoinOpForJoin
import org.apache.flink.table.runtime.batch.sql.join.JoinType.{JoinType, NestedLoopJoin, SortMergeJoin}
import org.apache.flink.table.runtime.utils.BatchTestBase
import org.apache.flink.table.runtime.utils.BatchTestBase.row
@@ -35,7 +36,7 @@ import scala.collection.Seq
import scala.util.Random
// @RunWith(classOf[Parameterized]) TODO
-class InnerJoinITCase extends BatchTestBase with JoinITCaseBase {
+class InnerJoinITCase extends BatchTestBase {
val expectedJoinType: JoinType = JoinType.SortMergeJoin
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/join/JoinConditionTypeCoerceRuleITCase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/join/JoinConditionTypeCoerceRuleITCase.scala
index 9b9f37c..15a2ae2 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/join/JoinConditionTypeCoerceRuleITCase.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/join/JoinConditionTypeCoerceRuleITCase.scala
@@ -27,7 +27,7 @@ import org.junit.{Before, Ignore, Test}
// @RunWith(classOf[Parameterized]) TODO
@Ignore // TODO support JoinConditionTypeCoerce
-class JoinConditionTypeCoerceRuleITCase extends BatchTestBase with JoinITCaseBase {
+class JoinConditionTypeCoerceRuleITCase extends BatchTestBase {
@Before
def before(): Unit = {
tEnv.getConfig.getConf.setInteger(TableConfigOptions.SQL_RESOURCE_DEFAULT_PARALLELISM, 3)
@@ -44,7 +44,7 @@ class JoinConditionTypeCoerceRuleITCase extends BatchTestBase with JoinITCaseBas
nullablesOfNumericData,
"a, b, c, d, e")
// Disable NestedLoopJoin.
- disableOtherJoinOpForJoin(tEnv, JoinType.SortMergeJoin)
+ JoinITCaseHelper.disableOtherJoinOpForJoin(tEnv, JoinType.SortMergeJoin)
}
@Test
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/join/JoinITCase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/join/JoinITCase.scala
index 8db382b..244189d 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/join/JoinITCase.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/join/JoinITCase.scala
@@ -38,7 +38,7 @@ import scala.collection.JavaConversions._
import scala.collection.Seq
// @RunWith(classOf[Parameterized]) TODO
-class JoinITCase() extends BatchTestBase with JoinITCaseBase {
+class JoinITCase() extends BatchTestBase {
val expectedJoinType: JoinType = SortMergeJoin
@@ -52,7 +52,7 @@ class JoinITCase() extends BatchTestBase with JoinITCaseBase {
registerCollection("l", data2_1, INT_DOUBLE, "a, b")
registerCollection("r", data2_2, INT_DOUBLE, "c, d")
registerCollection("t", data2_3, INT_DOUBLE, nullablesOfData2_3, "c, d")
- disableOtherJoinOpForJoin(tEnv, expectedJoinType)
+ JoinITCaseHelper.disableOtherJoinOpForJoin(tEnv, expectedJoinType)
}
@Test
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/join/JoinITCaseBase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/join/JoinITCaseHelper.scala
similarity index 98%
rename from flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/join/JoinITCaseBase.scala
rename to flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/join/JoinITCaseHelper.scala
index b07caac..d773133 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/join/JoinITCaseBase.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/join/JoinITCaseHelper.scala
@@ -24,7 +24,7 @@ import org.apache.flink.table.runtime.batch.sql.join.JoinType.{BroadcastHashJoin
/**
* providing join it case utility functions.
*/
-trait JoinITCaseBase {
+object JoinITCaseHelper {
def disableBroadcastHashJoin(tEnv: TableEnvironment): Unit = {
tEnv.getConfig.getConf.setLong(
@@ -43,6 +43,7 @@ trait JoinITCaseBase {
tEnv.getConfig.getConf.setString(
TableConfigOptions.SQL_EXEC_DISABLED_OPERATORS, disabledOperators)
}
+
}
object JoinType extends Enumeration {
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/join/OuterJoinITCase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/join/OuterJoinITCase.scala
index b21f322..a16b88e 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/join/OuterJoinITCase.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/join/OuterJoinITCase.scala
@@ -29,7 +29,7 @@ import org.junit.{Before, Test}
import scala.collection.Seq
//@RunWith(classOf[Parameterized]) TODO
-class OuterJoinITCase extends BatchTestBase with JoinITCaseBase {
+class OuterJoinITCase extends BatchTestBase {
val expectedJoinType: JoinType = JoinType.SortMergeJoin
@@ -65,7 +65,7 @@ class OuterJoinITCase extends BatchTestBase with JoinITCaseBase {
registerCollection("allnulls", allNulls, INT_ONLY, nullablesOfAllNulls, "a")
registerCollection("leftT", leftT, INT_DOUBLE, "a, b")
registerCollection("rightT", rightT, INT_DOUBLE, "c, d")
- disableOtherJoinOpForJoin(tEnv, expectedJoinType)
+ JoinITCaseHelper.disableOtherJoinOpForJoin(tEnv, expectedJoinType)
}
@Test
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/functions/SqlFunctionUtils.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/functions/SqlFunctionUtils.java
index a881ea2..0859119 100644
--- a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/functions/SqlFunctionUtils.java
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/functions/SqlFunctionUtils.java
@@ -219,7 +219,7 @@ public class SqlFunctionUtils {
* If str is longer than len, the return value is shortened to len characters.
*/
public static String lpad(String base, int len, String pad) {
- if (len < 0) {
+ if (len < 0 || "".equals(pad)) {
return null;
} else if (len == 0) {
return "";
@@ -254,7 +254,7 @@ public class SqlFunctionUtils {
* If str is longer than len, the return value is shortened to len characters.
*/
public static String rpad(String base, int len, String pad) {
- if (len < 0) {
+ if (len < 0 || "".equals(pad)) {
return null;
} else if (len == 0) {
return "";