| 533 | } |
| 534 | |
| 535 | Status Coordinator::StartBackendExec() { |
| 536 | int num_backends = backend_states_.size(); |
| 537 | backend_exec_complete_barrier_.reset(new CountingBarrier(num_backends)); |
| 538 | |
| 539 | DebugOptions debug_options(exec_params_.query_options()); |
| 540 | |
| 541 | VLOG_QUERY << "starting execution on " << num_backends << " backends for query_id=" |
| 542 | << PrintId(query_id()); |
| 543 | query_events_->MarkEvent(Substitute("Ready to start on $0 backends", num_backends)); |
| 544 | parent_query_driver_->SetExecTimeLimit(parent_request_state_); |
| 545 | |
| 546 | // Serialize the TQueryCtx once and pass it to each backend. The serialized buffer must |
| 547 | // stay valid until WaitOnExecRpcs() has returned. |
| 548 | ThriftSerializer serializer(true); |
| 549 | uint8_t* serialized_buf = nullptr; |
| 550 | uint32_t serialized_len = 0; |
| 551 | Status serialize_status = |
| 552 | serializer.SerializeToBuffer(&query_ctx(), &serialized_len, &serialized_buf); |
| 553 | if (UNLIKELY(!serialize_status.ok())) { |
| 554 | return UpdateExecState(serialize_status, nullptr, FLAGS_hostname); |
| 555 | } |
| 556 | kudu::Slice query_ctx_slice(serialized_buf, serialized_len); |
| 557 | // Incorporate the size of the shared TQueryCtx into the ExecQueryRpcStats. This also |
| 558 | // includes information about the size of the descriptor table, because that can |
| 559 | // be a major component of the TQueryCtx's size. |
| 560 | int64_t descriptor_table_size = query_ctx().desc_tbl_serialized.thrift_desc_tbl.size(); |
| 561 | execquery_rpc_stats_->SetSharedTQueryCtxSize(query_ctx_slice.size(), |
| 562 | descriptor_table_size); |
| 563 | |
| 564 | for (BackendState* backend_state: backend_states_) { |
| 565 | if (exec_rpcs_status_barrier_.pending() <= 0) { |
| 566 | // One of the backends has already indicated an error with Exec(). |
| 567 | break; |
| 568 | } |
| 569 | DebugActionNoFail(exec_params_.query_options(), "COORD_BEFORE_EXEC_RPC"); |
| 570 | // Safe for ExecAsync() to read 'filter_routing_table_' because it is complete |
| 571 | // at this point and won't be destroyed while this function is executing, |
| 572 | // because it won't be torn down until WaitOnExecRpcs() has returned. |
| 573 | DCHECK(filter_mode_ == TRuntimeFilterMode::OFF || filter_routing_table_->is_complete); |
| 574 | backend_state->ExecAsync(debug_options, *filter_routing_table_, query_ctx_slice, |
| 575 | &exec_rpcs_status_barrier_, execquery_rpc_stats_.get()); |
| 576 | } |
| 577 | // The ExecQueryRpcStats accumulated any warnings from sending the exec RPCs. Print the |
| 578 | // summary now. |
| 579 | execquery_rpc_stats_->PrintWarnings(); |
| 580 | Status exec_rpc_status = exec_rpcs_status_barrier_.Wait(); |
| 581 | if (!exec_rpc_status.ok()) { |
| 582 | // One of the backends failed to startup, so we cancel the other ones. |
| 583 | CancelBackends(/*fire_and_forget=*/ true); |
| 584 | WaitOnExecRpcs(); |
| 585 | vector<BackendState*> failed_backend_states; |
| 586 | for (BackendState* backend_state : backend_states_) { |
| 587 | // If Exec() rpc failed for a reason besides being aborted, blacklist the executor |
| 588 | // and retry the query. |
| 589 | if (!backend_state->exec_rpc_status().ok() |
| 590 | && !backend_state->exec_rpc_status().IsAborted()) { |
| 591 | failed_backend_states.push_back(backend_state); |
| 592 | LOG(INFO) << "Blacklisting " << backend_state->impalad_address() |
nothing calls this directly
no test coverage detected