@VisibleForTesting
  void prepareConcurrentMultiGetTasks() {
    multi_get_wait_cnt = 0;
    prepare_multi_get_start_time = DateTime.currentTimeMillis();


    final List<Long> row_base_time_list;
    if (RollupQuery.isValidQuery(rollup_query)) {
      row_base_time_list = prepareRowBaseTimesRollup();
    } else {
      row_base_time_list = prepareRowBaseTimes();
    }

    int next_concurrency_index = 0;
    List<GetRequest> gets_to_prepare = new ArrayList<GetRequest>(batch_size);
    Set<byte[]> tsuids = new ByteSet();
    
    final ByteMap<ByteMap<List<GetRequest>>> all_tsuids_gets;
    if (multiget_no_meta) {
      // prepare the tagvs combinations and base time list
      
      final List<byte[][]> tagv_compinations = prepareAllTagvCompounds();
      all_tsuids_gets = prepareRequestsNoMeta(tagv_compinations, row_base_time_list);
    } else {
      all_tsuids_gets = prepareRequests(row_base_time_list, tags);

    }
    // Iterate over all salts
    for (final Entry<byte[], ByteMap<List<GetRequest>>> salts_entry : all_tsuids_gets.entrySet()) {
      if (gets_to_prepare.size() > 0) { // if we have any gets_to_prepare for previous salt, create a 
                                        // request out of it.
        final MultiGetTask task = new MultiGetTask(tsuids, gets_to_prepare);
        final List<MultiGetTask> mulget_task_list = 
            multi_get_tasks.get((next_concurrency_index++) % concurrency_multi_get);
        mulget_task_list.add(task);
        ++multi_get_wait_cnt;
        multi_get_num_get_requests = multi_get_num_get_requests + gets_to_prepare.size();
        gets_to_prepare = new ArrayList<GetRequest>(batch_size);
        tsuids = new ByteSet();
      }
      byte[] curr_salt = salts_entry.getKey();
      // Iterate over all tsuid's in curr_salt and add them to gets_to_prepare
      for (final Entry<byte[], List<GetRequest>> gets_entry : salts_entry.getValue()) {
        byte[] tsuid = gets_entry.getKey();
      
          for (GetRequest request : gets_entry.getValue()) { 
            if (gets_to_prepare.size() >= batch_size) { // close batch and create a MultiGetTask
              final MultiGetTask task = new MultiGetTask(tsuids, gets_to_prepare);
              final List<MultiGetTask> mulget_task_list = 
                  multi_get_tasks.get((next_concurrency_index++) % concurrency_multi_get);
              mulget_task_list.add(task);
              ++multi_get_wait_cnt;
              multi_get_num_get_requests = multi_get_num_get_requests + gets_to_prepare.size();
              if (LOG.isDebugEnabled()) {
                LOG.debug("Finished preparing MultiGetRequest with " + gets_to_prepare.size() + " requests for salt " + curr_salt + 
                 " for tsuid " + Bytes.pretty(tsuid));
              }
              // prepare a new task list and tsuids
              gets_to_prepare = new ArrayList<GetRequest>(batch_size);
              tsuids = new ByteSet();
            }             
            gets_to_prepare.add(request);
            tsuids.add(gets_entry.getKey());
          }
          tsuids.add(gets_entry.getKey());
      }
    }
   
  
    if (gets_to_prepare.size() > 0) {
      
      final MultiGetTask task = new MultiGetTask(tsuids, gets_to_prepare);
      final List<MultiGetTask> mulget_task_list = 
          multi_get_tasks.get((next_concurrency_index++) % concurrency_multi_get);
      mulget_task_list.add(task);
      ++multi_get_wait_cnt;
      multi_get_num_get_requests = multi_get_num_get_requests + gets_to_prepare.size();
      LOG.debug("Finished preparing MultiGetRequest with " + gets_to_prepare.size());
      gets_to_prepare = new ArrayList<GetRequest>(batch_size);
      tsuids = new ByteSet();
    }

    prepare_multi_get_end_time = DateTime.currentTimeMillis();
    if (LOG.isDebugEnabled()) {
      LOG.debug("Finished preparing concurrency multi get task with " 
          + multi_get_wait_cnt + " tasks using "
          + (prepare_multi_get_end_time - prepare_multi_get_start_time) + "ms");
    }

  }
@ VisibleForTesting void prepareConcurrentMultiGetTasks ( ) { multi_get_wait_cnt = 0 ; prepare_multi_get_start_time = DateTime . currentTimeMillis ( ) ; final List < Long > row_base_time_list ; if ( RollupQuery . isValidQuery ( rollup_query ) ) { row_base_time_list = prepareRowBaseTimesRollup ( ) ; } else { row_base_time_list = prepareRowBaseTimes ( ) ; } int next_concurrency_index = 0 ; List < GetRequest > gets_to_prepare = new ArrayList < GetRequest > ( batch_size ) ; Set < byte [ ] > tsuids = new ByteSet ( ) ; final ByteMap < ByteMap < List < GetRequest > > > all_tsuids_gets ; if ( multiget_no_meta ) { final List < byte [ ] [ ] > tagv_compinations = prepareAllTagvCompounds ( ) ; all_tsuids_gets = prepareRequestsNoMeta ( tagv_compinations , row_base_time_list ) ; } else { all_tsuids_gets = prepareRequests ( row_base_time_list , tags ) ; } for ( final Entry < byte [ ] , ByteMap < List < GetRequest > > > salts_entry : all_tsuids_gets . entrySet ( ) ) { if ( gets_to_prepare . size ( ) > 0 ) { final MultiGetTask task = new MultiGetTask ( tsuids , gets_to_prepare ) ; final List < MultiGetTask > mulget_task_list = multi_get_tasks . get ( ( next_concurrency_index ++ ) % concurrency_multi_get ) ; mulget_task_list . add ( task ) ; ++ multi_get_wait_cnt ; multi_get_num_get_requests = multi_get_num_get_requests + gets_to_prepare . size ( ) ; gets_to_prepare = new ArrayList < GetRequest > ( batch_size ) ; tsuids = new ByteSet ( ) ; } byte [ ] curr_salt = salts_entry . getKey ( ) ; for ( final Entry < byte [ ] , List < GetRequest > > gets_entry : salts_entry . getValue ( ) ) { byte [ ] tsuid = gets_entry . getKey ( ) ; for ( GetRequest request : gets_entry . getValue ( ) ) { if ( gets_to_prepare . size ( ) >= batch_size ) { final MultiGetTask task = new MultiGetTask ( tsuids , gets_to_prepare ) ; final List < MultiGetTask > mulget_task_list = multi_get_tasks . get ( ( next_concurrency_index ++ ) % concurrency_multi_get ) ; mulget_task_list . add ( task ) ; ++ multi_get_wait_cnt ; multi_get_num_get_requests = multi_get_num_get_requests + gets_to_prepare . size ( ) ; if ( LOG . isDebugEnabled ( ) ) { LOG . debug ( "Finished preparing MultiGetRequest with " + gets_to_prepare . size ( ) + " requests for salt " + curr_salt + " for tsuid " + Bytes . pretty ( tsuid ) ) ; } gets_to_prepare = new ArrayList < GetRequest > ( batch_size ) ; tsuids = new ByteSet ( ) ; } gets_to_prepare . add ( request ) ; tsuids . add ( gets_entry . getKey ( ) ) ; } tsuids . add ( gets_entry . getKey ( ) ) ; } } if ( gets_to_prepare . size ( ) > 0 ) { final MultiGetTask task = new MultiGetTask ( tsuids , gets_to_prepare ) ; final List < MultiGetTask > mulget_task_list = multi_get_tasks . get ( ( next_concurrency_index ++ ) % concurrency_multi_get ) ; mulget_task_list . add ( task ) ; ++ multi_get_wait_cnt ; multi_get_num_get_requests = multi_get_num_get_requests + gets_to_prepare . size ( ) ; LOG . debug ( "Finished preparing MultiGetRequest with " + gets_to_prepare . size ( ) ) ; gets_to_prepare = new ArrayList < GetRequest > ( batch_size ) ; tsuids = new ByteSet ( ) ; } prepare_multi_get_end_time = DateTime . currentTimeMillis ( ) ; if ( LOG . isDebugEnabled ( ) ) { LOG . debug ( "Finished preparing concurrency multi get task with " + multi_get_wait_cnt + " tasks using " + ( prepare_multi_get_end_time - prepare_multi_get_start_time ) + "ms" ) ; } }
Compiles the list of TSUIDs and GetRequests to send to execute against
storage. Each batch will only have requests for one salt, i.e a batch
will not have requests with multiple salts.