| 565 | } |
| 566 | |
| 567 | void ImpalaServer::ExecuteStatementCommon(TExecuteStatementResp& return_val, |
| 568 | const TExecuteStatementReq& request, const TExecRequest* external_exec_request) { |
| 569 | HS2_RETURN_IF_ERROR(return_val, CheckNotShuttingDown(), SQLSTATE_GENERAL_ERROR); |
| 570 | // We ignore the runAsync flag here: Impala's queries will always run asynchronously, |
| 571 | // and will block on fetch. To the client, this looks like Hive's synchronous mode; the |
| 572 | // difference is that rows are not available when ExecuteStatement() returns. |
| 573 | TQueryCtx query_ctx; |
| 574 | Status status = TExecuteStatementReqToTQueryContext(request, &query_ctx); |
| 575 | HS2_RETURN_IF_ERROR(return_val, status, SQLSTATE_GENERAL_ERROR); |
| 576 | |
| 577 | TUniqueId session_id; |
| 578 | TUniqueId secret; |
| 579 | HS2_RETURN_IF_ERROR(return_val, THandleIdentifierToTUniqueId( |
| 580 | request.sessionHandle.sessionId, &session_id, &secret), SQLSTATE_GENERAL_ERROR); |
| 581 | ScopedSessionState session_handle(this); |
| 582 | shared_ptr<SessionState> session; |
| 583 | HS2_RETURN_IF_ERROR(return_val, |
| 584 | session_handle.WithSession(session_id, SecretArg::Session(secret), &session), |
| 585 | SQLSTATE_GENERAL_ERROR); |
| 586 | if (session == NULL) { |
| 587 | string err_msg = Substitute("Invalid session id: $0", PrintId(session_id)); |
| 588 | VLOG(1) << err_msg; |
| 589 | HS2_RETURN_IF_ERROR(return_val, Status::Expected(err_msg), SQLSTATE_GENERAL_ERROR); |
| 590 | } |
| 591 | |
| 592 | // Optionally enable result caching to allow restarting fetches. |
| 593 | int64_t cache_num_rows = -1; |
| 594 | if (request.__isset.confOverlay) { |
| 595 | map<string, string>::const_iterator iter = |
| 596 | request.confOverlay.find(IMPALA_RESULT_CACHING_OPT); |
| 597 | if (iter != request.confOverlay.end()) { |
| 598 | StringParser::ParseResult parse_result; |
| 599 | cache_num_rows = StringParser::StringToInt<int64_t>( |
| 600 | iter->second.c_str(), iter->second.size(), &parse_result); |
| 601 | if (parse_result != StringParser::PARSE_SUCCESS) { |
| 602 | HS2_RETURN_IF_ERROR( |
| 603 | return_val, Status::Expected(Substitute("Invalid value '$0' for '$1' option.", |
| 604 | iter->second, IMPALA_RESULT_CACHING_OPT)), SQLSTATE_GENERAL_ERROR); |
| 605 | } |
| 606 | } |
| 607 | } |
| 608 | |
| 609 | QueryHandle query_handle; |
| 610 | status = Execute(&query_ctx, session, &query_handle, external_exec_request); |
| 611 | |
| 612 | // Make query id available to the following HS2_RETURN_IF_ERROR(). |
| 613 | ScopedThreadContext scoped_tdi(GetThreadDebugInfo(), query_handle->query_id()); |
| 614 | |
| 615 | HS2_RETURN_IF_ERROR(return_val, status, SQLSTATE_GENERAL_ERROR); |
| 616 | |
| 617 | // Start thread to wait for results to become available. |
| 618 | status = query_handle->WaitAsync(); |
| 619 | if (!status.ok()) goto return_error; |
| 620 | |
| 621 | // Check if query return result set and optionally enable result caching on the |
| 622 | // ClientRequestState. |
| 623 | bool returns_result_set; |
| 624 | status = |
nothing calls this directly
no test coverage detected