fix a missing copy of the arity field in direct-memory copy() (#272)

Author: arnaud-daroussin

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

@@ -43,7 +43,7 @@ class CaseClassSerializer[T <: Product](
   private val nullPadding: Array[Byte] = if (super.getLength > 0) new Array(super.getLength) else EmptyByteArray
 
   override val isImmutableType: Boolean = isCaseClassImmutable && fieldSerializers.forall(_.isImmutableType)
-  val isImmutableSerializer: Boolean = fieldSerializers.forall(s => s.duplicate().eq(s))
+  val isImmutableSerializer: Boolean    = fieldSerializers.forall(s => s.duplicate().eq(s))
 
   // In Flink, serializers & serializer snapshotters have strict ser/de requirements.
   // Both need to be capable of creating one another.
@@ -131,6 +131,17 @@ class CaseClassSerializer[T <: Product](
     }
   }
 
+  override def copy(source: DataInputView, target: DataOutputView): Unit = {
+    val sourceArity = source.readInt()
+    target.writeInt(sourceArity)
+    if (sourceArity == -1) {
+      source.skipBytesToRead(nullPadding.length)
+      target.skipBytesToWrite(nullPadding.length)
+    } else {
+      super.copy(source, target)
+    }
+  }
+
   override def createInstance(fields: Array[AnyRef]): T = {
     constructor(fields)
   }

modules/flink-common-api/src/test/scala/org/apache/flinkx/api/serializer/CaseClassSerializerTest.scala

@@ -2,6 +2,7 @@ package org.apache.flinkx.api.serializer
 
 import org.apache.flink.api.common.typeutils.base.StringSerializer
 import org.apache.flink.api.java.typeutils.runtime.RowSerializer
+import org.apache.flink.core.memory.{DataInputDeserializer, DataOutputSerializer}
 import org.apache.flinkx.api.serializer.CaseClassSerializerTest.{Immutable, Mutable, OuterImmutable, OuterMutable}
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
@@ -90,6 +91,22 @@ class CaseClassSerializerTest extends AnyFlatSpec with Matchers {
     resultData shouldEqual expectedData
   }
 
+  it should "copy the serialized stream" in {
+    val serializer = new CaseClassSerializer[Immutable](classOf[Immutable], Array(StringSerializer.INSTANCE), true)
+    val outerSerializer = new CaseClassSerializer[OuterMutable](classOf[OuterMutable], Array(serializer), true)
+    val expectedData = OuterMutable(Immutable("a"))
+
+    val output = new DataOutputSerializer(1024)
+    outerSerializer.serialize(expectedData, output)
+    val input = new DataInputDeserializer(output.getSharedBuffer)
+    val newOutput = new DataOutputSerializer(1024)
+    outerSerializer.copy(input, newOutput)
+    val newInput = new DataInputDeserializer(newOutput.getSharedBuffer)
+    val resultData = outerSerializer.deserialize(newInput)
+
+    resultData shouldEqual expectedData
+  }
+
   "duplicate" should "return itself when the serializer is immutable" in {
     val serializer = new CaseClassSerializer[Mutable](classOf[Mutable], Array(StringSerializer.INSTANCE), false)
     serializer.duplicate() should be theSameInstanceAs serializer