feature: handle null field in case class (#267)

* feature: handle null field in case class

* fix 4 bytes missing in fixed length case class

* fix null value handling to respect the fixed length

Author: arnaud-daroussin

README.md

@@ -209,7 +209,7 @@ From Flink 1.19, a check is done to detect this misusage. To disable it, see [Di
 
 ### Flink ADT
 
-To derive a TypeInformation for a sealed trait, you can do:
+To derive a TypeInformation for a case class or sealed trait, you can do:
 
 ```scala mdoc:reset-object
 import org.apache.flinkx.api.serializers._
@@ -227,6 +227,31 @@ object Event {
 
 Be careful with a wildcard import of import `org.apache.flink.api.scala._`: it has a `createTypeInformation` implicit function, which may happily generate you a kryo-based serializer in a place you never expected. So in a case if you want to do this type of wildcard import, make sure that you explicitly called `deriveTypeInformation` for all the sealed traits in the current scope.
 
+#### Null value handling
+
+A case class can be null, the case class serializer natively handles the null case.
+
+A case class field can also be null, either:
+- the serializer of this field natively handles its nullability.
+- the field must be annotated with `@nullable` in order to be wrapped in Flink's `NullableSerializer`.
+
+In any cases, it can be a good hint to use `@nullable` annotation to indicate when fields are meant to be nullable.
+
+```scala mdoc:reset-object
+import org.apache.flinkx.api.serializers._
+import org.apache.flinkx.api.serializer.nullable
+import org.apache.flink.api.common.typeinfo.TypeInformation
+
+case class Click(id: String, clickEvent: ClickEvent)
+
+case class ClickEvent(
+    @nullable history: Array[String], // @nullable allows to handle null array
+    @nullable id: String) // Effectless here as null strings are natively handled
+
+Click("id1", null) // A case class can be null
+Click("id2", ClickEvent(null, null)) // Valid thanks to @nullable
+```
+
 ### Java types
 
 Built-in serializers are for Scala language abstractions and won't derive `TypeInformation` for Java classes (as they don't extend the `scala.Product` type). But you can always fall back to Flink's own POJO serializer in this way, so just make it implicit so this API can pick it up:

modules/flink-1-api/src/main/scala-2/org/apache/flinkx/api/LowPrioImplicits.scala

@@ -3,7 +3,8 @@ package org.apache.flinkx.api
 import magnolia1.{CaseClass, Magnolia, SealedTrait}
 import org.apache.flink.api.common.ExecutionConfig
 import org.apache.flink.api.common.typeinfo.TypeInformation
-import org.apache.flinkx.api.serializer.{CoproductSerializer, CaseClassSerializer, ScalaCaseObjectSerializer}
+import org.apache.flink.api.java.typeutils.runtime.NullableSerializer
+import org.apache.flinkx.api.serializer.{CaseClassSerializer, CoproductSerializer, ScalaCaseObjectSerializer, nullable}
 import org.apache.flinkx.api.typeinfo.{CoproductTypeInformation, ProductTypeInformation}
 import org.apache.flinkx.api.util.ClassUtil.isCaseClassImmutable
 
@@ -32,7 +33,12 @@ private[api] trait LowPrioImplicits {
         } else {
           new CaseClassSerializer[T](
             clazz = clazz,
-            scalaFieldSerializers = ctx.parameters.map(_.typeclass.createSerializer(config)).toArray,
+            scalaFieldSerializers = ctx.parameters.map { p =>
+              val ser = p.typeclass.createSerializer(config)
+              if (p.annotations.exists(_.isInstanceOf[nullable])) {
+                NullableSerializer.wrapIfNullIsNotSupported(ser, true)
+              } else ser
+            }.toArray,
             isCaseClassImmutable = isCaseClassImmutable(clazz, ctx.parameters.map(_.label))
           )
         }

modules/flink-1-api/src/main/scala-3/org/apache/flinkx/api/LowPrioImplicits.scala

@@ -8,7 +8,8 @@ import magnolia1.{CaseClass, SealedTrait}
 import org.apache.flink.api.common.typeinfo.TypeInformation
 import org.apache.flink.api.common.typeutils.TypeSerializer
 import org.apache.flink.api.common.ExecutionConfig
-import org.apache.flinkx.api.serializer.{CaseClassSerializer, CoproductSerializer, ScalaCaseObjectSerializer}
+import org.apache.flink.api.java.typeutils.runtime.NullableSerializer
+import org.apache.flinkx.api.serializer.{CaseClassSerializer, CoproductSerializer, ScalaCaseObjectSerializer, nullable}
 import org.apache.flinkx.api.typeinfo.{CoproductTypeInformation, ProductTypeInformation}
 import org.apache.flinkx.api.util.ClassUtil.isCaseClassImmutable
 
@@ -38,8 +39,12 @@ private[api] trait LowPrioImplicits extends TaggedDerivation[TypeInformation]:
           else
             new CaseClassSerializer[T & Product](
               clazz = clazz,
-              scalaFieldSerializers =
-                IArray.genericWrapArray(ctx.params.map(_.typeclass.createSerializer(config))).toArray,
+              scalaFieldSerializers = IArray.genericWrapArray(ctx.params.map { p =>
+                val ser = p.typeclass.createSerializer(config)
+                if (p.annotations.exists(_.isInstanceOf[nullable])) {
+                  NullableSerializer.wrapIfNullIsNotSupported(ser, true)
+                } else ser
+              }).toArray,
               isCaseClassImmutable = isCaseClassImmutable(clazz, ctx.params.map(_.label))
             )
         val ti = new ProductTypeInformation[T & Product](

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

@@ -3,7 +3,7 @@ package org.apache.flinkx.api
 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.SchemaEvolutionTest.{Click, ClickEvent, Event}
 import org.apache.flinkx.api.serializer.CaseClassSerializer
 import org.apache.flinkx.api.serializers._
 import org.scalatest.flatspec.AnyFlatSpec
@@ -13,11 +13,9 @@ import java.io.ByteArrayOutputStream
 import java.nio.file.{Files, Path}
 
 class SchemaEvolutionTest extends AnyFlatSpec with Matchers {
-  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                                                   =
+  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] =

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

@@ -3,10 +3,13 @@ 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.flink.api.common.typeutils.base.StringSerializer
+import org.apache.flink.api.java.typeutils.runtime.NullableSerializer
 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.flinkx.api.serializers._
+import org.apache.flinkx.api.serializer.{CaseClassSerializer, nullable}
 import org.scalatest.Inspectors
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
@@ -204,6 +207,11 @@ class SerializerTest extends AnyFlatSpec with Matchers with Inspectors with Test
     roundtrip(ser, ExtendingCaseClass("abc", "def"))
   }
 
+  it should "serialize a null case class" in {
+    val ser = implicitly[TypeInformation[Simple]].createSerializer(ec)
+    roundtrip(ser, null)
+  }
+
   it should "serialize a case class with nullable field" in {
     val ser = implicitly[TypeInformation[NullableField]].createSerializer(ec)
     roundtrip(ser, NullableField(null, Bar(1)))
@@ -214,12 +222,26 @@ class SerializerTest extends AnyFlatSpec with Matchers with Inspectors with Test
     roundtrip(ser, NullableFieldWithNoArity(null))
   }
 
+  it should "serialize nullable fields" in {
+    val ser = implicitly[TypeInformation[SimpleJava]].createSerializer(ec)
+    roundtrip(ser, SimpleJava(null, null))
+    val ccser = ser.asInstanceOf[CaseClassSerializer[SimpleJava]]
+    // IntSerializer doesn't handle null so it's wrapped in a NullableSerializer
+    ccser.getFieldSerializers()(0) shouldBe a[NullableSerializer[Integer]]
+    ccser.getFieldSerializers()(1) shouldBe a[StringSerializer] // StringSerializer natively handles null
+  }
+
+  it should "serialize a case class with a nullable field of a fixed size case class" in {
+    val ser = implicitly[TypeInformation[NullableFixedSizeCaseClass]].createSerializer(ec)
+    roundtrip(ser, NullableFixedSizeCaseClass(null))
+  }
+
 }
 
 object SerializerTest {
   case class Simple(a: Int, b: String)
   case class SimpleList(a: List[Int])
-  case class SimpleJava(a: Integer, b: String)
+  case class SimpleJava(@nullable a: Integer, @nullable b: String)
   case class JavaTime(a: Instant, b: LocalDate, c: LocalDateTime)
   case class Nested(a: Simple)
 
@@ -296,4 +318,6 @@ object SerializerTest {
 
   final case class NullableFieldWithNoArity(var a: NoArity)
 
+  final case class NullableFixedSizeCaseClass(@nullable javaTime: JavaTime)
+
 }

modules/flink-2-api/src/main/scala-2/org/apache/flinkx/api/LowPrioImplicits.scala

@@ -3,7 +3,8 @@ package org.apache.flinkx.api
 import magnolia1.{CaseClass, Magnolia, SealedTrait}
 import org.apache.flink.api.common.serialization.SerializerConfig
 import org.apache.flink.api.common.typeinfo.TypeInformation
-import org.apache.flinkx.api.serializer.{CoproductSerializer, CaseClassSerializer, ScalaCaseObjectSerializer}
+import org.apache.flink.api.java.typeutils.runtime.NullableSerializer
+import org.apache.flinkx.api.serializer.{CaseClassSerializer, CoproductSerializer, ScalaCaseObjectSerializer, nullable}
 import org.apache.flinkx.api.typeinfo.{CoproductTypeInformation, ProductTypeInformation}
 import org.apache.flinkx.api.util.ClassUtil.isCaseClassImmutable
 
@@ -32,7 +33,12 @@ private[api] trait LowPrioImplicits {
         } else {
           new CaseClassSerializer[T](
             clazz = clazz,
-            scalaFieldSerializers = ctx.parameters.map(_.typeclass.createSerializer(config)).toArray,
+            scalaFieldSerializers = ctx.parameters.map { p =>
+              val ser = p.typeclass.createSerializer(config)
+              if (p.annotations.exists(_.isInstanceOf[nullable])) {
+                NullableSerializer.wrapIfNullIsNotSupported(ser, true)
+              } else ser
+            }.toArray,
             isCaseClassImmutable = isCaseClassImmutable(clazz, ctx.parameters.map(_.label))
           )
         }

modules/flink-2-api/src/main/scala-3/org/apache/flinkx/api/LowPrioImplicits.scala

@@ -4,12 +4,12 @@ import scala.collection.mutable
 import scala.compiletime.summonInline
 import scala.deriving.Mirror
 import scala.reflect.ClassTag
-
 import magnolia1.{CaseClass, SealedTrait}
 import org.apache.flink.api.common.typeinfo.TypeInformation
 import org.apache.flink.api.common.typeutils.TypeSerializer
 import org.apache.flink.api.common.serialization.SerializerConfig
-import org.apache.flinkx.api.serializer.{CoproductSerializer, CaseClassSerializer, ScalaCaseObjectSerializer}
+import org.apache.flink.api.java.typeutils.runtime.NullableSerializer
+import org.apache.flinkx.api.serializer.{CaseClassSerializer, CoproductSerializer, ScalaCaseObjectSerializer, nullable}
 import org.apache.flinkx.api.typeinfo.{CoproductTypeInformation, ProductTypeInformation}
 import org.apache.flinkx.api.util.ClassUtil.isCaseClassImmutable
 
@@ -39,8 +39,12 @@ private[api] trait LowPrioImplicits extends TaggedDerivation[TypeInformation]:
           else
             new CaseClassSerializer[T & Product](
               clazz = clazz,
-              scalaFieldSerializers =
-                IArray.genericWrapArray(ctx.params.map(_.typeclass.createSerializer(config))).toArray,
+              scalaFieldSerializers = IArray.genericWrapArray(ctx.params.map { p =>
+                val ser = p.typeclass.createSerializer(config)
+                if (p.annotations.exists(_.isInstanceOf[nullable])) {
+                  NullableSerializer.wrapIfNullIsNotSupported(ser, true)
+                } else ser
+              }).toArray,
               isCaseClassImmutable = isCaseClassImmutable(clazz, ctx.params.map(_.label))
             )
         val ti = new ProductTypeInformation[T & Product](

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.serializer.CaseClassSerializer.EmptyByteArray
 import org.slf4j.{Logger, LoggerFactory}
 
 /** Serializer for Case Classes. Creation and access is different from our Java Tuples so we have to treat them
@@ -39,6 +40,8 @@ class CaseClassSerializer[T <: Product](
 
   @transient private lazy val log: Logger = LoggerFactory.getLogger(this.getClass)
 
+  private val nullPadding: Array[Byte] = if (super.getLength > 0) new Array(super.getLength) else EmptyByteArray
+
   override val isImmutableType: Boolean = isCaseClassImmutable &&
     fieldSerializers.forall(Option(_).exists(_.isImmutableType))
   val isImmutableSerializer: Boolean =
@@ -92,10 +95,13 @@ class CaseClassSerializer[T <: Product](
       createInstance(fields.toArray)
     }
 
+  override val getLength: Int = if (super.getLength == -1) -1 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
     target.writeInt(sourceArity)
+    if (value == null) target.write(nullPadding)
 
     (0 until sourceArity).foreach { i =>
       val serializer = fieldSerializers(i).asInstanceOf[TypeSerializer[Any]]
@@ -114,6 +120,7 @@ class CaseClassSerializer[T <: Product](
   def deserialize(source: DataInputView): T = {
     val sourceArity = source.readInt()
     if (sourceArity == -1) {
+      source.skipBytesToRead(nullPadding.length)
       null.asInstanceOf[T]
     } else {
       val fields = new Array[AnyRef](sourceArity)
@@ -134,3 +141,7 @@ class CaseClassSerializer[T <: Product](
     new ScalaCaseClassSerializerSnapshot[T](this)
 
 }
+
+object CaseClassSerializer {
+  private val EmptyByteArray: Array[Byte] = new Array(0)
+}

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

@@ -0,0 +1,5 @@
+package org.apache.flinkx.api.serializer
+
+import scala.annotation.StaticAnnotation
+
+final class nullable extends StaticAnnotation