Method ComputeRandomKrpcForAggregation

be/src/scheduling/scheduler.cc:398–492

typedef vector<pair<int, int>> InstanceToAggPairs;

void Scheduler::ComputeRandomKrpcForAggregation(const ExecutorConfig& executor_config,
    ScheduleState* state, FragmentScheduleState* src_state, int num_filters_per_host) {
  if (num_filters_per_host <= 1) return;
  // src_state->instance_states organize fragment instances as one dimension vector
  // where instances scheduled in same host is placed in adjacent indices.
  // Group instance indices from common host to 'instance_groups' by comparing
  // krpc address between adjacent element of instance_states.
  const NetworkAddressPB& coord_address = executor_config.coord_desc.krpc_address();
  vector<InstanceToAggPairs> instance_groups;
  InstanceToAggPairs coordinator_instances;
  for (int i = 0; i < src_state->instance_states.size(); ++i) {
    const NetworkAddressPB& krpc_address = src_state->instance_states[i].krpc_host;
    if (KrpcAddressEqual(krpc_address, coord_address)) {
      coordinator_instances.emplace_back(i, 0);
    } else {
      if (i == 0
          || !KrpcAddressEqual(
              krpc_address, src_state->instance_states[i - 1].krpc_host)) {
        // This is the first fragment instance scheduled in a host.
        // Append empty InstanceToAggPairs to 'instance_groups'.
        instance_groups.emplace_back();
      }
      instance_groups.back().emplace_back(i, 0);
    }
  }

  int num_non_coordinator_host = instance_groups.size();
  if (num_non_coordinator_host == 0) return;

  // Select number of intermediate aggregator so that each aggregator will receive
  // runtime filter update from at most 'num_filters_per_host' executors.
  int num_agg = (int)ceil((double)num_non_coordinator_host / num_filters_per_host);
  DCHECK_GT(num_agg, 0);

  if (UNLIKELY(FLAGS_sort_runtime_filter_aggregator_candidates)) {
    sort(instance_groups.begin(), instance_groups.end(),
        [src_state](InstanceToAggPairs a, InstanceToAggPairs b) {
          int idx_a = a[0].first;
          int idx_b = b[0].first;
          return CompareNetworkAddressPB(src_state->instance_states[idx_a].krpc_host,
                     src_state->instance_states[idx_b].krpc_host)
              < 0;
        });
  } else {
    std::shuffle(instance_groups.begin(), instance_groups.end(), *state->rng());
  }
  if (coordinator_instances.size() > 0) {
    // Put coordinator group behind so that coordinator won't be selected as intermediate
    // aggregator.
    instance_groups.push_back(coordinator_instances);
  }

  RuntimeFilterAggregatorInfoPB* agg_info =
      src_state->exec_params->mutable_filter_agg_info();
  agg_info->set_num_aggregators(num_agg);

  int group_idx = 0;
  int agg_idx = -1;