anchor.diff:              false ->
                   case Msg of
                       {bulk_distribute, Proc, N, Env, Data} ->
  -                        {RangeData, _Rest} = lists:partition(
  -                                fun(Entry) ->
  -                                    intervals:in(?RT:hash_key(element(1, Entry)),
  -                                                Range)
  -                                end, Data),
  +                        RangeData = get_range_data(Data, Range),
  +                        ?TRACE("~p bulk_distribute: sending to node ~p on interval ~p
  +                               Msg: ~p",
  +                               [self(), node:pidX(Head), Range, {bulkowner, Id, Range,
  +                              {bulk_distribute, Proc, N, Env, RangeData}, Parents}]),
                           comm:send(node:pidX(Head),
                                 {bulkowner, Id, Range,
                                 {bulk_distribute, Proc, N, Env, RangeData}, Parents});
  
positive.code:     MyDBRange = intervals:union(dht_node_state:get(State, my_range), DBRange),
      MyRange0 = lists:foldl(F, Range, MsgFwd),
      MyRange = intervals:intersection(MyRange0, MyDBRange),
      case intervals:is_empty(MyRange) of
          true -> ok;
          _ ->
              case Msg of
                  {bulk_read_entry, Issuer} ->
                      Data = db_dht:get_entries(dht_node_state:get(State, db), MyRange),
                      ReplyMsg = {bulk_read_entry_response, MyRange, Data},
                      % for aggregation using a tree, activate this instead:
                      % issue_send_reply(Id, Issuer, ReplyMsg, Parents);
                      comm:send(Issuer, {bulkowner, reply, Id, ReplyMsg});
                  {bulk_distribute, Proc, N, Msg1, Data} ->
                      RangeData =
                          [Entry || Entry <- Data,
                                    intervals:in(?RT:hash_key(element(1, Entry)),
                                                 MyRange)],
                      %% only deliver data in MyRange as data outside of it was
                      %% forwarded
                      comm:send_local(pid_groups:get_my(Proc),
                                      {bulk_distribute, Id, MyRange, 
                                       setelement(N, Msg1, RangeData), Parents});
                  {?send_to_group_member, Proc, Msg1} when Proc =/= dht_node ->
                      comm:send_local(pid_groups:get_my(Proc),
                                      {bulkowner, deliver, Id, MyRange, Msg1, Parents});
                  {do_snapshot, _SnapNo, _Leader} ->
                      comm:send_local(pid_groups:get_my(dht_node), Msg);
                  MrMsg when mr =:= element(1, MrMsg) ->
                      comm:send_local(pid_groups:get_my(dht_node), Msg)
              end
      end,
      RestRange = intervals:minus(MyRange0, MyRange),
  
negative.code: on({send_error, FailedTarget, {bulkowner, reply, Id, Target, Msg, Parents}, _Reason}, State) ->
      bulkowner:send_reply_failed(Id, Target, Msg, Parents, self(), FailedTarget),
      State.
  
  -spec get_range_data({ets, db_ets:db()} | [{nonempty_string(), term()},...],
                       intervals:interval()) -> [{nonempty_string(), term()}].
  get_range_data({ets, ETS}, Range) ->
      lists:foldl(fun(Interval, Acc1) ->
                      db_ets:foldl(ETS,
                                   fun(E, Acc) -> [db_ets:get(ETS,E) | Acc] end,
                                   Acc1, Interval)
                  end,
                  [], intervals:get_simple_intervals(Range));
  get_range_data(Data, Range) ->
      lists:filter(
        fun(Entry) ->
                intervals:in(element(1, Entry),
                             Range)
        end, Data).
  