anchor.diff:      id_to_index(VNodeID, RingSize);
   convert({SubpID, Bits}, subpartition_index) ->
       SubpID bsl Bits.
  +
  +%% @private
  +partition_to_preflist(Partition, NVal, Offset, UpNodes, CHBin) ->
  +    DocIdx = convert(Partition, partition_index),
  +    docidx_to_preflist(DocIdx, NVal, Offset, UpNodes, CHBin).
  +
  +%% @private
  +docidx_to_preflist(DocIdx, NVal, Offset, UpNodes, CHBin) ->
  +    %% `chashbin' would be fine with a straight integer, but
  +    %% `riak_core_apl' has a -spec that mandates binary(), so we'll
  +    %% play its game
  +    OrigPreflist =
  +        lists:map(fun({{Idx, Node}, primary}) -> {Idx, Node} end,
  +                  riak_core_apl:get_primary_apl_chbin(
  +                    <<DocIdx:160/integer>>, NVal, CHBin, UpNodes)),
  +    rotate_list(OrigPreflist, length(OrigPreflist), Offset).
  +
  +%% @private
  +rotate_list(List, Len, Offset) when Offset >= Len ->
  +    List;
  +rotate_list(List, _Len, Offset) ->
  +    {Head, Tail} = lists:split(Offset, List),
  +    Tail ++ Head.
  +
  +%% @private
  +index_to_id(Index, RingSize) ->
  +    RingIndexInc = chash:ring_increment(RingSize),
  +    Index div RingIndexInc.
  +
  +%% @private
  +id_to_index(Id, RingSize) when Id < RingSize->
  +    RingIndexInc = chash:ring_increment(RingSize),
  +    Id * RingIndexInc.
  +
  +%% @private
  +find_vnode_partitions(Index, N, RingSize) ->
  +    n_keyspaces({vnode_id, index_to_id(Index, RingSize), RingSize}, N).
  +
  +%% @private
  +partitions_by_index_or_filter(Idx, NVal, [], RingSize) ->
  +    find_vnode_partitions(Idx, NVal, RingSize);
  +partitions_by_index_or_filter(_Idx, _NVal, Filters, RingSize) ->
  +    %% Filters for traditional coverage plans are offset by 1, so when
  +    %% converting to partition indexes or IDs, we have to subtract 1
  +    lists:map(fun(P) ->
  +                      {partition_id,
  +                       add_offset(index_to_id(P, RingSize),
  +                                  -1, RingSize),
  +                       RingSize} end,
  +              Filters).
  +
  +
  +
  +%% @private
  +%% Must be able to comply with PVC if the target is 'all'
  +check_pvc(List, _PVC, allup) ->
  +    List;
  +check_pvc(List, PVC, all) ->
  +    check_pvc2(List, length(List), PVC).
  +
  +%% @private
  +check_pvc2(List, Len, PVC) when Len >= PVC ->
  +    List;
  +check_pvc2(_List, _Len, _PVC) ->
  +    [].
  +
  +%% @private
  +create_subpartition_plan(VNodeTarget, NVal, Count, PVC, ReqId, Service, CHBin, AvailNodeFun) ->
  +    MaskBSL = data_bits(Count),
  +
  +    %% Calculate an offset based on the request id to offer the
  +    %% possibility of different sets of VNodes being used even when
  +    %% all nodes are available.
  +    Offset = ReqId rem NVal,
  +
  +    RingSize = chashbin:num_partitions(CHBin),
  +
  +    UpNodes = AvailNodeFun(Service, CHBin),
  +    SubpList =
  +        lists:map(fun(SubpCounter) ->
  +                          SubpID = {SubpCounter, MaskBSL},
  +                          SubpIndex = convert(SubpID, subpartition_index),
  +                          PartID =
  +                              {partition_id,
  +                               chashbin:responsible_position(SubpIndex, CHBin),
  +                               RingSize},
  +
  +                          %% PVC is much like R; if the number of
  +                          %% available primary partitions won't reach
  +                          %% the specified PVC value, don't bother
  +                          %% including this partition at all. We can
  +                          %% decide later (based on all vs allup)
  +                          %% whether to return the successful
  +                          %% components of the coverage plan
  +                          {PartID, SubpID,
  +                           check_pvc(
  +                             docidx_to_preflist(SubpIndex, NVal, Offset, UpNodes, CHBin),
  +                             PVC, VNodeTarget)}
  +                  end,
  +                  lists:seq(0, Count - 1)),
  +
  +    %% Now we have a list of tuples; each subpartition maps to a
  +    %% partition ID, subpartition ID, and a list of zero or more
  +    %% {vnode_index, node} tuples for that partition
  +    maybe_create_subpartition_plan(VNodeTarget, SubpList, PVC).
  +
  +%% @private
  +maybe_create_subpartition_plan(allup, SubpList, PVC) ->
  +    map_subplist_to_plan(SubpList, PVC);
  +maybe_create_subpartition_plan(all, SubpList, PVC) ->
  +    maybe_create_subpartition_plan(all, lists:keyfind([], 3, SubpList), SubpList, PVC).
  +
  +%% @private
  +maybe_create_subpartition_plan(all, false, SubpList, PVC) ->
  +    map_subplist_to_plan(SubpList, PVC);
  +maybe_create_subpartition_plan(all, _, _SubpList, _PVC) ->
  +    {error, insufficient_vnodes_available}.
  +
  +%% @private
  +map_subplist_to_plan(SubpList, PVC) ->
  +    lists:flatten(
  +      lists:filtermap(fun({_PartID, _SubpID, []}) ->
  +                              false;
  +                         ({_PartID, SubpID, Vnodes}) ->
  +                              {true,
  +                               map_pvc_vnodes(Vnodes, SubpID, PVC)}
  +                    end, SubpList)).
  +
  +
  +%% @private
  +map_pvc_vnodes(Vnodes, SubpID, PVC) ->
  +    map_pvc_vnodes(Vnodes, SubpID, PVC, []).
  +
  +%% @private
  +map_pvc_vnodes(_Vnodes, _SubpID, 0, Accum) ->
  +    lists:reverse(Accum);
  +map_pvc_vnodes([{VnodeIdx, Node}|Tail], SubpID, PVC, Accum) ->
  +    map_pvc_vnodes(Tail, SubpID, PVC-1,
  +                   [{ VnodeIdx, Node, SubpID }] ++ Accum).
  +
  +
  +%% @private
  +%% Adding an offset while keeping the result inside [0, Top). Adding
  +%% Top to the left of `rem' allows offset to be negative without
  +%% violating the lower bound of zero
  +add_offset(Position, Offset, Top) ->
  +    (Position + (Top + Offset)) rem Top.
   
   %% @private
   increment_vnode({vnode_id, Position, RingSize}, Offset) ->
  
positive.code:             end
      end.
  
  %% ====================================================================
  %% Internal functions
  %% ====================================================================
  
  %% @private
  %% Adding an offset while keeping the result inside [0, Top). Adding
  %% Top to the left of `rem' allows offset to be negative without
  %% violating the lower bound of zero
  add_offset(Position, Offset, Top) ->
      (Position + (Top + Offset)) rem Top.
  
  %% @private
  %% Convert(from, to)
  %% We spend a lot of code mapping between data types, mostly
  %% integer-based. Consolidate that as much as possible here.
  -spec convert(partition_id() | vnode_id() | subpartition_id(), atom()) ->
  
negative.code:                               ReqId, DownNodes, Service, CHBin,
                                ?AVAIL_NODE_FUN).
  
  %% @doc Create a coverage plan with at least one slice per partition,
  %%      originally designed for parallel extraction of data via 2i.
  -spec create_subpartition_plan('all'|'allup',
                                 pos_integer(),
                                 pos_integer(),
                                 pos_integer(),
                                 req_id(), atom()) ->
                                        {error, term()} | subp_plan().
  create_subpartition_plan(VNodeTarget, NVal, Count, PVC, ReqId, Service) ->
      %% IMPORTANT: `Count' is assumed to be a power of 2 as determined
      %% by `riak_client'. Anything else will behave badly.
      {ok, ChashBin} = riak_core_ring_manager:get_chash_bin(),
      create_subpartition_plan(VNodeTarget, NVal, Count, PVC, ReqId, Service,
                               ChashBin, ?AVAIL_NODE_FUN).
  
  
  %% @doc Create a "mini" traditional coverage plan to replace
  %%      components of a previously-generated plan that are not useful
  %%      to the client because the node is unavailable
  -spec replace_subpartition_chunk(VnodeIdx :: index(),
                                   Node :: node(),
                                   {Mask :: non_neg_integer(),
  