https://github.com/apache/airflow/blob/6c458f29c0eeadb1282e524e76fdd379d6436824/airflow/providers/apache/hive/hooks/hive.py#L464

It should be:
```
if hasattr(self, 'sub_process')
```
airflow/providers/apache/hive/hooks/hive.py
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 load_df(