anchor.diff:      comm:send_local(DHTNode, {bulkowner, reply, Id, Target, Msg, Parents}).
   
   -spec issue_bulk_distribute(ID::uid:global_uid(), atom(), pos_integer(),
  -                           comm:message(), [tuple()]) -> ok.
  +                           comm:message(), [{string(), term()}] | {ets, ets:tab()}) -> ok.
   issue_bulk_distribute(Id, Proc, Pos, Msg, Data) ->
  +    issue_bulk_distribute(Id, Proc, Pos, Msg, Data, intervals:all()).
  +
  +-spec issue_bulk_distribute(ID::uid:global_uid(), atom(), pos_integer(),
  +                           comm:message(), [{string(), term()}] | {ets, ets:tab()},
  +                           intervals:interval()) -> ok.
  +issue_bulk_distribute(Id, Proc, Pos, Msg, Data, Interval) ->
  +    ?TRACE("~p ~p: bulk_distribute initiated to ~p", [self(), ?MODULE,
  +                                                      Interval]),
       DHTNode = pid_groups:find_a(dht_node),
  -    comm:send_local(DHTNode, {bulkowner, start, Id, intervals:all(), {bulk_distribute, Proc,
  +    comm:send_local(DHTNode, {bulkowner, start, Id, Interval, {bulk_distribute, Proc,
                                                           Pos, Msg, Data}}).
   
   -spec send_reply(Id::uid:global_uid(), Target::comm:mypid(), Msg::comm:message(),
  
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:             false ->
                  case Msg of
                      {bulk_distribute, Proc, N, Env, 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});
  