anchor.diff:      pop_next_query(),
       {ok, new_state(RegisteredName)}.
   
  -handle_call({execute, {QId, [Qry|_] = SubQueries, DDL}}, ReceiverPid, 
  -            State = #state{status        = void,
  -                           run_sub_qs_fn = RunSubQs}) ->
  -    Indices = lists:seq(1, length(SubQueries)),
  -    ZQueries = lists:zip(Indices, SubQueries),
  -    SubQs = [{{qry, Q}, {qid, {I, QId}}} || {I, Q} <- ZQueries],
  -    ok = RunSubQs(SubQs),
  -    {reply, ok, State#state{qid          = QId,
  -                            receiver_pid = ReceiverPid,
  -                            qry          = Qry,
  -                            ddl          = DDL,
  -                            sub_qrys     = Indices}};
  -handle_call({execute, {QId, _, _}}, _, State = #state{status = Status}) 
  -  when Status =/= void ->
  -    lager:error("Qry queue manager should have cleared the status before assigning new query ~p", [QId]),
  -    {reply, {error, mismanagement}, State}.
  +handle_call(_, _, State) ->
  +    {reply, {error, not_handled}, State}.
   
   -spec handle_cast(term(), #state{}) -> {noreply, #state{}}.
   %% @private
  
positive.code: 
  -record(state, {
            name                                :: atom(),
            ddl                                 :: undefined | #ddl_v1{},
            qry           = none                :: none | #riak_sql_v1{},
            qid           = undefined           :: undefined | {node(), non_neg_integer()},
            sub_qrys      = []                  :: [integer()],
  
negative.code: pop_next_query() ->
      self() ! pop_next_query.
  
  %% 
  execute_query(ReceiverPid, QId, [Qry|_] = SubQueries,
                #state{ run_sub_qs_fn = RunSubQs } = State) ->
      Indices = lists:seq(1, length(SubQueries)),
      ZQueries = lists:zip(Indices, SubQueries),
      SubQs = [{{qry, Q}, {qid, {I, QId}}} || {I, Q} <- ZQueries],
      ok = RunSubQs(SubQs),
      {ok, State#state{qid          = QId,
                       receiver_pid = ReceiverPid,
                       qry          = Qry,
                       sub_qrys     = Indices}}.
  
  %%
  add_subquery_result(SubQId, Chunk, 
                      #state{qry      = Qry,
                             result   = IndexedChunks,
                             sub_qrys = SubQs} = State) ->
      #riak_sql_v1{'SELECT' = SelectSpec} = Qry,
      case lists:member(SubQId, SubQs) of
          true ->
              Decoded = decode_results(lists:flatten(Chunk), SelectSpec),
              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.
  
  subqueries_done(SubQId, QId,
                  #state{qid          = QId,
                         receiver_pid = ReceiverPid,
                         result       = IndexedChunks,
                         sub_qrys     = SubQQ} = State) ->
      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
              new_state(State#state.name);
          _MoreSubQueriesNotDone ->
              State
      end.
  
  %%%===================================================================
  %%% Unit tests
  %%%===================================================================
  