Call duplicate when type is not immutable (#114)

call duplicate when serializer is not immutable for some of the type infos

---------

Co-authored-by: Suno <suno@hpcnt.com>

Author: novakov-alexey

src/main/scala/org/apache/flinkx/api/serializer/CaseClassSerializer.scala

@@ -39,9 +39,11 @@ abstract class CaseClassSerializer[T <: Product](
 
   @transient lazy val log: Logger = LoggerFactory.getLogger(this.getClass)
 
-  override def duplicate: CaseClassSerializer[T] = {
+  override def isImmutableType: Boolean =
+    scalaFieldSerializers.forall(_.isImmutableType)
+
+  override def duplicate: CaseClassSerializer[T] =
     clone().asInstanceOf[CaseClassSerializer[T]]
-  }
 
   @throws[CloneNotSupportedException]
   override protected def clone(): Object = {

src/main/scala/org/apache/flinkx/api/typeinfo/CollectionTypeInformation.scala

@@ -8,11 +8,13 @@ import scala.reflect.{ClassTag, classTag}
 
 case class CollectionTypeInformation[T: ClassTag](serializer: TypeSerializer[T]) extends TypeInformation[T] {
   val clazz = classTag[T].runtimeClass.asInstanceOf[Class[T]]
-  override def createSerializer(config: ExecutionConfig): TypeSerializer[T] = serializer
-  override def isBasicType: Boolean                                         = false
-  override def isTupleType: Boolean                                         = false
-  override def isKeyType: Boolean                                           = false
-  override def getTotalFields: Int                                          = 1
-  override def getTypeClass: Class[T]                                       = clazz
-  override def getArity: Int                                                = 1
+  override def createSerializer(config: ExecutionConfig): TypeSerializer[T] =
+    if (serializer.isImmutableType) serializer
+    else serializer.duplicate()
+  override def isBasicType: Boolean   = false
+  override def isTupleType: Boolean   = false
+  override def isKeyType: Boolean     = false
+  override def getTotalFields: Int    = 1
+  override def getTypeClass: Class[T] = clazz
+  override def getArity: Int          = 1
 }

src/main/scala/org/apache/flinkx/api/typeinfo/CoproductTypeInformation.scala

@@ -5,11 +5,13 @@ import org.apache.flink.api.common.typeinfo.TypeInformation
 import org.apache.flink.api.common.typeutils.TypeSerializer
 
 case class CoproductTypeInformation[T](c: Class[T], ser: TypeSerializer[T]) extends TypeInformation[T] {
-  override def createSerializer(config: ExecutionConfig): TypeSerializer[T] = ser
-  override def isBasicType: Boolean                                         = false
-  override def isTupleType: Boolean                                         = false
-  override def isKeyType: Boolean                                           = false
-  override def getTotalFields: Int                                          = 1
-  override def getTypeClass: Class[T]                                       = c
-  override def getArity: Int                                                = 1
+  override def createSerializer(config: ExecutionConfig): TypeSerializer[T] =
+    if (ser.isImmutableType) ser
+    else ser.duplicate()
+  override def isBasicType: Boolean   = false
+  override def isTupleType: Boolean   = false
+  override def isKeyType: Boolean     = false
+  override def getTotalFields: Int    = 1
+  override def getTypeClass: Class[T] = c
+  override def getArity: Int          = 1
 }

src/main/scala/org/apache/flinkx/api/typeinfo/ProductTypeInformation.scala

@@ -15,5 +15,7 @@ class ProductTypeInformation[T <: Product](
       fieldTypes,
       fieldNames
     ) {
-  override def createSerializer(config: ExecutionConfig): TypeSerializer[T] = ser
+  override def createSerializer(config: ExecutionConfig): TypeSerializer[T] =
+    if (ser.isImmutableType) ser
+    else ser.duplicate()
 }

src/main/scala/org/apache/flinkx/api/typeinfo/SimpleTypeInformation.scala

@@ -7,11 +7,15 @@ import org.apache.flink.api.common.typeutils.TypeSerializer
 import scala.reflect.{classTag, ClassTag}
 
 abstract class SimpleTypeInformation[T: ClassTag: TypeSerializer] extends TypeInformation[T] {
-  override def createSerializer(config: ExecutionConfig): TypeSerializer[T] = implicitly[TypeSerializer[T]]
-  override def isBasicType: Boolean                                         = false
-  override def isTupleType: Boolean                                         = false
-  override def isKeyType: Boolean                                           = false
-  override def getTotalFields: Int                                          = 1
+  override def createSerializer(config: ExecutionConfig): TypeSerializer[T] = {
+    val ser = implicitly[TypeSerializer[T]]
+    if (ser.isImmutableType) ser
+    else ser.duplicate()
+  }
+  override def isBasicType: Boolean   = false
+  override def isTupleType: Boolean   = false
+  override def isKeyType: Boolean     = false
+  override def getTotalFields: Int    = 1
   override def getTypeClass: Class[T] = classTag[T].runtimeClass.asInstanceOf[Class[T]]
   override def getArity: Int          = 1
 }