Update sbt, scripted-plugin to 1.11.0 (#257)

* Improve reliability of TypeInformation.createSerializer() and TypeSerializer.duplicate() (#254)

* Improve reliability of TypeInformation.createSerializer() and TypeSerializer.duplicate()

* Update magnolia to 1.3.18 (#256)

* Update sbt, scripted-plugin to 1.11.0

---------

Co-authored-by: Arnaud <[email protected]>

Author: scala-steward

build.sbt

@@ -29,7 +29,7 @@ lazy val commonSettings = Seq(
       )
     } else {
       Seq(
-        "com.softwaremill.magnolia1_3" %% "magnolia"        % "1.3.16",
+        "com.softwaremill.magnolia1_3" %% "magnolia"        % "1.3.18",
         "org.scala-lang"               %% "scala3-compiler" % scalaVersion.value % Provided
       )
     }

modules/flink-1-api/src/main/scala/org/apache/flinkx/api/serializer/ArraySerializer.scala

@@ -29,6 +29,15 @@ class ArraySerializer[T](val child: TypeSerializer[T], clazz: Class[T]) extends
     }
   }
 
+  override def duplicate(): ArraySerializer[T] = {
+    val duplicatedChild = child.duplicate()
+    if (duplicatedChild.eq(child)) {
+      this
+    } else {
+      new ArraySerializer[T](duplicatedChild, clazz)
+    }
+  }
+
   override def getLength: Int = -1
 
   override def deserialize(source: DataInputView): Array[T] = {

modules/flink-1-api/src/main/scala/org/apache/flinkx/api/serializer/CoproductSerializer.scala

@@ -8,7 +8,8 @@ import org.apache.flinkx.api.serializer.CoproductSerializer.CoproductSerializerS
 class CoproductSerializer[T](subtypeClasses: Array[Class[_]], subtypeSerializers: Array[TypeSerializer[_]])
     extends MutableSerializer[T] {
 
-  override val isImmutableType: Boolean = subtypeSerializers.forall(_.isImmutableType)
+  override val isImmutableType: Boolean = subtypeSerializers.forall(Option(_).exists(_.isImmutableType))
+  val isImmutableSerializer: Boolean    = subtypeSerializers.forall(Option(_).forall(s => s.duplicate().eq(s)))
 
   override def copy(from: T): T = {
     if (from == null || isImmutableType) {
@@ -19,6 +20,14 @@ class CoproductSerializer[T](subtypeClasses: Array[Class[_]], subtypeSerializers
     }
   }
 
+  override def duplicate(): CoproductSerializer[T] = {
+    if (isImmutableSerializer) {
+      this
+    } else {
+      new CoproductSerializer[T](subtypeClasses, subtypeSerializers.map(_.duplicate()))
+    }
+  }
+
   override def createInstance(): T =
     // this one may be used for later reuse, but we never reuse coproducts due to their unclear concrete type
     subtypeSerializers.head.createInstance().asInstanceOf[T]
@@ -74,7 +83,7 @@ object CoproductSerializer {
 
       subtypeSerializers = (0 until len).map { _ =>
         if (
-        /* - The old code was calling getCurrentVersion() just before calling readSnapshot().
+          /* - The old code was calling getCurrentVersion() just before calling readSnapshot().
              If only getCurrentVersion() is called, we know we must deserialize with old behavior.
            - The new code calls getCurrentVersion() only before calling writeSnapshot().
              getCurrentVersion() is not called before calling readSnapshot()

modules/flink-1-api/src/main/scala/org/apache/flinkx/api/serializer/ListCCSerializer.scala

@@ -16,6 +16,15 @@ class ListCCSerializer[T](child: TypeSerializer[T], clazz: Class[T]) extends Mut
     }
   }
 
+  override def duplicate(): ListCCSerializer[T] = {
+    val duplicatedChild = child.duplicate()
+    if (duplicatedChild.eq(child)) {
+      this
+    } else {
+      new ListCCSerializer[T](duplicatedChild, clazz)
+    }
+  }
+
   override def createInstance(): ::[T] = throw new IllegalArgumentException("cannot create instance of non-empty list")
   override def getLength: Int          = -1
   override def deserialize(source: DataInputView): ::[T] = {

modules/flink-1-api/src/main/scala/org/apache/flinkx/api/serializer/ListSerializer.scala

@@ -15,6 +15,15 @@ class ListSerializer[T](child: TypeSerializer[T], clazz: Class[T]) extends Mutab
     }
   }
 
+  override def duplicate(): ListSerializer[T] = {
+    val duplicatedChild = child.duplicate()
+    if (duplicatedChild.eq(child)) {
+      this
+    } else {
+      new ListSerializer[T](duplicatedChild, clazz)
+    }
+  }
+
   override def createInstance(): List[T] = List.empty[T]
   override def getLength: Int            = -1
   override def deserialize(source: DataInputView): List[T] = {

modules/flink-1-api/src/main/scala/org/apache/flinkx/api/serializer/MapSerializer.scala

@@ -17,6 +17,16 @@ class MapSerializer[K, V](ks: TypeSerializer[K], vs: TypeSerializer[V]) extends
     }
   }
 
+  override def duplicate(): MapSerializer[K, V] = {
+    val duplicatedKS = ks.duplicate()
+    val duplicatedVS = vs.duplicate()
+    if (duplicatedKS.eq(ks) && duplicatedVS.eq(vs)) {
+      this
+    } else {
+      new MapSerializer[K, V](duplicatedKS, duplicatedVS)
+    }
+  }
+
   override def createInstance(): Map[K, V] = Map.empty[K, V]
   override def getLength: Int              = -1
   override def deserialize(source: DataInputView): Map[K, V] = {
@@ -56,7 +66,7 @@ object MapSerializer {
 
     override def readSnapshot(readVersion: Int, in: DataInputView, userCodeClassLoader: ClassLoader): Unit = {
       if (
-      /* - The old code was calling getCurrentVersion() just before calling readSnapshot().
+        /* - The old code was calling getCurrentVersion() just before calling readSnapshot().
            If only getCurrentVersion() is called, we know we must deserialize with old behavior.
          - The new code calls getCurrentVersion() only before calling writeSnapshot().
            getCurrentVersion() is not called before calling readSnapshot()

modules/flink-1-api/src/main/scala/org/apache/flinkx/api/serializer/MappedSerializer.scala

@@ -17,12 +17,27 @@ case class MappedSerializer[A, B](mapper: TypeMapper[A, B], ser: TypeSerializer[
     }
   }
 
-  override def equals(obj: Any): Boolean = ser.equals(obj)
+  override def duplicate(): MappedSerializer[A, B] = {
+    val duplicatedSer = ser.duplicate()
+    if (duplicatedSer.eq(ser)) {
+      this
+    } else {
+      MappedSerializer[A, B](mapper, duplicatedSer)
+    }
+  }
+
+  override def equals(other: Any): Boolean = other match {
+    case that: MappedSerializer[_, _] =>
+      mapper == that.mapper &&
+      ser == that.ser
+    case _ => false
+  }
+
+  override def toString = s"MappedSerializer($mapper, $ser)"
 
-  override def toString: String = ser.toString
+  override def hashCode(): Int = 31 * mapper.hashCode + ser.hashCode
 
-  override def hashCode(): Int = ser.hashCode()
-  override def getLength: Int  = ser.getLength
+  override def getLength: Int = ser.getLength
 
   override def serialize(record: A, target: DataOutputView): Unit = {
     ser.serialize(mapper.map(record), target)
@@ -60,15 +75,15 @@ object MappedSerializer {
       val mapperClazz = InstantiationUtil.resolveClassByName[TypeMapper[A, B]](in, userCodeClassLoader)
       mapper = InstantiationUtil.instantiate(mapperClazz)
       if (
-      /* - The old code was calling getCurrentVersion() just before calling readSnapshot().
+        /* - The old code was calling getCurrentVersion() just before calling readSnapshot().
            If only getCurrentVersion() is called, we know we must deserialize with old behavior.
          - The new code calls getCurrentVersion() only before calling writeSnapshot().
            getCurrentVersion() is not called before calling readSnapshot()
            or both getCurrentVersion() and writeSnapshot() are called,
            so in these cases we know the readVersion parameter is trustable to determine which behavior to apply. */
         (!currentVersionCalled || writeSnapshotCalled) &&
-          // readVersion is trustable
-          readVersion == 2
+        // readVersion is trustable
+        readVersion == 2
       ) {
         ser = TypeSerializerSnapshot.readVersionedSnapshot[B](in, userCodeClassLoader).restoreSerializer()
       } else {

modules/flink-1-api/src/main/scala/org/apache/flinkx/api/serializer/SeqSerializer.scala

@@ -15,6 +15,15 @@ class SeqSerializer[T](child: TypeSerializer[T], clazz: Class[T]) extends Mutabl
     }
   }
 
+  override def duplicate(): SeqSerializer[T] = {
+    val duplicatedChild = child.duplicate()
+    if (duplicatedChild.eq(child)) {
+      this
+    } else {
+      new SeqSerializer[T](duplicatedChild, clazz)
+    }
+  }
+
   override def createInstance(): Seq[T] = Seq.empty[T]
   override def getLength: Int           = -1
   override def deserialize(source: DataInputView): Seq[T] = {

modules/flink-1-api/src/main/scala/org/apache/flinkx/api/serializer/SetSerializer.scala

@@ -15,6 +15,15 @@ class SetSerializer[T](child: TypeSerializer[T], clazz: Class[T]) extends Mutabl
     }
   }
 
+  override def duplicate(): SetSerializer[T] = {
+    val duplicatedChild = child.duplicate()
+    if (duplicatedChild.eq(child)) {
+      this
+    } else {
+      new SetSerializer[T](duplicatedChild, clazz)
+    }
+  }
+
   override def createInstance(): Set[T] = Set.empty[T]
   override def getLength: Int           = -1
   override def deserialize(source: DataInputView): Set[T] = {

modules/flink-1-api/src/main/scala/org/apache/flinkx/api/serializer/VectorSerializer.scala

@@ -15,6 +15,15 @@ class VectorSerializer[T](child: TypeSerializer[T], clazz: Class[T]) extends Mut
     }
   }
 
+  override def duplicate(): VectorSerializer[T] = {
+    val duplicatedChild = child.duplicate()
+    if (duplicatedChild.eq(child)) {
+      this
+    } else {
+      new VectorSerializer[T](duplicatedChild, clazz)
+    }
+  }
+
   override def createInstance(): Vector[T] = Vector.empty[T]
   override def getLength: Int              = -1
   override def deserialize(source: DataInputView): Vector[T] = {