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
['-hiveconf', f'mapreduce.job.priority={self.mapred_queue_priority}']
                    )
                if self.mapred_job_name:
                    hive_conf_params.extend(['-hiveconf', f'mapred.job.name={self.mapred_job_name}'])
                hive_cmd.extend(hive_conf_params)
                hive_cmd.extend(['-f', f.name])
                if verbose:
                    self.log.info("%s", " ".join(hive_cmd))
                sub_process: Any = subprocess.Popen(
                    hive_cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, cwd=tmp_dir, close_fds=True
                )
                self.sub_process = sub_process
                stdout = ''
                while True:
                    line = sub_process.stdout.readline()
                    if not line:
                        break
                    stdout += line.decode('UTF-8')
                    if verbose:
                        self.log.info(line.decode('UTF-8').strip())
                sub_process.wait()
                if sub_process.returncode:
                    raise AirflowException(stdout)
                return stdout
    def test_hql(self, hql: Union[str, Text]) -> None:
        """Test an hql statement using the hive cli and EXPLAIN"""
        create, insert, other = [], [], []
        for query in hql.split(';'):  
            query_original = query
            query = query.lower().strip()