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 2018/01/05 16:15:00 UTC
[1/2] flink git commit: [FLINK-8139] [table] Check types for
hashCode/equals only for stateful operations
Repository: flink
Updated Branches:
refs/heads/master 9dd3a859b -> f88da4d04
[FLINK-8139] [table] Check types for hashCode/equals only for stateful operations
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/f88da4d0
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/f88da4d0
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/f88da4d0
Branch: refs/heads/master
Commit: f88da4d04c328c46b9c94dc76d3e7a3e4e87b2ea
Parents: 3c91de5
Author: twalthr <tw...@apache.org>
Authored: Fri Jan 5 16:38:01 2018 +0100
Committer: twalthr <tw...@apache.org>
Committed: Fri Jan 5 17:14:30 2018 +0100
----------------------------------------------------------------------
.../flink/table/api/BatchTableEnvironment.scala | 1 -
.../table/api/StreamTableEnvironment.scala | 1 -
.../flink/table/api/TableEnvironment.scala | 29 +---------
.../flink/table/typeutils/TypeCheckUtils.scala | 58 +++++++++++++++++++-
.../table/typeutils/TypeCheckUtilsTest.scala | 54 ++++++++++++++++++
5 files changed, 112 insertions(+), 31 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/f88da4d0/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/BatchTableEnvironment.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/BatchTableEnvironment.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/BatchTableEnvironment.scala
index e0e00d4..c25007b 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/BatchTableEnvironment.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/BatchTableEnvironment.scala
@@ -100,7 +100,6 @@ abstract class BatchTableEnvironment(
tableSource match {
case batchTableSource: BatchTableSource[_] =>
- TableEnvironment.checkValidTableType(tableSource)
registerTableInternal(name, new BatchTableSourceTable(batchTableSource))
case _ =>
throw new TableException("Only BatchTableSource can be registered in " +
http://git-wip-us.apache.org/repos/asf/flink/blob/f88da4d0/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/StreamTableEnvironment.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/StreamTableEnvironment.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/StreamTableEnvironment.scala
index 84d5784..920da2e 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/StreamTableEnvironment.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/StreamTableEnvironment.scala
@@ -118,7 +118,6 @@ abstract class StreamTableEnvironment(
s"A rowtime attribute requires an EventTime time characteristic in stream " +
s"environment. But is: ${execEnv.getStreamTimeCharacteristic}")
}
- TableEnvironment.checkValidTableType(tableSource)
registerTableInternal(name, new StreamTableSourceTable(streamTableSource))
case _ =>
throw new TableException("Only StreamTableSource can be registered in " +
http://git-wip-us.apache.org/repos/asf/flink/blob/f88da4d0/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala
index 8782477..c3cab13 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala
@@ -36,7 +36,7 @@ import org.apache.calcite.sql.parser.SqlParser
import org.apache.calcite.sql.util.ChainedSqlOperatorTable
import org.apache.calcite.tools._
import org.apache.flink.api.common.functions.MapFunction
-import org.apache.flink.api.common.typeinfo.{AtomicType, TypeInformation, PrimitiveArrayTypeInfo}
+import org.apache.flink.api.common.typeinfo.{AtomicType, TypeInformation}
import org.apache.flink.api.common.typeutils.CompositeType
import org.apache.flink.api.java.typeutils.{RowTypeInfo, _}
import org.apache.flink.api.java.{ExecutionEnvironment => JavaBatchExecEnv}
@@ -1148,33 +1148,6 @@ object TableEnvironment {
}
/**
- * Checks if the chosen table type is valid.
- * @param table The table to check
- */
- def checkValidTableType(table: Any): Unit = {
- val types = table match {
- case t: Table => table.asInstanceOf[Table].getSchema.getTypes
- case st: TableSource[_] => table.asInstanceOf[TableSource[_]].getTableSchema.getTypes
- }
- for (typeInfo <- types) {
- if(!typeInfo.asInstanceOf[TypeInformation[_]].isBasicType &&
- !typeInfo.isInstanceOf[PrimitiveArrayTypeInfo[_]]) {
- val typeClass = typeInfo.asInstanceOf[TypeInformation[_]].getTypeClass
- if (typeClass.getMethod("hashCode").getDeclaringClass eq classOf[Object]) {
- throw new ValidationException(s"Illegal Table type." +
- s"Please make sure type ${typeClass.getCanonicalName}" +
- s" implement own hashCode method")
- }
- if (typeClass.getMethod("equals", classOf[Any]).getDeclaringClass eq classOf[Object]) {
- throw new ValidationException(s"Illegal Table type." +
- s"Please make sure type ${typeClass.getCanonicalName}" +
- s" implement own equals method")
- }
- }
- }
- }
-
- /**
* Returns field indexes for a given [[TypeInformation]].
*
* @param inputType The TypeInformation extract the field positions from.
http://git-wip-us.apache.org/repos/asf/flink/blob/f88da4d0/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/typeutils/TypeCheckUtils.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/typeutils/TypeCheckUtils.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/typeutils/TypeCheckUtils.scala
index fa7abab..278ae18 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/typeutils/TypeCheckUtils.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/typeutils/TypeCheckUtils.scala
@@ -19,7 +19,9 @@ package org.apache.flink.table.typeutils
import org.apache.flink.api.common.typeinfo.BasicTypeInfo._
import org.apache.flink.api.common.typeinfo._
-import org.apache.flink.api.java.typeutils.{MapTypeInfo, ObjectArrayTypeInfo}
+import org.apache.flink.api.common.typeutils.CompositeType
+import org.apache.flink.api.java.typeutils.{MapTypeInfo, ObjectArrayTypeInfo, PojoTypeInfo}
+import org.apache.flink.table.api.ValidationException
import org.apache.flink.table.typeutils.TimeIntervalTypeInfo.{INTERVAL_MILLIS, INTERVAL_MONTHS}
import org.apache.flink.table.validate._
@@ -104,4 +106,58 @@ object TypeCheckUtils {
ValidationFailure(s"$caller requires orderable types, get $dataType here")
}
}
+
+ /**
+ * Checks whether a type implements own hashCode() and equals() methods for storing an instance
+ * in Flink's state.
+ *
+ * @param t type information to be validated
+ */
+ def validateStateType(t: TypeInformation[_]): Unit = t match {
+ // make sure that a POJO class is a valid state type
+ case pt: PojoTypeInfo[_] =>
+ // we don't check the types recursively to give a chance of wrapping
+ // proper hashCode/equals methods around an immutable type
+ validateStateType(pt.getClass)
+ // recursively check composite types
+ case ct: CompositeType[_] =>
+ validateStateType(t.getTypeClass)
+ // we check recursively for entering Flink types such as tuples and rows
+ for (i <- 0 until ct.getArity) {
+ val subtype = ct.getTypeAt(i)
+ validateStateType(subtype)
+ }
+ // check other type information only based on the type class
+ case _: TypeInformation[_] =>
+ validateStateType(t.getTypeClass)
+ }
+
+ /**
+ * Checks whether a class implements own hashCode() and equals() methods for storing an instance
+ * in Flink's state.
+ *
+ * @param c class to be validated
+ */
+ def validateStateType(c: Class[_]): Unit = {
+ // skip primitives
+ if (!c.isPrimitive) {
+ // check the component type of arrays
+ if (c.isArray) {
+ validateStateType(c.getComponentType)
+ }
+ // check type for methods
+ else {
+ if (c.getMethod("hashCode").getDeclaringClass eq classOf[Object]) {
+ throw new ValidationException(
+ s"Type '${c.getCanonicalName}' cannot be used in a stateful operation because it " +
+ s"does not implement a proper hashCode() method.")
+ }
+ if (c.getMethod("equals", classOf[Object]).getDeclaringClass eq classOf[Object]) {
+ throw new ValidationException(
+ s"Type '${c.getCanonicalName}' cannot be used in a stateful operation because it " +
+ s"does not implement a proper equals() method.")
+ }
+ }
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/f88da4d0/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/typeutils/TypeCheckUtilsTest.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/typeutils/TypeCheckUtilsTest.scala b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/typeutils/TypeCheckUtilsTest.scala
new file mode 100644
index 0000000..65a7dbd
--- /dev/null
+++ b/flink-libraries/flink-table/src/test/scala/org/apache/flink/table/typeutils/TypeCheckUtilsTest.scala
@@ -0,0 +1,54 @@
+/*
+ * 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.typeutils
+
+import org.apache.flink.api.scala._
+import org.apache.flink.api.scala.typeutils.{Types => ScalaTypes}
+import org.apache.flink.table.api.{Types, ValidationException}
+import org.apache.flink.table.typeutils.TypeCheckUtils.validateStateType
+import org.junit.Test
+
+class TypeCheckUtilsTest {
+
+ @Test
+ def testValidateStateType(): Unit = {
+ validateStateType(Types.STRING)
+ validateStateType(Types.LONG)
+ validateStateType(Types.SQL_TIMESTAMP)
+ validateStateType(Types.ROW(Types.LONG, Types.DECIMAL))
+ validateStateType(ScalaTypes.CASE_CLASS[(Long, Int)])
+ validateStateType(Types.OBJECT_ARRAY(Types.LONG))
+ validateStateType(Types.PRIMITIVE_ARRAY(Types.LONG))
+ }
+
+ @Test(expected = classOf[ValidationException])
+ def testInvalidType(): Unit = {
+ validateStateType(ScalaTypes.NOTHING)
+ }
+
+ @Test(expected = classOf[ValidationException])
+ def testInvalidType2(): Unit = {
+ validateStateType(Types.ROW(ScalaTypes.NOTHING))
+ }
+
+ @Test(expected = classOf[ValidationException])
+ def testInvalidType3(): Unit = {
+ validateStateType(Types.OBJECT_ARRAY[Nothing](ScalaTypes.NOTHING))
+ }
+}
[2/2] flink git commit: [FLINK-8139] [table] Check for proper
equals() and hashCode() for stateful operators
Posted by tw...@apache.org.
[FLINK-8139] [table] Check for proper equals() and hashCode() for stateful operators
This closes #5065.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/3c91de51
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/3c91de51
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/3c91de51
Branch: refs/heads/master
Commit: 3c91de518c4ce95d180400ea36a840be10fe6a86
Parents: 9dd3a85
Author: Aegeaner <xi...@gmail.com>
Authored: Fri Nov 24 16:11:28 2017 +0800
Committer: twalthr <tw...@apache.org>
Committed: Fri Jan 5 17:14:30 2018 +0100
----------------------------------------------------------------------
.../main/java/org/apache/flink/types/Row.java | 4 +--
.../flink/table/api/BatchTableEnvironment.scala | 1 +
.../table/api/StreamTableEnvironment.scala | 1 +
.../flink/table/api/TableEnvironment.scala | 29 +++++++++++++++++++-
.../table/runtime/utils/CommonTestData.scala | 4 +++
5 files changed, 36 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/3c91de51/flink-core/src/main/java/org/apache/flink/types/Row.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/types/Row.java b/flink-core/src/main/java/org/apache/flink/types/Row.java
index 4783314..b8bdbf9 100644
--- a/flink-core/src/main/java/org/apache/flink/types/Row.java
+++ b/flink-core/src/main/java/org/apache/flink/types/Row.java
@@ -106,12 +106,12 @@ public class Row implements Serializable{
Row row = (Row) o;
- return Arrays.equals(fields, row.fields);
+ return Arrays.deepEquals(fields, row.fields);
}
@Override
public int hashCode() {
- return Arrays.hashCode(fields);
+ return Arrays.deepHashCode(fields);
}
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/3c91de51/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/BatchTableEnvironment.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/BatchTableEnvironment.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/BatchTableEnvironment.scala
index c25007b..e0e00d4 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/BatchTableEnvironment.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/BatchTableEnvironment.scala
@@ -100,6 +100,7 @@ abstract class BatchTableEnvironment(
tableSource match {
case batchTableSource: BatchTableSource[_] =>
+ TableEnvironment.checkValidTableType(tableSource)
registerTableInternal(name, new BatchTableSourceTable(batchTableSource))
case _ =>
throw new TableException("Only BatchTableSource can be registered in " +
http://git-wip-us.apache.org/repos/asf/flink/blob/3c91de51/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/StreamTableEnvironment.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/StreamTableEnvironment.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/StreamTableEnvironment.scala
index 920da2e..84d5784 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/StreamTableEnvironment.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/StreamTableEnvironment.scala
@@ -118,6 +118,7 @@ abstract class StreamTableEnvironment(
s"A rowtime attribute requires an EventTime time characteristic in stream " +
s"environment. But is: ${execEnv.getStreamTimeCharacteristic}")
}
+ TableEnvironment.checkValidTableType(tableSource)
registerTableInternal(name, new StreamTableSourceTable(streamTableSource))
case _ =>
throw new TableException("Only StreamTableSource can be registered in " +
http://git-wip-us.apache.org/repos/asf/flink/blob/3c91de51/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala
----------------------------------------------------------------------
diff --git a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala
index c3cab13..8782477 100644
--- a/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala
+++ b/flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala
@@ -36,7 +36,7 @@ import org.apache.calcite.sql.parser.SqlParser
import org.apache.calcite.sql.util.ChainedSqlOperatorTable
import org.apache.calcite.tools._
import org.apache.flink.api.common.functions.MapFunction
-import org.apache.flink.api.common.typeinfo.{AtomicType, TypeInformation}
+import org.apache.flink.api.common.typeinfo.{AtomicType, TypeInformation, PrimitiveArrayTypeInfo}
import org.apache.flink.api.common.typeutils.CompositeType
import org.apache.flink.api.java.typeutils.{RowTypeInfo, _}
import org.apache.flink.api.java.{ExecutionEnvironment => JavaBatchExecEnv}
@@ -1148,6 +1148,33 @@ object TableEnvironment {
}
/**
+ * Checks if the chosen table type is valid.
+ * @param table The table to check
+ */
+ def checkValidTableType(table: Any): Unit = {
+ val types = table match {
+ case t: Table => table.asInstanceOf[Table].getSchema.getTypes
+ case st: TableSource[_] => table.asInstanceOf[TableSource[_]].getTableSchema.getTypes
+ }
+ for (typeInfo <- types) {
+ if(!typeInfo.asInstanceOf[TypeInformation[_]].isBasicType &&
+ !typeInfo.isInstanceOf[PrimitiveArrayTypeInfo[_]]) {
+ val typeClass = typeInfo.asInstanceOf[TypeInformation[_]].getTypeClass
+ if (typeClass.getMethod("hashCode").getDeclaringClass eq classOf[Object]) {
+ throw new ValidationException(s"Illegal Table type." +
+ s"Please make sure type ${typeClass.getCanonicalName}" +
+ s" implement own hashCode method")
+ }
+ if (typeClass.getMethod("equals", classOf[Any]).getDeclaringClass eq classOf[Object]) {
+ throw new ValidationException(s"Illegal Table type." +
+ s"Please make sure type ${typeClass.getCanonicalName}" +
+ s" implement own equals method")
+ }
+ }
+ }
+ }
+
+ /**
* Returns field indexes for a given [[TypeInformation]].
*
* @param inputType The TypeInformation extract the field positions from.
http://git-wip-us.apache.org/repos/asf/flink/blob/3c91de51/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 9223887..e8a5cba 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
@@ -181,5 +181,9 @@ object CommonTestData {
val x = new java.util.HashMap[String, String]()
override def toString: String = x.toString
+
+ override def hashCode(): Int = super.hashCode()
+
+ override def equals(obj: scala.Any): Boolean = super.equals(obj)
}
}