Connection to Impala using the HiveServer2 (HS2) protocol. This class does not use Impyla's DB-API cursors. Instead, it is built directly on the HS2 RPC layer to support manipulating one operation from multiple connections concurrently. This class is designed to be minimalistic to facil
| 996 | |
| 997 | |
| 998 | class MinimalHS2Connection(ImpalaConnection): |
| 999 | """ |
| 1000 | Connection to Impala using the HiveServer2 (HS2) protocol. |
| 1001 | |
| 1002 | This class does not use Impyla's DB-API cursors. Instead, it is built directly on the |
| 1003 | HS2 RPC layer to support manipulating one operation from multiple connections |
| 1004 | concurrently. |
| 1005 | |
| 1006 | This class is designed to be minimalistic to facilitate testing. Each method is mapped |
| 1007 | to only one Thrift RPC. |
| 1008 | """ |
| 1009 | def __init__(self, host_port, user=None): |
| 1010 | self.__host_port = host_port |
| 1011 | host, port = host_port.split(":") |
| 1012 | self.__conn = hs2.connect(host, port, auth_mechanism='NOSASL') |
| 1013 | self.__user = user if user is not None else getpass.getuser() |
| 1014 | self.__session = self.__conn.open_session(self.__user) |
| 1015 | self.__query_options = dict() |
| 1016 | |
| 1017 | def connect(self): |
| 1018 | pass # Do nothing |
| 1019 | |
| 1020 | def close(self): |
| 1021 | self.log_client("closing connection to: %s" % self.__host_port) |
| 1022 | try: |
| 1023 | self.__session.close() |
| 1024 | finally: |
| 1025 | self.__conn.close() |
| 1026 | |
| 1027 | def __log_execute(self, sql_stmt): |
| 1028 | session_id = session_handle_to_session_id(self.__session.handle) |
| 1029 | self.log_client( |
| 1030 | u"executing at {0}. session: {1} user: {2}\n{3}".format( |
| 1031 | self.__host_port, session_id, self.__user, format_sql_for_logging(sql_stmt)) |
| 1032 | ) |
| 1033 | |
| 1034 | def log_client(self, message): |
| 1035 | """Log 'message' at INFO level, prefixed wih the protocol name of this connection.""" |
| 1036 | LOG.info(u"minimal_{0}: {1}".format(self.get_test_protocol(), message)) |
| 1037 | |
| 1038 | def execute(self, sql_stmt, user=None, fetch_profile_after_close=False, # noqa: U100 |
| 1039 | fetch_exec_summary=False, # noqa: U100 |
| 1040 | profile_format=TRuntimeProfileFormat.STRING): # noqa: U100 |
| 1041 | raise NotImplementedError() |
| 1042 | |
| 1043 | def execute_async(self, sql_stmt): |
| 1044 | self.__log_execute(sql_stmt) |
| 1045 | hs2_operation = self.__session.execute(sql_stmt, configuration=self.__query_options) |
| 1046 | operation_handle = MinimalHS2OperationHandle(hs2_operation.handle, sql_stmt) |
| 1047 | self.log_handle(operation_handle, "query started") |
| 1048 | return operation_handle |
| 1049 | |
| 1050 | def __get_operation(self, operation_handle): |
| 1051 | return hs2.Operation(self.__session, operation_handle.get_handle()) |
| 1052 | |
| 1053 | def fetch(self, sql_stmt, operation_handle, max_rows=-1): # noqa: U100 |
| 1054 | """ |
| 1055 | Fetch the results of the query. It will block the current connection if the results |
no outgoing calls