anchor.diff:  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]) ->
  +run_sub_qs_fn([]) -> ok;
  +run_sub_qs_fn([{{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).
  +    Opts = [Bucket, none, Q, Timeout, all, undefined, CoverageFn],
  +    {ok, _PID} = riak_kv_index_fsm_sup:start_index_fsm(node(), [{raw, QId, Me}, Opts]),
  +    run_sub_qs_fn(T).
   
   decode_results(KVList, SelectSpec) ->
       lists:append(
  
positive.code: 
  -spec handle_call(term(), {pid(), term()}, #state{}) ->
                           {reply, Reply::ok | {error, atom()} | list(), #state{}}.
  %% @private
  handle_call(Request, _, State) ->
      case handle_req(Request, State) of
          {{error, _} = Error, _SEs, NewState} ->
              {reply, Error, NewState};
          {Reply, SEs, NewState} ->
              ok = handle_side_effects(SEs),
              {reply, Reply, NewState}
      end.
  
  
  -spec handle_cast(term(), #state{}) -> {noreply, #state{}}.
  %% @private
  
negative.code: -define(NO_MAX_RESULTS, no_max_results).
  -define(NO_PG_SORT, undefined).
  
  
  -record(state, {
            name                                :: atom(),
            ddl                                 :: undefined | #ddl_v1{},
            qry           = none                :: none | #riak_sql_v1{},
            qid           = undefined           :: undefined | {node(), non_neg_integer()},
            sub_qrys      = []                  :: [integer()],
            status        = void                :: void | accumulating_chunks,
            receiver_pid                        :: pid(),
            result        = []                  :: [{non_neg_integer(), list()}] | [{binary(), term()}],
            run_sub_qs_fn = fun run_sub_qs_fn/1 :: fun()
           }).
  
  %%%===================================================================
  