| 289 | } |
| 290 | |
| 291 | void Coordinator::BackendState::ExecAsync(const DebugOptions& debug_options, |
| 292 | const FilterRoutingTable& filter_routing_table, |
| 293 | const kudu::Slice& serialized_query_ctx, |
| 294 | TypedCountingBarrier<Status>* exec_status_barrier, |
| 295 | Coordinator::ExecQueryRpcStats* execquery_rpc_stats) { |
| 296 | { |
| 297 | lock_guard<mutex> l(lock_); |
| 298 | DCHECK(!exec_done_); |
| 299 | DCHECK(status_.ok()); |
| 300 | // Do not issue an ExecQueryFInstances RPC if there are no fragment instances |
| 301 | // scheduled to run on this backend. |
| 302 | if (IsEmptyBackend()) { |
| 303 | DCHECK(backend_exec_params_.is_coord_backend()); |
| 304 | exec_done_ = true; |
| 305 | exec_status_barrier->Notify(Status::OK()); |
| 306 | goto done; |
| 307 | } |
| 308 | |
| 309 | std::unique_ptr<ControlServiceProxy> proxy; |
| 310 | Status get_proxy_status = |
| 311 | ControlService::GetProxy(krpc_host_, host_.hostname(), &proxy); |
| 312 | if (!get_proxy_status.ok()) { |
| 313 | SetExecError(get_proxy_status, exec_status_barrier); |
| 314 | goto done; |
| 315 | } |
| 316 | |
| 317 | ExecQueryFInstancesRequestPB request; |
| 318 | TExecPlanFragmentInfo fragment_info; |
| 319 | int64_t total_scan_ranges; |
| 320 | SetRpcParams(debug_options, filter_routing_table, &request, &fragment_info, |
| 321 | &total_scan_ranges); |
| 322 | |
| 323 | exec_rpc_controller_.set_timeout( |
| 324 | MonoDelta::FromMilliseconds(FLAGS_backend_client_rpc_timeout_ms)); |
| 325 | |
| 326 | // Serialize the sidecar and add it to the rpc controller. |
| 327 | Status serialize_debug_status = |
| 328 | DebugAction(exec_params_.query_options(), "EXEC_SERIALIZE_FRAGMENT_INFO"); |
| 329 | if (UNLIKELY(!serialize_debug_status.ok())) { |
| 330 | SetExecError(serialize_debug_status, exec_status_barrier); |
| 331 | goto done; |
| 332 | } |
| 333 | |
| 334 | int sidecar_idx; |
| 335 | int64_t fragment_info_length = 0; |
| 336 | // TODO: eliminate the extra copy here by using a Slice |
| 337 | Status sidecar_status = |
| 338 | SetFaststringSidecar(fragment_info, &exec_rpc_controller_, &sidecar_idx, |
| 339 | &fragment_info_length); |
| 340 | if (UNLIKELY(!sidecar_status.ok())) { |
| 341 | SetExecError(sidecar_status, exec_status_barrier); |
| 342 | goto done; |
| 343 | } |
| 344 | request.set_plan_fragment_info_sidecar_idx(sidecar_idx); |
| 345 | |
| 346 | // Add the serialized TQueryCtx as a sidecar. |
| 347 | unique_ptr<RpcSidecar> query_ctx_sidecar = |
| 348 | RpcSidecar::FromSlice(serialized_query_ctx); |
no test coverage detected