Method ExecQueryFInstances

be/src/service/control-service.cc:132–177

}

void ControlService::ExecQueryFInstances(const ExecQueryFInstancesRequestPB* request,
    ExecQueryFInstancesResponsePB* response, RpcContext* rpc_context) {
  DebugActionNoFail(FLAGS_debug_actions, "EXEC_QUERY_FINSTANCES_DELAY");
  DCHECK(request->has_coord_state_idx());
  DCHECK(request->has_plan_fragment_info_sidecar_idx());
  DCHECK(request->has_query_ctx_sidecar_idx());
  // Deserialize the sidecars. The QueryState will make a copy of the TQueryCtx and
  // TExecPlanFragmentInfo, so we can deallocate the deserialized values after
  // StartQuery(). TODO: can we avoid this extra copy?
  TExecPlanFragmentInfo fragment_info;
  const Status& fragment_info_sidecar_status =
      GetSidecar(request->plan_fragment_info_sidecar_idx(), rpc_context, &fragment_info);
  if (!fragment_info_sidecar_status.ok()) {
    RespondAndReleaseRpc(fragment_info_sidecar_status, response, rpc_context);
    return;
  }
  TQueryCtx query_ctx;
  const Status& query_ctx_sidecar_status =
      GetSidecar(request->query_ctx_sidecar_idx(), rpc_context, &query_ctx);
  if (!query_ctx_sidecar_status.ok()) {
    RespondAndReleaseRpc(query_ctx_sidecar_status, response, rpc_context);
    return;
  }
  ScopedThreadContext scoped_tdi(GetThreadDebugInfo(), query_ctx.query_id);
  VLOG_QUERY << "ExecQueryFInstances():"
             << " query_id=" << PrintId(query_ctx.query_id)
             << " coord=" << query_ctx.coord_hostname << ":"
             << query_ctx.coord_ip_address.port
             << " #instances=" << fragment_info.fragment_instance_ctxs.size();
  Status resp_status;
  if (UNLIKELY(fragment_info.fragments.size() == 0
      || fragment_info.fragment_instance_ctxs.size() == 0)) {
    resp_status = Status(Substitute("ExecQueryFInstances() failed: query_id=: $0, "
        "no instance in TExecPlanFragmentInfo", PrintId(query_ctx.query_id)));
    LOG(ERROR) << resp_status.msg().msg();
    RespondAndReleaseRpc(resp_status, response, rpc_context);
    return;
  }
  resp_status = ExecEnv::GetInstance()->query_exec_mgr()->StartQuery(
      request, query_ctx, fragment_info);
  if (!resp_status.ok()) {
    LOG(INFO) << "ExecQueryFInstances() failed: query_id=" << PrintId(query_ctx.query_id)
              << ": " << resp_status.GetDetail();
  }
  RespondAndReleaseRpc(resp_status, response, rpc_context);
}

void ControlService::ReportExecStatus(const ReportExecStatusRequestPB* request,
    ReportExecStatusResponsePB* response, RpcContext* rpc_context) {