<SYSTEM_TASK:>
Solve the following problem using Python, implementing the functions described below, one line at a time
<END_TASK>
<USER_TASK:>
Description:
 def _submit_query_job(self, query_data):
 """ Submit a query job to BigQuery. This is similar to BigQueryClient.query, but gives the user direct access to the query method on the offical BigQuery python client. For fine-grained control over a query job, see: https://google-api-client-libraries.appspot.com/documentation/bigquery/v2/python/latest/bigquery_v2.jobs.html#query Parameters query_data query object as per "configuration.query" in https://cloud.google.com/bigquery/docs/reference/v2/jobs#configuration.query Returns ------- tuple job id and query results if query completed. If dry_run is True, job id will be None and results will be empty if the query is valid or a dict containing the response if invalid. Raises ------ BigQueryTimeoutException On timeout """


        logger.debug('Submitting query job: %s' % query_data)

        job_collection = self.bigquery.jobs()

        try:
            query_reply = job_collection.query(
                projectId=self.project_id, body=query_data).execute(
                num_retries=self.num_retries)
        except HttpError as e:
            if query_data.get("dryRun", False):
                return None, json.loads(e.content.decode('utf8'))
            raise

        job_id = query_reply['jobReference'].get('jobId')
        schema = query_reply.get('schema', {'fields': None})['fields']
        rows = query_reply.get('rows', [])
        job_complete = query_reply.get('jobComplete', False)

        # raise exceptions if it's not an async query
        # and job is not completed after timeout
        if not job_complete and query_data.get("timeoutMs", False):
            logger.error('BigQuery job %s timeout' % job_id)
            raise BigQueryTimeoutException()

        return job_id, [self._transform_row(row, schema) for row in rows]