[fast-avro] Fixing case when schema changed and complex fields became…

… nullable.

Scenario: Record is serialized using schema with not-null complex fields (record, map, array)
and later deserialized with changed schema having these fields nullable (union of "null" complex type).

fastserde/avro-fastserde/src/main/java/com/linkedin/avro/fastserde/FastDeserializerGenerator.java

@@ -239,10 +239,12 @@ private void processRecord(JVar recordSchemaVar, String recordName, final Schema
       // as effectiveRecordReaderSchema and recordAction needs to be adjusted.
       if (Schema.Type.UNION.equals(recordReaderSchema.getType())) {
         effectiveRecordReaderSchema = schemaAssistant.compatibleUnionSchema(recordWriterSchema, recordReaderSchema);
-        recordSchemaVar = declareSchemaVar(effectiveRecordReaderSchema, recordName + "RecordSchema",
-                recordSchemaVar.invoke("getTypes").invoke("get").arg(JExpr.lit(
-                        schemaAssistant.compatibleUnionSchemaIndex(recordWriterSchema, recordReaderSchema)
-                )));
+        if (recordSchemaVar != null) {
+          recordSchemaVar = declareSchemaVar(effectiveRecordReaderSchema, recordName + "RecordSchema",
+                  recordSchemaVar.invoke("getTypes").invoke("get").arg(JExpr.lit(
+                          schemaAssistant.compatibleUnionSchemaIndex(recordWriterSchema, recordReaderSchema)
+                  )));
+        }
 
         Symbol symbol = null;
         ListIterator<Symbol> symbolIterator = recordAction.getSymbolIterator() != null ? recordAction.getSymbolIterator()
@@ -726,15 +728,16 @@ private void processArray(JVar arraySchemaVar, final String name, final Schema a
 
     Schema effectiveArrayReaderSchema = arrayReaderSchema;
     if (action.getShouldRead()) {
-
       // if reader schema is a union then the compatible union array type must be selected
       // as effectiveArrayReaderSchema and action needs to be adjusted.
       if (Schema.Type.UNION.equals(arrayReaderSchema.getType())) {
         effectiveArrayReaderSchema = schemaAssistant.compatibleUnionSchema(arraySchema, arrayReaderSchema);
-        arraySchemaVar = declareSchemaVar(effectiveArrayReaderSchema, name + "ArraySchema",
-                arraySchemaVar.invoke("getTypes").invoke("get").arg(JExpr.lit(
-                        schemaAssistant.compatibleUnionSchemaIndex(arraySchema, arrayReaderSchema)
-                )));
+        if (arraySchemaVar != null) {
+          arraySchemaVar = declareSchemaVar(effectiveArrayReaderSchema, name + "ArraySchema",
+                  arraySchemaVar.invoke("getTypes").invoke("get").arg(JExpr.lit(
+                          schemaAssistant.compatibleUnionSchemaIndex(arraySchema, arrayReaderSchema)
+                  )));
+        }
 
         Symbol symbol = action.getSymbol();
         if (!(symbol instanceof Symbol.UnionAdjustAction)) {
@@ -914,7 +917,7 @@ private void processMap(JVar mapSchemaVar, final String name, final Schema mapSc
      * Determine the action symbol for Map value. {@link ResolvingGrammarGenerator} generates
      * resolving grammar symbols with reversed order of production sequence. If this symbol is
      * a terminal, its production list will be <code>null</code>. Otherwise the production list
-     * holds the the sequence of the symbols that forms this symbol.
+     * holds the sequence of the symbols that forms this symbol.
      *
      * The {@link FastDeserializerGenerator.generateDeserializer} tries to proceed as a depth-first,
      * left-to-right traversal of the schema. So for a nested Map, we need to iterate production list
@@ -926,21 +929,23 @@ private void processMap(JVar mapSchemaVar, final String name, final Schema mapSc
       // as effectiveMapReaderSchema and action needs to be adjusted.
       if (Schema.Type.UNION.equals(mapReaderSchema.getType())) {
         effectiveMapReaderSchema = schemaAssistant.compatibleUnionSchema(mapSchema, mapReaderSchema);
-        mapSchemaVar = declareSchemaVar(effectiveMapReaderSchema, name + "MapSchema",
-                mapSchemaVar.invoke("getTypes").invoke("get").arg(JExpr.lit(
-                        schemaAssistant.compatibleUnionSchemaIndex(mapSchema, mapReaderSchema)
-                )));
+        if (mapSchemaVar != null) {
+          mapSchemaVar = declareSchemaVar(effectiveMapReaderSchema, name + "MapSchema",
+                  mapSchemaVar.invoke("getTypes").invoke("get").arg(JExpr.lit(
+                          schemaAssistant.compatibleUnionSchemaIndex(mapSchema, mapReaderSchema)
+                  )));
+        }
 
         Symbol symbol = action.getSymbol();
-        if (!(symbol instanceof Symbol.UnionAdjustAction)) {
+        if (!(symbol instanceof Symbol.UnionAdjustAction)) { // TODO: maybe check if (symbol != null) as well?
           for (Symbol aSymbol: symbol.production) {
             if (aSymbol instanceof Symbol.UnionAdjustAction) {
               symbol = aSymbol;
               break;
             }
           }
         }
-        if (symbol == null) {
+        if (symbol == null) { // TODO always false(?) - if symbol is null then symbol.production from above throws NPE
           throw new FastDeserializerGeneratorException("Symbol.UnionAdjustAction is expected but was not found");
         }
         action = FieldAction.fromValues(action.getType(), action.getShouldRead(),