| 131 | } |
| 132 | |
| 133 | void Coordinator::BackendState::SetRpcParams(const DebugOptions& debug_options, |
| 134 | const FilterRoutingTable& filter_routing_table, ExecQueryFInstancesRequestPB* request, |
| 135 | TExecPlanFragmentInfo* fragment_info, int64_t* total_scan_ranges) { |
| 136 | *total_scan_ranges = 0; |
| 137 | request->set_coord_state_idx(state_idx_); |
| 138 | request->set_min_mem_reservation_bytes( |
| 139 | backend_exec_params_.min_mem_reservation_bytes()); |
| 140 | request->set_initial_mem_reservation_total_claims( |
| 141 | backend_exec_params_.initial_mem_reservation_total_claims()); |
| 142 | request->set_per_backend_mem_limit( |
| 143 | exec_params_.query_schedule().per_backend_mem_limit()); |
| 144 | |
| 145 | // set fragment_ctxs and fragment_instance_ctxs |
| 146 | fragment_info->__isset.fragments = true; |
| 147 | fragment_info->__isset.fragment_instance_ctxs = true; |
| 148 | fragment_info->fragment_instance_ctxs.resize( |
| 149 | backend_exec_params_.instance_params().size()); |
| 150 | DCHECK_GT(fragment_info->fragment_instance_ctxs.size(), 0); |
| 151 | for (int i = 0; i < backend_exec_params_.instance_params().size(); ++i) { |
| 152 | TPlanFragmentInstanceCtx& instance_ctx = fragment_info->fragment_instance_ctxs[i]; |
| 153 | PlanFragmentInstanceCtxPB* instance_ctx_pb = request->add_fragment_instance_ctxs(); |
| 154 | const FInstanceExecParamsPB& params = backend_exec_params_.instance_params(i); |
| 155 | int fragment_idx = params.fragment_idx(); |
| 156 | DCHECK_LT(fragment_idx, exec_params_.query_schedule().fragment_exec_params().size()); |
| 157 | const FragmentExecParamsPB& fragment_exec_params = |
| 158 | exec_params_.query_schedule().fragment_exec_params(fragment_idx); |
| 159 | |
| 160 | // add a TPlanFragment, if we don't already have it |
| 161 | if (fragment_info->fragments.empty() |
| 162 | || fragment_info->fragments.back().idx != fragment_idx) { |
| 163 | const TPlanFragment* fragment = exec_params_.GetFragments()[fragment_idx]; |
| 164 | fragment_info->fragments.push_back(*fragment); |
| 165 | PlanFragmentCtxPB* fragment_ctx = request->add_fragment_ctxs(); |
| 166 | fragment_ctx->set_fragment_idx(fragment_idx); |
| 167 | *fragment_ctx->mutable_destinations() = fragment_exec_params.destinations(); |
| 168 | } |
| 169 | |
| 170 | instance_ctx.fragment_idx = fragment_idx; |
| 171 | instance_ctx_pb->set_fragment_idx(fragment_idx); |
| 172 | UniqueIdPBToTUniqueId(params.instance_id(), &instance_ctx.fragment_instance_id); |
| 173 | instance_ctx.per_fragment_instance_idx = params.per_fragment_instance_idx(); |
| 174 | *instance_ctx_pb->mutable_per_node_scan_ranges() = params.per_node_scan_ranges(); |
| 175 | // Sum the scan ranges over all the fragments |
| 176 | for (const auto& pair : params.per_node_scan_ranges()) { |
| 177 | *total_scan_ranges += pair.second.scan_ranges().size(); |
| 178 | } |
| 179 | for (const auto& entry : fragment_exec_params.per_exch_num_senders()) { |
| 180 | instance_ctx.per_exch_num_senders[entry.first] = entry.second; |
| 181 | } |
| 182 | instance_ctx.__set_sender_id(params.sender_id()); |
| 183 | *instance_ctx_pb->mutable_join_build_inputs() = params.join_build_inputs(); |
| 184 | if (params.num_join_build_outputs() != -1) { |
| 185 | instance_ctx.__set_num_join_build_outputs(params.num_join_build_outputs()); |
| 186 | } |
| 187 | if (debug_options.enabled() |
| 188 | && (debug_options.instance_idx() == -1 |
| 189 | || debug_options.instance_idx() == GetInstanceIdx(params.instance_id()))) { |
| 190 | instance_ctx.__set_debug_options(debug_options.ToThrift()); |
nothing calls this directly
no test coverage detected