Impala client. Uses the HS2 protocol plus Impala-specific extensions.
| 593 | |
| 594 | |
| 595 | class ImpalaHS2Client(ImpalaClient): |
| 596 | """Impala client. Uses the HS2 protocol plus Impala-specific extensions.""" |
| 597 | def __init__(self, *args, **kwargs): |
| 598 | super(ImpalaHS2Client, self).__init__(*args, **kwargs) |
| 599 | self.FINISHED_STATE = TOperationState._NAMES_TO_VALUES["FINISHED_STATE"] |
| 600 | self.ERROR_STATE = TOperationState._NAMES_TO_VALUES["ERROR_STATE"] |
| 601 | self.CANCELED_STATE = TOperationState._NAMES_TO_VALUES["CANCELED_STATE"] |
| 602 | self._clear_current_query_handle() |
| 603 | |
| 604 | # If connected, this is the handle returned by the OpenSession RPC that needs |
| 605 | # to be passed into most HS2 RPCs. |
| 606 | self.session_handle = None |
| 607 | # Enable retries only for hs2-http protocol. |
| 608 | if self.use_http_base_transport: |
| 609 | # Maximum number of tries for idempotent rpcs. |
| 610 | self.max_tries = self.connect_max_tries |
| 611 | else: |
| 612 | self.max_tries = 1 |
| 613 | # Minimum sleep interval between retry attempts. |
| 614 | self.min_sleep_interval = 1 |
| 615 | |
| 616 | # In case of direct instantiation of the client where the converter is |
| 617 | # not set, there should be a default value converter assigned |
| 618 | if self.value_converter is None: |
| 619 | self.value_converter = HS2ValueConverter() |
| 620 | |
| 621 | if self.rpc_stdout or self.rpc_stdout is not None: |
| 622 | self.thrift_printer = ThriftPrettyPrinter() |
| 623 | |
| 624 | self._base_request_id = str(uuid.uuid1()) |
| 625 | self._request_num = 0 |
| 626 | |
| 627 | def _get_thrift_client(self, protocol): |
| 628 | return ImpalaHiveServer2Service.Client(protocol) |
| 629 | |
| 630 | def _get_sleep_interval_for_retries(self, num_tries): |
| 631 | """Returns the sleep interval in seconds for the 'num_tries' retry attempt.""" |
| 632 | assert num_tries > 0 and num_tries < self.max_tries |
| 633 | return self.min_sleep_interval * (num_tries - 1) |
| 634 | |
| 635 | def _open_session(self): |
| 636 | def OpenSession(req): |
| 637 | return self.imp_service.OpenSession(req) |
| 638 | # OpenSession rpcs are idempotent and so ok to retry. If the client gets disconnected |
| 639 | # and the server successfully opened a session, the client will retry and rely on |
| 640 | # server to clean up the session. |
| 641 | req = TOpenSessionReq(TProtocolVersion.HIVE_CLI_SERVICE_PROTOCOL_V6, |
| 642 | username=self.user) |
| 643 | resp = self._do_hs2_rpc(OpenSession, req, retry_on_error=True) |
| 644 | self._check_hs2_rpc_status(resp.status) |
| 645 | assert (resp.serverProtocolVersion |
| 646 | == TProtocolVersion.HIVE_CLI_SERVICE_PROTOCOL_V6), resp.serverProtocolVersion |
| 647 | # TODO: ensure it's closed if needed |
| 648 | self.session_handle = resp.sessionHandle |
| 649 | |
| 650 | self._populate_query_options() |
| 651 | |
| 652 | def get_custom_http_headers(self): |
no outgoing calls