Legacy Beeswax client. Uses the Beeswax protocol plus Impala-specific extensions. TODO: remove once we've phased out beeswax.
| 1275 | |
| 1276 | |
| 1277 | class ImpalaBeeswaxClient(ImpalaClient): |
| 1278 | """Legacy Beeswax client. Uses the Beeswax protocol plus Impala-specific extensions. |
| 1279 | TODO: remove once we've phased out beeswax.""" |
| 1280 | def __init__(self, *args, **kwargs): |
| 1281 | super(ImpalaBeeswaxClient, self).__init__(*args, **kwargs) |
| 1282 | assert not self.use_http_base_transport |
| 1283 | self.FINISHED_STATE = QueryState._NAMES_TO_VALUES["FINISHED"] |
| 1284 | self.ERROR_STATE = QueryState._NAMES_TO_VALUES["EXCEPTION"] |
| 1285 | self.CANCELED_STATE = QueryState._NAMES_TO_VALUES["EXCEPTION"] |
| 1286 | |
| 1287 | def _get_thrift_client(self, protocol): |
| 1288 | return ImpalaService.Client(protocol) |
| 1289 | |
| 1290 | def _options_to_string_list(self, set_query_options): |
| 1291 | if sys.version_info.major < 3: |
| 1292 | key_value_pairs = set_query_options.iteritems() |
| 1293 | else: |
| 1294 | key_value_pairs = set_query_options.items() |
| 1295 | return [utf8_encode_if_needed("%s=%s" % (k, v)) for (k, v) in key_value_pairs] |
| 1296 | |
| 1297 | def _open_session(self): |
| 1298 | # Beeswax doesn't have a "session" concept independent of connections, so |
| 1299 | # we do not need to explicitly open a sesion. We still need to set up the |
| 1300 | # query options. |
| 1301 | # |
| 1302 | # The default query options are retrieved from a rpc call, and are dependent |
| 1303 | # on the impalad to which a connection has been established. They need to be |
| 1304 | # refreshed each time a connection is made. This is particularly helpful when |
| 1305 | # there is a version mismatch between the shell and the impalad. |
| 1306 | try: |
| 1307 | get_default_query_options = self.imp_service.get_default_configuration(False) |
| 1308 | except Exception: |
| 1309 | return |
| 1310 | rpc_result = self._do_beeswax_rpc(lambda: get_default_query_options) |
| 1311 | options, status = rpc_result |
| 1312 | if status != RpcStatus.OK: |
| 1313 | raise RPCException("Unable to retrieve default query options") |
| 1314 | |
| 1315 | for option in options: |
| 1316 | self.default_query_options[option.key.upper()] = option.value |
| 1317 | # If connected to an Impala that predates IMPALA-2181 then the received options |
| 1318 | # wouldn't contain a level attribute. In this case the query_option_levels |
| 1319 | # map is left empty. |
| 1320 | if option.level is not None: |
| 1321 | self.query_option_levels[option.key.upper()] = option.level |
| 1322 | |
| 1323 | def close_connection(self): |
| 1324 | # Beeswax sessions are scoped to the connection, so we only need to close transport. |
| 1325 | self._close_transport() |
| 1326 | |
| 1327 | def _ping_impala_service(self): |
| 1328 | try: |
| 1329 | resp = self.imp_service.PingImpalaService() |
| 1330 | except TApplicationException as t: |
| 1331 | if t.type == TApplicationException.UNKNOWN_METHOD: |
| 1332 | raise MissingThriftMethodException(t.message) |
| 1333 | raise |
| 1334 | except TTransportException as e: |
no outgoing calls