[FLINK-14642] Add support for copying null values to the TupleSeriali…

…zer and CaseClassSerializer
is · Nov 14, 2019 · 84c96b3 · 84c96b3
1 parent a8868dd
commit 84c96b3
Show file tree

Hide file tree

Showing 2 changed files with 19 additions and 6 deletions.
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializer.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/TupleSerializer.java
@@ -106,6 +106,10 @@ public T createOrReuseInstance(Object[] fields, T reuse) {
 
  @Override
  public T copy(T from) {
+ if (from == null) {
+ return null;
+ }
+
  T target = instantiateRaw();
  for (int i = 0; i < arity; i++) {
  Object copy = fieldSerializers[i].copy(from.getField(i));
@@ -116,6 +120,10 @@ public T copy(T from) {
 
  @Override
  public T copy(T from, T reuse) {
+ if (from == null) {
+ return null;
+ }
+
  for (int i = 0; i < arity; i++) {
  Object copy = fieldSerializers[i].copy((Object)from.getField(i), reuse.getField(i));
  reuse.setField(copy, i);

diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/CaseClassSerializer.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/CaseClassSerializer.scala
@@ -86,13 +86,18 @@ abstract class CaseClassSerializer[T <: Product](
  }
 
  def copy(from: T): T = {
- initArray()
- var i = 0
- while (i < arity) {
- fields(i) = fieldSerializers(i).copy(from.productElement(i).asInstanceOf[AnyRef])
- i += 1
+ if (from == null) {
+ null.asInstanceOf[T]
+ }
+ else {
+ initArray()
+ var i = 0
+ while (i < arity) {
+ fields(i) = fieldSerializers(i).copy(from.productElement(i).asInstanceOf[AnyRef])
+ i += 1
+ }
+ createInstance(fields)
  }
- createInstance(fields)
  }
 
  def serialize(value: T, target: DataOutputView) {