[fast-avro][bugfix] Delegating setSchema() call to coldDeserializer f…

…rom FastGenericDatumReader.

It's needed to deserialize 1st record(s) from file using DataFileStream.

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

@@ -2,6 +2,8 @@
 
 import com.linkedin.avro.fastserde.customized.DatumReaderCustomization;
 import java.io.IOException;
+
+import org.apache.avro.Schema;
 import org.apache.avro.io.Decoder;
 
 import static com.linkedin.avro.fastserde.customized.DatumReaderCustomization.*;
@@ -17,5 +19,15 @@ default T deserialize(T reuse, Decoder d) throws IOException {
     return deserialize(reuse, d, DEFAULT_DATUM_READER_CUSTOMIZATION);
   }
 
+  /**
+   * Set the writer's schema.
+   * @see org.apache.avro.io.DatumReader#setSchema(Schema)
+   */
+  default void setSchema(Schema writerSchema) {
+    // Implement this method only in vanilla-avro-based classes (e.g. fallback scenario).
+    // Normally for generated deserializers it doesn't make sense.
+    throw new UnsupportedOperationException("Can't change schema for already generated class.");
+  }
+
   T deserialize(T reuse, Decoder d, DatumReaderCustomization customization) throws IOException;
 }

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

@@ -87,6 +87,8 @@ public void setSchema(Schema schema) {
     if (readerSchema == null) {
       readerSchema = writerSchema;
     }
+
+    coldDeserializer.setSchema(schema);
   }
 
   @Override

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

@@ -52,6 +52,11 @@ public FastDeserializerWithAvroSpecificImpl(Schema writerSchema, Schema readerSc
       this.runtimeClassGenerationDone = runtimeClassGenerationDone;
     }
 
+    @Override
+    public void setSchema(Schema writerSchema) {
+      this.customizedDatumReader.setSchema(writerSchema);
+    }
+
     @Override
     public V deserialize(V reuse, Decoder d, DatumReaderCustomization customization) throws IOException {
       if (failFast) {
@@ -103,6 +108,11 @@ public FastDeserializerWithAvroGenericImpl(Schema writerSchema, Schema readerSch
       this.runtimeClassGenerationDone = runtimeClassGenerationDone;
     }
 
+    @Override
+    public void setSchema(Schema writerSchema) {
+      customizedDatumReader.setSchema(writerSchema);
+    }
+
     @Override
     public V deserialize(V reuse, Decoder d, DatumReaderCustomization customization) throws IOException {
       if (failFast) {