| 74 | QueryExecMgr::~QueryExecMgr() {} |
| 75 | |
| 76 | Status QueryExecMgr::StartQuery(const ExecQueryFInstancesRequestPB* request, |
| 77 | const TQueryCtx& query_ctx, const TExecPlanFragmentInfo& fragment_info) { |
| 78 | TUniqueId query_id = query_ctx.query_id; |
| 79 | VLOG(2) << "StartQueryFInstances() query_id=" << PrintId(query_id) |
| 80 | << " coord=" << query_ctx.coord_hostname << ":" |
| 81 | << query_ctx.coord_ip_address.port; |
| 82 | bool dummy; |
| 83 | QueryState* qs = |
| 84 | GetOrCreateQueryState(query_ctx, request->per_backend_mem_limit(), &dummy); |
| 85 | RETURN_IF_ERROR(DebugAction(query_ctx.client_request.query_options.debug_action, |
| 86 | "QUERY_STATE_BEFORE_INIT_GLOBAL")); |
| 87 | RETURN_IF_ERROR(DebugAction(query_ctx.client_request.query_options.debug_action, |
| 88 | "QUERY_STATE_BEFORE_INIT", {std::to_string(FLAGS_krpc_port)})); |
| 89 | Status status = qs->Init(request, fragment_info); |
| 90 | if (!status.ok()) { |
| 91 | qs->ReleaseBackendResourceRefcount(); // Release refcnt acquired in Init(). |
| 92 | ReleaseQueryState(qs); |
| 93 | return status; |
| 94 | } |
| 95 | // avoid blocking the rpc handler thread for too long by starting a new thread for |
| 96 | // query startup (which takes ownership of the QueryState reference) |
| 97 | unique_ptr<Thread> t; |
| 98 | status = Thread::Create("query-exec-mgr", |
| 99 | Substitute("query-state-$0", PrintId(query_id)), |
| 100 | &QueryExecMgr::ExecuteQueryHelper, this, qs, &t, true); |
| 101 | if (!status.ok()) { |
| 102 | // decrement refcount taken in QueryState::Init() |
| 103 | qs->ReleaseBackendResourceRefcount(); |
| 104 | // decrement refcount taken in GetOrCreateQueryState() |
| 105 | ReleaseQueryState(qs); |
| 106 | return status; |
| 107 | } |
| 108 | t->Detach(); |
| 109 | return Status::OK(); |
| 110 | } |
| 111 | |
| 112 | QueryState* QueryExecMgr::CreateQueryState( |
| 113 | const TQueryCtx& query_ctx, int64_t mem_limit) { |
no test coverage detected