You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2018/10/31 15:16:30 UTC
[flink] branch release-1.6 updated (1f90649 -> 5aa07cd)
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a change to branch release-1.6
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 1f90649 [FLINK-10690][tests] Fix Files.list resource leaks
new 04fb292 [hotfix] [Scala API] Simplify generics in EitherSerializer
new 5aa07cd [FLINK-10693] [Scala API] Fix incorrect duplication in EitherSerializer
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../api/scala/typeutils/EitherSerializer.scala | 50 +++++++-----
.../flink/api/scala/typeutils/EitherTypeInfo.scala | 10 +--
.../api/scala/typeutils/EitherSerializerTest.scala | 90 ++++++++++++++++++++++
3 files changed, 124 insertions(+), 26 deletions(-)
create mode 100644 flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/EitherSerializerTest.scala
[flink] 02/02: [FLINK-10693] [Scala API] Fix incorrect duplication
in EitherSerializer
Posted by tr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch release-1.6
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 5aa07cd277bad087bbcf901afaac6133c87b6812
Author: Stephan Ewen <se...@apache.org>
AuthorDate: Fri Oct 26 15:35:14 2018 +0200
[FLINK-10693] [Scala API] Fix incorrect duplication in EitherSerializer
This closes #6951.
---
.../api/scala/typeutils/EitherSerializer.scala | 11 ++-
.../api/scala/typeutils/EitherSerializerTest.scala | 90 ++++++++++++++++++++++
2 files changed, 100 insertions(+), 1 deletion(-)
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EitherSerializer.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EitherSerializer.scala
index 2974aaf..fbcddc2 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EitherSerializer.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EitherSerializer.scala
@@ -32,7 +32,16 @@ class EitherSerializer[A, B](
val rightSerializer: TypeSerializer[B])
extends TypeSerializer[Either[A, B]] {
- override def duplicate: EitherSerializer[A,B] = this
+ override def duplicate: EitherSerializer[A,B] = {
+ val leftDup = leftSerializer.duplicate()
+ val rightDup = rightSerializer.duplicate()
+
+ if (leftDup.eq(leftSerializer) && rightDup.eq(rightSerializer)) {
+ this
+ } else {
+ new EitherSerializer[A, B](leftDup, rightDup)
+ }
+ }
override def createInstance: Either[A, B] = {
Left(null).asInstanceOf[Left[A, B]]
diff --git a/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/EitherSerializerTest.scala b/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/EitherSerializerTest.scala
new file mode 100644
index 0000000..f62d9a9
--- /dev/null
+++ b/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/EitherSerializerTest.scala
@@ -0,0 +1,90 @@
+/*
+ * 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.api.scala.typeutils
+
+import org.apache.flink.api.common.ExecutionConfig
+import org.apache.flink.api.common.typeutils.SerializerTestBase
+import org.apache.flink.api.common.typeutils.base.{IntSerializer, StringSerializer}
+import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer
+
+import org.junit.Test
+import org.junit.Assert.assertNotSame
+import org.junit.Assert.assertSame
+
+/**
+ * Test suite for the [[EitherSerializer]]
+ */
+class EitherSerializerTest extends SerializerTestBase[Either[String, Integer]] {
+
+ // --------------------------------------------------------------------------
+ // test suite
+ // --------------------------------------------------------------------------
+
+ override protected def createSerializer() =
+ new EitherSerializer[String, Integer](
+ StringSerializer.INSTANCE,
+ IntSerializer.INSTANCE)
+
+ override protected def getLength: Int = -1
+
+ override protected def getTypeClass: Class[Either[String, Integer]] =
+ classOf[Either[String, Integer]]
+
+ override protected def getTestData: Array[Either[String, Integer]] =
+ Array[Either[String, Integer]](
+ Left("hello"),
+ Right(17),
+ Right(0),
+ Left("friend"),
+ Right(200),
+ Right(100),
+ Left("foo"),
+ Right(1060876234),
+ Left("bar")
+ )
+
+ // --------------------------------------------------------------------------
+ // either serializer specific tests
+ // --------------------------------------------------------------------------
+
+ @Test
+ def testDuplication(): Unit = {
+ val serializerSS: EitherSerializer[String, String] =
+ new EitherSerializer[String, String](
+ StringSerializer.INSTANCE,
+ StringSerializer.INSTANCE
+ )
+
+ val serializerSO: EitherSerializer[String, Object] =
+ new EitherSerializer[String, Object](
+ StringSerializer.INSTANCE,
+ new KryoSerializer[Object](classOf[Object], new ExecutionConfig())
+ )
+
+ val serializerOS: EitherSerializer[Object, String] =
+ new EitherSerializer[Object, String](
+ new KryoSerializer[Object](classOf[Object], new ExecutionConfig()),
+ StringSerializer.INSTANCE
+ )
+
+ assertSame(serializerSS, serializerSS.duplicate())
+ assertNotSame(serializerSO, serializerSO.duplicate())
+ assertNotSame(serializerOS, serializerOS.duplicate())
+ }
+}
[flink] 01/02: [hotfix] [Scala API] Simplify generics in
EitherSerializer
Posted by tr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch release-1.6
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 04fb292e60c6b52179391a4fa7ce02083cadcbf8
Author: Stephan Ewen <se...@apache.org>
AuthorDate: Fri Oct 26 15:32:43 2018 +0200
[hotfix] [Scala API] Simplify generics in EitherSerializer
---
.../api/scala/typeutils/EitherSerializer.scala | 41 +++++++++++-----------
.../flink/api/scala/typeutils/EitherTypeInfo.scala | 10 +++---
2 files changed, 25 insertions(+), 26 deletions(-)
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EitherSerializer.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EitherSerializer.scala
index 439e0c2..2974aaf 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EitherSerializer.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EitherSerializer.scala
@@ -27,15 +27,15 @@ import org.apache.flink.core.memory.{DataInputView, DataOutputView}
*/
@Internal
@SerialVersionUID(9219995873023657525L)
-class EitherSerializer[A, B, T <: Either[A, B]](
+class EitherSerializer[A, B](
val leftSerializer: TypeSerializer[A],
val rightSerializer: TypeSerializer[B])
- extends TypeSerializer[T] {
+ extends TypeSerializer[Either[A, B]] {
- override def duplicate: EitherSerializer[A,B,T] = this
+ override def duplicate: EitherSerializer[A,B] = this
- override def createInstance: T = {
- Left(null).asInstanceOf[T]
+ override def createInstance: Either[A, B] = {
+ Left(null).asInstanceOf[Left[A, B]]
}
override def isImmutableType: Boolean = {
@@ -45,12 +45,12 @@ class EitherSerializer[A, B, T <: Either[A, B]](
override def getLength: Int = -1
- override def copy(from: T): T = from match {
- case Left(a) => Left(leftSerializer.copy(a)).asInstanceOf[T]
- case Right(b) => Right(rightSerializer.copy(b)).asInstanceOf[T]
+ override def copy(from: Either[A, B]): Either[A, B] = from match {
+ case Left(a) => Left(leftSerializer.copy(a))
+ case Right(b) => Right(rightSerializer.copy(b))
}
- override def copy(from: T, reuse: T): T = copy(from)
+ override def copy(from: Either[A, B], reuse: Either[A, B]): Either[A, B] = copy(from)
override def copy(source: DataInputView, target: DataOutputView): Unit = {
val isLeft = source.readBoolean()
@@ -62,7 +62,7 @@ class EitherSerializer[A, B, T <: Either[A, B]](
}
}
- override def serialize(either: T, target: DataOutputView): Unit = either match {
+ override def serialize(either: Either[A, B], target: DataOutputView): Unit = either match {
case Left(a) =>
target.writeBoolean(true)
leftSerializer.serialize(a, target)
@@ -71,27 +71,27 @@ class EitherSerializer[A, B, T <: Either[A, B]](
rightSerializer.serialize(b, target)
}
- override def deserialize(source: DataInputView): T = {
+ override def deserialize(source: DataInputView): Either[A, B] = {
val isLeft = source.readBoolean()
if (isLeft) {
- Left(leftSerializer.deserialize(source)).asInstanceOf[T]
+ Left(leftSerializer.deserialize(source))
} else {
- Right(rightSerializer.deserialize(source)).asInstanceOf[T]
+ Right(rightSerializer.deserialize(source))
}
}
- override def deserialize(reuse: T, source: DataInputView): T = {
+ override def deserialize(reuse: Either[A, B], source: DataInputView): Either[A, B] = {
val isLeft = source.readBoolean()
if (isLeft) {
- Left(leftSerializer.deserialize(source)).asInstanceOf[T]
+ Left(leftSerializer.deserialize(source))
} else {
- Right(rightSerializer.deserialize(source)).asInstanceOf[T]
+ Right(rightSerializer.deserialize(source))
}
}
override def equals(obj: Any): Boolean = {
obj match {
- case eitherSerializer: EitherSerializer[_, _, _] =>
+ case eitherSerializer: EitherSerializer[_, _] =>
eitherSerializer.canEqual(this) &&
leftSerializer.equals(eitherSerializer.leftSerializer) &&
rightSerializer.equals(eitherSerializer.rightSerializer)
@@ -100,7 +100,7 @@ class EitherSerializer[A, B, T <: Either[A, B]](
}
override def canEqual(obj: Any): Boolean = {
- obj.isInstanceOf[EitherSerializer[_, _, _]]
+ obj.isInstanceOf[EitherSerializer[_, _]]
}
override def hashCode(): Int = {
@@ -116,7 +116,7 @@ class EitherSerializer[A, B, T <: Either[A, B]](
}
override def ensureCompatibility(
- configSnapshot: TypeSerializerConfigSnapshot): CompatibilityResult[T] = {
+ configSnapshot: TypeSerializerConfigSnapshot): CompatibilityResult[Either[A, B]] = {
configSnapshot match {
case eitherSerializerConfig: EitherSerializerConfigSnapshot[A, B] =>
@@ -142,7 +142,7 @@ class EitherSerializer[A, B, T <: Either[A, B]](
&& rightCompatResult.getConvertDeserializer != null) {
CompatibilityResult.requiresMigration(
- new EitherSerializer[A, B, T](
+ new EitherSerializer[A, B](
new TypeDeserializerAdapter(leftCompatResult.getConvertDeserializer),
new TypeDeserializerAdapter(rightCompatResult.getConvertDeserializer)
)
@@ -154,7 +154,6 @@ class EitherSerializer[A, B, T <: Either[A, B]](
} else {
CompatibilityResult.compatible()
}
-
case _ => CompatibilityResult.requiresMigration()
}
}
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EitherTypeInfo.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EitherTypeInfo.scala
index e897309..b2a055e 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EitherTypeInfo.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EitherTypeInfo.scala
@@ -52,18 +52,18 @@ class EitherTypeInfo[A, B, T <: Either[A, B]](
@PublicEvolving
def createSerializer(executionConfig: ExecutionConfig): TypeSerializer[T] = {
- val leftSerializer = if (leftTypeInfo != null) {
+ val leftSerializer: TypeSerializer[A] = if (leftTypeInfo != null) {
leftTypeInfo.createSerializer(executionConfig)
} else {
- new NothingSerializer
+ (new NothingSerializer).asInstanceOf[TypeSerializer[A]]
}
- val rightSerializer = if (rightTypeInfo != null) {
+ val rightSerializer: TypeSerializer[B] = if (rightTypeInfo != null) {
rightTypeInfo.createSerializer(executionConfig)
} else {
- new NothingSerializer
+ (new NothingSerializer).asInstanceOf[TypeSerializer[B]]
}
- new EitherSerializer(leftSerializer, rightSerializer)
+ new EitherSerializer[A, B](leftSerializer, rightSerializer).asInstanceOf[TypeSerializer[T]]
}
override def equals(obj: Any): Boolean = {