anchor.diff:                           <- riak_core_apl:offline_owners(Service, CHBin)],
       [(DownVNode div PartitionSize) || DownVNode <- DownVNodes].
   
  -
  -%% @private
  -find_coverage(_AllKeySpaces, _Offset, _NVal, _PartitionCount,
  +%% @private
  +merge_coverage_results({VnodeId, PartitionIds}, Acc) ->
  +    case proplists:get_value(VnodeId, Acc) of
  +        undefined ->
  +            [{VnodeId, PartitionIds} | Acc];
  +        MorePartitionIds ->
  +            UniqueValues =
  +                lists:usort(PartitionIds ++ MorePartitionIds),
  +            [{VnodeId, UniqueValues} |
  +             proplists:delete(VnodeId, Acc)]
  +    end.
  +
  +
  +%% @private
  +%% @doc Generates a minimal set of vnodes and partitions to find the requested data
  +-spec find_minimal_coverage(list(partition_id()), non_neg_integer(), non_neg_integer(), non_neg_integer(), list(partition_id()), non_neg_integer(), list({vnode_id(), list(partition_id())})) -> {ok, list({vnode_id(), list(partition_id())})} | {error, term()}.
  +find_minimal_coverage(_AllKeySpaces, _Offset, _NVal, _PartitionCount,
                 _UnavailableKeySpaces, 0, Results) ->
       {ok, Results};
  -find_coverage(AllKeySpaces, Offset, NVal, PartitionCount, UnavailableKeySpaces, PVC, []) ->
  -    %% Calculate the available keyspaces.
  -    AvailableKeySpaces = [{((VNode+Offset) rem PartitionCount),
  -                           VNode,
  -                           n_keyspaces(VNode, NVal, PartitionCount)}
  -                          || VNode <- (AllKeySpaces -- UnavailableKeySpaces)],
  -    case find_coverage_vnodes(
  -           ordsets:from_list(AllKeySpaces),
  -           AvailableKeySpaces,
  -           []) of
  -        {ok, CoverageResults} ->
  -            find_coverage(AllKeySpaces,
  -                          Offset,
  -                          NVal,
  -                          PartitionCount,
  -                          UnavailableKeySpaces,
  -                          PVC-1,
  -                          CoverageResults);
  -        Error ->
  -            Error
  -    end;
  -find_coverage(AllKeySpaces,
  -              Offset,
  -              NVal,
  -              PartitionCount,
  -              UnavailableKeySpaces,
  -              PVC,
  -              ResultsAcc) ->
  +find_minimal_coverage(AllKeySpaces,
  +                      Offset,
  +                      NVal,
  +                      PartitionCount,
  +                      UnavailableKeySpaces,
  +                      PVC,
  +                      ResultsAcc) ->
       %% Calculate the available keyspaces. The list of
       %% keyspaces for each vnode that have already been
       %% covered by the plan are subtracted from the complete
  
positive.code:                               AvailableKeySpaces,
                                ResultsAcc) of
          {ok, CoverageResults} ->
              UpdateResultsFun =
                  fun({Key, NewValues}, Results) ->
                          case proplists:get_value(Key, Results) of
                              undefined ->
                                  [{Key, NewValues} | Results];
                              Values ->
                                  UniqueValues = lists:usort(Values ++ NewValues),
                                  [{Key, UniqueValues} |
                                   proplists:delete(Key, Results)]
                          end
                  end,
              UpdatedResults =
                  lists:foldl(UpdateResultsFun, ResultsAcc, CoverageResults),
              find_coverage(AllKeySpaces,
                            Offset,
                            NVal,
                            PartitionCount,
                            UnavailableKeySpaces,
                            PVC-1,
                            UpdatedResults);
          Error ->
              Error
      end.
  
  %% @private
  %% @doc Find the N key spaces for a VNode
  n_keyspaces(VNode, N, PartitionCount) ->
      ordsets:from_list([X rem PartitionCount ||
                            X <- lists:seq(PartitionCount + VNode - N,
  
negative.code:                       [{2, 0, ?SET([6, 7, 0, 1, 2])},
                         {3, 7, ?SET([6, 7, 0, 1, 2])}])).
  
  create_plan_test_() ->
      {setup,
       fun cpsetup/0,
       fun cpteardown/1,
       fun test_create_plan/1}.
  
  cpsetup() ->
      meck:new(riak_core_node_watcher, []),
      meck:expect(riak_core_node_watcher, nodes, 1, [mynode]),
      CHash = chash:fresh(8, mynode),
      chashbin:create(CHash).
  
  cpteardown(_) ->
      meck:unload().
  
  test_create_plan(CHBin) ->
      Plan =
          {[{1278813932664540053428224228626747642198940975104,
             mynode},
            {730750818665451459101842416358141509827966271488,
             mynode},
            {365375409332725729550921208179070754913983135744,
             mynode}],
           [{730750818665451459101842416358141509827966271488,
             [548063113999088594326381812268606132370974703616,
              730750818665451459101842416358141509827966271488]}]},
      [?_assertEqual(Plan,
                     create_plan(all, 3, 1, 1234, riak_kv, CHBin))].
  
  -endif.
  