| 536 | } |
| 537 | |
| 538 | void QueryState::ConstructReport(bool instances_started, |
| 539 | ReportExecStatusRequestPB* report, TRuntimeProfileForest* profiles_forest) { |
| 540 | report->Clear(); |
| 541 | report->set_backend_report_seq_no(++last_report_seq_no_); |
| 542 | TUniqueIdToUniqueIdPB(query_id(), report->mutable_query_id()); |
| 543 | DCHECK(exec_rpc_params_.has_coord_state_idx()); |
| 544 | report->set_coord_state_idx(exec_rpc_params_.coord_state_idx()); |
| 545 | Status report_overall_status; |
| 546 | { |
| 547 | unique_lock<SpinLock> l(status_lock_); |
| 548 | |
| 549 | Status debug_action_status = |
| 550 | DebugAction(query_options(), "CONSTRUCT_QUERY_STATE_REPORT"); |
| 551 | if (UNLIKELY(!debug_action_status.ok())) overall_status_ = debug_action_status; |
| 552 | |
| 553 | StatusToProto(overall_status_, report->mutable_overall_status()); |
| 554 | report_overall_status = overall_status_; |
| 555 | if (IsValidFInstanceId(failed_finstance_id_)) { |
| 556 | TUniqueIdToUniqueIdPB(failed_finstance_id_, report->mutable_fragment_instance_id()); |
| 557 | } |
| 558 | } |
| 559 | if (!report_overall_status.ok() && query_spilled_.Load() == 1 && file_group_ != nullptr |
| 560 | && file_group_->IsSpillingDiskFaulty()) { |
| 561 | report->set_local_disk_faulty(true); |
| 562 | } |
| 563 | |
| 564 | // Add profile to report |
| 565 | host_profile_->ToThrift(&profiles_forest->host_profile); |
| 566 | profiles_forest->__isset.host_profile = true; |
| 567 | |
| 568 | // Free resources in chunked counters in the profile |
| 569 | host_profile_->ClearChunkedTimeSeriesCounters(); |
| 570 | |
| 571 | if (instances_started) { |
| 572 | // Map from fragment idx to the averaged profile. When aggregated profiles are |
| 573 | // enabled (IMPALA-9382), we populate this map with profiles that are aggregated |
| 574 | // from all the instances on this backend. |
| 575 | unordered_map<int, AggregatedRuntimeProfile*> agg_profiles; |
| 576 | ObjectPool agg_profile_pool; |
| 577 | |
| 578 | // Stats that we aggregate across the instances. |
| 579 | int64_t cpu_user_ns = AsyncCodegenThreadUserTime(); |
| 580 | int64_t cpu_sys_ns = AsyncCodegenThreadSysTime(); |
| 581 | int64_t bytes_read = 0; |
| 582 | int64_t scan_ranges_complete = 0; |
| 583 | int64_t exchange_bytes_sent = 0; |
| 584 | int64_t scan_bytes_sent = 0; |
| 585 | std::map<int32_t, int64_t> per_join_rows_produced; |
| 586 | |
| 587 | for (const auto& entry : fis_map_) { |
| 588 | FragmentInstanceState* fis = entry.second; |
| 589 | |
| 590 | // If this fragment instance has already sent its last report, skip it. |
| 591 | if (fis->final_report_sent()) { |
| 592 | DCHECK(fis->IsDone()); |
| 593 | } else { |
| 594 | // Update the status and profiles of this fragment instance. |
| 595 | FragmentInstanceExecStatusPB* instance_status = |
nothing calls this directly
no test coverage detected