def query_and_wait(
    connector,
    client: bigquery.Client,
    query: str,
    *,
    job_config: bigquery.QueryJobConfig,
    location: Optional[str],
    project_id: Optional[str],
    max_results: Optional[int],
    timeout_ms: Optional[int],
):
    """Start a query and wait for it to complete.

    Args:
        connector (GbqConnector):
            General pandas-gbq "connector" with helpers for stateful progress
            logs and error raising.

        client (bigquery.Client):
            A connection to BigQuery, used to make API requests.

        query (str):
            The text of the query to run.

        job_config (bigquery.QueryJobConfig):
            Options for running the query.

        location (Optional[str]):
            BigQuery location to run the query. Uses the default if not set.

        project (Optional[str]):
            GCP project ID where to run the query. Uses the default if not set.

        max_results (Optional[int]):
            Maximum number of rows in the result set.

        timeout_ms (Optional[int]):
            How long to wait before cancelling the query.

    Returns:
        bigquery.RowIterator:
            Result iterator from which we can download the results in the
            desired format (pandas.DataFrame).
    """
    query_reply = try_query(
        connector,
        functools.partial(
            client.query,
            query,
            job_config=job_config,
            location=location,
            project=project_id,
        ),
    )
    logger.debug("Query running...")

    job_id = query_reply.job_id
    logger.debug("Job ID: %s" % job_id)

    _wait_for_query_job(connector, connector.client, query_reply, timeout_ms)

    if query_reply.cache_hit:
        logger.debug("Query done.\nCache hit.\n")
    else:
        bytes_processed = query_reply.total_bytes_processed or 0
        bytes_billed = query_reply.total_bytes_billed or 0
        logger.debug(
            "Query done.\nProcessed: {} Billed: {}".format(
                sizeof_fmt(bytes_processed),
                sizeof_fmt(bytes_billed),
            )
        )
        logger.debug(
            "Standard price: ${:,.2f} USD\n".format(bytes_billed * QUERY_PRICE_FOR_TB)
        )

    # As of google-cloud-bigquery 2.3.0, QueryJob.result() uses
    # getQueryResults() instead of tabledata.list, which returns the correct
    # response with DML/DDL queries.
    try:
        return query_reply.result(max_results=max_results)
    except connector.http_error as ex:
        connector.process_http_error(ex)
Start a query and wait for it to complete.

    Args:
        connector (GbqConnector):
            General pandas-gbq "connector" with helpers for stateful progress
            logs and error raising.

        client (bigquery.Client):
            A connection to BigQuery, used to make API requests.

        query (str):
            The text of the query to run.

        job_config (bigquery.QueryJobConfig):
            Options for running the query.

        location (Optional[str]):
            BigQuery location to run the query. Uses the default if not set.

        project (Optional[str]):
            GCP project ID where to run the query. Uses the default if not set.

        max_results (Optional[int]):
            Maximum number of rows in the result set.

        timeout_ms (Optional[int]):
            How long to wait before cancelling the query.

    Returns:
        bigquery.RowIterator:
            Result iterator from which we can download the results in the
            desired format (pandas.DataFrame).
    
pandas_gbq/query.py