| 2874 | } |
| 2875 | |
| 2876 | void AdmissionController::AdmitQuery( |
| 2877 | QueueNode* node, string& user, bool was_queued, bool is_trivial) { |
| 2878 | ScheduleState* state = node->admitted_schedule.get(); |
| 2879 | VLOG_RPC << "Admitting query " << PrintId(state->query_id()) |
| 2880 | << " ScheduleState=" << PrintScheduleStateMemInfo(state); |
| 2881 | |
| 2882 | // Update memory and number of queries. |
| 2883 | bool track_per_user = HasQuotaConfig(node->pool_cfg) || HasQuotaConfig(node->root_cfg); |
| 2884 | PerUserTracking per_user_tracking{user, was_queued, track_per_user}; |
| 2885 | UpdateStatsOnAdmission(*state, is_trivial, per_user_tracking); |
| 2886 | UpdateExecGroupMetric(state->executor_group(), 1); |
| 2887 | // Update summary profile. |
| 2888 | const string& admission_result = was_queued ? |
| 2889 | PROFILE_INFO_VAL_ADMIT_QUEUED : |
| 2890 | (is_trivial ? PROFILE_INFO_VAL_ADMIT_TRIVIAL : PROFILE_INFO_VAL_ADMIT_IMMEDIATELY); |
| 2891 | state->summary_profile()->AddInfoString( |
| 2892 | PROFILE_INFO_KEY_ADMISSION_RESULT, admission_result); |
| 2893 | state->summary_profile()->AddInfoString( |
| 2894 | PROFILE_INFO_KEY_ADMITTED_MEM, PrintBytes(state->GetClusterMemoryToAdmit())); |
| 2895 | state->summary_profile()->AddInfoString( |
| 2896 | PROFILE_INFO_KEY_EXECUTOR_GROUP, state->executor_group()); |
| 2897 | state->summary_profile()->AddInfoString(PROFILE_INFO_KEY_EXECUTOR_GROUP_QUERY_LOAD, |
| 2898 | std::to_string(GetExecGroupQueryLoad(state->executor_group()))); |
| 2899 | // We may have admitted based on stale information. Include a warning in the profile |
| 2900 | // if this might be the case. |
| 2901 | int64_t time_since_update_ms; |
| 2902 | string staleness_detail = GetStalenessDetailLocked("", &time_since_update_ms); |
| 2903 | // IMPALA-8235: convert to TIME_NS because of issues with tools consuming TIME_MS. |
| 2904 | COUNTER_SET(ADD_COUNTER(state->summary_profile(), |
| 2905 | PROFILE_TIME_SINCE_LAST_UPDATE_COUNTER_NAME, TUnit::TIME_NS), |
| 2906 | static_cast<int64_t>(time_since_update_ms * NANOS_PER_MICRO * MICROS_PER_MILLI)); |
| 2907 | if (!staleness_detail.empty()) { |
| 2908 | state->summary_profile()->AddInfoString( |
| 2909 | PROFILE_INFO_KEY_STALENESS_WARNING, staleness_detail); |
| 2910 | } |
| 2911 | DCHECK(num_released_backends_.find(state->query_id()) == num_released_backends_.end()); |
| 2912 | num_released_backends_[state->query_id()] = state->per_backend_schedule_states().size(); |
| 2913 | |
| 2914 | // Store info about the admitted resources so that we can release them. |
| 2915 | auto it = running_queries_.find(node->admission_request.coord_id); |
| 2916 | if (it == running_queries_.end()) { |
| 2917 | auto insert_result = |
| 2918 | running_queries_.insert(make_pair(node->admission_request.coord_id, |
| 2919 | std::unordered_map<UniqueIdPB, RunningQuery>())); |
| 2920 | DCHECK(insert_result.second); |
| 2921 | it = insert_result.first; |
| 2922 | } |
| 2923 | DCHECK(it->second.find(state->query_id()) == it->second.end()); |
| 2924 | RunningQuery& running_query = it->second[state->query_id()]; |
| 2925 | running_query.request_pool = state->request_pool(); |
| 2926 | running_query.executor_group = state->executor_group(); |
| 2927 | running_query.is_trivial = is_trivial; |
| 2928 | if (track_per_user) { |
| 2929 | // Do not set user if user quotas are not configured. |
| 2930 | running_query.user = user; |
| 2931 | } |
| 2932 | for (const auto& entry : state->per_backend_schedule_states()) { |
| 2933 | BackendAllocation& allocation = running_query.per_backend_resources[entry.first]; |
no test coverage detected