Method CreateCollocatedAndScanInstances

be/src/scheduling/scheduler.cc:795–970

Maybe the easiest way to understand the objective of this algorithm is as a generalization of two simpler instance creation algorithms that decide how many instances of a fragment to create on each node, given a set of nodes that were already chosen by previous scheduling steps: 1. Instance creation for an interior fragment (i.e. a fragment without scans) without a UNION, where we create one finst

//    b) Create the finstances, based on the computed parallelism and assign the scan
//       ranges to it.
void Scheduler::CreateCollocatedAndScanInstances(const ExecutorConfig& executor_config,
    FragmentScheduleState* fragment_state, ScheduleState* state) {
  const TPlanFragment& fragment = fragment_state->fragment;
  bool has_union = ContainsUnionNode(fragment.plan);
  DCHECK(has_union || ContainsScanNode(fragment.plan));
  // Build a map of hosts to the num instances this fragment should have, before we take
  // into account scan ranges. If this fragment has input fragments, we always run with
  // at least the same num instances as the input fragment.
  std::unordered_map<NetworkAddressPB, int> instances_per_host;

  // Add hosts of input fragments, counting the number of instances of the fragment.
  // Only do this if there's a union - otherwise only consider the parallelism of
  // the input scan, for consistency with the previous behaviour of only using
  // the parallelism of the scan.
  if (has_union) {
    for (FragmentIdx idx : fragment_state->exchange_input_fragments) {
      std::unordered_map<NetworkAddressPB, int> input_fragment_instances_per_host;
      const FragmentScheduleState& input_state = *state->GetFragmentScheduleState(idx);
      for (const FInstanceScheduleState& instance_state : input_state.instance_states) {
        ++input_fragment_instances_per_host[instance_state.host];
      }
      // Merge with the existing hosts by taking the max num instances.
      if (instances_per_host.empty()) {
        // Optimization for the common case of one input fragment.
        instances_per_host = move(input_fragment_instances_per_host);
      } else {
        for (auto& entry : input_fragment_instances_per_host) {
          int& num_instances = instances_per_host[entry.first];
          num_instances = max(num_instances, entry.second);
        }
      }
    }
  }

  // Add hosts of scan nodes.
  vector<TPlanNodeId> scan_node_ids = FindScanNodes(fragment.plan);
  DCHECK(has_union || scan_node_ids.size() == 1) << "This method may need revisiting "
      << "for plans with no union and multiple scans per fragment";
  vector<NetworkAddressPB> scan_hosts;
  GetScanHosts(scan_node_ids, *fragment_state, &scan_hosts);
  if (scan_hosts.empty() && instances_per_host.empty()) {
    // None of the scan nodes have any scan ranges and there is no input fragment feeding
    // into this fragment; run it on a random executor.
    // TODO TODO: the TODO below seems partially stale
    // TODO: we'll need to revisit this strategy once we can partition joins
    // (in which case this fragment might be executing a right outer join
    // with a large build table)
    vector<BackendDescriptorPB> all_executors =
        executor_config.group.GetAllExecutorDescriptors();
    int idx = std::uniform_int_distribution<int>(0, all_executors.size() - 1)(
        *state->rng());
    const BackendDescriptorPB& be_desc = all_executors[idx];
    scan_hosts.push_back(be_desc.address());
  }
  for (const NetworkAddressPB& host_addr : scan_hosts) {
    // Ensure that the num instances is at least as many as input fragments. We don't
    // want to increment if there were already some instances from the input fragment,
    // since that could result in too high a num_instances.