anchor.diff:  -export([example/0, example_bucket/0, example_reduce/0,
            example_setup/0, example_setup/1]).
   
  +-include("riak_kv_vnode.hrl").
   -include_lib("riak_pipe/include/riak_pipe.hrl").
   -include_lib("riak_pipe/include/riak_pipe_log.hrl").
   
  
positive.code:               end
        end).
  
  %% @equiv send_inputs(Pipe, Inputs, 60000)
  send_inputs(Pipe, Inputs) ->
      send_inputs(Pipe, Inputs, ?DEFAULT_TIMEOUT).
  
negative.code:     riak_pipe:eoi(Pipe),
      ok;
  send_inputs(Pipe, Bucket, _Timeout) when is_binary(Bucket) ->
      %% start a new short pipe in order to use a coverage-enqueue
      %% that will trigger pipe workers to list keys from their twin
      %% kv vnodes, and then enqueue those keys in the MR pipe
      [{_Name, Head}|_] = Pipe#pipe.fittings,
      {ok, LKP} = riak_pipe:exec([#fitting_spec{name=listkeys,
                                                module=riak_pipe_w_xform,
                                                arg=fun kvlistkeys_xform/3,
                                                nval=1}],
                                 [{sink, Head}]),
      ReqId = erlang:phash2(erlang:now()), %% stolen from riak_client
      BucketProps = riak_core_bucket:get_bucket(Bucket),
      NVal = proplists:get_value(n_val, BucketProps),
      {ok, Sender} = riak_pipe_qcover_sup:start_qcover_fsm(
                       [{raw, ReqId, self()},
                        [LKP, {Bucket, []}, NVal]]),
      erlang:link(Sender),
      receive
          {ReqId, done} ->
              riak_pipe:eoi(LKP),
              ok;
          {ReqId, Error} ->
              Error
      end;
  send_inputs(Pipe, {Bucket, FilterExprs}, _Timeout) ->
      {ok, C} = riak:local_client(),
      case C:stream_list_keys({Bucket, FilterExprs}) of
  