anchor.diff:  -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,
  -
  -    %% Stream Searching...
  -    stream_search/2,
  -    collect_result/2,
   
       %% Explain...
       explain/2,
  
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:     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,
  