feat(#286): add serializers for DurationSerializer, FiniteDurationSerializer, Ordering and SortedSet

Author: arnaud-daroussin

README.md

@@ -352,6 +352,50 @@ implicit val mapper2: TypeMapper[WrappedString, String] = new TypeMapper[Wrapped
 }  
 ```
 
+### Ordering
+
+`SortedSet` requires a type-information for its elements and also for the ordering of the elements. Type-information of default orderings are not implicitly available in the context because we cannot make the assumption the user wants to use the natural ordering or a custom one.
+
+Type-information of default ordering are available in `org.apache.flinkx.api.serializer.OrderingTypeInfo` and can be used as follows:
+```scala mdoc:reset-object
+import org.apache.flinkx.api._
+import org.apache.flinkx.api.serializers._
+
+case class Foo(bars: SortedSet[String])
+
+object Foo {
+  implicit val fooInfo: TypeInformation[Foo] = {
+    // type-information for Ordering need to be explicitly put in the context
+    implicit val orderingStringInfo: TypeInformation[Ordering[String]] =
+      OrderingTypeInfo.DefaultStringOrderingInfo
+    deriveTypeInformation
+  }
+}
+```
+
+It's also possible to derive the type-information of a custom ordering if it's an ADT:
+```scala mdoc:reset-object
+import org.apache.flinkx.api._
+import org.apache.flinkx.api.serializers._
+
+case class Bar(a: Int, b: String)
+
+case object BarOrdering extends Ordering[Bar] {
+  override def compare(x: Bar, y: Bar): Int = x.a.compare(y.a)
+}
+
+case class Foo(bar: SortedSet[Bar])
+
+object Foo {
+  implicit val fooInfo: TypeInformation[Foo] = {
+    // Derive the type-information of custom Bar ordering
+    implicit val barOrderingInfo: TypeInformation[Ordering[Bar]] =
+      OrderingTypeInfo.deriveOrdering[BarOrdering.type, Bar]
+    deriveTypeInformation
+  }
+}
+```
+
 ### Schema evolution
 
 #### ADT

modules/flink-common-api/src/main/scala/org/apache/flinkx/api/serializer/CaseClassSerializer.scala

@@ -22,6 +22,7 @@ import org.apache.flink.api.common.typeutils.{TypeSerializer, TypeSerializerSnap
 import org.apache.flink.api.java.typeutils.runtime.TupleSerializerBase
 import org.apache.flink.core.memory.{DataInputView, DataOutputView}
 import org.apache.flink.types.NullFieldException
+import org.apache.flinkx.api.{NullMarker, VariableLengthDataType}
 import org.apache.flinkx.api.serializer.CaseClassSerializer.EmptyByteArray
 import org.slf4j.{Logger, LoggerFactory}
 
@@ -103,11 +104,13 @@ class CaseClassSerializer[T <: Product](
       createInstance(fields)
     }
 
-  override val getLength: Int = if (super.getLength == -1) -1 else super.getLength + 4 // +4 bytes for the arity field
+  override val getLength: Int =
+    if (super.getLength == VariableLengthDataType) VariableLengthDataType
+    else super.getLength + 4 // +4 bytes for the arity field
 
   def serialize(value: T, target: DataOutputView): Unit = {
-    // Write an arity of -1 to indicate null value
-    val sourceArity = if (value == null) -1 else arity
+    // Write a negative arity to indicate null value
+    val sourceArity = if (value == null) NullMarker else arity
     target.writeInt(sourceArity)
     if (value == null) target.write(nullPadding)
 
@@ -129,7 +132,7 @@ class CaseClassSerializer[T <: Product](
 
   def deserialize(source: DataInputView): T = {
     val sourceArity = source.readInt()
-    if (sourceArity == -1) {
+    if (sourceArity < 0) {
       source.skipBytesToRead(nullPadding.length)
       null.asInstanceOf[T]
     } else {
@@ -146,7 +149,7 @@ class CaseClassSerializer[T <: Product](
   override def copy(source: DataInputView, target: DataOutputView): Unit = {
     val sourceArity = source.readInt()
     target.writeInt(sourceArity)
-    if (sourceArity == -1) {
+    if (sourceArity < 0) {
       source.skipBytesToRead(nullPadding.length)
       target.skipBytesToWrite(nullPadding.length)
     } else {

modules/flink-common-api/src/main/scala/org/apache/flinkx/api/serializer/CollectionSerializerSnapshot.scala

@@ -69,8 +69,9 @@ class CollectionSerializerSnapshot[F[_], T, S <: TypeSerializer[F[T]]](
       case "long"    => out.writeUTF("java.lang.Long")
       case "byte"    => out.writeUTF("java.lang.Byte")
       case "short"   => out.writeUTF("java.lang.Short")
-      case "char"    => out.writeUTF("java.lang.Char")
+      case "char"    => out.writeUTF("java.lang.Character")
       case "boolean" => out.writeUTF("java.lang.Boolean")
+      case "void"    => out.writeUTF("java.lang.Void")
       case other     => out.writeUTF(other)
     }
     TypeSerializerSnapshot.writeVersionedSnapshot(out, nestedSerializer.snapshotConfiguration())

modules/flink-common-api/src/main/scala/org/apache/flinkx/api/serializer/SortedCollectionSerializerSnapshot.scala

@@ -0,0 +1,50 @@
+package org.apache.flinkx.api.serializer
+
+import org.apache.flink.api.common.typeutils.{TypeSerializer, TypeSerializerSnapshot}
+import org.apache.flink.core.memory.{DataInputView, DataOutputView}
+
+/** Generic serializer snapshot for sorted collection.
+  * @param aSerializer
+  *   the serializer of `A`
+  * @param aOrderingSerializer
+  *   the serializer of `Ordering[A]`
+  * @param sClass
+  *   the class of `S`
+  * @param aClass
+  *   the class of `A`
+  * @tparam F
+  *   the type of the serialized collection
+  * @tparam A
+  *   the type of the collection's elements
+  * @tparam S
+  *   the type of the collection serializer
+  */
+class SortedCollectionSerializerSnapshot[F[_], A, S <: TypeSerializer[F[A]]](
+    aSerializer: TypeSerializer[A],
+    sClass: Class[S],
+    aClass: Class[A],
+    private var aOrderingSerializer: TypeSerializer[Ordering[A]]
+) extends CollectionSerializerSnapshot[F, A, S](aSerializer, sClass, aClass) {
+
+  // Empty constructor is required to instantiate this class during deserialization.
+  def this() = this(null, null, null, null)
+
+  override def getCurrentVersion: Int = super.getCurrentVersion // Must be aligned with CollectionSerializerSnapshot
+
+  override def writeSnapshot(out: DataOutputView): Unit = {
+    super.writeSnapshot(out)
+    TypeSerializerSnapshot.writeVersionedSnapshot(out, aOrderingSerializer.snapshotConfiguration())
+  }
+
+  override def readSnapshot(readVersion: Int, in: DataInputView, userCodeClassLoader: ClassLoader): Unit = {
+    super.readSnapshot(readVersion, in, userCodeClassLoader)
+    aOrderingSerializer =
+      TypeSerializerSnapshot.readVersionedSnapshot[Ordering[A]](in, userCodeClassLoader).restoreSerializer()
+  }
+
+  override def restoreSerializer(): TypeSerializer[F[A]] = {
+    val constructor = clazz.getConstructors()(0)
+    constructor.newInstance(nestedSerializer, vclazz, aOrderingSerializer).asInstanceOf[TypeSerializer[F[A]]]
+  }
+
+}

modules/flink-common-api/src/main/scala/org/apache/flinkx/api/serializer/SortedSetSerializer.scala

@@ -0,0 +1,100 @@
+package org.apache.flinkx.api.serializer
+
+import org.apache.flink.api.common.typeutils.{TypeSerializer, TypeSerializerSnapshot}
+import org.apache.flink.core.memory.{DataInputView, DataOutputView}
+import org.apache.flinkx.api.{NullMarker, VariableLengthDataType}
+
+import java.util.Objects
+import scala.collection.immutable.SortedSet
+import scala.collection.mutable
+
+/** Serializer for [[SortedSet]]. Handle nullable value. */
+class SortedSetSerializer[A](
+    val aSerializer: TypeSerializer[A],
+    val aClass: Class[A],
+    val aOrderingSerializer: TypeSerializer[Ordering[A]]
+) extends MutableSerializer[SortedSet[A]] { // SortedSet is immutable, but its elements can be mutable
+
+  override val isImmutableType: Boolean = aSerializer.isImmutableType && aOrderingSerializer.isImmutableType
+
+  override def duplicate(): SortedSetSerializer[A] = {
+    val duplicatedASerializer         = aSerializer.duplicate()
+    val duplicatedAOrderingSerializer = aOrderingSerializer.duplicate()
+    if (duplicatedASerializer.eq(aSerializer) && duplicatedAOrderingSerializer.eq(aOrderingSerializer)) {
+      this
+    } else {
+      new SortedSetSerializer[A](duplicatedASerializer, aClass, duplicatedAOrderingSerializer)
+    }
+  }
+
+  override def createInstance(): SortedSet[A] = SortedSet.empty[A](aOrderingSerializer.createInstance())
+
+  override def copy(from: SortedSet[A]): SortedSet[A] =
+    if (from == null || isImmutableType) {
+      from
+    } else {
+      implicit val ordering: Ordering[A] =
+        if (aOrderingSerializer.isImmutableType) from.ordering else aOrderingSerializer.copy(from.ordering)
+      if (aSerializer.isImmutableType) {
+        SortedSet.from(from)
+      } else {
+        from.map(aSerializer.copy)
+      }
+    }
+
+  override def getLength: Int = VariableLengthDataType
+
+  override def serialize(records: SortedSet[A], target: DataOutputView): Unit =
+    if (records == null) {
+      target.writeInt(NullMarker)
+    } else {
+      target.writeInt(records.size)
+      records.foreach(element => aSerializer.serialize(element, target))
+      aOrderingSerializer.serialize(records.ordering, target)
+    }
+
+  override def deserialize(source: DataInputView): SortedSet[A] = {
+    var remaining = source.readInt()
+    if (remaining == NullMarker) {
+      null
+    } else {
+      val buffer = mutable.Buffer[A]()
+      while (remaining > 0) {
+        buffer.append(aSerializer.deserialize(source))
+        remaining -= 1
+      }
+      implicit val ordering: Ordering[A] = aOrderingSerializer.deserialize(source)
+      buffer.to(SortedSet)
+    }
+  }
+
+  override def copy(source: DataInputView, target: DataOutputView): Unit = {
+    var remaining = source.readInt()
+    target.writeInt(remaining)
+    if (remaining >= 0) {
+      while (remaining > 0) {
+        aSerializer.copy(source, target)
+        remaining -= 1
+      }
+      aOrderingSerializer.copy(source, target)
+    }
+  }
+
+  override def snapshotConfiguration(): TypeSerializerSnapshot[SortedSet[A]] =
+    new SortedCollectionSerializerSnapshot[SortedSet, A, SortedSetSerializer[A]](
+      aSerializer,
+      classOf[SortedSetSerializer[A]],
+      aClass,
+      aOrderingSerializer
+    )
+
+  override def hashCode(): Int = Objects.hash(aSerializer, aClass, aOrderingSerializer)
+
+  override def equals(obj: Any): Boolean =
+    obj match {
+      case other: SortedSetSerializer[_] =>
+        aSerializer == other.aSerializer && aClass == other.aClass && aOrderingSerializer == other.aOrderingSerializer
+      case _ => false
+    }
+
+}

modules/flink-common-api/src/main/scala/org/apache/flinkx/api/serializer/durationSerializer.scala

@@ -0,0 +1,92 @@
+package org.apache.flinkx.api.serializer
+
+import org.apache.flink.api.common.typeutils.{SimpleTypeSerializerSnapshot, TypeSerializer, TypeSerializerSnapshot}
+import org.apache.flink.core.memory.{DataInputView, DataOutputView}
+import org.apache.flinkx.api.{VariableLengthDataType, serializers}
+
+import java.lang.Long.{BYTES => LongBytes}
+import scala.concurrent.duration.{Duration, FiniteDuration, TimeUnit}
+
+/** Serializer for [[Duration]]. Handle nullable value. */
+object DurationSerializer extends ImmutableSerializer[Duration] {
+
+  val NullDiscriminant                     = -1
+  val FiniteDurationDiscriminant           = 0
+  val UndefinedDurationDiscriminant        = 1
+  val PositiveInfiniteDurationDiscriminant = 2
+  val NegativeInfiniteDurationDiscriminant = 3
+
+  override def createInstance: Duration = Duration.Zero
+
+  override def getLength: Int = VariableLengthDataType
+
+  override def serialize(duration: Duration, target: DataOutputView): Unit =
+    duration match {
+      case null => target.writeByte(NullDiscriminant) // Cannot use NullMarker here as we are writing a byte
+      case finiteDuration: FiniteDuration =>
+        target.writeByte(FiniteDurationDiscriminant)
+        FiniteDurationSerializer.serialize(finiteDuration, target)
+      case Duration.Inf         => target.writeByte(PositiveInfiniteDurationDiscriminant)
+      case Duration.MinusInf    => target.writeByte(NegativeInfiniteDurationDiscriminant)
+      case _: Duration.Infinite => // Last to handle Duration.Undefined which doesn't equal itself
+        target.writeByte(UndefinedDurationDiscriminant)
+    }
+
+  override def deserialize(source: DataInputView): Duration =
+    source.readByte() match {
+      case NullDiscriminant                     => null
+      case FiniteDurationDiscriminant           => FiniteDurationSerializer.deserialize(source)
+      case UndefinedDurationDiscriminant        => Duration.Undefined
+      case PositiveInfiniteDurationDiscriminant => Duration.Inf
+      case NegativeInfiniteDurationDiscriminant => Duration.MinusInf
+      case _                                    => throw new IllegalArgumentException("Unknown duration type")
+    }
+
+  override def copy(source: DataInputView, target: DataOutputView): Unit = {
+    val discriminant = source.readByte()
+    target.writeByte(discriminant)
+    if (discriminant == FiniteDurationDiscriminant) FiniteDurationSerializer.copy(source, target)
+  }
+
+  override def snapshotConfiguration(): TypeSerializerSnapshot[Duration] = new DurationSerializerSnapshot()
+
+}
+
+/** Serializer snapshot for [[Duration]]. */
+class DurationSerializerSnapshot extends SimpleTypeSerializerSnapshot[Duration](() => DurationSerializer)
+
+/** Serializer for [[FiniteDuration]].
+  *
+  * Don't handle null value. If you need a nullable finite duration, use [[DurationSerializer]].
+  */
+object FiniteDurationSerializer extends ImmutableSerializer[FiniteDuration] {
+
+  private val timeUnitSerializer: TypeSerializer[TimeUnit] = serializers.infoToSer(serializers.timeUnitInfo)
+
+  override def createInstance: FiniteDuration = Duration.Zero
+
+  override def getLength: Int = LongBytes + timeUnitSerializer.getLength // 1 Long + TimeUnit
+
+  override def serialize(finiteDuration: FiniteDuration, target: DataOutputView): Unit = {
+    target.writeLong(finiteDuration.length)
+    timeUnitSerializer.serialize(finiteDuration.unit, target)
+  }
+
+  override def deserialize(source: DataInputView): FiniteDuration = {
+    val length = source.readLong()
+    val unit   = timeUnitSerializer.deserialize(source)
+    new FiniteDuration(length, unit)
+  }
+
+  override def copy(source: DataInputView, target: DataOutputView): Unit = {
+    target.writeLong(source.readLong())
+    timeUnitSerializer.copy(source, target)
+  }
+
+  override def snapshotConfiguration(): TypeSerializerSnapshot[FiniteDuration] = new FiniteDurationSerializerSnapshot()
+
+}
+
+/** Serializer snapshot for [[FiniteDuration]]. */
+class FiniteDurationSerializerSnapshot
+    extends SimpleTypeSerializerSnapshot[FiniteDuration](() => FiniteDurationSerializer)