Feature: handle null case class (#265)

* Feature: handle null case class

* Removes DISABLE_CASE_CLASS_ARITY_USAGE flag

* Removes special condition for case class with no arity: it's handled like any other case class

* fix IllegalArgumentException: argument type mismatch

* fix case class instantiation: call constructor with all deserialized fields and append default values to complete if necessary

* fix constructor reflection lookup being done too early

* switch from apply() to constructor() call to fix conflict with other overloaded apply()

Author: arnaud-daroussin

README.md

@@ -413,26 +413,6 @@ env
 
 ## Feature Flags
 
-### Disable Class Arity Usage
-
-In the [1.1.5 release](https://github.com/flink-extended/flink-scala-api/releases/tag/v1.18.1_1.1.5) the Case Class serialization process also [stores case class arity](https://github.com/flink-extended/flink-scala-api/pull/98/files#diff-e896c210d6a754cb3afb462aea34cca08f090330f6f3c663a64dfb5584fc3727R106) number to
-a savepoint. This was introduced to support Case Class schema evolution and allow to add new
-class fields with default values. However, unfortunately this is the breaking change to the Flink job state restore process. Flink job will fail, if
-a savepoint used for the job restore was created by 1.1.4 or earlier releases.
-
-In order migrate to the 1.1.5 release version, one can use specially added environment variable:
-`DISABLE_CASE_CLASS_ARITY_USAGE`.
-
-To disable new savepoint format and be able to restore a Flink job with a savepoint created before 1.1.5 release set the variable to `true`. 
-
-Example: `DISABLE_CASE_CLASS_ARITY_USAGE = true`
-
-To enable new serialization logic set this variable to `false` or simply do not define this environment variable.
-
-Example: `DISABLE_CASE_CLASS_ARITY_USAGE = false`
-
-P.S. this flag can be deprecated in future when most of the users migrate to the latest library version.
-
 ### Disable fail-fast on Scala type resolution with Class
 
 From [1.2.3 release](https://github.com/flink-extended/flink-scala-api/releases/tag/v1.20.0_1.2.3), a check is done to prevent misusage of Scala type resolution with `Class` which may lead to silently fallback to generic Kryo serializers.

modules/flink-1-api/src/test/resources/without-arity-test.dat

@@ -1,26 +1,56 @@
 package org.apache.flinkx.api
 
-import org.apache.flinkx.api.SchemaEvolutionTest.{Click, ClickEvent, Event, NoArityTest}
-import org.apache.flinkx.api.serializers._
-import org.apache.flinkx.api.serializer.CaseClassSerializer
-import org.apache.flink.api.common.typeinfo.TypeInformation
-import org.apache.flink.core.memory.{DataInputViewStreamWrapper, DataOutputViewStreamWrapper}
 import org.apache.flink.api.common.ExecutionConfig
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.core.memory._
+import org.apache.flinkx.api.SchemaEvolutionTest.{Click, ClickEvent, Event, View}
+import org.apache.flinkx.api.serializer.CaseClassSerializer
+import org.apache.flinkx.api.serializers._
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 
 import java.io.ByteArrayOutputStream
 import java.nio.file.{Files, Path}
 
 class SchemaEvolutionTest extends AnyFlatSpec with Matchers {
-  private implicit val eventTypeInfo: TypeInformation[Event]       = deriveTypeInformation[Event]
-  private implicit val arityTestInfo: TypeInformation[NoArityTest] = deriveTypeInformation[NoArityTest]
-  private val clicks                                               =
+  private implicit val clickEventTypeInfo: TypeInformation[ClickEvent] = deriveTypeInformation[ClickEvent]
+  private implicit val viewInfo: TypeInformation[View]                 = deriveTypeInformation[View]
+  private implicit val newClickTypeInfo: TypeInformation[Click]        = deriveTypeInformation[Click]
+  private implicit val eventTypeInfo: TypeInformation[Event]           = deriveTypeInformation[Event]
+  private val clicks                                                   =
     List(ClickEvent("a", "2021-01-01"), ClickEvent("b", "2021-01-01"), ClickEvent("c", "2021-01-01"))
 
   def createSerializer[T: TypeInformation] =
     implicitly[TypeInformation[T]].createSerializer(new ExecutionConfig())
 
+  it should "serialize click with old serializer and deserialize it with new serializer" in {
+    // Serializer before schema change: without serializers for the "new" default fields
+    val oldClickSerializer = new CaseClassSerializer[Click](
+      clazz = classOf[Click],
+      scalaFieldSerializers = Array(stringSerializer, createSerializer[List[ClickEvent]]),
+      isCaseClassImmutable = true
+    )
+    val newClickSerializer = createSerializer[Click] // Serializer derived from the "new" case class
+    val expected           = Click(null, clicks)
+
+    // serialize "old" Click with old serializer
+    val out = new DataOutputSerializer(1024 * 1024)
+    oldClickSerializer.serialize(expected, out)
+
+    // deserialize old Click with new serializer
+    val in     = new DataInputDeserializer(out.getSharedBuffer)
+    val result = newClickSerializer.deserialize(in)
+    result shouldBe expected
+
+    // serialize modified Click with new serializer
+    val modifiedExpected = expected.copy(fieldInFile = "modified1", fieldNotInFile = "modified2")
+    newClickSerializer.serialize(modifiedExpected, out)
+
+    // deserialize modified Click with new serializer
+    val modifiedResult = newClickSerializer.deserialize(in)
+    modifiedResult shouldBe modifiedExpected
+  }
+
   ignore should "generate blob for event=click+purchase" in {
     val buffer          = new ByteArrayOutputStream()
     val eventSerializer = createSerializer[Event]
@@ -34,28 +64,9 @@ class SchemaEvolutionTest extends AnyFlatSpec with Matchers {
     click shouldBe Click("p1", clicks)
   }
 
-  ignore should "generate blob for no arity test" in {
-    val buffer          = new ByteArrayOutputStream()
-    val eventSerializer = createSerializer[NoArityTest]
-    eventSerializer.serialize(NoArityTest(4, 3, List("test")), new DataOutputViewStreamWrapper(buffer))
-    Files.write(Path.of("src/test/resources/without-arity-test.dat"), buffer.toByteArray)
-  }
-
-  it should "decode class without arity info" in {
-    val buffer     = this.getClass.getResourceAsStream("/without-arity-test.dat")
-    val serializer = createSerializer[NoArityTest] match {
-      case s: CaseClassSerializer[_] => s
-      case s                         => fail(s"Derived serializer must be of CaseClassSerializer type, but was $s")
-    }
-    val decoded =
-      serializer.deserializeFromSource(new DataInputViewStreamWrapper(buffer), classArityUsageDisabled = true)
-    decoded shouldBe NoArityTest(4, 3, List("test"))
-  }
 }
 
 object SchemaEvolutionTest {
-  case class NoArityTest(field1: Long, field2: Long, field3: List[String] = Nil)
-
   sealed trait Event
   case class Click(
       id: String,

modules/flink-1-api/src/test/scala/org/apache/flinkx/api/SchemaEvolutionTest.scala

@@ -6,6 +6,7 @@ import org.apache.flink.api.common.typeutils.{TypeSerializer, TypeSerializerSnap
 import org.apache.flink.core.memory._
 import org.apache.flink.util.ChildFirstClassLoader
 import org.apache.flinkx.api.SerializerSnapshotTest._
+import org.apache.flinkx.api.serializer.CaseClassSerializer
 import org.apache.flinkx.api.serializers._
 import org.scalatest.Assertion
 import org.scalatest.flatspec.AnyFlatSpec
@@ -61,7 +62,7 @@ class SerializerSnapshotTest extends AnyFlatSpec with Matchers {
   }
 
   it should "do map ser snapshot adt " in {
-    implicit val ti: Typeclass[OuterTrait] = deriveTypeInformation[OuterTrait]
+    implicit val ti: TypeInformation[OuterTrait] = deriveTypeInformation[OuterTrait]
     drop(ti)
     assertRoundtripSerializer(createSerializer[TraitMap])
   }
@@ -133,6 +134,42 @@ class SerializerSnapshotTest extends AnyFlatSpec with Matchers {
     deserializedData should be(expectedData)
   }
 
+  it should "serialize and deserialize case class with null and default" in {
+    val expectedData = WithDefault(null)
+    // Serializer before schema change: without serializers for the second "new" default fields
+    val oldSerializer = new CaseClassSerializer[WithDefault](
+      clazz = classOf[WithDefault],
+      scalaFieldSerializers = Array(implicitly[TypeSerializer[SimpleClass1]]),
+      isCaseClassImmutable = false
+    )
+    val oldSnapshot = oldSerializer.snapshotConfiguration()
+
+    // Serialize the old snapshot
+    val output = new DataOutputSerializer(1024 * 1024)
+    TypeSerializerSnapshot.writeVersionedSnapshot(output, oldSnapshot) // Flink always calls this
+
+    // Serialize the "old" data with "old" serializer
+    oldSerializer.serialize(expectedData, output)
+
+    // Deserialize the old snapshot
+    val input                   = new DataInputDeserializer(output.getSharedBuffer)
+    val deserializedOldSnapshot = TypeSerializerSnapshot
+      .readVersionedSnapshot[WithDefault](input, getClass.getClassLoader) // Flink always calls this
+
+    // Deserialize the old data with the new serializer
+    val newSerializer = implicitly[TypeSerializer[WithDefault]]
+    val deserializedData = newSerializer.deserialize(input)
+    deserializedData should be(expectedData)
+
+    // serialize modified data with new serializer
+    expectedData.sc2 = SimpleClass2("c", 3)
+    newSerializer.serialize(expectedData, output)
+
+    // deserialize modified data with new serializer
+    val modifiedResult = newSerializer.deserialize(input)
+    modifiedResult shouldBe expectedData
+  }
+
   def roundtripSerializer[T](ser: TypeSerializer[T], cl: ClassLoader = getClass.getClassLoader): TypeSerializer[T] = {
     val snap   = ser.snapshotConfiguration()
     val buffer = new ByteArrayOutputStream()
@@ -174,4 +211,6 @@ object SerializerSnapshotTest {
 
   case class OuterClass(map: Map[UUID, List[OuterTrait]])
 
+  case class WithDefault(var sc1: SimpleClass1 = SimpleClass1("a", 1), var sc2: SimpleClass2 = SimpleClass2("b", 2))
+
 }

modules/flink-1-api/src/test/scala/org/apache/flinkx/api/SerializerSnapshotTest.scala

@@ -1,15 +1,15 @@
 package org.apache.flinkx.api
 
 import cats.data.NonEmptyList
+import org.apache.flink.api.common.ExecutionConfig
+import org.apache.flink.api.common.typeinfo.TypeInformation
 import org.apache.flinkx.api.SerializerTest.DeeplyNested.ModeNested.SuperNested.{Egg, Food}
 import org.apache.flinkx.api.SerializerTest.NestedRoot.NestedMiddle.NestedBottom
 import org.apache.flinkx.api.SerializerTest._
-import org.apache.flink.api.common.typeinfo.TypeInformation
-import org.apache.flink.api.common.ExecutionConfig
+import org.apache.flinkx.api.serializers._
 import org.scalatest.Inspectors
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
-import org.apache.flinkx.api.serializers._
 
 import java.time.{Instant, LocalDate, LocalDateTime}
 import java.util.UUID
@@ -204,6 +204,16 @@ class SerializerTest extends AnyFlatSpec with Matchers with Inspectors with Test
     roundtrip(ser, ExtendingCaseClass("abc", "def"))
   }
 
+  it should "serialize a case class with nullable field" in {
+    val ser = implicitly[TypeInformation[NullableField]].createSerializer(ec)
+    roundtrip(ser, NullableField(null, Bar(1)))
+  }
+
+  it should "serialize a case class with a nullable field of a case class with no arity" in {
+    val ser = implicitly[TypeInformation[NullableFieldWithNoArity]].createSerializer(ec)
+    roundtrip(ser, NullableFieldWithNoArity(null))
+  }
+
 }
 
 object SerializerTest {
@@ -276,4 +286,14 @@ object SerializerTest {
 
   final case class ExtendingCaseClass(override val a: String, b: String) extends AbstractClass(a)
 
+  final case class NullableField(var a: Foo = null, var b: Bar = null)
+
+  object NullableField {
+    def apply(a: Foo): NullableField = new NullableField(a, null)
+  }
+
+  final case class NoArity()
+
+  final case class NullableFieldWithNoArity(var a: NoArity)
+
 }

modules/flink-1-api/src/test/scala/org/apache/flinkx/api/SerializerTest.scala

@@ -1,14 +1,13 @@
 package org.apache.flinkx.api.serializer
 
 import scala.annotation.nowarn
-import scala.reflect.runtime.{currentMirror => cm}
-import scala.reflect.runtime.universe
 import scala.reflect.runtime.universe._
+import scala.reflect.runtime.{universe, currentMirror => cm}
 
 private[serializer] trait ConstructorCompat {
 
   @nowarn("msg=(eliminated by erasure)|(explicit array)")
-  final def lookupConstructor[T <: Product](cls: Class[T], numFields: Int): Array[AnyRef] => T = {
+  final def lookupConstructor[T <: Product](cls: Class[T]): Array[AnyRef] => T = {
     val rootMirror  = universe.runtimeMirror(cls.getClassLoader)
     val classSymbol = rootMirror.classSymbol(cls)
 
@@ -32,41 +31,26 @@ private[serializer] trait ConstructorCompat {
       .head
       .asMethod
 
-    val classMirror             = rootMirror.reflectClass(classSymbol)
-    val constructorMethodMirror = classMirror.reflectConstructor(primaryConstructorSymbol)
-
-    lazy val claas  = cm.classSymbol(cls)
-    lazy val module = claas.companion.asModule
-    lazy val im     = cm.reflect(cm.reflectModule(module).instance)
-
-    def withDefault(im: InstanceMirror, name: String, givenArgs: Int): List[Any] = {
-      val at     = TermName(name)
-      val ts     = im.symbol.typeSignature
-      val method = ts.member(at).asMethod
-
-      // either defarg or default val for type of p
-      def valueFor(p: Symbol, i: Int): Any = {
-        val defarg = ts member TermName(s"$name$$default$$${i + 1}")
+    val classMirror     = rootMirror.reflectClass(classSymbol)
+    val constructor     = classMirror.reflectConstructor(primaryConstructorSymbol)
+    val claas           = cm.classSymbol(cls)
+    val module          = claas.companion.asModule
+    val im              = cm.reflect(cm.reflectModule(module).instance)
+    val ts              = im.symbol.typeSignature
+    val constructorSize = primaryConstructorSymbol.paramLists.flatten.size
+    val defaultValues   = (1 to constructorSize)
+      .flatMap { i =>
+        val defarg = ts.member(TermName(s"$$lessinit$$greater$$default$$$i"))
         if (defarg != NoSymbol)
-          im.reflectMethod(defarg.asMethod)()
-        else
-          p.typeSignature match {
-            case t if t =:= typeOf[String]                                                                => null
-            case t if t =:= typeOf[Int] | t =:= typeOf[Long] | t =:= typeOf[Double] | t =:= typeOf[Float] => 0
-            case x => throw new IllegalArgumentException(x.toString)
-          }
+          Some(im.reflectMethod(defarg.asMethod)())
+        else None
       }
 
-      val defaultArgs = method.paramLists.flatten.splitAt(givenArgs)._2
-      defaultArgs.zipWithIndex.map(p => valueFor(p._1, p._2 + givenArgs))
-    }
-
-    { (args: Array[AnyRef]) =>
-      {
-        lazy val defaultArgs = withDefault(im, "apply", args.length)
-        val allArgs          = args.toList ++ (if (args.length == numFields) Nil else defaultArgs)
-        constructorMethodMirror.apply(allArgs: _*).asInstanceOf[T]
-      }
+    (args: Array[AnyRef]) => {
+      // Append default values for missing arguments
+      val allArgs = args ++ defaultValues.takeRight(constructorSize - args.length)
+      constructor.apply(allArgs: _*).asInstanceOf[T]
     }
   }
+
 }

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

@@ -1,20 +1,25 @@
 package org.apache.flinkx.api.serializer
 
-import java.lang.reflect.Modifier
-
+import java.lang.reflect.Constructor
 import scala.util.control.NonFatal
 
 private[serializer] trait ConstructorCompat:
   // As of Scala version 3.1.2, there is no direct support for runtime reflection.
   // This is in contrast to Scala 2, which has its own APIs for reflecting on classes.
   // Thus, fallback to Java reflection and look up the constructor matching the required signature.
-  final def lookupConstructor[T](cls: Class[T], numFields: Int): Array[AnyRef] => T =
+  final def lookupConstructor[T](cls: Class[T]): Array[AnyRef] => T =
     // Types of parameters can fail to match when (un)boxing is used.
     // Say you have a class `final case class Foo(a: String, b: Int)`.
     // The first parameter is an alias for `java.lang.String`, which the constructor uses.
     // The second parameter is an alias for `java.lang.Integer`, but the constructor actually takes an unboxed `int`.
     val constructor =
-      try cls.getConstructors.collectFirst { case c if c.getParameterCount == numFields => c }.get
+      try
+        cls.getConstructors
+          .foldLeft[Option[Constructor[?]]](None) {
+            case (Some(longest), c) if longest.getParameterCount < c.getParameterCount => Some(c)
+            case (_, c)                                                                => Some(c)
+          }
+          .get
       catch
         case NonFatal(e) =>
           throw new IllegalArgumentException(
@@ -28,19 +33,13 @@ private[serializer] trait ConstructorCompat:
             e
           )
 
-    { (args: Array[AnyRef]) =>
-      {
-        lazy val defaultArgs = cls
-          .getMethods()
-          .filter(
-            _.getName()
-              .startsWith("$lessinit$greater$default")
-          )
-          .sortBy(_.getName())
-          .map(_.invoke(null))
-          .takeRight(numFields - args.length) // read default values for missing arguments
+    lazy val defaultArgs = cls.getMethods
+      .filter(_.getName.startsWith("$lessinit$greater$default"))
+      .sortBy(_.getName())
+      .map(_.invoke(null))
 
-        val allArgs = args.toList ++ (if (args.length == numFields) Nil else defaultArgs)
-        constructor.newInstance(allArgs*).asInstanceOf[T]
-      }
+    (args: Array[AnyRef]) => {
+      // Append default values for missing arguments
+      val allArgs = args ++ defaultArgs.takeRight(constructor.getParameterCount - args.length)
+      constructor.newInstance(allArgs*).asInstanceOf[T]
     }