| 153 | } |
| 154 | |
| 155 | Status Coordinator::Exec() { |
| 156 | const TQueryExecRequest& request = exec_params_.query_exec_request(); |
| 157 | DCHECK(request.plan_exec_info.size() > 0); |
| 158 | |
| 159 | VLOG_QUERY << "Exec() query_id=" << PrintId(query_id()) |
| 160 | << " stmt=" << request.query_ctx.client_request.stmt; |
| 161 | stmt_type_ = request.stmt_type; |
| 162 | |
| 163 | query_profile_ = RuntimeProfile::Create( |
| 164 | obj_pool(), "Execution Profile " + PrintId(query_id()), false); |
| 165 | finalization_timer_ = PROFILE_FinalizationTimer.Instantiate(query_profile_); |
| 166 | filter_updates_received_ = PROFILE_FiltersReceived.Instantiate(query_profile_); |
| 167 | execquery_rpc_stats_ = make_unique<ExecQueryRpcStats>( |
| 168 | PROFILE_ExecQueryRPCSizes.Instantiate(query_profile_)); |
| 169 | |
| 170 | host_profiles_ = RuntimeProfile::Create(obj_pool(), "Per Node Profiles", false); |
| 171 | query_profile_->AddChild(host_profiles_); |
| 172 | |
| 173 | SCOPED_TIMER(query_profile_->total_time_counter()); |
| 174 | |
| 175 | // initialize progress updater |
| 176 | const string& str = Substitute("Query $0", PrintId(query_id())); |
| 177 | scan_progress_.Init(str, exec_params_.query_schedule().num_scan_ranges()); |
| 178 | |
| 179 | query_state_ = ExecEnv::GetInstance()->query_exec_mgr()->CreateQueryState( |
| 180 | query_ctx(), exec_params_.query_schedule().coord_backend_mem_limit()); |
| 181 | filter_mem_tracker_ = query_state_->obj_pool()->Add(new MemTracker( |
| 182 | -1, "Runtime Filter (Coordinator)", query_state_->query_mem_tracker(), false)); |
| 183 | |
| 184 | InitFragmentStats(); |
| 185 | // create BackendStates and per-instance state, including profiles, and install |
| 186 | // the latter in the FragmentStats' root profile |
| 187 | InitBackendStates(); |
| 188 | exec_summary_.Init(exec_params_); |
| 189 | |
| 190 | int64_t total_finstances = 0; |
| 191 | for (BackendState* backend_state : backend_states_) { |
| 192 | total_finstances += backend_state->exec_params().instance_params().size(); |
| 193 | } |
| 194 | const string& query_progress_str = |
| 195 | Substitute("Query $0 progress", PrintId(query_id())); |
| 196 | query_progress_.Init(query_progress_str, total_finstances); |
| 197 | |
| 198 | if (filter_mode_ != TRuntimeFilterMode::OFF) { |
| 199 | // Populate the runtime filter routing table. This should happen before starting the |
| 200 | // fragment instances. This code anticipates the indices of the instance states |
| 201 | // created later on in ExecRemoteFragment() |
| 202 | InitFilterRoutingTable(); |
| 203 | } |
| 204 | |
| 205 | // At this point, all static setup is done and all structures are initialized. Only |
| 206 | // runtime-related state changes past this point (examples: fragment instance |
| 207 | // profiles, etc.) |
| 208 | |
| 209 | RETURN_IF_ERROR(StartBackendExec()); |
| 210 | RETURN_IF_ERROR(FinishBackendStartup()); |
| 211 | |
| 212 | // set coord_instance_ and coord_sink_ |
no test coverage detected