public static Object convertToAvroObject(Object relObj, Schema schema) {
    if (relObj == null) {
      return null;
    }
    switch (schema.getType()) {
      case RECORD:
        return convertToGenericRecord((SamzaSqlRelRecord) relObj, getNonNullUnionSchema(schema));
      case ARRAY:
        List<Object> avroList = ((List<Object>) relObj).stream()
            .map(o -> convertToAvroObject(o, getNonNullUnionSchema(schema).getElementType()))
            .collect(Collectors.toList());
        return avroList;
      case MAP:
        return ((Map<String, ?>) relObj).entrySet()
            .stream()
            .collect(Collectors.toMap(Map.Entry::getKey,
              e -> convertToAvroObject(e.getValue(), getNonNullUnionSchema(schema).getValueType())));
      case UNION:
        for (Schema unionSchema : schema.getTypes()) {
          if (isSchemaCompatibleWithRelObj(relObj, unionSchema)) {
            return convertToAvroObject(relObj, unionSchema);
          }
        }
        return null;
      case ENUM:
        return new GenericData.EnumSymbol(schema, (String) relObj);
      case FIXED:
        return new GenericData.Fixed(schema, ((ByteString) relObj).getBytes());
      case BYTES:
        return ByteBuffer.wrap(((ByteString) relObj).getBytes());
      default:
        return relObj;
    }
  }
It is possible that the destination Avro schema doesn't have all the fields that are projected from the
SQL. This is especially possible in SQL statements like
       insert into kafka.outputTopic select id, company from profile
where company is an avro record in itself whose schema can evolve. When this happens we will end up with
fields in the SamzaSQLRelRecord for company field which doesn't have equivalent fields in the outputTopic's schema
for company. To support this scenario where the input schemas and output schemas can evolve in their own cadence,
We ignore the fields which doesn't have corresponding schema in the output topic.
samza-sql/src/main/java/org/apache/samza/sql/avro/AvroRelConverter.java