You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by fh...@apache.org on 2017/11/08 17:42:02 UTC
[5/5] flink git commit: [FLINK-7922] [table] Fix
FlinkTypeFactory.leastRestrictive for composite types.
[FLINK-7922] [table] Fix FlinkTypeFactory.leastRestrictive for composite types.
This closes #4929.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/949ae79d
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/949ae79d
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/949ae79d
Branch: refs/heads/master
Commit: 949ae79dad149a97f4703daa7e58d7fbd0957dac
Parents: 9d20e89
Author: Rong Rong <ro...@uber.com>
Authored: Tue Oct 31 11:05:38 2017 -0700
Committer: Fabian Hueske <fh...@apache.org>
Committed: Wed Nov 8 17:12:18 2017 +0100
----------------------------------------------------------------------
.../flink/table/calcite/FlinkTypeFactory.scala | 39 +++++----
.../table/api/batch/sql/SetOperatorsTest.scala | 55 +++++++++++++
.../api/batch/table/SetOperatorsTest.scala | 59 ++++++++++++++
.../flink/table/api/stream/sql/UnionTest.scala | 83 ++++++++++++++++++++
.../table/expressions/ScalarOperatorsTest.scala | 5 +-
.../utils/ScalarOperatorsTestBase.scala | 9 ++-
.../stream/table/SetOperatorsITCase.scala | 20 ++++-
.../table/runtime/utils/CommonTestData.scala | 6 ++
.../flink/table/utils/TableTestBase.scala | 26 ++++++
9 files changed, 280 insertions(+), 22 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/949ae79d/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/FlinkTypeFactory.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/FlinkTypeFactory.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/FlinkTypeFactory.scala
index 2874e61..04fab76 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/FlinkTypeFactory.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/calcite/FlinkTypeFactory.scala
@@ -268,30 +268,37 @@ class FlinkTypeFactory(typeSystem: RelDataTypeSystem) extends JavaTypeFactoryImp
override def leastRestrictive(types: util.List[RelDataType]): RelDataType = {
val type0 = types.get(0)
if (type0.getSqlTypeName != null) {
- val resultType = resolveAny(types)
- if (resultType != null) {
- return resultType
+ val resultType = resolveAllIdenticalTypes(types)
+ if (resultType.isDefined) {
+ // result type for identical types
+ return resultType.get
}
}
+ // fall back to super
super.leastRestrictive(types)
}
- private def resolveAny(types: util.List[RelDataType]): RelDataType = {
+ private def resolveAllIdenticalTypes(types: util.List[RelDataType]): Option[RelDataType] = {
val allTypes = types.asScala
- val hasAny = allTypes.exists(_.getSqlTypeName == SqlTypeName.ANY)
- if (hasAny) {
- val head = allTypes.head
- // only allow ANY with exactly the same GenericRelDataType for all types
- if (allTypes.forall(_ == head)) {
- val nullable = allTypes.exists(
- sqlType => sqlType.isNullable || sqlType.getSqlTypeName == SqlTypeName.NULL
- )
- createTypeWithNullability(head, nullable)
- } else {
+
+ val head = allTypes.head
+ // check if all types are the same
+ if (allTypes.forall(_ == head)) {
+ // types are the same, check nullability
+ val nullable = allTypes
+ .exists(sqlType => sqlType.isNullable || sqlType.getSqlTypeName == SqlTypeName.NULL)
+ // return type with nullability
+ Some(createTypeWithNullability(head, nullable))
+ } else {
+ // types are not all the same
+ if (allTypes.exists(_.getSqlTypeName == SqlTypeName.ANY)) {
+ // one of the type was ANY.
+ // we cannot generate a common type if it differs from other types.
throw TableException("Generic ANY types must have a common type information.")
+ } else {
+ // cannot resolve a common type for different input types
+ None
}
- } else {
- null
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/949ae79d/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/batch/sql/SetOperatorsTest.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/batch/sql/SetOperatorsTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/batch/sql/SetOperatorsTest.scala
index bc9b453..bff0b78 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/batch/sql/SetOperatorsTest.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/batch/sql/SetOperatorsTest.scala
@@ -18,8 +18,11 @@
package org.apache.flink.table.api.batch.sql
+import org.apache.flink.api.java.typeutils.{GenericTypeInfo, RowTypeInfo}
import org.apache.flink.api.scala._
+import org.apache.flink.table.api.Types
import org.apache.flink.table.api.scala._
+import org.apache.flink.table.runtime.utils.CommonTestData.NonPojo
import org.apache.flink.table.utils.TableTestUtil._
import org.apache.flink.table.utils.TableTestBase
import org.junit.{Ignore, Test}
@@ -178,4 +181,56 @@ class SetOperatorsTest extends TableTestBase {
expected
)
}
+
+ @Test
+ def testUnionNullableTypes(): Unit = {
+ val util = batchTestUtil()
+ util.addTable[((Int, String), (Int, String), Int)]("A", 'a, 'b, 'c)
+
+ val expected = binaryNode(
+ "DataSetUnion",
+ unaryNode(
+ "DataSetCalc",
+ batchTableNode(0),
+ term("select", "a")
+ ),
+ unaryNode(
+ "DataSetCalc",
+ batchTableNode(0),
+ term("select", "CASE(>(c, 0), b, null) AS EXPR$0")
+ ),
+ term("union", "a")
+ )
+
+ util.verifySql(
+ "SELECT a FROM A UNION ALL SELECT CASE WHEN c > 0 THEN b ELSE NULL END FROM A",
+ expected
+ )
+ }
+
+ @Test
+ def testUnionAnyType(): Unit = {
+ val util = batchTestUtil()
+ val typeInfo = Types.ROW(
+ new GenericTypeInfo(classOf[NonPojo]),
+ new GenericTypeInfo(classOf[NonPojo]))
+ util.addJavaTable(typeInfo, "A", "a, b")
+
+ val expected = binaryNode(
+ "DataSetUnion",
+ unaryNode(
+ "DataSetCalc",
+ batchTableNode(0),
+ term("select", "a")
+ ),
+ unaryNode(
+ "DataSetCalc",
+ batchTableNode(0),
+ term("select", "b")
+ ),
+ term("union", "a")
+ )
+
+ util.verifyJavaSql("SELECT a FROM A UNION ALL SELECT b FROM A", expected)
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/949ae79d/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/batch/table/SetOperatorsTest.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/batch/table/SetOperatorsTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/batch/table/SetOperatorsTest.scala
index 686973e..2d4e205 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/batch/table/SetOperatorsTest.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/batch/table/SetOperatorsTest.scala
@@ -20,8 +20,12 @@ package org.apache.flink.table.api.batch.table
import java.sql.Timestamp
+import org.apache.flink.api.java.typeutils.GenericTypeInfo
import org.apache.flink.api.scala._
+import org.apache.flink.table.api.Types
import org.apache.flink.table.api.scala._
+import org.apache.flink.table.expressions.Null
+import org.apache.flink.table.runtime.utils.CommonTestData.NonPojo
import org.apache.flink.table.utils.TableTestBase
import org.apache.flink.table.utils.TableTestUtil._
import org.junit.Test
@@ -76,4 +80,59 @@ class SetOperatorsTest extends TableTestBase {
util.verifyTable(in, expected)
}
+
+ @Test
+ def testUnionNullableTypes(): Unit = {
+ val util = batchTestUtil()
+ val t = util.addTable[((Int, String), (Int, String), Int)]("A", 'a, 'b, 'c)
+
+ val in = t.select('a)
+ .unionAll(
+ t.select(('c > 0) ? ('b, Null(createTypeInformation[(Int, String)]))))
+
+ val expected = binaryNode(
+ "DataSetUnion",
+ unaryNode(
+ "DataSetCalc",
+ batchTableNode(0),
+ term("select", "a")
+ ),
+ unaryNode(
+ "DataSetCalc",
+ batchTableNode(0),
+ term("select", "CASE(>(c, 0), b, null) AS _c0")
+ ),
+ term("union", "a")
+ )
+
+ util.verifyTable(in, expected)
+ }
+
+ @Test
+ def testUnionAnyType(): Unit = {
+ val util = batchTestUtil()
+ val typeInfo = Types.ROW(
+ new GenericTypeInfo(classOf[NonPojo]),
+ new GenericTypeInfo(classOf[NonPojo]))
+ val t = util.addJavaTable(typeInfo, "A", "a, b")
+
+ val in = t.select('a).unionAll(t.select('b))
+
+ val expected = binaryNode(
+ "DataSetUnion",
+ unaryNode(
+ "DataSetCalc",
+ batchTableNode(0),
+ term("select", "a")
+ ),
+ unaryNode(
+ "DataSetCalc",
+ batchTableNode(0),
+ term("select", "b")
+ ),
+ term("union", "a")
+ )
+
+ util.verifyJavaTable(in, expected)
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/949ae79d/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/UnionTest.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/UnionTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/UnionTest.scala
new file mode 100644
index 0000000..7e807f6
--- /dev/null
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/api/stream/sql/UnionTest.scala
@@ -0,0 +1,83 @@
+/*
+ * 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.java.typeutils.GenericTypeInfo
+import org.apache.flink.api.scala._
+import org.apache.flink.table.api.Types
+import org.apache.flink.table.api.scala._
+import org.apache.flink.table.runtime.utils.CommonTestData.NonPojo
+import org.apache.flink.table.utils.TableTestUtil._
+import org.apache.flink.table.utils.TableTestBase
+import org.junit.Test
+
+class UnionTest extends TableTestBase {
+
+ @Test
+ def testUnionAllNullableCompositeType() = {
+ val streamUtil = streamTestUtil()
+ streamUtil.addTable[((Int, String), (Int, String), Int)]("A", 'a, 'b, 'c)
+
+ val expected = binaryNode(
+ "DataStreamUnion",
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "a")
+ ),
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "CASE(>(c, 0), b, null) AS EXPR$0")
+ ),
+ term("union all", "a")
+ )
+
+ streamUtil.verifySql(
+ "SELECT a FROM A UNION ALL SELECT CASE WHEN c > 0 THEN b ELSE NULL END FROM A",
+ expected
+ )
+ }
+
+ @Test
+ def testUnionAnyType(): Unit = {
+ val streamUtil = streamTestUtil()
+ val typeInfo = Types.ROW(
+ new GenericTypeInfo(classOf[NonPojo]),
+ new GenericTypeInfo(classOf[NonPojo]))
+ streamUtil.addJavaTable(typeInfo, "A", "a, b")
+
+ val expected = binaryNode(
+ "DataStreamUnion",
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "a")
+ ),
+ unaryNode(
+ "DataStreamCalc",
+ streamTableNode(0),
+ term("select", "b")
+ ),
+ term("union all", "a")
+ )
+
+ streamUtil.verifyJavaSql("SELECT a FROM A UNION ALL SELECT b FROM A", expected)
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/949ae79d/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/ScalarOperatorsTest.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/ScalarOperatorsTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/ScalarOperatorsTest.scala
index cbdce8b..6dd2afc 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/ScalarOperatorsTest.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/ScalarOperatorsTest.scala
@@ -211,7 +211,6 @@ class ScalarOperatorsTest extends ScalarOperatorsTestBase {
@Test
def testOtherExpressions(): Unit = {
-
// nested field null type
testSqlApi("CASE WHEN f13.f1 IS NULL THEN 'a' ELSE 'b' END", "a")
testSqlApi("CASE WHEN f13.f1 IS NOT NULL THEN 'a' ELSE 'b' END", "b")
@@ -222,6 +221,10 @@ class ScalarOperatorsTest extends ScalarOperatorsTestBase {
testAllApis('f13.get("f1").isNull, "f13.get('f1').isNull", "f13.f1 IS NULL", "true")
testAllApis('f13.get("f1").isNotNull, "f13.get('f1').isNotNull", "f13.f1 IS NOT NULL", "false")
+ // array element access test
+ testSqlApi("CASE WHEN f18 IS NOT NULL THEN f18[1] ELSE NULL END", "1")
+ testSqlApi("CASE WHEN f19 IS NOT NULL THEN f19[1] ELSE NULL END", "(1,a)")
+
// boolean literals
testAllApis(
true,
http://git-wip-us.apache.org/repos/asf/flink/blob/949ae79d/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/utils/ScalarOperatorsTestBase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/utils/ScalarOperatorsTestBase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/utils/ScalarOperatorsTestBase.scala
index 149d8c1..3eeb215 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/utils/ScalarOperatorsTestBase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/expressions/utils/ScalarOperatorsTestBase.scala
@@ -22,6 +22,7 @@ import java.sql.Date
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.java.typeutils.RowTypeInfo
+import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.table.api.Types
import org.apache.flink.table.functions.ScalarFunction
import org.apache.flink.types.Row
@@ -29,7 +30,7 @@ import org.apache.flink.types.Row
class ScalarOperatorsTestBase extends ExpressionTestBase {
def testData: Row = {
- val testData = new Row(18)
+ val testData = new Row(20)
testData.setField(0, 1: Byte)
testData.setField(1, 1: Short)
testData.setField(2, 1)
@@ -48,6 +49,8 @@ class ScalarOperatorsTestBase extends ExpressionTestBase {
testData.setField(15, Date.valueOf("1996-11-10"))
testData.setField(16, BigDecimal("0.00000000").bigDecimal)
testData.setField(17, BigDecimal("10.0").bigDecimal)
+ testData.setField(18, Array[Integer](1,2))
+ testData.setField(19, Array[(Int, String)]((1,"a"), (2, "b")))
testData
}
@@ -70,7 +73,9 @@ class ScalarOperatorsTestBase extends ExpressionTestBase {
Types.STRING,
Types.SQL_DATE,
Types.DECIMAL,
- Types.DECIMAL
+ Types.DECIMAL,
+ Types.OBJECT_ARRAY(Types.INT),
+ Types.OBJECT_ARRAY(createTypeInformation[(Int, String)])
).asInstanceOf[TypeInformation[Any]]
}
http://git-wip-us.apache.org/repos/asf/flink/blob/949ae79d/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/stream/table/SetOperatorsITCase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/stream/table/SetOperatorsITCase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/stream/table/SetOperatorsITCase.scala
index cf195a5..5e15e14 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/stream/table/SetOperatorsITCase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/stream/table/SetOperatorsITCase.scala
@@ -23,6 +23,7 @@ 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.apache.flink.table.runtime.utils.CommonTestData.NonPojo
import org.apache.flink.table.runtime.utils.{StreamITCase, StreamTestData}
import org.apache.flink.types.Row
import org.junit.Assert._
@@ -88,9 +89,22 @@ class SetOperatorsITCase extends StreamingMultipleProgramsTestBase {
assertEquals(expected.sorted, StreamITCase.testResults.sorted)
}
- class NonPojo {
- val x = new java.util.HashMap[String, String]()
+ @Test
+ def testUnionWithCompositeType(): Unit = {
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ val tEnv = TableEnvironment.getTableEnvironment(env)
+
+ StreamITCase.testResults = mutable.MutableList()
+ val s1 = env.fromElements((1, (1, "a")), (2, (2, "b")))
+ .toTable(tEnv, 'a, 'b)
+ val s2 = env.fromElements(((3, "c"), 3), ((4, "d"), 4))
+ .toTable(tEnv, 'a, 'b)
- override def toString: String = x.toString
+ val result = s1.unionAll(s2.select('b, 'a)).toAppendStream[Row]
+ result.addSink(new StreamITCase.StringSink[Row])
+ env.execute()
+
+ val expected = mutable.MutableList("1,(1,a)", "2,(2,b)", "3,(3,c)", "4,(4,d)")
+ assertEquals(expected.sorted, StreamITCase.testResults.sorted)
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/949ae79d/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/utils/CommonTestData.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/utils/CommonTestData.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/utils/CommonTestData.scala
index e8568ca..9223887 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/utils/CommonTestData.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/utils/CommonTestData.scala
@@ -176,4 +176,10 @@ object CommonTestData {
this(null, null)
}
}
+
+ class NonPojo {
+ val x = new java.util.HashMap[String, String]()
+
+ override def toString: String = x.toString
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/949ae79d/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/utils/TableTestBase.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/utils/TableTestBase.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/utils/TableTestBase.scala
index 0a0d12e..4042f50 100644
--- a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/utils/TableTestBase.scala
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/utils/TableTestBase.scala
@@ -190,6 +190,19 @@ case class BatchTableTestUtil() extends TableTestUtil {
actual.split("\n").map(_.trim).mkString("\n"))
}
+ def verifyJavaSql(query: String, expected: String): Unit = {
+ verifyJavaTable(javaTableEnv.sqlQuery(query), expected)
+ }
+
+ def verifyJavaTable(resultTable: Table, expected: String): Unit = {
+ val relNode = resultTable.getRelNode
+ val optimized = javaTableEnv.optimize(relNode)
+ val actual = RelOptUtil.toString(optimized)
+ assertEquals(
+ expected.split("\n").map(_.trim).mkString("\n"),
+ actual.split("\n").map(_.trim).mkString("\n"))
+ }
+
def printTable(resultTable: Table): Unit = {
val relNode = resultTable.getRelNode
val optimized = tableEnv.optimize(relNode)
@@ -268,6 +281,19 @@ case class StreamTableTestUtil() extends TableTestUtil {
actual.split("\n").map(_.trim).mkString("\n"))
}
+ def verifyJavaSql(query: String, expected: String): Unit = {
+ verifyJavaTable(javaTableEnv.sqlQuery(query), expected)
+ }
+
+ def verifyJavaTable(resultTable: Table, expected: String): Unit = {
+ val relNode = resultTable.getRelNode
+ val optimized = javaTableEnv.optimize(relNode, updatesAsRetraction = false)
+ val actual = RelOptUtil.toString(optimized)
+ assertEquals(
+ expected.split("\n").map(_.trim).mkString("\n"),
+ actual.split("\n").map(_.trim).mkString("\n"))
+ }
+
// the print methods are for debugging purposes only
def printTable(resultTable: Table): Unit = {
val relNode = resultTable.getRelNode