anchor.diff:      from_binary/1,
       to_binary/1
   ]).
  +
  +mapred_search(Index, SearchString, Query, ResultTransformer, Timeout) ->
  +    Me = self(),
  +    {ok,MR_ReqId} = mapred_search_stream(Index, SearchString, Query, Me, ResultTransformer, Timeout),
  +    Results = luke_flow:collect_output(MR_ReqId, Timeout),
  +    Results.
  +
  +mapred_search_stream(Index, SearchString, Query, ClientPid, ResultTransformer, Timeout) ->
  +    %% NOTE: Not proud of this code, as it creates a circular dependency on the 
  +    %% Parse the query...
  +    case parse_query(Index, SearchString) of
  +        {ok, Ops} ->
  +            QueryOps = Ops;
  +        {error, ParseError} ->
  +            M = "Error running query '~s': ~p~n",
  +            error_logger:error_msg(M, [SearchString, ParseError]),
  +            throw({search_stream, SearchString, ParseError}),
  +            QueryOps = undefined % Make compiler happy.
  +    end,
  +
  +    %% Set up the mapreduce job...
  +    case RiakClient:mapred_stream(Query, ClientPid, ResultTransformer, Timeout) of
  +        {ok, {ReqId, FlowPid}} ->
  +            %% Perform a search, funnel results to the mapred job...
  +            F = fun(Results, Acc) ->
  +                %% Make the list of BKeys...
  +                BKeys = [{Index, DocID} || {DocID, _Props} <- Results],
  +                luke_flow:add_inputs(FlowPid, BKeys),
  +                Acc
  +            end,
  +            ok = search_fold(Index, QueryOps, F, ok, Timeout),
  +            luke_flow:finish_inputs(FlowPid),
  +            {ok, ReqId};
  +        Error ->
  +            Error
  +    end.
   
   
   %% Parse the provided query. Returns either {ok, QueryOps} or {error,
  
positive.code:         id=Ref, termcount=NumTerms,
          inputcount=NumInputs, querynorm=QueryNorm }.
  
  %% Gather all results from the provided SearchRef, return the list of
  %% results sorted in descending order by score.
  collect_results(SearchRef, Timeout, Acc) ->
      M = collect_result(SearchRef, Timeout),
      case M of
          {done, _} ->
              sort_by_score(SearchRef, Acc);
          {[], Ref} ->
              collect_results(Ref, Timeout, Acc);
          {Results, Ref} ->
              collect_results(Ref, Timeout, Acc ++ Results)
          %% Dialyzer says this clause is impossible
          % Error ->
          %     Error
      end.
  
  %% Collect one or more individual results in as non-blocking of a
  
negative.code: -define(OPTIMIZER_PROC_CT, 32).
  
  -export([
      %% MapReduce Searching...
      mapred_search/5, mapred_search_stream/6,
  
      %% Searching...
      parse_query/2,
      search/5,
      search_fold/5,
      search_doc/5,
  
      %% Explain...
      explain/2,
  