anchor.diff:  %% extract keydata from the input
   erlang_keydata({_OkError, _Input, KeyData}) -> KeyData.
   
  +%% @doc Evaluate Javascript map functions ... if the input is ok.
   map_js(_JS, _Arg, {{error, notfound}, {Bucket, Key}, KeyData}) ->
       {ok, [{not_found,
              {Bucket, Key},
  
positive.code:         
  %% @doc Evaluate the map function.
  -spec map(riak_kv_mrc_pipe:map_query_fun(), term(), term())
           -> {ok, [term()]} | {forward_preflist, Reason :: term()}.
  map({modfun, Module, Function}, Arg, Input0) ->
      Input = erlang_input(Input0),
      KeyData = erlang_keydata(Input0),
  
negative.code:             {error, Error}
      end.
  
  %% @doc Perform any one-time initialization of the phase that's
  %% possible/needed.  This currently includes looking up functions from
  %% Riak KV (for `{jsanon, {Bucket, Key}}' and `{strfun, {Bucket,
  %% Key}}', as well as compiling `strfun' specs to Erlang functions.
  %%
  %% <ul>
  %%   <li>`{jsanon, {Bucket, Key}}' is converted to `{jsanon, Source}'</li>
  %%
  %%   <li>`{strfun, {Bucket, Key}}' and `{strfun, Source}' are both
  %%   converted to `{qfun, Fun}' after compiling</li>
  %% </ul>
  -spec init_phase(PhaseSpec :: term()) ->
           {ok, PhaseSpec :: term()} | {error, Reason :: term()}.
  init_phase({Anon, {Bucket, Key}})
    when Anon =:= jsanon; Anon =:= strfun ->
      %% lookup source for stored functions only at fitting worker startup
      {ok, C} = riak:local_client(),
      case C:get(Bucket, Key, 1) of
          {ok, Object} ->
  