anchor.diff:            ddl                  :: undefined | #ddl_v1{},
             qry      = none      :: none | #riak_sql_v1{},
             qid      = undefined :: undefined | {node(), non_neg_integer()},
  -          n_chunks = 0         :: non_neg_integer(),
  -	  sub_qrys = []        :: [integer()],
  -          status   = void      :: void | accumulating_chunks | finished,
  -          result   = []        :: [binary()]
  +          sub_qrys  = []       :: [integer()],
  +          status    = void     :: void | accumulating_chunks | finished,
  +          result    = []       :: [{non_neg_integer(), list()}] | [{binary(), term()}]
            }).
   
   %%%===================================================================
  
positive.code:     handle_side_effects(T).
  
  decode_results(KVList, SelectSpec) ->
      [extract_riak_object(SelectSpec, V) || {_, V} <- KVList].
  
  extract_riak_object(SelectSpec, V) when is_binary(V) ->
      % don't care about bkey
  
negative.code: %% @private
  %% sometimes we get finish statements after we have finished accumulating
  handle_info({{_SubQId, _QId}, done},
              State = #state{status = finished}) ->
      lager:debug("Received stray done on QId ~p (~p)", [_QId, _SubQId]),
      {noreply, State};
  handle_info({{SubQId, QId}, done},
              State = #state{qid       = QId,
                             result    = IndexedChunks,
                             sub_qrys  = SubQQ}) ->
      lager:debug("Received done on QId ~p (~p); SubQQ: ~p", [QId, SubQId, SubQQ]),
      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)),
              %% drop indexes, serialize
              {noreply, State#state{status = finished,
                                    result = lists:append(R2)}};
          _MoreSubQueriesNotDone ->
              {noreply, State}
      end;
  
  handle_info({{SubQId, QId}, {results, Chunk}},
              State = #state{qid      = QId,
                             qry      = Qry,
                             result   = IndexedChunks,
                             sub_qrys = SubQs}) ->
      #riak_sql_v1{'SELECT' = SelectSpec} = Qry,
      NewS = case lists:member(SubQId, SubQs) of
                 true ->
                     Decoded = decode_results(lists:flatten(Chunk), SelectSpec),
                     lager:debug("Got chunk on QId ~p (~p); SubQQ: ~p", [QId, SubQId, SubQs]),
                     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,
      {noreply, NewS};
  
  %% late chunks: warn
  handle_info({{SubQId, QId}, {results, _LateData}},
              State = #state{qid = QId,
                             status = finished}) ->
      lager:debug("Discarding late chunk (~b bytes) on qid ~p (subqid ~p)",
                  [length(lists:flatten(_LateData)), QId, SubQId]),
      {noreply, State};
  
  %% other error conditions
  handle_info({{_SubQId, QId1}, _}, State = #state{qid = QId2})
    when QId1 =/= QId2 ->
      lager:warning("Bad query id ~p (expected ~p)", [QId1, QId2]),
      {noreply, State}.
  