Execute the query 'query_str' asynchronously on the server with options dictionary 'set_query_options' and return a query handle that can be used for subsequent ImpalaClient method calls for the query.
(self, query_str, set_query_options)
| 755 | return query |
| 756 | |
| 757 | def execute_query(self, query_str, set_query_options): |
| 758 | """Execute the query 'query_str' asynchronously on the server with options dictionary |
| 759 | 'set_query_options' and return a query handle that can be used for subsequent |
| 760 | ImpalaClient method calls for the query.""" |
| 761 | self._clear_current_query_handle() |
| 762 | self.is_query_cancelled = False |
| 763 | |
| 764 | def ExecuteStatement(req): |
| 765 | return self.imp_service.ExecuteStatement(req) |
| 766 | # Read queries should be idempotent but most dml queries are not. Also retrying |
| 767 | # query execution from client could be expensive and so likely makes sense to do |
| 768 | # it if server is also aware of the retries. |
| 769 | req = self._create_query_req(query_str, set_query_options) |
| 770 | resp = self._do_hs2_rpc(ExecuteStatement, req) |
| 771 | if resp.status.statusCode != TStatusCode.SUCCESS_STATUS: |
| 772 | msg = utf8_decode_if_needed(resp.status.errorMessage) |
| 773 | raise QueryStateException("ERROR: {0}".format(msg)) |
| 774 | handle = resp.operationHandle |
| 775 | |
| 776 | try: |
| 777 | self._set_current_query_handle(handle) |
| 778 | if handle.hasResultSet: |
| 779 | def GetResultSetMetadata(req): |
| 780 | return self.imp_service.GetResultSetMetadata(req) |
| 781 | # GetResultSetMetadata rpc is idempotent and should be safe to retry. |
| 782 | req = TGetResultSetMetadataReq(handle) |
| 783 | resp = self._do_hs2_rpc(GetResultSetMetadata, req, retry_on_error=True) |
| 784 | self._check_hs2_rpc_status(resp.status) |
| 785 | assert resp.schema is not None, resp |
| 786 | # Attach the schema to the handle for convenience. |
| 787 | handle.schema = resp.schema |
| 788 | handle.is_closed = False |
| 789 | return handle |
| 790 | finally: |
| 791 | self._clear_current_query_handle() |
| 792 | |
| 793 | def get_query_id_str(self, last_query_handle): |
| 794 | if last_query_handle is None: |