Method GetExecutorGroupsForQuery

be/src/scheduling/admission-controller.cc:3228–3272

}

vector<const ExecutorGroup*> AdmissionController::GetExecutorGroupsForQuery(
    const ClusterMembershipMgr::ExecutorGroups& all_groups,
    const TQueryExecRequest& exec_req) {
  vector<const ExecutorGroup*> matching_groups;
  if (scheduler_->IsCoordinatorOnlyQuery(exec_req)) {
    // Coordinator only queries can run regardless of the presence of exec groups. This
    // empty group works as a proxy to schedule coordinator only queries.
    matching_groups.push_back(cluster_membership_mgr_->GetEmptyExecutorGroup());
    return matching_groups;
  }
  const string& pool_name = exec_req.query_ctx.request_pool;
  string prefix(pool_name + POOL_GROUP_DELIMITER);
  // We search for matching groups before the health check so that we don't fall back to
  // the default group in case there are matching but unhealthy groups.
  for (const auto& it : all_groups) {
    StringPiece name(it.first);
    if (name.starts_with(prefix)) matching_groups.push_back(&it.second);
  }
  if (matching_groups.empty()) {
    auto default_it = all_groups.find(ImpalaServer::DEFAULT_EXECUTOR_GROUP_NAME);
    if (default_it == all_groups.end()) return matching_groups;
    VLOG(3) << "Checking default executor group for pool " << pool_name;
    matching_groups.push_back(&default_it->second);
  }
  // Filter out unhealthy groups.
  auto erase_from = std::remove_if(matching_groups.begin(), matching_groups.end(),
      [](const ExecutorGroup* g) { return !g->IsHealthy(); });
  matching_groups.erase(erase_from, matching_groups.end());
  // Sort executor groups by name.
  auto cmp = [](const ExecutorGroup* a, const ExecutorGroup* b) {
    return a->name() < b->name();
  };
  sort(matching_groups.begin(), matching_groups.end(), cmp);
  if (FLAGS_balance_queries_across_executor_groups) {
    // Sort executor groups by available memory and slots in descending order, we
    // prioritize executor groups that with more available memory, when their available
    // memory are same we choose the one with more available slots, when their available
    // memory and slots are same we choose on an alphabetical basis.
    auto available_resource_cmp = [this](const ExecutorGroup* a, const ExecutorGroup* b) {
      return GetAvailableMemAndSlots(*a) > GetAvailableMemAndSlots(*b);
    };
    sort(matching_groups.begin(), matching_groups.end(), available_resource_cmp);
  }
  return matching_groups;
}

int64_t AdmissionController::GetClusterSize(
    const ClusterMembershipMgr::Snapshot& membership_snapshot) {