def test_hql(self, hql):
        """
        Test an hql statement using the hive cli and EXPLAIN

        """
        create, insert, other = [], [], []
        for query in hql.split(';'):  # naive
            query_original = query
            query = query.lower().strip()

            if query.startswith('create table'):
                create.append(query_original)
            elif query.startswith(('set ',
                                   'add jar ',
                                   'create temporary function')):
                other.append(query_original)
            elif query.startswith('insert'):
                insert.append(query_original)
        other = ';'.join(other)
        for query_set in [create, insert]:
            for query in query_set:

                query_preview = ' '.join(query.split())[:50]
                self.log.info("Testing HQL [%s (...)]", query_preview)
                if query_set == insert:
                    query = other + '; explain ' + query
                else:
                    query = 'explain ' + query
                try:
                    self.run_cli(query, verbose=False)
                except AirflowException as e:
                    message = e.args[0].split('\n')[-2]
                    self.log.info(message)
                    error_loc = re.search(r'(\d+):(\d+)', message)
                    if error_loc and error_loc.group(1).isdigit():
                        lst = int(error_loc.group(1))
                        begin = max(lst - 2, 0)
                        end = min(lst + 3, len(query.split('\n')))
                        context = '\n'.join(query.split('\n')[begin:end])
                        self.log.info("Context :\n %s", context)
                else:
                    self.log.info("SUCCESS")
def test_hql ( self , hql ) : create , insert , other = [ ] , [ ] , [ ] for query in hql . split ( ';' ) : query_original = query query = query . lower ( ) . strip ( ) if query . startswith ( 'create table' ) : create . append ( query_original ) elif query . startswith ( ( 'set ' , 'add jar ' , 'create temporary function' ) ) : other . append ( query_original ) elif query . startswith ( 'insert' ) : insert . append ( query_original ) other = ';' . join ( other ) for query_set in [ create , insert ] : for query in query_set : query_preview = ' ' . join ( query . split ( ) ) [ : 50 ] self . log . info ( "Testing HQL [%s (...)]" , query_preview ) if query_set == insert : query = other + '; explain ' + query else : query = 'explain ' + query try : self . run_cli ( query , verbose = False ) except AirflowException as e : message = e . args [ 0 ] . split ( '\n' ) [ - 2 ] self . log . info ( message ) error_loc = re . search ( r'(\d+):(\d+)' , message ) if error_loc and error_loc . group ( 1 ) . isdigit ( ) : lst = int ( error_loc . group ( 1 ) ) begin = max ( lst - 2 , 0 ) end = min ( lst + 3 , len ( query . split ( '\n' ) ) ) context = '\n' . join ( query . split ( '\n' ) [ begin : end ] ) self . log . info ( "Context :\n %s" , context ) else : self . log . info ( "SUCCESS" )
Test an hql statement using the hive cli and EXPLAIN