anchor.diff:            qid           = undefined           :: undefined | {node(), non_neg_integer()},
             sub_qrys      = []                  :: [integer()],
             receiver_pid                        :: pid(),
  -          fsm_queue          = []             :: [list()],
  -          n_running_fsms     = 0              :: non_neg_integer(),
  -          max_n_running_fsms = 5              :: pos_integer(),
  -          result             = []             :: rows_acc() | aggregate_acc() | group_by_acc(),
  -          qbuf_ref                            :: undefined | riak_kv_qry_buffers:qbuf_ref(),
  -          orig_qry                            :: undefined | ?SQL_SELECT{}
  +          %% overload protection:
  +          %% 1. Maintain a limited number of running fsms (i.e.,
  +          %%    process at most that many subqueries at a time):
  +          fsm_queue          = []                   :: [list()],
  +          n_running_fsms     = 0                    :: non_neg_integer(),
  +          max_running_fsms   = ?MAX_RUNNING_FSMS    :: pos_integer(),
  +          %% 2. Estimate query size (wget-style):
  +          n_subqueries_done  = 0                    :: non_neg_integer(),
  +          total_query_data   = 0                    :: non_neg_integer(),
  +          max_query_data     = ?MAX_QUERY_DATA_SIZE :: non_neg_integer(),
  +          %% For queries not backed by query buffers, results are
  +          %% accumulated in memory:
  +          result             = []                   :: rows_acc() | aggregate_acc() | group_by_acc(),
  +          %% Query buffer support
  +          qbuf_ref                 :: undefined | riak_kv_qry_buffers:qbuf_ref(),
  +          orig_qry                 :: undefined | ?SQL_SELECT{}
            }).
   
   %%%===================================================================
  
positive.code: 
  handle_info({{SubQId, QId}, {results, Chunk}}, #state{qid = QId} = State) ->
      {noreply, throttling_spawn_index_fsms(
                  add_subquery_result(SubQId, Chunk, State))};
  
  handle_info({{SubQId, QId}, {error, Reason} = Error},
              #state{receiver_pid = ReceiverPid,
  
negative.code: 
  -include("riak_kv_ts.hrl").
  
  -define(MAX_RUNNING_FSMS, 5).
  -define(MAX_QUERY_DATA_SIZE, 5*1024*1024*1024).
  
  %% accumulators for different types of query results.
  -type rows_acc()      :: [{non_neg_integer(), list()}].
  -type aggregate_acc() :: [{binary(), term()}].
  