anchor.diff:  
   -record(state, {
             name                                :: atom(),
  -          ddl                                 :: undefined | #ddl_v1{},
             qry           = none                :: none | #riak_sql_v1{},
             qid           = undefined           :: undefined | {node(), non_neg_integer()},
             sub_qrys      = []                  :: [integer()],
  
positive.code: 
  %% @private
  -spec handle_info(term(), #state{}) -> {noreply, #state{}}.
  handle_info(pop_next_query, State) ->
      {query, ReceiverPid, QId, Qry, DDL} = riak_kv_qry_queue:blocking_pop(),
      Request = {execute, {QId, Qry, DDL}},
      case handle_call(Request, ReceiverPid, State) of
           {reply, ok, NewState}                -> {noreply, NewState};
           {reply, {error, _} = Error, NewState} -> ReceiverPid ! Error,
                                                {noreply, new_state(NewState#state.name)}
      end;
  handle_info({{SubQId, QId}, done},
              State = #state{qid          = QId,
                             receiver_pid = ReceiverPid,
                             result       = IndexedChunks,
                             sub_qrys     = SubQQ}) ->
      lager:debug("Received done on QId ~p (~p); SubQQ: ~p", [QId, SubQId, SubQQ]),
      case SubQQ of
          [] ->
              lager:debug("Done collecting on QId ~p (~p): ~p", [QId, SubQId, IndexedChunks]),
              %% sort by index, to reassemble according to coverage plan
              {_, R2} = lists:unzip(lists:sort(IndexedChunks)),
              Results = lists:append(R2),
              % send the results to the waiting client process
              ReceiverPid ! {ok, Results},
              pop_next_query(),
              %% drop indexes, serialize
              {noreply, new_state(State#state.name)};
          _MoreSubQueriesNotDone ->
              {noreply, State}
      end;
  
  handle_info({{SubQId, QId}, {results, Chunk}},
              State = #state{qid      = QId,
                             qry      = Qry,
                             result   = IndexedChunks,
                             sub_qrys = SubQs}) ->
      #riak_sql_v1{'SELECT' = SelectSpec} = Qry,
      NewS = case lists:member(SubQId, SubQs) of
                 true ->
                     Decoded = decode_results(lists:flatten(Chunk), SelectSpec),
                     lager:debug("Got chunk on QId ~p (~p); SubQQ: ~p", [QId, SubQId, SubQs]),
                     NSubQ = lists:delete(SubQId, SubQs),
                     State#state{status   = accumulating_chunks,
                                 result   = [{SubQId, Decoded} | IndexedChunks],
                                 sub_qrys = NSubQ};
                 false ->
                     %% discard;
                     %% Don't touch state as it may have already 'finished'.
                     State
             end,
      {noreply, NewS};
  
  handle_info({{SubQId, QId}, {error, Reason} = Error},
              State = #state{receiver_pid = ReceiverPid,
                             qid    = QId,
  
negative.code:     pop_next_query(),
      {ok, new_state(RegisteredName)}.
  
  handle_call(_, _, State) ->
      {reply, {error, not_handled}, State}.
  
  -spec handle_cast(term(), #state{}) -> {noreply, #state{}}.
  %% @private
  