anchor.diff:                  [StreamRef, OutputPid, OutputRef, Msg]),
               OutputPid ! []
       end,
  -ok.
  -
  +    ok.
  +
  +receive_catalog_query_results(StreamRef, OutputPid, OutputRef) ->
  +    receive
  +        {catalog_query, _ReqId, "$end_of_results", _, _, _, _} ->
  +            io:format("catalog_query_results -> $end_of_results~n", []),
  +            OutputPid ! {catalog_query_response, done, OutputRef};
  +        {catalog_query, StreamRef, Partition, Index, 
  +                        Field, Term, JSONProps} ->
  +            OutputPid ! {catalog_query_response, 
  +                            {Partition, Index, Field, Term, JSONProps},
  +                            OutputRef},
  +            receive_catalog_query_results(StreamRef, OutputPid, OutputRef);
  +        Msg ->
  +            io:format("receive_catalog_query_results: unknown message: ~p~n", [Msg]),
  +            receive_catalog_query_results(StreamRef, OutputPid, OutputRef)
  +    end,
  +    ok.
   
   %% @spec get(state(), BKey :: riak_object:bkey()) ->
   %%         {ok, Val :: binary()} | {error, Reason :: term()}
  
positive.code:             list_to_binary(StartTerm), 
              list_to_binary(EndTerm), 
              Partition),
          receive_info_range_results(StreamRef, OutputPid, OutputRef, []),
          raptor_conn:close(Conn)
      end),
      ok;
  
negative.code: list(_State) ->
      throw({error, not_supported}).
  
  %% @spec list_bucket(state(), riak_object:bucket()) ->
  %%           [riak_object:key()]
  %% @doc Get a list of the keys in a bucket
  list_bucket(_State, _Bucket) ->
      throw({error, not_supported}).    
  
  %% spawn a process to kick off the catalog listing
  %%   for the this partition, e.g.,
  fold(State, Fun0, Acc) ->
      Partition = integer_to_list(State#state.partition),
      {ok, Conn} = raptor_conn_sup:new_conn(),
      Me = self(), 
      CatalogResultsPid = spawn(fun() ->
          fold_catalog_process(Me, Fun0, Acc, false, 0, 0) end),
      spawn(fun() ->
          {ok, StreamRef} = raptor_conn:catalog_query(
              Conn,
              ["partition_id:\"", Partition , "\""]),
          receive_catalog_query_results(StreamRef, CatalogResultsPid, erlang:make_ref()),
          raptor_conn:close(Conn) end),
      receive_fold_results(Acc, 0).
  
  %% receive catalog entries for current partition & kick
  %%   off a stream process for each one in parallel    
  fold_catalog_process(FoldResultPid,
                       Fun0, Acc, 
                       CatalogDone,
                       StreamProcessCount,
                       FinishedStreamProcessCount) ->
      %% kick off one fold_stream_process per catalog entry that comes back
      %%   increment StreamProcessCount
      %% when receive done from fold_stream_process processes, increment
      %%   FinishedStreamProcessCount
      %% when CatalogDone = true && StreamProcessCount == FinishedStreamProcessCount,
      %%   exit
      Me = self(),
      receive
          {catalog_query_response, done, _OutputRef} ->
              case StreamProcessCount of
                  0 -> 
                      FoldResultPid ! {fold_result, done},
                      true;
                  _SPC -> 
                      fold_catalog_process(FoldResultPid, Fun0, Acc, true,
                          StreamProcessCount, FinishedStreamProcessCount)
              end;
                                   
          {catalog_query_response, {Partition, Index, Field, Term, _JSONProps}, _OutputRef} ->
              %% kick off stream for this PIFT
              spawn(fun() ->
                  {ok, Conn} = raptor_conn_sup:new_conn(),
                  {ok, StreamRef} = raptor_conn:stream(
                      Conn, 
                      list_to_binary(Index), 
                      list_to_binary(Field), 
                      list_to_binary(Term), 
                      <<"0">>, <<"0">>, <<"0">>, 
                      list_to_binary(Partition)),
                  fold_stream_process(Me, FoldResultPid, StreamRef, Fun0, Acc, Index, Field, Term),
                  raptor_conn:close(Conn) end),
              fold_catalog_process(FoldResultPid, Fun0, Acc, CatalogDone,
                                   StreamProcessCount+1, FinishedStreamProcessCount);
  
          {fold_stream, done, _StreamRef1} ->
              case FinishedStreamProcessCount >= (StreamProcessCount-1) of
                  true ->
                      case CatalogDone of
                          true ->
                              FoldResultPid ! {fold_result, done},
                              fold_catalog_process(FoldResultPid, Fun0, Acc, CatalogDone,
                                  StreamProcessCount, FinishedStreamProcessCount+1);
                          false ->
                              fold_catalog_process(FoldResultPid, Fun0, Acc, CatalogDone,
                                  StreamProcessCount, FinishedStreamProcessCount+1)
                      end;
                  _ -> 
                      fold_catalog_process(FoldResultPid, Fun0, Acc, CatalogDone,
                          StreamProcessCount, FinishedStreamProcessCount+1)
              end,
              fold_catalog_process(FoldResultPid, Fun0, Acc, CatalogDone,
                  StreamProcessCount, FinishedStreamProcessCount+1);
          Msg ->
              io:format("fold_catalog_process: unknown message: ~p~n", [Msg]),
              fold_catalog_process(FoldResultPid, Fun0, Acc, CatalogDone,
                  StreamProcessCount, FinishedStreamProcessCount)
      end.
  
  %% for each result of a stream process, package the entry in the
  %%   form of an infamously shady put-embedded command and forward
  %%   to receive_fold_results process
  fold_stream_process(CatalogProcessPid, FoldResultPid, StreamRef, Fun0, Acc, Index, Field, Term) ->
      receive
          {stream, StreamRef, "$end_of_table", _} ->
              CatalogProcessPid ! {fold_stream, done, StreamRef};
          {stream, StreamRef, Value, Props} ->
              Props2 = binary_to_term(Props),
              IndexBin = riak_search_utils:to_binary(Index),
              FieldTermBin = riak_search_utils:to_binary([Field, ".", Term]),
              Payload = {index, Index, Field, Term, 0, 0, Value, Props2, erlang:now()},
              BObj = term_to_binary(riak_object:new(IndexBin, FieldTermBin, Payload)),
              FoldResultPid ! {fold_result, Fun0({IndexBin, FieldTermBin}, BObj, Acc)},
              fold_stream_process(CatalogProcessPid, FoldResultPid, StreamRef, Fun0, Acc, Index, Field, Term);
          Msg ->
              io:format("fold_stream_process: unknown message: ~p~n", [Msg]),
              fold_stream_process(CatalogProcessPid, FoldResultPid, StreamRef, Fun0, Acc, Index, Field, Term)
      end.
  
  %% receive the Fun0(processed) objects from all the "buckets" on this partition, accumulate them
  %%   and return them
  receive_fold_results(Acc, Count) ->
      receive 
          {fold_result, done} ->
              Acc;
          {fold_result, Obj} ->
              %%io:format("receive_fold_results: Count = ~p~n", [Count]),
              receive_fold_results(Acc ++ [Obj], Count+1)
      end.
  
  %% test fold
  test_fold() ->
      Fun0 = fun(BKey, Obj, Acc) ->
          io:format("test_fold: Fun0: ~p, ~p~n", [BKey, length(Acc)]),
          Obj
      end,
      Partition = 0, %% 182687704666362864775460604089535377456991567872,
      State = #state { partition=Partition, conn=undefined },
      fold(State, Fun0, []).
  