| 304 | } |
| 305 | |
| 306 | void FragmentInstanceState::GetStatusReport(FragmentInstanceExecStatusPB* instance_status, |
| 307 | TRuntimeProfileTree* unagg_profile, AggregatedRuntimeProfile* agg_profile, |
| 308 | const Status& overall_status) { |
| 309 | DCHECK_NE(unagg_profile == nullptr, agg_profile == nullptr); |
| 310 | DFAKE_SCOPED_LOCK(report_status_lock_); |
| 311 | DCHECK(!final_report_sent_); |
| 312 | // Update the counter for the peak per host mem usage. |
| 313 | if (per_host_mem_usage_ != nullptr) { |
| 314 | per_host_mem_usage_->Set(runtime_state()->query_mem_tracker()->peak_consumption()); |
| 315 | } |
| 316 | if (final_report_generated_) { |
| 317 | // Since execution was already finished, the contents of this report will be identical |
| 318 | // to the last report, so don't advance the sequence number. |
| 319 | instance_status->set_report_seq_no(report_seq_no_); |
| 320 | } else { |
| 321 | instance_status->set_report_seq_no(AdvanceReportSeqNo()); |
| 322 | } |
| 323 | const TUniqueId& finstance_id = instance_id(); |
| 324 | TUniqueIdToUniqueIdPB(finstance_id, instance_status->mutable_fragment_instance_id()); |
| 325 | // For failed fragment instance, report it as "done" when overall_statue is reported |
| 326 | // with error. This avoid coordinator to ignore the last status report. |
| 327 | const bool done = (IsDone() && !ExecFailed()) || (ExecFailed() && !overall_status.ok()); |
| 328 | instance_status->set_done(done); |
| 329 | instance_status->set_current_state(current_state()); |
| 330 | DCHECK(profile() != nullptr); |
| 331 | if (agg_profile != nullptr) { |
| 332 | // Figure out the index of this instance relative to other instances of this fragment |
| 333 | // on the backend. |
| 334 | int instance_idx = instance_ctx_.per_fragment_instance_idx - |
| 335 | fragment_state_->min_per_fragment_instance_idx(); |
| 336 | agg_profile->UpdateAggregatedFromInstance(profile(), instance_idx); |
| 337 | } else { |
| 338 | DCHECK(unagg_profile != nullptr); |
| 339 | profile()->ToThrift(unagg_profile); |
| 340 | } |
| 341 | |
| 342 | // Pull out and aggregate counters from the profile. |
| 343 | RuntimeProfile::Counter* user_time = profile()->GetCounter("TotalThreadsUserTime"); |
| 344 | if (user_time != nullptr) cpu_user_ns_ = user_time->value(); |
| 345 | |
| 346 | RuntimeProfile::Counter* system_time = profile()->GetCounter("TotalThreadsSysTime"); |
| 347 | if (system_time != nullptr) cpu_sys_ns_ = system_time->value(); |
| 348 | |
| 349 | // Compute local_time for use below. |
| 350 | profile()->ComputeTimeInProfile(); |
| 351 | vector<RuntimeProfileBase*> nodes; |
| 352 | profile()->GetAllChildren(&nodes); |
| 353 | int64_t bytes_read = 0; |
| 354 | int64_t scan_ranges_complete = 0; |
| 355 | int64_t total_bytes_sent = 0; |
| 356 | std::map<int32_t, int64_t> per_join_rows_produced; |
| 357 | for (RuntimeProfileBase* node : nodes) { |
| 358 | RuntimeProfile::Counter* c = node->GetCounter(PROFILE_BytesRead.name()); |
| 359 | if (c != nullptr) bytes_read += c->value(); |
| 360 | c = node->GetCounter(PROFILE_ScanRangesComplete.name()); |
| 361 | if (c != nullptr) scan_ranges_complete += c->value(); |
| 362 | c = node->GetCounter(KrpcDataStreamSender::TOTAL_BYTES_SENT_COUNTER); |
| 363 | if (c != nullptr) total_bytes_sent += c->value(); |
no test coverage detected