MapField having null Row value throws NullPointerException during encoding

[damccorm]

Below is the scenario tried:

public void testMapWithNullRowAsValue() {

Schema primitiveFieldsScema =
Schema.builder().addNullableField("color",
FieldType.STRING).build();
      
FieldType valueType = FieldType.row(primitiveFieldsScema).withNullable(true);
    
 
Schema inputSchema =
                Schema.builder()
                      .addNullableField("mapWithValueAsRow",

                            FieldType.map(FieldType.STRING, valueType))
                      .build();

Map<String,
Row> mapWithValueAsNullRow = new HashMap<>();
mapWithValueAsNullRow.put("key", null);   // Setting
value of map as NULL

Row rowOfMap =  Row.withSchema(inputSchema)
                   .addValue(mapWithValueAsNullRow)

                  .build();

pipeline.apply(Create.of(rowOfMap)).setRowSchema(inputSchema);

pipeline.run().waitUntilFinish(Duration.standardMinutes(1));

}

 

Below NPE is seen:

java.lang.NullPointerException
    at org.apache.beam.sdk.coders.RowCoderGenerator$EncodeInstruction.encodeDelegate(RowCoderGenerator.java:225)
  
 at org.apache.beam.sdk.coders.Coder$ByteBuddy$DifqoQx1.encode(Unknown Source)
    at org.apache.beam.sdk.coders.Coder$ByteBuddy$DifqoQx1.encode(Unknown
Source)
    at org.apache.beam.sdk.coders.MapCoder.encode(MapCoder.java:96)
    at org.apache.beam.sdk.coders.MapCoder.encode(MapCoder.java:69)
  
 at org.apache.beam.sdk.coders.MapCoder.encode(MapCoder.java:43)
    at org.apache.beam.sdk.coders.RowCoderGenerator$EncodeInstruction.encodeDelegate(RowCoderGenerator.java:234)
  
 at org.apache.beam.sdk.coders.Coder$ByteBuddy$9dwdE80N.encode(Unknown Source)
    at org.apache.beam.sdk.coders.Coder$ByteBuddy$9dwdE80N.encode(Unknown
Source)
    at org.apache.beam.sdk.coders.RowCoder.encode(RowCoder.java:145)
    at org.apache.beam.sdk.schemas.SchemaCoder.encode(SchemaCoder.java:81)
  
 at org.apache.beam.sdk.coders.Coder.encode(Coder.java:136)
    at org.apache.beam.sdk.util.CoderUtils.encodeToSafeStream(CoderUtils.java:82)
  
 at org.apache.beam.sdk.util.CoderUtils.encodeToByteArray(CoderUtils.java:66)
    at org.apache.beam.sdk.util.CoderUtils.encodeToByteArray(CoderUtils.java:51)
  
 at org.apache.beam.sdk.util.CoderUtils.clone(CoderUtils.java:141)
    at org.apache.beam.sdk.util.MutationDetectors$CodedValueMutationDetector.<init>(MutationDetectors.java:113)
  
 at org.apache.beam.sdk.util.MutationDetectors.forValueWithCoder(MutationDetectors.java:44)
    at
org.apache.beam.runners.direct.ImmutabilityCheckingBundleFactory$ImmutabilityEnforcingBundle.add(ImmutabilityCheckingBundleFactory.java:112)
  
 at org.apache.beam.runners.direct.BoundedReadEvaluatorFactory$BoundedReadEvaluator.processElement(BoundedReadEvaluatorFactory.java:151)
  
 at org.apache.beam.runners.direct.DirectTransformExecutor.processElements(DirectTransformExecutor.java:160)
  
 at org.apache.beam.runners.direct.DirectTransformExecutor.run(DirectTransformExecutor.java:124)
  
 at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
  
 at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
  
 at java.lang.Thread.run(Thread.java:748)

Imported from Jira BEAM-7841. Original Jira may contain additional context.
Reported by: vbm.