anchor.diff:              %% discard; Don't touch state as it may have already 'finished'.
               State
       end.
  +
  +%%
  +run_select_on_chunk(SubQId, Chunk, #state{ qry = Query,
  +                                           result = QueryResult1 }) ->
  +  DecodedChunk = decode_results(lists:flatten(Chunk)),
  +  SelClause = sql_select_clause(Query),
  +  case sql_select_calc_type(Query) of
  +      rows ->
  +          run_select_on_rows_chunk(SubQId, SelClause, DecodedChunk, QueryResult1);
  +      aggregate ->
  +          run_select_on_aggregate_chunk(SelClause, QueryResult1, DecodedChunk)
  +  end.
  +
  +%% Run the selection clause on results that accumulate rows
  +run_select_on_rows_chunk(SubQId, SelClause, DecodedChunk, QueryResult1) ->
  +    IndexedChunks =
  +        [riak_kv_qry_compiler:run_select(SelClause, Row) || Row <- DecodedChunk],
  +    [{SubQId, IndexedChunks} | QueryResult1].
  +
  +%%
  +run_select_on_aggregate_chunk(SelClause, DecodedChunk, QueryResult1) ->
  +    lists:foldl(
  +        fun(E, Acc) ->
  +            riak_kv_qry_compiler:run_select(SelClause, E, Acc)
  +        end, QueryResult1, DecodedChunk).
   
   %%
   -spec cancel_error_query(Error::any(), State1::#state{}) ->
  
positive.code:                                            col_names = ColNames}, Rows) ->
      {ColNames, ColTypes, Rows}.
  
  %%%===================================================================
  %%% Unit tests
  %%%===================================================================
  
negative.code:                      result       = InitialState }}.
  
  %%
  add_subquery_result(SubQId, Chunk, #state{ sub_qrys = SubQs} = State) ->
      case lists:member(SubQId, SubQs) of
          true ->
              try
                QueryResult2 = run_select_on_chunk(SubQId, Chunk, State),
                NSubQ = lists:delete(SubQId, SubQs),
                State#state{status   = accumulating_chunks,
                            result   = QueryResult2,
  