<s> import logging <EOL> import itertools <EOL> import json <EOL> import re <EOL> from itertools import imap , izip <EOL> from operator import itemgetter <EOL> from django . utils . translation import ugettext as _ <EOL> from desktop . lib import thrift_util <EOL> from desktop . conf import DEFAULT_USER <EOL> from hadoop import cluster <EOL> from TCLIService import TCLIService <EOL> from TCLIService . ttypes import TOpenSessionReq , TGetTablesReq , TFetchResultsReq , TStatusCode , TGetResultSetMetadataReq , TGetColumnsReq , TTypeId , TExecuteStatementReq , TGetOperationStatusReq , TFetchOrientation , TCloseSessionReq , TGetSchemasReq , TGetLogReq , TCancelOperationReq , TCloseOperationReq , TFetchResultsResp , TRowSet , TProtocolVersion <EOL> from beeswax import conf as beeswax_conf <EOL> from beeswax import hive_site <EOL> from beeswax . hive_site import hiveserver2_use_ssl <EOL> from beeswax . conf import CONFIG_WHITELIST , LIST_PARTITIONS_LIMIT <EOL> from beeswax . models import Session , HiveServerQueryHandle , HiveServerQueryHistory <EOL> from beeswax . server . dbms import Table , NoSuchObjectException , DataTable , QueryServerException <EOL> LOG = logging . getLogger ( __name__ ) <EOL> IMPALA_RESULTSET_CACHE_SIZE = '<STR_LIT>' <EOL> DEFAULT_USER = DEFAULT_USER . get ( ) <EOL> class HiveServerTable ( Table ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , table_results , table_schema , desc_results , desc_schema ) : <EOL> if beeswax_conf . THRIFT_VERSION . get ( ) >= <NUM_LIT:7> : <EOL> if not table_results . columns : <EOL> raise NoSuchObjectException ( ) <EOL> self . table = table_results . columns <EOL> else : <EOL> if not table_results . rows : <EOL> raise NoSuchObjectException ( ) <EOL> self . table = table_results . rows and table_results . rows [ <NUM_LIT:0> ] or '<STR_LIT>' <EOL> self . table_schema = table_schema <EOL> self . desc_results = desc_results <EOL> self . desc_schema = desc_schema <EOL> self . describe = HiveServerTTableSchema ( self . desc_results , self . desc_schema ) . cols ( ) <EOL> self . _details = None <EOL> @ property <EOL> def name ( self ) : <EOL> return HiveServerTRow ( self . table , self . table_schema ) . col ( '<STR_LIT>' ) <EOL> @ property <EOL> def is_view ( self ) : <EOL> return HiveServerTRow ( self . table , self . table_schema ) . col ( '<STR_LIT>' ) == '<STR_LIT>' <EOL> @ property <EOL> def partition_keys ( self ) : <EOL> try : <EOL> return [ PartitionKeyCompatible ( row [ '<STR_LIT>' ] , row [ '<STR_LIT>' ] , row [ '<STR_LIT>' ] ) for row in self . _get_partition_column ( ) ] <EOL> except : <EOL> LOG . exception ( '<STR_LIT>' ) <EOL> return [ ] <EOL> @ property <EOL> def path_location ( self ) : <EOL> try : <EOL> rows = self . describe <EOL> rows = [ row for row in rows if row [ '<STR_LIT>' ] . startswith ( '<STR_LIT>' ) ] <EOL> if rows : <EOL> return rows [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> except : <EOL> LOG . exception ( '<STR_LIT>' ) <EOL> return None <EOL> @ property <EOL> def cols ( self ) : <EOL> rows = self . describe <EOL> col_row_index = <NUM_LIT:2> <EOL> try : <EOL> end_cols_index = map ( itemgetter ( '<STR_LIT>' ) , rows [ col_row_index : ] ) . index ( '<STR_LIT>' ) <EOL> return rows [ col_row_index : ] [ : end_cols_index ] + self . _get_partition_column ( ) <EOL> except ValueError : <EOL> return rows [ col_row_index : ] <EOL> except : <EOL> return rows <EOL> def _get_partition_column ( self ) : <EOL> rows = self . describe <EOL> try : <EOL> col_row_index = map ( itemgetter ( '<STR_LIT>' ) , rows ) . index ( '<STR_LIT>' ) + <NUM_LIT:3> <EOL> end_cols_index = map ( itemgetter ( '<STR_LIT>' ) , rows [ col_row_index : ] ) . index ( '<STR_LIT>' ) <EOL> return rows [ col_row_index : ] [ : end_cols_index ] <EOL> except : <EOL> return [ ] <EOL> @ property <EOL> def comment ( self ) : <EOL> return HiveServerTRow ( self . table , self . table_schema ) . col ( '<STR_LIT>' ) <EOL> @ property <EOL> def properties ( self ) : <EOL> rows = self . describe <EOL> col_row_index = <NUM_LIT:2> <EOL> end_cols_index = map ( itemgetter ( '<STR_LIT>' ) , rows [ col_row_index : ] ) . index ( '<STR_LIT>' ) <EOL> return [ { <EOL> '<STR_LIT>' : prop [ '<STR_LIT>' ] . strip ( ) if prop [ '<STR_LIT>' ] else prop [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : prop [ '<STR_LIT>' ] . strip ( ) if prop [ '<STR_LIT>' ] else prop [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : prop [ '<STR_LIT>' ] . strip ( ) if prop [ '<STR_LIT>' ] else prop [ '<STR_LIT>' ] <EOL> } for prop in rows [ col_row_index + end_cols_index + <NUM_LIT:1> : ] <EOL> ] <EOL> @ property <EOL> def stats ( self ) : <EOL> rows = self . properties <EOL> col_row_index = map ( itemgetter ( '<STR_LIT>' ) , rows ) . index ( '<STR_LIT>' ) + <NUM_LIT:1> <EOL> end_cols_index = map ( itemgetter ( '<STR_LIT>' ) , rows [ col_row_index : ] ) . index ( None ) <EOL> return rows [ col_row_index : ] [ : end_cols_index ] <EOL> @ property <EOL> def has_complex ( self ) : <EOL> has_complex = False <EOL> complex_types = [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] <EOL> patterns = [ re . compile ( typ ) for typ in complex_types ] <EOL> for column in self . cols : <EOL> if isinstance ( column , dict ) and '<STR_LIT>' in column : <EOL> column_type = column [ '<STR_LIT>' ] <EOL> else : <EOL> column_type = column . type <EOL> if column_type and any ( p . match ( column_type . lower ( ) ) for p in patterns ) : <EOL> has_complex = True <EOL> break <EOL> return has_complex <EOL> @ property <EOL> def details ( self ) : <EOL> if self . _details is None : <EOL> props = dict ( [ ( stat [ '<STR_LIT>' ] , stat [ '<STR_LIT>' ] ) for stat in self . properties if stat [ '<STR_LIT>' ] != '<STR_LIT>' ] ) <EOL> serde = props . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . _details = { <EOL> '<STR_LIT>' : dict ( [ ( stat [ '<STR_LIT>' ] , stat [ '<STR_LIT>' ] ) for stat in self . stats ] ) , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : props . get ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : props . get ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : props . get ( '<STR_LIT>' , '<STR_LIT>' ) != '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' if '<STR_LIT>' in serde else ( '<STR_LIT:text>' if '<STR_LIT>' in serde else serde . rsplit ( '<STR_LIT:.>' , <NUM_LIT:1> ) [ - <NUM_LIT:1> ] ) <EOL> } <EOL> } <EOL> return self . _details <EOL> class HiveServerTRowSet2 : <EOL> def __init__ ( self , row_set , schema ) : <EOL> self . row_set = row_set <EOL> self . rows = row_set . rows <EOL> self . schema = schema <EOL> self . startRowOffset = row_set . startRowOffset <EOL> def is_empty ( self ) : <EOL> return not self . row_set . columns or not HiveServerTColumnValue2 ( self . row_set . columns [ <NUM_LIT:0> ] ) . val <EOL> def cols ( self , col_names ) : <EOL> cols_rows = [ ] <EOL> rs = HiveServerTRow2 ( self . row_set . columns , self . schema ) <EOL> cols = [ rs . full_col ( name ) for name in col_names ] <EOL> for cols_row in itertools . izip ( * cols ) : <EOL> cols_rows . append ( dict ( itertools . izip ( col_names , cols_row ) ) ) <EOL> return cols_rows <EOL> def __iter__ ( self ) : <EOL> return self <EOL> def next ( self ) : <EOL> if self . row_set . columns : <EOL> return HiveServerTRow2 ( self . row_set . columns , self . schema ) <EOL> else : <EOL> raise StopIteration <EOL> class HiveServerTRow2 : <EOL> def __init__ ( self , cols , schema ) : <EOL> self . cols = cols <EOL> self . schema = schema <EOL> def col ( self , colName ) : <EOL> pos = self . _get_col_position ( colName ) <EOL> return HiveServerTColumnValue2 ( self . cols [ pos ] ) . val [ <NUM_LIT:0> ] <EOL> def full_col ( self , colName ) : <EOL> pos = self . _get_col_position ( colName ) <EOL> return HiveServerTColumnValue2 ( self . cols [ pos ] ) . val <EOL> def _get_col_position ( self , column_name ) : <EOL> return filter ( lambda ( i , col ) : col . columnName == column_name , enumerate ( self . schema . columns ) ) [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> def fields ( self ) : <EOL> try : <EOL> return [ HiveServerTColumnValue2 ( field ) . val . pop ( <NUM_LIT:0> ) for field in self . cols ] <EOL> except IndexError : <EOL> raise StopIteration <EOL> class HiveServerTColumnValue2 : <EOL> def __init__ ( self , tcolumn_value ) : <EOL> self . column_value = tcolumn_value <EOL> @ property <EOL> def val ( self ) : <EOL> if self . column_value . stringVal : <EOL> return self . _get_val ( self . column_value . stringVal ) <EOL> elif self . column_value . i16Val is not None : <EOL> return self . _get_val ( self . column_value . i16Val ) <EOL> elif self . column_value . i32Val is not None : <EOL> return self . _get_val ( self . column_value . i32Val ) <EOL> elif self . column_value . i64Val is not None : <EOL> return self . _get_val ( self . column_value . i64Val ) <EOL> elif self . column_value . doubleVal is not None : <EOL> return self . _get_val ( self . column_value . doubleVal ) <EOL> elif self . column_value . boolVal is not None : <EOL> return self . _get_val ( self . column_value . boolVal ) <EOL> elif self . column_value . byteVal is not None : <EOL> return self . _get_val ( self . column_value . byteVal ) <EOL> elif self . column_value . binaryVal is not None : <EOL> return self . _get_val ( self . column_value . binaryVal ) <EOL> @ classmethod <EOL> def _get_val ( cls , column ) : <EOL> column . values = cls . set_nulls ( column . values , column . nulls ) <EOL> column . nulls = '<STR_LIT>' <EOL> return column . values <EOL> @ classmethod <EOL> def mark_nulls ( cls , values , bytestring ) : <EOL> mask = bytearray ( bytestring ) <EOL> for n in mask : <EOL> yield n & <NUM_LIT> <EOL> yield n & <NUM_LIT> <EOL> yield n & <NUM_LIT> <EOL> yield n & <NUM_LIT> <EOL> yield n & <NUM_LIT> <EOL> yield n & <NUM_LIT> <EOL> yield n & <NUM_LIT> <EOL> yield n & <NUM_LIT> <EOL> @ classmethod <EOL> def set_nulls ( cls , values , bytestring ) : <EOL> if bytestring == '<STR_LIT>' or re . match ( '<STR_LIT>' , bytestring ) : <EOL> return values <EOL> else : <EOL> _values = [ None if is_null else value for value , is_null in itertools . izip ( values , cls . mark_nulls ( values , bytestring ) ) ] <EOL> if len ( values ) != len ( _values ) : <EOL> _values . extend ( values [ len ( _values ) : ] ) <EOL> return _values <EOL> class HiveServerDataTable ( DataTable ) : <EOL> def __init__ ( self , results , schema , operation_handle , query_server ) : <EOL> self . schema = schema and schema . schema <EOL> self . row_set = HiveServerTRowSet ( results . results , schema ) <EOL> self . operation_handle = operation_handle <EOL> if query_server [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> self . has_more = results . hasMoreRows <EOL> else : <EOL> self . has_more = not self . row_set . is_empty ( ) <EOL> self . startRowOffset = self . row_set . startRowOffset <EOL> @ property <EOL> def ready ( self ) : <EOL> return True <EOL> def cols ( self ) : <EOL> if self . schema : <EOL> return [ HiveServerTColumnDesc ( col ) for col in self . schema . columns ] <EOL> else : <EOL> return [ ] <EOL> def rows ( self ) : <EOL> for row in self . row_set : <EOL> yield row . fields ( ) <EOL> class HiveServerTTableSchema : <EOL> def __init__ ( self , columns , schema ) : <EOL> self . columns = columns <EOL> self . schema = schema <EOL> def cols ( self ) : <EOL> try : <EOL> return HiveServerTRowSet ( self . columns , self . schema ) . cols ( ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> except : <EOL> cols = HiveServerTRowSet ( self . columns , self . schema ) . cols ( ( '<STR_LIT:name>' , '<STR_LIT:type>' , '<STR_LIT>' ) ) <EOL> for col in cols : <EOL> col [ '<STR_LIT>' ] = col . pop ( '<STR_LIT:name>' ) <EOL> col [ '<STR_LIT>' ] = col . pop ( '<STR_LIT:type>' ) <EOL> return cols <EOL> def col ( self , colName ) : <EOL> pos = self . _get_col_position ( colName ) <EOL> return HiveServerTColumnDesc ( self . columns [ pos ] ) . val <EOL> def _get_col_position ( self , column_name ) : <EOL> return filter ( lambda ( i , col ) : col . columnName == column_name , enumerate ( self . schema . columns ) ) [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> if beeswax_conf . THRIFT_VERSION . get ( ) >= <NUM_LIT:7> : <EOL> HiveServerTRow = HiveServerTRow2 <EOL> HiveServerTRowSet = HiveServerTRowSet2 <EOL> else : <EOL> class HiveServerTRow : <EOL> def __init__ ( self , row , schema ) : <EOL> self . row = row <EOL> self . schema = schema <EOL> def col ( self , colName ) : <EOL> pos = self . _get_col_position ( colName ) <EOL> return HiveServerTColumnValue ( self . row . colVals [ pos ] ) . val <EOL> def _get_col_position ( self , column_name ) : <EOL> return filter ( lambda ( i , col ) : col . columnName == column_name , enumerate ( self . schema . columns ) ) [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> def fields ( self ) : <EOL> return [ HiveServerTColumnValue ( field ) . val for field in self . row . colVals ] <EOL> class HiveServerTRowSet : <EOL> def __init__ ( self , row_set , schema ) : <EOL> self . row_set = row_set <EOL> self . rows = row_set . rows <EOL> self . schema = schema <EOL> self . startRowOffset = row_set . startRowOffset <EOL> def is_empty ( self ) : <EOL> return len ( self . rows ) == <NUM_LIT:0> <EOL> def cols ( self , col_names ) : <EOL> cols_rows = [ ] <EOL> for row in self . rows : <EOL> row = HiveServerTRow ( row , self . schema ) <EOL> cols = { } <EOL> for col_name in col_names : <EOL> cols [ col_name ] = row . col ( col_name ) <EOL> cols_rows . append ( cols ) <EOL> return cols_rows <EOL> def __iter__ ( self ) : <EOL> return self <EOL> def next ( self ) : <EOL> if self . rows : <EOL> return HiveServerTRow ( self . rows . pop ( <NUM_LIT:0> ) , self . schema ) <EOL> else : <EOL> raise StopIteration <EOL> class HiveServerTColumnValue : <EOL> def __init__ ( self , tcolumn_value ) : <EOL> self . column_value = tcolumn_value <EOL> @ property <EOL> def val ( self ) : <EOL> if self . column_value . boolVal is not None : <EOL> return self . column_value . boolVal . value <EOL> elif self . column_value . byteVal is not None : <EOL> return self . column_value . byteVal . value <EOL> elif self . column_value . i16Val is not None : <EOL> return self . column_value . i16Val . value <EOL> elif self . column_value . i32Val is not None : <EOL> return self . column_value . i32Val . value <EOL> elif self . column_value . i64Val is not None : <EOL> return self . column_value . i64Val . value <EOL> elif self . column_value . doubleVal is not None : <EOL> return self . column_value . doubleVal . value <EOL> elif self . column_value . stringVal is not None : <EOL> return self . column_value . stringVal . value <EOL> class HiveServerTColumnDesc : <EOL> def __init__ ( self , column ) : <EOL> self . column = column <EOL> @ property <EOL> def name ( self ) : <EOL> return self . column . columnName <EOL> @ property <EOL> def comment ( self ) : <EOL> return self . column . comment <EOL> @ property <EOL> def type ( self ) : <EOL> return self . get_type ( self . column . typeDesc ) <EOL> @ classmethod <EOL> def get_type ( self , typeDesc ) : <EOL> for ttype in typeDesc . types : <EOL> if ttype . primitiveEntry is not None : <EOL> return TTypeId . _VALUES_TO_NAMES [ ttype . primitiveEntry . type ] <EOL> elif ttype . mapEntry is not None : <EOL> return ttype . mapEntry <EOL> elif ttype . unionEntry is not None : <EOL> return ttype . unionEntry <EOL> elif ttype . arrayEntry is not None : <EOL> return ttype . arrayEntry <EOL> elif ttype . structEntry is not None : <EOL> return ttype . structEntry <EOL> elif ttype . userDefinedTypeEntry is not None : <EOL> return ttype . userDefinedTypeEntry <EOL> class HiveServerClient : <EOL> HS2_MECHANISMS = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> DEFAULT_TABLE_TYPES = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ] <EOL> def __init__ ( self , query_server , user ) : <EOL> self . query_server = query_server <EOL> self . user = user <EOL> use_sasl , mechanism , kerberos_principal_short_name , impersonation_enabled , auth_username , auth_password = self . get_security ( ) <EOL> LOG . info ( <EOL> '<STR_LIT>' % ( <EOL> self . query_server [ '<STR_LIT>' ] , use_sasl , mechanism , kerberos_principal_short_name , impersonation_enabled , auth_username ) <EOL> ) <EOL> self . use_sasl = use_sasl <EOL> self . kerberos_principal_short_name = kerberos_principal_short_name <EOL> self . impersonation_enabled = impersonation_enabled <EOL> if self . query_server [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> from impala import conf as impala_conf <EOL> ssl_enabled = impala_conf . SSL . ENABLED . get ( ) <EOL> ca_certs = impala_conf . SSL . CACERTS . get ( ) <EOL> keyfile = impala_conf . SSL . KEY . get ( ) <EOL> certfile = impala_conf . SSL . CERT . get ( ) <EOL> validate = impala_conf . SSL . VALIDATE . get ( ) <EOL> timeout = impala_conf . SERVER_CONN_TIMEOUT . get ( ) <EOL> else : <EOL> ssl_enabled = hiveserver2_use_ssl ( ) <EOL> ca_certs = beeswax_conf . SSL . CACERTS . get ( ) <EOL> keyfile = beeswax_conf . SSL . KEY . get ( ) <EOL> certfile = beeswax_conf . SSL . CERT . get ( ) <EOL> validate = beeswax_conf . SSL . VALIDATE . get ( ) <EOL> timeout = beeswax_conf . SERVER_CONN_TIMEOUT . get ( ) <EOL> if auth_username : <EOL> username = auth_username <EOL> password = auth_password <EOL> else : <EOL> username = user . username <EOL> password = None <EOL> thrift_class = TCLIService <EOL> if self . query_server [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> from ImpalaService import ImpalaHiveServer2Service <EOL> thrift_class = ImpalaHiveServer2Service <EOL> self . _client = thrift_util . get_client ( thrift_class . Client , <EOL> query_server [ '<STR_LIT>' ] , <EOL> query_server [ '<STR_LIT>' ] , <EOL> service_name = query_server [ '<STR_LIT>' ] , <EOL> kerberos_principal = kerberos_principal_short_name , <EOL> use_sasl = use_sasl , <EOL> mechanism = mechanism , <EOL> username = username , <EOL> password = password , <EOL> timeout_seconds = timeout , <EOL> use_ssl = ssl_enabled , <EOL> ca_certs = ca_certs , <EOL> keyfile = keyfile , <EOL> certfile = certfile , <EOL> validate = validate , <EOL> transport_mode = query_server . get ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> http_url = query_server . get ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> def get_security ( self ) : <EOL> principal = self . query_server [ '<STR_LIT>' ] <EOL> impersonation_enabled = False <EOL> auth_username = self . query_server [ '<STR_LIT>' ] <EOL> auth_password = self . query_server [ '<STR_LIT>' ] <EOL> if principal : <EOL> kerberos_principal_short_name = principal . split ( '<STR_LIT:/>' , <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> else : <EOL> kerberos_principal_short_name = None <EOL> if self . query_server [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> if auth_password : <EOL> use_sasl = True <EOL> mechanism = HiveServerClient . HS2_MECHANISMS [ '<STR_LIT>' ] <EOL> else : <EOL> cluster_conf = cluster . get_cluster_conf_for_job_submission ( ) <EOL> use_sasl = cluster_conf is not None and cluster_conf . SECURITY_ENABLED . get ( ) <EOL> mechanism = HiveServerClient . HS2_MECHANISMS [ '<STR_LIT>' ] <EOL> impersonation_enabled = self . query_server [ '<STR_LIT>' ] <EOL> else : <EOL> hive_mechanism = hive_site . get_hiveserver2_authentication ( ) <EOL> if hive_mechanism not in HiveServerClient . HS2_MECHANISMS : <EOL> raise Exception ( _ ( '<STR_LIT>' ) % ( hive_mechanism , HiveServerClient . HS2_MECHANISMS . keys ( ) ) ) <EOL> use_sasl = hive_mechanism in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> mechanism = HiveServerClient . HS2_MECHANISMS [ hive_mechanism ] <EOL> impersonation_enabled = hive_site . hiveserver2_impersonation_enabled ( ) <EOL> return use_sasl , mechanism , kerberos_principal_short_name , impersonation_enabled , auth_username , auth_password <EOL> def open_session ( self , user ) : <EOL> kwargs = { <EOL> '<STR_LIT>' : beeswax_conf . THRIFT_VERSION . get ( ) - <NUM_LIT:1> , <EOL> '<STR_LIT:username>' : user . username , <EOL> '<STR_LIT>' : { } , <EOL> } <EOL> if self . impersonation_enabled : <EOL> kwargs . update ( { '<STR_LIT:username>' : DEFAULT_USER } ) <EOL> if self . query_server [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> kwargs [ '<STR_LIT>' ] . update ( { '<STR_LIT>' : user . username } ) <EOL> if self . query_server [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> kwargs [ '<STR_LIT>' ] . update ( { '<STR_LIT>' : user . username } ) <EOL> kwargs [ '<STR_LIT>' ] . update ( { '<STR_LIT>' : '<STR_LIT:true>' } ) <EOL> kwargs [ '<STR_LIT>' ] . update ( { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> if self . query_server [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> kwargs [ '<STR_LIT>' ] . update ( { '<STR_LIT>' : user . username } ) <EOL> if self . query_server [ '<STR_LIT>' ] == '<STR_LIT>' and self . query_server [ '<STR_LIT>' ] > <NUM_LIT:0> : <EOL> kwargs [ '<STR_LIT>' ] . update ( { '<STR_LIT>' : str ( self . query_server [ '<STR_LIT>' ] ) } ) <EOL> LOG . info ( '<STR_LIT>' % ( self . query_server [ '<STR_LIT>' ] , user . username ) ) <EOL> req = TOpenSessionReq ( ** kwargs ) <EOL> res = self . _client . OpenSession ( req ) <EOL> if res . status is not None and res . status . statusCode not in ( TStatusCode . SUCCESS_STATUS , ) : <EOL> if hasattr ( res . status , '<STR_LIT>' ) and res . status . errorMessage : <EOL> message = res . status . errorMessage <EOL> else : <EOL> message = '<STR_LIT>' <EOL> raise QueryServerException ( Exception ( '<STR_LIT>' % ( req , res ) ) , message = message ) <EOL> sessionId = res . sessionHandle . sessionId <EOL> LOG . info ( '<STR_LIT>' % repr ( sessionId . guid ) ) <EOL> encoded_status , encoded_guid = HiveServerQueryHandle ( secret = sessionId . secret , guid = sessionId . guid ) . get ( ) <EOL> properties = json . dumps ( res . configuration ) <EOL> session = Session . objects . create ( owner = user , <EOL> application = self . query_server [ '<STR_LIT>' ] , <EOL> status_code = res . status . statusCode , <EOL> secret = encoded_status , <EOL> guid = encoded_guid , <EOL> server_protocol_version = res . serverProtocolVersion , <EOL> properties = properties ) <EOL> if not session . get_properties ( ) : <EOL> session . properties = json . dumps ( self . get_configuration ( ) ) <EOL> session . save ( ) <EOL> return session <EOL> def call ( self , fn , req , status = TStatusCode . SUCCESS_STATUS ) : <EOL> session = Session . objects . get_session ( self . user , self . query_server [ '<STR_LIT>' ] ) <EOL> if session is None : <EOL> session = self . open_session ( self . user ) <EOL> if hasattr ( req , '<STR_LIT>' ) and req . sessionHandle is None : <EOL> req . sessionHandle = session . get_handle ( ) <EOL> res = fn ( req ) <EOL> if res . status . statusCode == TStatusCode . ERROR_STATUS and re . search ( '<STR_LIT>' , res . status . errorMessage or '<STR_LIT>' , re . I ) : <EOL> LOG . info ( '<STR_LIT>' % ( self . user , res ) ) <EOL> session = self . open_session ( self . user ) <EOL> req . sessionHandle = session . get_handle ( ) <EOL> res = getattr ( self . _client , fn . attr ) ( req ) <EOL> if status is not None and res . status . statusCode not in ( <EOL> TStatusCode . SUCCESS_STATUS , TStatusCode . SUCCESS_WITH_INFO_STATUS , TStatusCode . STILL_EXECUTING_STATUS ) : <EOL> if hasattr ( res . status , '<STR_LIT>' ) and res . status . errorMessage : <EOL> message = res . status . errorMessage <EOL> else : <EOL> message = '<STR_LIT>' <EOL> raise QueryServerException ( Exception ( '<STR_LIT>' % ( req , res ) ) , message = message ) <EOL> else : <EOL> return res <EOL> def close_session ( self , sessionHandle ) : <EOL> req = TCloseSessionReq ( sessionHandle = sessionHandle ) <EOL> return self . _client . CloseSession ( req ) <EOL> def get_databases ( self , schemaName = None ) : <EOL> req = TGetSchemasReq ( ) <EOL> if schemaName is not None : <EOL> req . schemaName = schemaName <EOL> if self . query_server [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> req . schemaName = None <EOL> res = self . call ( self . _client . GetSchemas , req ) <EOL> results , schema = self . fetch_result ( res . operationHandle , orientation = TFetchOrientation . FETCH_NEXT , max_rows = <NUM_LIT> ) <EOL> self . close_operation ( res . operationHandle ) <EOL> col = '<STR_LIT>' <EOL> return HiveServerTRowSet ( results . results , schema . schema ) . cols ( ( col , ) ) <EOL> def get_database ( self , database ) : <EOL> if self . query_server [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> raise NotImplementedError ( _ ( "<STR_LIT>" ) % { <EOL> '<STR_LIT>' : "<STR_LIT>" <EOL> } ) <EOL> query = '<STR_LIT>' % ( database ) <EOL> ( desc_results , desc_schema ) , operation_handle = self . execute_statement ( query , max_rows = <NUM_LIT> , orientation = TFetchOrientation . FETCH_NEXT ) <EOL> self . close_operation ( operation_handle ) <EOL> cols = ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:location>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> if len ( HiveServerTRowSet ( desc_results . results , desc_schema . schema ) . cols ( cols ) ) != <NUM_LIT:1> : <EOL> raise ValueError ( _ ( "<STR_LIT>" ) % { '<STR_LIT>' : query } ) <EOL> return HiveServerTRowSet ( desc_results . results , desc_schema . schema ) . cols ( cols ) [ <NUM_LIT:0> ] <EOL> def get_tables_meta ( self , database , table_names , table_types = None ) : <EOL> if not table_types : <EOL> table_types = self . DEFAULT_TABLE_TYPES <EOL> req = TGetTablesReq ( schemaName = database , tableName = table_names , tableTypes = table_types ) <EOL> res = self . call ( self . _client . GetTables , req ) <EOL> results , schema = self . fetch_result ( res . operationHandle , orientation = TFetchOrientation . FETCH_NEXT , max_rows = <NUM_LIT> ) <EOL> self . close_operation ( res . operationHandle ) <EOL> cols = ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> return HiveServerTRowSet ( results . results , schema . schema ) . cols ( cols ) <EOL> def get_tables ( self , database , table_names , table_types = None ) : <EOL> if not table_types : <EOL> table_types = self . DEFAULT_TABLE_TYPES <EOL> req = TGetTablesReq ( schemaName = database , tableName = table_names , tableTypes = table_types ) <EOL> res = self . call ( self . _client . GetTables , req ) <EOL> results , schema = self . fetch_result ( res . operationHandle , orientation = TFetchOrientation . FETCH_NEXT , max_rows = <NUM_LIT> ) <EOL> self . close_operation ( res . operationHandle ) <EOL> return HiveServerTRowSet ( results . results , schema . schema ) . cols ( ( '<STR_LIT>' , ) ) <EOL> def get_table ( self , database , table_name , partition_spec = None ) : <EOL> req = TGetTablesReq ( schemaName = database , tableName = table_name ) <EOL> res = self . call ( self . _client . GetTables , req ) <EOL> table_results , table_schema = self . fetch_result ( res . operationHandle , orientation = TFetchOrientation . FETCH_NEXT ) <EOL> self . close_operation ( res . operationHandle ) <EOL> if partition_spec : <EOL> query = '<STR_LIT>' % ( database , table_name , partition_spec ) <EOL> else : <EOL> query = '<STR_LIT>' % ( database , table_name ) <EOL> ( desc_results , desc_schema ) , operation_handle = self . execute_statement ( query , max_rows = <NUM_LIT> , orientation = TFetchOrientation . FETCH_NEXT ) <EOL> self . close_operation ( operation_handle ) <EOL> return HiveServerTable ( table_results . results , table_schema . schema , desc_results . results , desc_schema . schema ) <EOL> def execute_query ( self , query , max_rows = <NUM_LIT:1000> ) : <EOL> configuration = self . _get_query_configuration ( query ) <EOL> return self . execute_query_statement ( statement = query . query [ '<STR_LIT>' ] , max_rows = max_rows , configuration = configuration ) <EOL> def execute_query_statement ( self , statement , max_rows = <NUM_LIT:1000> , configuration = { } , orientation = TFetchOrientation . FETCH_FIRST ) : <EOL> ( results , schema ) , operation_handle = self . execute_statement ( statement = statement , max_rows = max_rows , configuration = configuration , orientation = orientation ) <EOL> return HiveServerDataTable ( results , schema , operation_handle , self . query_server ) <EOL> def execute_async_query ( self , query , statement = <NUM_LIT:0> ) : <EOL> if statement == <NUM_LIT:0> : <EOL> if self . query_server [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> for resource in query . get_configuration_statements ( ) : <EOL> self . execute_statement ( resource . strip ( ) ) <EOL> configuration = { } <EOL> if self . query_server [ '<STR_LIT>' ] == '<STR_LIT>' and self . query_server [ '<STR_LIT>' ] > <NUM_LIT:0> : <EOL> configuration [ IMPALA_RESULTSET_CACHE_SIZE ] = str ( self . query_server [ '<STR_LIT>' ] ) <EOL> configuration . update ( self . _get_query_configuration ( query ) ) <EOL> query_statement = query . get_query_statement ( statement ) <EOL> return self . execute_async_statement ( statement = query_statement , confOverlay = configuration ) <EOL> def execute_statement ( self , statement , max_rows = <NUM_LIT:1000> , configuration = { } , orientation = TFetchOrientation . FETCH_NEXT ) : <EOL> if self . query_server [ '<STR_LIT>' ] == '<STR_LIT>' and self . query_server [ '<STR_LIT>' ] > <NUM_LIT:0> : <EOL> configuration [ '<STR_LIT>' ] = str ( self . query_server [ '<STR_LIT>' ] ) <EOL> req = TExecuteStatementReq ( statement = statement . encode ( '<STR_LIT:utf-8>' ) , confOverlay = configuration ) <EOL> res = self . call ( self . _client . ExecuteStatement , req ) <EOL> return self . fetch_result ( res . operationHandle , max_rows = max_rows , orientation = orientation ) , res . operationHandle <EOL> def execute_async_statement ( self , statement , confOverlay ) : <EOL> if self . query_server [ '<STR_LIT>' ] == '<STR_LIT>' and self . query_server [ '<STR_LIT>' ] > <NUM_LIT:0> : <EOL> confOverlay [ '<STR_LIT>' ] = str ( self . query_server [ '<STR_LIT>' ] ) <EOL> req = TExecuteStatementReq ( statement = statement . encode ( '<STR_LIT:utf-8>' ) , confOverlay = confOverlay , runAsync = True ) <EOL> res = self . call ( self . _client . ExecuteStatement , req ) <EOL> return HiveServerQueryHandle ( secret = res . operationHandle . operationId . secret , <EOL> guid = res . operationHandle . operationId . guid , <EOL> operation_type = res . operationHandle . operationType , <EOL> has_result_set = res . operationHandle . hasResultSet , <EOL> modified_row_count = res . operationHandle . modifiedRowCount ) <EOL> def fetch_data ( self , operation_handle , orientation = TFetchOrientation . FETCH_NEXT , max_rows = <NUM_LIT:1000> ) : <EOL> results , schema = self . fetch_result ( operation_handle , orientation , max_rows ) <EOL> return HiveServerDataTable ( results , schema , operation_handle , self . query_server ) <EOL> def cancel_operation ( self , operation_handle ) : <EOL> req = TCancelOperationReq ( operationHandle = operation_handle ) <EOL> return self . call ( self . _client . CancelOperation , req ) <EOL> def close_operation ( self , operation_handle ) : <EOL> req = TCloseOperationReq ( operationHandle = operation_handle ) <EOL> return self . call ( self . _client . CloseOperation , req ) <EOL> def get_columns ( self , database , table ) : <EOL> req = TGetColumnsReq ( schemaName = database , tableName = table ) <EOL> res = self . call ( self . _client . GetColumns , req ) <EOL> res , schema = self . fetch_result ( res . operationHandle , orientation = TFetchOrientation . FETCH_NEXT ) <EOL> self . close_operation ( res . operationHandle ) <EOL> return res , schema <EOL> def fetch_result ( self , operation_handle , orientation = TFetchOrientation . FETCH_FIRST , max_rows = <NUM_LIT:1000> ) : <EOL> if operation_handle . hasResultSet : <EOL> fetch_req = TFetchResultsReq ( operationHandle = operation_handle , orientation = orientation , maxRows = max_rows ) <EOL> res = self . call ( self . _client . FetchResults , fetch_req ) <EOL> else : <EOL> res = TFetchResultsResp ( results = TRowSet ( startRowOffset = <NUM_LIT:0> , rows = [ ] , columns = [ ] ) ) <EOL> if operation_handle . hasResultSet and TFetchOrientation . FETCH_FIRST : <EOL> meta_req = TGetResultSetMetadataReq ( operationHandle = operation_handle ) <EOL> schema = self . call ( self . _client . GetResultSetMetadata , meta_req ) <EOL> else : <EOL> schema = None <EOL> return res , schema <EOL> def fetch_log ( self , operation_handle , orientation = TFetchOrientation . FETCH_NEXT , max_rows = <NUM_LIT:1000> ) : <EOL> req = TFetchResultsReq ( operationHandle = operation_handle , orientation = orientation , maxRows = max_rows , fetchType = <NUM_LIT:1> ) <EOL> res = self . call ( self . _client . FetchResults , req ) <EOL> if beeswax_conf . THRIFT_VERSION . get ( ) >= <NUM_LIT:7> : <EOL> lines = res . results . columns [ <NUM_LIT:0> ] . stringVal . values <EOL> else : <EOL> lines = imap ( lambda r : r . colVals [ <NUM_LIT:0> ] . stringVal . value , res . results . rows ) <EOL> return '<STR_LIT:\n>' . join ( lines ) <EOL> def get_operation_status ( self , operation_handle ) : <EOL> req = TGetOperationStatusReq ( operationHandle = operation_handle ) <EOL> return self . call ( self . _client . GetOperationStatus , req ) <EOL> def explain ( self , query ) : <EOL> query_statement = query . get_query_statement ( <NUM_LIT:0> ) <EOL> configuration = self . _get_query_configuration ( query ) <EOL> return self . execute_query_statement ( statement = '<STR_LIT>' % query_statement , configuration = configuration , orientation = TFetchOrientation . FETCH_NEXT ) <EOL> def get_log ( self , operation_handle ) : <EOL> try : <EOL> req = TGetLogReq ( operationHandle = operation_handle ) <EOL> res = self . call ( self . _client . GetLog , req ) <EOL> return res . log <EOL> except : <EOL> LOG . exception ( '<STR_LIT>' ) <EOL> return '<STR_LIT>' <EOL> def get_partitions ( self , database , table_name , partition_spec = None , max_parts = None , reverse_sort = True ) : <EOL> table = self . get_table ( database , table_name ) <EOL> query = '<STR_LIT>' % ( database , table_name ) <EOL> if self . query_server [ '<STR_LIT>' ] == '<STR_LIT>' and partition_spec : <EOL> query += '<STR_LIT>' % partition_spec <EOL> partition_table = self . execute_query_statement ( query , max_rows = <NUM_LIT> , orientation = TFetchOrientation . FETCH_NEXT ) <EOL> if self . query_server [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> try : <EOL> cols = [ col . name for col in partition_table . cols ( ) ] <EOL> stop = cols . index ( '<STR_LIT>' ) <EOL> partition_keys = cols [ : stop ] <EOL> num_parts = len ( partition_keys ) <EOL> rows = partition_table . rows ( ) <EOL> partition_values = [ partition [ : num_parts ] for partition in rows ] <EOL> partition_values = partition_values [ : - <NUM_LIT:1> ] <EOL> partitions_formatted = [ ] <EOL> for values in partition_values : <EOL> zipped_parts = izip ( partition_keys , values ) <EOL> partitions_formatted . append ( [ '<STR_LIT:/>' . join ( [ '<STR_LIT>' % ( part [ <NUM_LIT:0> ] , part [ <NUM_LIT:1> ] ) for part in zipped_parts ] ) ] ) <EOL> partitions = [ PartitionValueCompatible ( partition , table ) for partition in partitions_formatted ] <EOL> except Exception , e : <EOL> raise ValueError ( _ ( '<STR_LIT>' ) % ( database , table_name ) ) <EOL> else : <EOL> partitions = [ PartitionValueCompatible ( partition , table ) for partition in partition_table . rows ( ) ] <EOL> if reverse_sort : <EOL> partitions . reverse ( ) <EOL> if max_parts is None or max_parts <= <NUM_LIT:0> : <EOL> max_parts = LIST_PARTITIONS_LIMIT . get ( ) <EOL> return partitions [ : max_parts ] <EOL> def get_configuration ( self ) : <EOL> configuration = { } <EOL> if self . query_server [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> query = '<STR_LIT>' <EOL> results = self . execute_query_statement ( query , orientation = TFetchOrientation . FETCH_NEXT ) <EOL> configuration = dict ( ( row [ <NUM_LIT:0> ] , row [ <NUM_LIT:1> ] ) for row in results . rows ( ) ) <EOL> else : <EOL> query = '<STR_LIT>' <EOL> results = self . execute_query_statement ( query , orientation = TFetchOrientation . FETCH_FIRST ) <EOL> config_whitelist = [ config . lower ( ) for config in CONFIG_WHITELIST . get ( ) ] <EOL> properties = [ ( row [ <NUM_LIT:0> ] . split ( '<STR_LIT:=>' ) [ <NUM_LIT:0> ] , row [ <NUM_LIT:0> ] . split ( '<STR_LIT:=>' ) [ <NUM_LIT:1> ] ) for row in results . rows ( ) if '<STR_LIT:=>' in row [ <NUM_LIT:0> ] ] <EOL> configuration = dict ( ( prop , value ) for prop , value in properties if prop . lower ( ) in config_whitelist ) <EOL> return configuration <EOL> def _get_query_configuration ( self , query ) : <EOL> return dict ( [ ( setting [ '<STR_LIT:key>' ] , setting [ '<STR_LIT:value>' ] ) for setting in query . settings ] ) <EOL> class HiveServerTableCompatible ( HiveServerTable ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , hive_table ) : <EOL> self . table = hive_table . table <EOL> self . table_schema = hive_table . table_schema <EOL> self . desc_results = hive_table . desc_results <EOL> self . desc_schema = hive_table . desc_schema <EOL> self . describe = HiveServerTTableSchema ( self . desc_results , self . desc_schema ) . cols ( ) <EOL> self . _details = None <EOL> @ property <EOL> def cols ( self ) : <EOL> return [ <EOL> type ( '<STR_LIT>' , ( object , ) , { <EOL> '<STR_LIT:name>' : col . get ( '<STR_LIT>' , '<STR_LIT>' ) . strip ( ) if col . get ( '<STR_LIT>' ) else '<STR_LIT>' , <EOL> '<STR_LIT:type>' : col . get ( '<STR_LIT>' , '<STR_LIT>' ) . strip ( ) if col . get ( '<STR_LIT>' ) else '<STR_LIT>' , <EOL> '<STR_LIT>' : col . get ( '<STR_LIT>' , '<STR_LIT>' ) . strip ( ) if col . get ( '<STR_LIT>' ) else '<STR_LIT>' <EOL> } ) for col in HiveServerTable . cols . fget ( self ) <EOL> ] <EOL> class ResultCompatible : <EOL> def __init__ ( self , data_table ) : <EOL> self . data_table = data_table <EOL> self . rows = data_table . rows <EOL> self . has_more = data_table . has_more <EOL> self . start_row = data_table . startRowOffset <EOL> self . ready = True <EOL> @ property <EOL> def columns ( self ) : <EOL> return self . cols ( ) <EOL> def cols ( self ) : <EOL> return [ col . name for col in self . data_table . cols ( ) ] <EOL> class PartitionKeyCompatible : <EOL> def __init__ ( self , name , type , comment ) : <EOL> self . name = name <EOL> self . type = type <EOL> self . comment = comment <EOL> def __eq__ ( self , other ) : <EOL> return isinstance ( other , PartitionKeyCompatible ) and self . name == other . name and self . type == other . type and self . comment == other . comment <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( self . name , self . type , self . comment ) <EOL> class PartitionValueCompatible : <EOL> def __init__ ( self , partition_row , table , properties = None ) : <EOL> if properties is None : <EOL> properties = { } <EOL> partition = partition_row [ <NUM_LIT:0> ] <EOL> parts = partition . split ( '<STR_LIT:/>' ) <EOL> self . partition_spec = '<STR_LIT:U+002C>' . join ( [ "<STR_LIT>" % ( pv [ <NUM_LIT:0> ] , pv [ <NUM_LIT:1> ] ) for pv in [ part . split ( '<STR_LIT:=>' ) for part in parts ] ] ) <EOL> self . values = [ pv [ <NUM_LIT:1> ] for pv in [ part . split ( '<STR_LIT:=>' ) for part in parts ] ] <EOL> self . sd = type ( '<STR_LIT>' , ( object , ) , properties , ) <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( self . partition_spec , self . values , self . sd ) <EOL> class ExplainCompatible : <EOL> def __init__ ( self , data_table ) : <EOL> self . textual = '<STR_LIT:\n>' . join ( [ line [ <NUM_LIT:0> ] for line in data_table . rows ( ) ] ) <EOL> class ResultMetaCompatible : <EOL> def __init__ ( self ) : <EOL> self . in_tablename = True <EOL> class HiveServerClientCompatible ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , client ) : <EOL> self . _client = client <EOL> self . user = client . user <EOL> self . query_server = client . query_server <EOL> def query ( self , query , statement = <NUM_LIT:0> ) : <EOL> return self . _client . execute_async_query ( query , statement ) <EOL> def get_state ( self , handle ) : <EOL> operationHandle = handle . get_rpc_handle ( ) <EOL> res = self . _client . get_operation_status ( operationHandle ) <EOL> return HiveServerQueryHistory . STATE_MAP [ res . operationState ] <EOL> def get_operation_status ( self , handle ) : <EOL> operationHandle = handle . get_rpc_handle ( ) <EOL> return self . _client . get_operation_status ( operationHandle ) <EOL> def use ( self , query ) : <EOL> data = self . _client . execute_query ( query ) <EOL> self . _client . close_operation ( data . operation_handle ) <EOL> return data <EOL> def explain ( self , query ) : <EOL> data_table = self . _client . explain ( query ) <EOL> data = ExplainCompatible ( data_table ) <EOL> self . _client . close_operation ( data_table . operation_handle ) <EOL> return data <EOL> def fetch ( self , handle , start_over = False , max_rows = None ) : <EOL> operationHandle = handle . get_rpc_handle ( ) <EOL> if max_rows is None : <EOL> max_rows = <NUM_LIT:1000> <EOL> if start_over and not ( self . query_server [ '<STR_LIT>' ] == '<STR_LIT>' and self . query_server [ '<STR_LIT>' ] == <NUM_LIT:0> ) : <EOL> orientation = TFetchOrientation . FETCH_FIRST <EOL> else : <EOL> orientation = TFetchOrientation . FETCH_NEXT <EOL> data_table = self . _client . fetch_data ( operationHandle , orientation = orientation , max_rows = max_rows ) <EOL> return ResultCompatible ( data_table ) <EOL> def cancel_operation ( self , handle ) : <EOL> operationHandle = handle . get_rpc_handle ( ) <EOL> return self . _client . cancel_operation ( operationHandle ) <EOL> def close ( self , handle ) : <EOL> return self . close_operation ( handle ) <EOL> def close_operation ( self , handle ) : <EOL> operationHandle = handle . get_rpc_handle ( ) <EOL> return self . _client . close_operation ( operationHandle ) <EOL> def close_session ( self , session ) : <EOL> operationHandle = session . get_handle ( ) <EOL> return self . _client . close_session ( operationHandle ) <EOL> def dump_config ( self ) : <EOL> return '<STR_LIT>' <EOL> def get_log ( self , handle , start_over = True ) : <EOL> operationHandle = handle . get_rpc_handle ( ) <EOL> if beeswax_conf . USE_GET_LOG_API . get ( ) or self . query_server [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> return self . _client . get_log ( operationHandle ) <EOL> else : <EOL> if start_over : <EOL> orientation = TFetchOrientation . FETCH_FIRST <EOL> else : <EOL> orientation = TFetchOrientation . FETCH_NEXT <EOL> return self . _client . fetch_log ( operationHandle , orientation = orientation , max_rows = - <NUM_LIT:1> ) <EOL> def get_databases ( self , schemaName = None ) : <EOL> col = '<STR_LIT>' <EOL> return [ table [ col ] for table in self . _client . get_databases ( schemaName ) ] <EOL> def get_database ( self , database ) : <EOL> return self . _client . get_database ( database ) <EOL> def get_tables_meta ( self , database , table_names , table_types = None ) : <EOL> tables = self . _client . get_tables_meta ( database , table_names , table_types ) <EOL> massaged_tables = [ ] <EOL> for table in tables : <EOL> massaged_tables . append ( { <EOL> '<STR_LIT:name>' : table [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : table [ '<STR_LIT>' ] , <EOL> '<STR_LIT:type>' : table [ '<STR_LIT>' ] . capitalize ( ) } <EOL> ) <EOL> return massaged_tables <EOL> def get_tables ( self , database , table_names , table_types = None ) : <EOL> tables = [ table [ '<STR_LIT>' ] for table in self . _client . get_tables ( database , table_names , table_types ) ] <EOL> tables . sort ( ) <EOL> return tables <EOL> def get_table ( self , database , table_name , partition_spec = None ) : <EOL> table = self . _client . get_table ( database , table_name , partition_spec ) <EOL> return HiveServerTableCompatible ( table ) <EOL> def get_columns ( self , database , table ) : <EOL> return self . _client . get_columns ( database , table ) <EOL> def get_default_configuration ( self , * args , ** kwargs ) : <EOL> return [ ] <EOL> def get_results_metadata ( self , handle ) : <EOL> return ResultMetaCompatible ( ) <EOL> def create_database ( self , name , description ) : raise NotImplementedError ( ) <EOL> def alter_table ( self , dbname , tbl_name , new_tbl ) : raise NotImplementedError ( ) <EOL> def open_session ( self , user ) : <EOL> return self . _client . open_session ( user ) <EOL> def add_partition ( self , new_part ) : raise NotImplementedError ( ) <EOL> def get_partition ( self , * args , ** kwargs ) : raise NotImplementedError ( ) <EOL> def get_partitions ( self , database , table_name , partition_spec , max_parts , reverse_sort = True ) : <EOL> return self . _client . get_partitions ( database , table_name , partition_spec , max_parts , reverse_sort ) <EOL> def alter_partition ( self , db_name , tbl_name , new_part ) : raise NotImplementedError ( ) <EOL> def get_configuration ( self ) : <EOL> return self . _client . get_configuration ( ) </s>
