Method prepareConcurrentMultiGetTasks

src/core/MultiGetQuery.java:659–748

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.

()

   * will not have requests with multiple salts.
   */
  @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