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:31 UTC
[flink] 01/02: [hotfix] [Scala API] Simplify generics in
EitherSerializer
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 = {