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
port=db.port,
            auth=auth_mechanism,
            kerberos_service_name=kerberos_service_name,
            username=db.login or username,
            password=password,
            database=schema or db.schema or 'default',
        )
        
    def _get_results(
        self,
        hql: Union[str, Text, List[str]],
        schema: str = 'default',
        fetch_size: Optional[int] = None,
        hive_conf: Optional[Dict[Any, Any]] = None,
    ) -> Any:
        from pyhive.exc import ProgrammingError
        if isinstance(hql, str):
            hql = [hql]
        previous_description = None
        with contextlib.closing(self.get_conn(schema)) as conn, contextlib.closing(conn.cursor()) as cur:
            cur.arraysize = fetch_size or 1000
            
            
            db = self.get_connection(self.hiveserver2_conn_id)  
            
            if db.extra_dejson.get('run_set_variable_statements', True):
                env_context = get_context_from_env_var()
                if hive_conf:
                    env_context.update(hive_conf)
                for k, v in env_context.items():