[flink] 02/06: [FLINK-11334] Don't use IntSerializer in EnumValueSerializer

[al...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

aljoscha pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git

commit 6d9b45b94ef3f12ecb9be38fe8c7662eeb7ea72c
Author: Aljoscha Krettek <al...@gmail.com>
AuthorDate: Tue Feb 19 15:59:13 2019 +0100

    [FLINK-11334] Don't use IntSerializer in EnumValueSerializer
    
    Using an IntSerialize technically makes EnumValueSerializer a composite
    serializer, which it doesn't have to be. This decouples
    EnumValueSerializer from any future changes to IntSerializer.
---
 .../flink/api/scala/typeutils/EnumValueSerializer.scala     | 13 ++++++-------
 1 file changed, 6 insertions(+), 7 deletions(-)

diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializer.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializer.scala
index 84ce017..6733099 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializer.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/EnumValueSerializer.scala
@@ -21,7 +21,6 @@ import java.io.IOException
 
 import org.apache.flink.annotation.Internal
 import org.apache.flink.api.common.typeutils.{TypeSerializer, TypeSerializerConfigSnapshot, TypeSerializerSchemaCompatibility}
-import org.apache.flink.api.common.typeutils.base.IntSerializer
 import org.apache.flink.api.java.typeutils.runtime.{DataInputViewStream, DataOutputViewStream}
 import org.apache.flink.core.memory.{DataInputView, DataOutputView}
 import org.apache.flink.util.{InstantiationUtil, Preconditions}
@@ -37,25 +36,25 @@ class EnumValueSerializer[E <: Enumeration](val enum: E) extends TypeSerializer[
 
   type T = E#Value
 
-  val intSerializer = new IntSerializer()
-
   override def duplicate: EnumValueSerializer[E] = this
 
   override def createInstance: T = enum(0)
 
   override def isImmutableType: Boolean = true
 
-  override def getLength: Int = intSerializer.getLength
+  override def getLength: Int = 4
 
   override def copy(from: T): T = enum.apply(from.id)
 
   override def copy(from: T, reuse: T): T = copy(from)
 
-  override def copy(src: DataInputView, tgt: DataOutputView): Unit = intSerializer.copy(src, tgt)
+  override def copy(src: DataInputView, tgt: DataOutputView): Unit = {
+    tgt.writeInt(src.readInt())
+  }
 
-  override def serialize(v: T, tgt: DataOutputView): Unit = intSerializer.serialize(v.id, tgt)
+  override def serialize(v: T, tgt: DataOutputView): Unit = tgt.writeInt(v.id)
 
-  override def deserialize(source: DataInputView): T = enum(intSerializer.deserialize(source))
+  override def deserialize(source: DataInputView): T = enum(source.readInt())
 
   override def deserialize(reuse: T, source: DataInputView): T = deserialize(source)