| 505 | } |
| 506 | |
| 507 | bool Coordinator::BackendState::ApplyExecStatusReport( |
| 508 | const ReportExecStatusRequestPB& backend_exec_status, |
| 509 | const TRuntimeProfileForest& thrift_profiles, ExecSummary* exec_summary, |
| 510 | ProgressUpdater* scan_range_progress, ProgressUpdater* query_progress, |
| 511 | DmlExecState* dml_exec_state, vector<AuxErrorInfoPB>* aux_error_info, |
| 512 | const vector<FragmentStats*>& fragment_stats) { |
| 513 | DCHECK(!IsEmptyBackend()); |
| 514 | CHECK(FLAGS_gen_experimental_profile || |
| 515 | backend_exec_status.fragment_exec_status().empty()) |
| 516 | << "Received pre-aggregated profile but --gen_experimental_profile=false"; |
| 517 | // Hold the exec_summary's lock to avoid exposing it half-way through |
| 518 | // the update loop below. |
| 519 | lock_guard<SpinLock> l1(exec_summary->lock); |
| 520 | unique_lock<mutex> lock(lock_); |
| 521 | last_report_time_ms_ = GenerateReportTimestamp(); |
| 522 | |
| 523 | // If this backend completed previously, don't apply the update. This ensures that |
| 524 | // the profile doesn't change during or after Coordinator::ComputeQuerySummary(), but |
| 525 | // can mean that we lose profile information for failed queries, since the Coordinator |
| 526 | // will call Cancel() on all BackendStates and we may stop accepting reports before some |
| 527 | // backends send their final report. |
| 528 | // TODO: revisit ComputeQuerySummary() |
| 529 | if (IsDoneLocked(lock)) return false; |
| 530 | |
| 531 | // Use empty profile in case profile serialization/deserialization failed. |
| 532 | // Depending on the --gen_experimental_profile value, there is one profile tree per |
| 533 | // fragment (if true) or per fragment instance (if false). |
| 534 | vector<TRuntimeProfileTree> empty_profiles; |
| 535 | // We iterate over the profiles in order. The profile trees include the instance |
| 536 | // profiles in the same order as 'instance_exec_status', then the aggregated fragment |
| 537 | // profiles in the same order as 'fragment_exec_status'. |
| 538 | vector<TRuntimeProfileTree>::const_iterator profile_iter; |
| 539 | int expected_num_profiles = FLAGS_gen_experimental_profile ? |
| 540 | backend_exec_status.fragment_exec_status().size() : |
| 541 | backend_exec_status.instance_exec_status().size(); |
| 542 | if (UNLIKELY(thrift_profiles.profile_trees.size() == 0)) { |
| 543 | empty_profiles.resize(expected_num_profiles); |
| 544 | profile_iter = empty_profiles.begin(); |
| 545 | } else { |
| 546 | DCHECK_EQ(expected_num_profiles, thrift_profiles.profile_trees.size()); |
| 547 | profile_iter = thrift_profiles.profile_trees.begin(); |
| 548 | } |
| 549 | |
| 550 | int64_t report_time_ms = UnixMillis(); |
| 551 | for (auto status_iter = backend_exec_status.instance_exec_status().begin(); |
| 552 | status_iter != backend_exec_status.instance_exec_status().end(); |
| 553 | ++status_iter) { |
| 554 | const FragmentInstanceExecStatusPB& instance_exec_status = *status_iter; |
| 555 | int64_t report_seq_no = instance_exec_status.report_seq_no(); |
| 556 | int instance_idx = GetInstanceIdx(instance_exec_status.fragment_instance_id()); |
| 557 | DCHECK_EQ(instance_stats_map_.count(instance_idx), 1); |
| 558 | InstanceStats* instance_stats = instance_stats_map_[instance_idx]; |
| 559 | int64_t last_report_seq_no = instance_stats->last_report_seq_no_; |
| 560 | DCHECK_EQ(instance_stats->exec_params_.instance_id(), |
| 561 | instance_exec_status.fragment_instance_id()); |
| 562 | // Ignore duplicate or out-of-order messages. |
| 563 | if (report_seq_no <= last_report_seq_no) { |
| 564 | VLOG_QUERY << "Ignoring stale update for query instance " |
no test coverage detected