anchor.diff:           [ riak_pipe:fitting_spec() ].
   reduce2pipe(FunSpec, Arg, Keep, I, ConstHashCookie) ->
       Hash = chash:key_of(ConstHashCookie),
  -    ConstantFun = fun(_) -> Hash end,
       [#fitting_spec{name={reduce,I},
                      module=riak_kv_w_reduce,
                      arg={rct,
                           riak_kv_w_reduce:reduce_compat(FunSpec),
                           Arg},
  -                   chashfun=ConstantFun}
  +                   chashfun=Hash}
        |[#fitting_spec{name=I,
                        module=riak_pipe_w_tee,
                        arg=sink,
  
positive.code: query_type(Idx, QueryT) ->
      element(1, element(Idx, QueryT)).
  
  -spec query_arg(integer(), tuple()) -> term().
  query_arg(Idx, QueryT) ->
      element(3, element(Idx, QueryT)).
  
  %% @doc Convert a reduce phase to its equivalent pipe fittings.
  %%
  %% Reduce converts to:
  
negative.code:                            X)) ||
          X <- lists:seq(1, Num)],
      ok.
  
  %% @doc For Riak 1.0 compatibility, provide a translation from old
  %% anonymous functions to new ones.  This function should have a
  %% limited-use lifetime: it will only be evaluated while a cluster is
  %% in the middle of a rolling-upgrade from 1.0.x to 1.1.
  %%
  %% Yes, the return value is a new anonymous function.  This shouldn't
  %% be a problem with a future upgrade, though, as no one should be
  %% running a cluster that includes three Riak versions.  Therefor, the
  %% node the spread this old Fun around the cluster should have been
  %% stopped, along with the pipe defined by the old fun before this new
  %% fun would itself be considered old.
  compat_fun(Fun) ->
      {uniq, Uniq} = erlang:fun_info(Fun, uniq),
      {index, I} = erlang:fun_info(Fun, index),
      compat_fun(Uniq, I, Fun).
  
  %% Riak 1.0.1 and 1.0.2 funs
  compat_fun(120571329, 1, _Fun) ->
      %% chash used for kv_get in map
      {ok, fun riak_kv_pipe_get:bkey_chash/1};
  compat_fun(112900629, 2, _Fun) ->
      %% nval used for kv_get in map
      {ok, fun riak_kv_pipe_get:bkey_nval/1};
  compat_fun(19126064, 3, Fun) ->
      %% constant chash used for reduce
      {env, [Hash]} = erlang:fun_info(Fun, env),
      {ok, fun(_) -> Hash end};
  compat_fun(29992360, 4, _Fun) ->
      %% chash used for kv_get in link
      {ok, fun riak_kv_pipe_get:bkey_chash/1};
  compat_fun(22321692, 5, _Fun) ->
      %% nval used for kv_get in link
      {ok, fun riak_kv_pipe_get:bkey_nval/1};
  compat_fun(66856669, 6, Fun) ->
      %% link extraction function
      %% Yes, the env really does have bucket and tag the reverse of the spec
      {env, [Tag, Bucket]} = erlang:fun_info(Fun, env),
      {ok, fun({ok, Input, _Keydata}, Partition, FittingDetails) ->
                   Results = riak_kv_mrc_map:link_phase(
                               Input, undefined, {Bucket, Tag}),
                   [ riak_pipe_vnode_worker:send_output(
                       R, Partition, FittingDetails)
                     || R <- Results ],
                   ok;
              ({{error, _},_,_}, _, _) ->
                   ok
           end};
  
  %% dunno
  compat_fun(_, _, _) ->
      error.
  