anchor.diff:              {ok, Output}
       end.
   
  -reduce_local_chashfun(_) ->
  -    {ok, Ring} = riak_core_ring_manager:get_my_ring(),
  -    riak_pipe_vnode:hash_for_partition(
  -      hd(riak_core_ring:my_indices(Ring))).
  +%% chashfun for 0 or 1 reducer per node.
  +%% reduce_local_chashfun(_) ->
  +%%     {ok, Ring} = riak_core_ring_manager:get_my_ring(),
  +%%     riak_pipe_vnode:hash_for_partition(
  +%%       hd(riak_core_ring:my_indices(Ring))).
   
   bkey_nval({Bucket, _Key}) ->
       BucketProps = riak_core_bucket:get_bucket(Bucket),
  
positive.code:                    module=riak_pipe_w_pass,
                     chashfun=follow}];
  mr2pipe_phases(Query) ->
      Numbered = lists:zip(Query, lists:seq(0, length(Query)-1)),
      Fittings0 = lists:flatten([mr2pipe_phase(P,I) || {P,I} <- Numbered]),
      Fs = fix_final_fitting(Fittings0),
      case lists:last(Query) of
          {_, _, _, false} ->
  
negative.code:                      chashfun=follow}
         ||Keep]].
  
  reduce2pipe(FunSpec, Arg, Keep, I, ConstHashCookie) ->
      ConstantFun = fun(_) -> chash:key_of(ConstHashCookie) end,
      [#fitting_spec{name={reduce,I},
                     module=riak_kv_w_mapred,
                     arg=reduce_compat(FunSpec, Arg),
                     chashfun=ConstantFun}
       |[#fitting_spec{name=I,
                       module=riak_pipe_w_tee,
                       arg=sink,
  