| 155 | } |
| 156 | |
| 157 | Status TestEnv::CreateQueryState( |
| 158 | int64_t query_id, const TQueryOptions* query_options, RuntimeState** runtime_state) { |
| 159 | TQueryCtx query_ctx; |
| 160 | if (query_options != nullptr) query_ctx.client_request.query_options = *query_options; |
| 161 | query_ctx.query_id.hi = 0; |
| 162 | query_ctx.query_id.lo = query_id; |
| 163 | query_ctx.request_pool = "test-pool"; |
| 164 | query_ctx.coord_hostname = exec_env_->configured_backend_address_.hostname; |
| 165 | query_ctx.coord_ip_address = FromNetworkAddressPB(exec_env_->krpc_address_); |
| 166 | TUniqueId backend_id; |
| 167 | UniqueIdPBToTUniqueId(exec_env_->backend_id(), &backend_id); |
| 168 | query_ctx.__set_coord_backend_id(backend_id); |
| 169 | TQueryOptions* query_options_to_use = &query_ctx.client_request.query_options; |
| 170 | int64_t mem_limit = |
| 171 | query_options_to_use->__isset.mem_limit && query_options_to_use->mem_limit > 0 ? |
| 172 | query_options_to_use->mem_limit : |
| 173 | -1; |
| 174 | |
| 175 | // CreateQueryState() enforces the invariant that 'query_id' must be unique. |
| 176 | QueryState* qs = exec_env_->query_exec_mgr()->CreateQueryState(query_ctx, mem_limit); |
| 177 | query_states_.push_back(qs); |
| 178 | // make sure to initialize data structures unrelated to the TExecQueryFInstancesParams |
| 179 | // param |
| 180 | ExecQueryFInstancesRequestPB rpc_params; |
| 181 | // create dummy -Ctx fields, we need them for FragmentInstance-/RuntimeState |
| 182 | rpc_params.set_coord_state_idx(0); |
| 183 | rpc_params.add_fragment_ctxs(); |
| 184 | rpc_params.add_fragment_instance_ctxs(); |
| 185 | TExecPlanFragmentInfo fragment_info; |
| 186 | fragment_info.__set_fragments(vector<TPlanFragment>({TPlanFragment()})); |
| 187 | fragment_info.__set_fragment_instance_ctxs( |
| 188 | vector<TPlanFragmentInstanceCtx>({TPlanFragmentInstanceCtx()})); |
| 189 | RETURN_IF_ERROR(qs->Init(&rpc_params, fragment_info)); |
| 190 | FragmentState* frag_state = qs->obj_pool()->Add(new FragmentState( |
| 191 | qs, qs->fragment_info_.fragments[0], qs->exec_rpc_params_.fragment_ctxs(0))); |
| 192 | FragmentInstanceState* fis = qs->obj_pool()->Add(new FragmentInstanceState(qs, |
| 193 | frag_state, qs->fragment_info_.fragment_instance_ctxs[0], |
| 194 | qs->exec_rpc_params_.fragment_instance_ctxs(0))); |
| 195 | RuntimeState* rs = |
| 196 | qs->obj_pool()->Add(new RuntimeState(qs, fis->fragment(), fis->instance_ctx(), |
| 197 | fis->fragment_ctx(), fis->instance_ctx_pb(), exec_env_.get())); |
| 198 | runtime_states_.push_back(rs); |
| 199 | |
| 200 | *runtime_state = rs; |
| 201 | return Status::OK(); |
| 202 | } |
| 203 | } |
no test coverage detected