anchor.diff:                          %%     [Partition, Index, Field, Term, JSONProps]),
                           {ok, Conn} = riak_sock_pool:checkout(?CONN_POOL),
                           try
  -                          {ok, StreamRef} = raptor_conn:stream(
  -                              Conn,
  -                              to_binary(Index),
  -                              to_binary(Field),
  -                              to_binary(Term),
  -                              to_binary(Partition)),
  -                          fold_stream_process(Me, FoldResultPid, StreamRef, Fun0, Acc, 
  -                                              Index, Field, Term)
  +                            IndexBin = to_binary(Index),
  +                            FieldBin = to_binary(Field),
  +                            TermBin = to_binary(Term),
  +                            {ok, StreamRef} = raptor_conn:stream(
  +                                                Conn,
  +                                                IndexBin,
  +                                                FieldBin,
  +                                                TermBin,
  +                                                to_binary(Partition)),
  +                          fold_stream_process(Me, FoldResultPid, StreamRef,
  +                                              IndexBin, FieldBin, TermBin)
                           after 
                               riak_sock_pool:checkin(?CONN_POOL, Conn)
                           end end),
  -                    fold_catalog_process(CatRef, FoldResultPid, Fun0, Acc, CatalogDone,
  +                    fold_catalog_process(CatRef, FoldResultPid, CatalogDone,
                                            StreamProcessCount+1, FinishedStreamProcessCount, false)
                   end;
   
  
positive.code: %%   off a stream process for each one in parallel
  fold_catalog_process(CatRef,
                       FoldResultPid,
                       Fun0, Acc,
                       CatalogDone,
                       StreamProcessCount,
                       FinishedStreamProcessCount,
  
negative.code:             CatalogProcessPid ! {fold_stream, done, StreamRef};
          {stream, StreamRef, Value, Props}=_Msg2 ->
              case Props of
                  <<>> ->
                      Props2 = [];
                  _ ->
                      Props2 = binary_to_term(Props)
              end,
              FieldTermBin = <<FieldBin/binary, ".", TermBin/binary>>,
              BObj = term_to_binary({FieldBin, TermBin, Value, Props2}),
              FoldResultPid ! {fold_result, IndexBin, FieldTermBin, BObj},
              fold_stream_process(CatalogProcessPid, FoldResultPid, StreamRef, 
                                  IndexBin, FieldBin, TermBin);
          _Msg ->
              %% io:format("fold_stream_process: unknown message: ~p~n", [_Msg]),
              fold_stream_process(CatalogProcessPid, FoldResultPid, StreamRef, 
                                  IndexBin, FieldBin, TermBin)
          after ?FOLD_TIMEOUT ->
              CatalogProcessPid ! {fold_stream, done, StreamRef},
              error_logger:warning_msg("fold_stream_process: table timed out: ~p.~p.~p~n",
                                        [IndexBin, FieldBin, TermBin])
      end.
  
  %% receive the Fun0(processed) objects from all the "buckets" on this partition, accumulate them
  %%   and return them
  receive_fold_results(Fun, Acc) ->
      receive
          {fold_result, done} ->
              %% io:format("receive_fold_results: fold complete [~p objects].~n",
              %%           [Count]),
              Acc;
          {fold_result, IndexBin, FieldTermBin, BObj} ->
              receive_fold_results(Fun, Fun({IndexBin,FieldTermBin},BObj,Acc))
      after ?FOLD_TIMEOUT ->
              throw({timeout,Acc})
      end.
  
  %%%
  