| 903 | } |
| 904 | |
| 905 | void Coordinator::BackendState::InstanceStats::Update( |
| 906 | const FragmentInstanceExecStatusPB& exec_status, |
| 907 | const TRuntimeProfileTree* thrift_profile, int64_t report_time_ms, |
| 908 | ExecSummary* exec_summary) { |
| 909 | last_report_time_ms_ = report_time_ms; |
| 910 | DCHECK_GT(exec_status.report_seq_no(), last_report_seq_no_); |
| 911 | last_report_seq_no_ = exec_status.report_seq_no(); |
| 912 | if (exec_status.done()) stopwatch_.Stop(); |
| 913 | profile_->UpdateInfoString(LAST_REPORT_TIME_DESC, |
| 914 | ToStringFromUnixMillis(last_report_time_ms_)); |
| 915 | if (!FLAGS_gen_experimental_profile) { |
| 916 | DCHECK(thrift_profile != nullptr); |
| 917 | profile_->Update(*thrift_profile); |
| 918 | profile_->ComputeTimeInProfile(); |
| 919 | } |
| 920 | |
| 921 | // update exec_summary |
| 922 | TExecSummary& thrift_exec_summary = exec_summary->thrift_exec_summary; |
| 923 | for (const ExecSummaryDataPB& exec_summary_entry : exec_status.exec_summary_data()) { |
| 924 | bool is_plan_node = exec_summary_entry.has_plan_node_id(); |
| 925 | bool is_data_sink = exec_summary_entry.has_data_sink_id(); |
| 926 | DCHECK(is_plan_node || is_data_sink) << "Invalid exec summary entry sent by executor"; |
| 927 | int exec_summary_idx; |
| 928 | if (is_plan_node) { |
| 929 | exec_summary_idx = |
| 930 | exec_summary->node_id_to_idx_map[exec_summary_entry.plan_node_id()]; |
| 931 | } else { |
| 932 | exec_summary_idx = |
| 933 | exec_summary->data_sink_id_to_idx_map[exec_summary_entry.data_sink_id()]; |
| 934 | } |
| 935 | TPlanNodeExecSummary& node_exec_summary = thrift_exec_summary.nodes[exec_summary_idx]; |
| 936 | DCHECK_EQ(node_exec_summary.fragment_idx, exec_params_.fragment_idx()); |
| 937 | int per_fragment_instance_idx = exec_params_.per_fragment_instance_idx(); |
| 938 | DCHECK_LT(per_fragment_instance_idx, node_exec_summary.exec_stats.size()) |
| 939 | << " instance_id=" << PrintId(exec_params_.instance_id()) |
| 940 | << " fragment_idx=" << exec_params_.fragment_idx(); |
| 941 | TExecStats& instance_stats = node_exec_summary.exec_stats[per_fragment_instance_idx]; |
| 942 | |
| 943 | if (exec_summary_entry.has_rows_returned()) { |
| 944 | instance_stats.__set_cardinality(exec_summary_entry.rows_returned()); |
| 945 | } |
| 946 | if (exec_summary_entry.has_peak_mem_usage()) { |
| 947 | instance_stats.__set_memory_used(exec_summary_entry.peak_mem_usage()); |
| 948 | } |
| 949 | DCHECK(exec_summary_entry.has_local_time_ns()); |
| 950 | instance_stats.__set_latency_ns(exec_summary_entry.local_time_ns()); |
| 951 | if (exec_summary_entry.has_last_batch_returned()) { |
| 952 | instance_stats.__set_last_batch_returned(exec_summary_entry.last_batch_returned()); |
| 953 | } |
| 954 | node_exec_summary.__isset.exec_stats = true; |
| 955 | } |
| 956 | |
| 957 | // extract the current execution state of this instance |
| 958 | current_state_ = exec_status.current_state(); |
| 959 | agg_profile_up_to_date_ = false; |
| 960 | } |
| 961 | |
| 962 | void Coordinator::BackendState::InstanceStats::UpdateExecStats( |
no test coverage detected