anchor.diff:                             qry      = Qry,
                              result   = IndexedChunks,
                              sub_qrys = SubQs}) ->
  -    #riak_sql_v1{'SELECT' = SelectSpec} = Qry,
  +    #riak_sql_v1{'SELECT' = {plain_row_select, SelectSpec}} = Qry,
       NewS = case lists:member(SubQId, SubQs) of
                  true ->
                      Decoded = decode_results(lists:flatten(Chunk), SelectSpec),
  
positive.code: new_state(RegisteredName) ->
      #state{name = RegisteredName}.
  
  -spec handle_req({atom(), term()}, #state{}) ->
                          {ok | {ok | error, term()},
                           list(), #state{}}.
  handle_req({execute, {QId, [Qry|_] = SubQueries, DDL}}, State = #state{status = void}) ->
      %% TODO make this run with multiple sub-queries
      %% limit sub-queries and throw error
      Indices = lists:seq(1, length(SubQueries)),
      ZQueries = lists:zip(Indices, SubQueries),
      SEs = [{run_sub_query, {qry, Q}, {qid, {I, QId}}} || {I, Q} <- ZQueries],
      {ok, SEs, State#state{qid = QId, qry = Qry, ddl = DDL, sub_qrys = Indices}};
  
  handle_req({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]),
      {{error, mismanagement}, [], State};
  
  handle_req(_Request, State) ->
      {ok, ?NO_SIDEEFFECTS, State}.
  
  handle_side_effects([]) ->
      ok;
  handle_side_effects([{run_sub_query, {qry, Q}, {qid, QId}} | T]) ->
      Table = Q#riak_sql_v1.'FROM',
      Bucket = riak_kv_pb_timeseries:table_to_bucket(Table),
      %% fix these up too
      Timeout = {timeout, 10000},
      Me = self(),
      CoverageFn = {colocated, riak_kv_qry_coverage_plan},
      {ok, _PID} = riak_kv_index_fsm_sup:start_index_fsm(node(), [{raw, QId, Me}, [Bucket, none, Q, Timeout, all, undefined, CoverageFn]]),
      handle_side_effects(T);
  handle_side_effects([H | T]) ->
      lager:warning("in riak_kv_qry:handle_side_effects not handling ~p", [H]),
      handle_side_effects(T).
  
  decode_results(KVList, SelectSpec) ->
      lists:append(
  
negative.code: 
  -spec handle_call(term(), {pid(), term()}, #state{}) ->
                           {reply, Reply::ok | {error, atom()} | list(), #state{}}.
  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}.
  
  -spec handle_cast(term(), #state{}) -> {noreply, #state{}}.
  %% @private
  