anchor.diff:  -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()}]
  +          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()
            }).
   
   %%%===================================================================
  
positive.code: -compile(export_all).
  -include_lib("eunit/include/eunit.hrl").
  
  %%
  %% Test runner
  %%
  
  -define(NO_OUTPUTS, []).
  
  runner_TEST(Tests) -> io:format("running tests ~p~n", [Tests]),
                        test_r2(Tests, #state{}, 1, [], [], []).
  
  test_r2([], _State, _LineNo, SideEffects, Replies, Errs) ->
      {lists:reverse(SideEffects), lists:reverse(Replies), lists:reverse(Errs)};
  test_r2([H | T], State, LineNo, SideEffects, Replies, Errs) ->
      io:format("Before ~p~n- SideEffects is ~p~n- Replies is ~p~n- Errors is ~p~n", [H, SideEffects, Replies, Errs]),
      {NewSt, NewSEs, NewRs, NewE} = run(H, State, LineNo, SideEffects, Replies, Errs),
      io:format("After ~p~n- NewSEs is ~p~n- NewRs is ~p~n- NewE is ~p~n", [H, NewSEs, NewRs, NewE]),
      test_r2(T, NewSt, LineNo + 1, NewSEs, NewRs, NewE).
  
  run({run, {init, Arg}}, _State, _LNo, SideEffects, Replies, Errs) ->
      {ok, NewState} = init(Arg),
      {NewState, SideEffects, Replies, Errs};
  run({clear, side_effects}, State, _LNo, _SideEffects, Replies, Errs) ->
      {State, [], Replies, Errs};
  run({clear, replies}, State, _LNo, SideEffects, _Replies, Errs) ->
      {State, SideEffects, [], Errs};
  run({dump, state}, State, LNo, SideEffects, Replies, Errs) ->
      io:format("On Line No ~p State is~n- ~p~n", [LNo, State]),
      {State, SideEffects, Replies, Errs};
  run({dump, Replies}, State, LNo, SideEffects, Replies, Errs) ->
      io:format("On Line No ~p Replies is~n- ~p~n", [LNo, Replies]),
      {State, SideEffects, Replies, Errs};
  run({dump, errors}, State, LNo, SideEffects, Replies, Errs) ->
      io:format("On Line No ~p Errs is~n- ~p~n", [LNo, Errs]),
      {State, SideEffects, Replies, Errs};
  run({msg, Msg}, State, _LNo, SideEffects, Replies, Errs) ->
      {Reply, SEs, NewState} = handle_req(Msg, State),
      NewSEs = case SEs of
                   [] -> SideEffects;
                   _  -> lists:flatten(SEs, SideEffects)
               end,
      {NewState, NewSEs, [Reply | Replies], Errs};
  run({side_effect, G}, State, LNo, SEs, Replies, Errs) ->
      {NewSE, NewErrs}
          = case SEs of
                []      -> Err = {error, {line_no, LNo}, {side_effect, {expected, []}, {got, G}}},
                           {[], [Err | Errs]};
                [G | T] -> {T, Errs};
                [E | T] -> Err = {error, {line_no, LNo}, {side_effect, {expected, E}, {got, G}}},
                           {T, [Err | Errs]}
            end,
      {State, NewSE, Replies, NewErrs};
  run({reply, G}, State, LNo, SideEffects, Replies, Errs) ->
      {NewRs, NewErrs}
          = case Replies of
                []      -> Err = {error, {line_no, LNo}, {reply, {expected, []}, {got, G}}},
                           {[], [Err | Errs]};
                [G | T] -> {T, Errs};
                [E | T] -> Err = {error, {line_no, LNo}, {reply, {expected, E}, {got, G}}},
                           {T, [Err | Errs]}
            end,
      {State, SideEffects, NewRs, NewErrs};
  run(H, State, LNo, SideEffects, Replies, Errs) ->
      Err = {error, {line_no, LNo}, {unknown_test_state, H}},
      {State, SideEffects, Replies, [Err | Errs]}.
  
  -define(MAX_Q_LEN, 5).
  
  simple_init_test() ->
      Tests = [
               {run, {init, [fms1]}}
             ],
      Results = runner_TEST(Tests),
      ?assertEqual({[], [], []}, Results).
  
  -endif.
  
negative.code: new_state(RegisteredName) ->
      #state{name = RegisteredName}.
  
  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},
      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(
  