anchor.diff:  
   %% Common code for sending a single bucket or multiple inputs map/request
   %% @private
  -send_mapred_req(Pid, MapRed, ClientPid, _CallTimeout) ->
  +send_mapred_req(Pid, MapRed, ClientPid) ->
       ReqMsg = #rpbmapredreq{request = encode_mapred_req(MapRed),
                              content_type = <<"application/x-erlang-binary">>},
       ReqId = mk_reqid(),
  
positive.code: %%      The ClientPid will receive messages in this format:
  %% ```  {ReqId::req_id(), {mapred, Phase::non_neg_integer(), mapred_result()}}
  %%      {ReqId::req_id(), done}'''
  -spec mapred_bucket_stream(ConnectionPid::pid(), bucket(), [mapred_queryterm()], ClientPid::pid(), timeout(), timeout()) ->
                                    {ok, req_id()} | {error, term()}.
  mapred_bucket_stream(Pid, Bucket, Query, ClientPid, Timeout, CallTimeout) ->
      MapRed = [{'inputs', Bucket},
                {'query', Query},
                {'timeout', Timeout}],
      send_mapred_req(Pid, MapRed, ClientPid, CallTimeout).
  
  
  %% @doc Execute a search query. This command will return an error
  
negative.code: mapred_stream(Pid, {index,Bucket,Name,StartKey,EndKey}, Query, ClientPid, Timeout, CallTimeout) when is_binary(Name) andalso is_integer(EndKey) ->
      BinEndKey = list_to_binary(integer_to_list(EndKey)),
      mapred_stream(Pid, {index,Bucket,Name,StartKey,BinEndKey}, Query, ClientPid, Timeout, CallTimeout);
  mapred_stream(Pid, Inputs, Query, ClientPid, Timeout, _CallTimeout) ->
      MapRed = [{'inputs', Inputs},
                {'query', Query},
                {'timeout', Timeout}],
      send_mapred_req(Pid, MapRed, ClientPid).
  
  %% @doc Perform a MapReduce job against a bucket across the cluster.
  %%      See the MapReduce documentation for explanation of behavior.
  