anchor.diff:                     module=riak_pipe_w_pass,
                      chashfun=follow}];
   mr2pipe_phases(Query) ->
  +    Now = now(),
       Numbered = lists:zip(Query, lists:seq(0, length(Query)-1)),
  -    Fittings0 = lists:flatten([mr2pipe_phase(P,I) || {P,I} <- Numbered]),
  +    Fittings0 = lists:flatten([mr2pipe_phase(P,I,Now) || {P,I} <- Numbered]),
       Fs = fix_final_fitting(Fittings0),
       case lists:last(Query) of
           {_, _, _, false} ->
  
positive.code:             {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))).
  
  bkey_nval({Bucket, _Key}) ->
      BucketProps = riak_core_bucket:get_bucket(Bucket),
  
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,
  