feat: add ability to retry query job automatically if the job fails

[JakeSummers]

Environment details

• OS type and version: Mac (10.15.7) and Python Base Docker Image
• Python version: Python 3.7.8
• Poetry version: Poetry version 1.0.10
• google-cloud-bigquery: 1.28.0 Google BigQuery API client library

Issue

I am using the retry parameter in biquery.Client.query but it isn't retrying my request.

Code example

Setup code:

import structlog
from google.api_core import retry, exceptions
from google.api_core.retry import if_exception_type, Retry
from google.cloud import bigquery

from common.logging.setup_logging import set_logger_config

set_logger_config() # Enables debug level logging for retry

project = "myproject"
source_table = f"{project}.mydataset.TABLE_NOT_REALLY_HERE"
statement = f"SELECT * FROM `{source_table}`"
client = bigquery.Client(project=project)

Now if you run:

retry_policy = retry.Retry(predicate=if_exception_type(exceptions.NotFound))
query_job = client.query(statement, retry=retry_policy)
result = list(query_job.result(retry=retry_policy))
print(result)

Only one request is made:

Traceback (most recent call last):
  File "/Users/.../lib/python3.7/site-packages/google/cloud/bigquery/job.py", line 3230, in result
    super(QueryJob, self).result(retry=retry, timeout=timeout)
  File "/Users/.../lib/python3.7/site-packages/google/cloud/bigquery/job.py", line 835, in result
    return super(_AsyncJob, self).result(timeout=timeout)
  File "/Users/.../lib/python3.7/site-packages/google/api_core/future/polling.py", line 134, in result
    raise self._exception
google.api_core.exceptions.NotFound: 404 Not found: Table <TABLE_NAME> was not found in location US

(job ID: 5aeb5865-7bde-4795-ae80-75410f8b9282)

                                              -----Query Job SQL Follows-----                                              

    |    .    |    .    |    .    |    .    |    .    |    .    |    .    |    .    |    .    |    .    |    .    |
   1:SELECT *
   2:                FROM `<TABLE_NAME>`
   3:                
    |    .    |    .    |    .    |    .    |    .    |    .    |    .    |    .    |    .    |    .    |    .    |
python-BaseException

Process finished with exit code 1

Workaround

If instead, I run:

@Retry(predicate=if_exception_type(exceptions.NotFound))
def do():
    query_job = client.query(statement)
    result = list(query_job.result())
    print(result)
do()

This works as expected and retries the request:

{"logger": "google.api_core.retry", "timestamp": "2021-03-02T19:07:48.000317Z", "severity": "debug", "message": "Retrying due to 404 Not found: Table <table_name> was not found in location US\n\n(job ID: <REMOVED>, sleeping 1.0s ..."}
{"logger": "google.api_core.retry", "timestamp": "2021-03-02T19:07:49.363842Z", "severity": "debug", "message": "Retrying due to 404 Not found: Table <table_name> was not found in location US\n\n(job ID: <REMOVED>, sleeping 3.2s ..."}
{"logger": "google.api_core.retry", "timestamp": "2021-03-02T19:07:52.741042Z", "severity": "debug", "message": "Retrying due to 404 Not found: Table <table_name> was not found in location US\n\n(job ID: <REMOVED>, sleeping 2.8s ..."}
{"logger": "google.api_core.retry", "timestamp": "2021-03-02T19:07:55.969272Z", "severity": "debug", "message": "Retrying due to 404 Not found: Table <table_name> was not found in location US\n\n(job ID: <REMOVED>, sleeping 9.2s ..."}
{"logger": "google.api_core.retry", "timestamp": "2021-03-02T19:08:05.493544Z", "severity": "debug", "message": "Retrying due to 404 Not found: Table <table_name> was not found in location US\n\n(job ID: <REMOVED>, sleeping 4.2s ..."}
{"logger": "google.api_core.retry", "timestamp": "2021-03-02T19:08:09.960265Z", "severity": "debug", "message": "Retrying due to 404 Not found: Table <table_name> was not found in location US\n\n(job ID: <REMOVED>, sleeping 44.0s ..."}

[plamut]

Confirmed, passing in a custom Retry object via the retry argument does not work, but it works if the same object is applied directly as a decorator to the calling function.

If I had to guess, I'd say the custom Retry instance gets lost somewhere in the call chain and a default Retry is used instead (which does not repeat NotFound errors).

[plamut]

OK, so here's the thing - the custom Retry is passed down to where the actual API call is made and wraps it, but the call itself does not raise an exception.

Instead, the response is OK, but the response body, among other things, contains the error info, which is parsed and stored on the job instance (job instance is also a future and it's resolved with the exception parsed from the response body). Then, when inquiring for the job result:

query_job.result(retry=retry_policy)

...the job instance (read: future) figures out it has already been resolved with an error and raises it, it does not make any more requests and the retry passed to result() also has no effect.

TL; DR - considering that job instances are futures, this could actually be considered expected behavior, since that's the way futures are supposed to work. I do agree, however, that it's not that intuitive.

@tswast Do you agree with this reasoning?

We should at least document it better when a Retry actually kicks in (e.g. "external" errors). Right now a job.begin request "succeeds", even if it creates a job that immediately fails, but retries don't see this, and they don't do deep inspection of the response.

[plamut]

@JakeSummers I saw your reaction. :)

The client.query() call makes an API request to the jobs.insert endpoint. An OK response means that a job has been successfully created on the backend, even if the job itself stopped running immediately due to an error. As far as Retry is concerned, the jobs.insert request succeeded and there's nothing to retry.

Once a job is done running, its state will not change, and it cannot be retried/restarted, the backend doesn't support that. The only thing clients can do is to try and create another job using the same arguments, but it would certainly not be desired if a client repeatedly spawned numerous new jobs every time a successfully created job terminated with an error.

Does the behavior make more sense now?

[tswast]

@JakeSummers Do I understand your request correctly that you'd like to retry the whole query?

We have some discussion at #14 and googleapis/google-cloud-python#5555 about this.

There are some classes of errors for which it would make sense to restart a job from the beginning. For example, a query job fails with "INTERNAL ERROR" might succeed the next time you run the same query. For these cases, we provide a bqclient.create_job() method which can take the job configuration of any kind of job and attempt to re-run it.

In general, I agree with @plamut that "it would certainly not be desired if a client repeatedly spawned numerous new jobs every time a successfully created job terminated with an error."

I'll discuss with some BQ engineers to see if there are known errors where it is safe to restart the job.

[JakeSummers]

RE: @plamut

The client.query() call makes an API request to the jobs.insert endpoint. An OK response means that a job has been successfully created on the backend, even if the job itself stopped running immediately due to an error. As far as Retry is concerned, the jobs.insert request succeeded and there's nothing to retry.

Maybe it is naive of me to say this, but I view this as an implementation-detail of the client library. As a user, my feeling is that the API is leaking its implementation into my code.

---

RE: @tswast

@JakeSummers Do I understand your request correctly that you'd like to retry the whole query?

Yes!

Maybe providing more context about the situation that I am hitting would make this more clear. I am querying a table that is managed by an up-stream ETL. Every hour the table is deleted and re-created. There is a very-breif-window for which the table does not exist. For this reason, I want to retry the query when I receive a 404.

With the current API, it looks like the retry parameter is to handle http errors that are encounter while the query is running not just a subset of errors that are encountered while creating the job.

There are some classes of errors for which it would make sense to restart a job from the beginning. For example, a query job fails with "INTERNAL ERROR" might succeed the next time you run the same query. For these cases, we provide a bqclient.create_job() method which can take the job configuration of any kind of job and attempt to re-run it.

Thanks for the pointer. I had a quick look at this, for the documentation, it is unclear to me that this is expected to behave differently. Right now, I am just wrapping my call in a retry but if I experience issues with that, I suppose that I will explore this.

---

Other question: Is this behaviour consistent with how the retry parameter works on the storage API? For example, the list_blobs (link) takes a retry param. Does this only retry in certain senarios?

[tswast]

Other question: Is this behaviour consistent with how the retry parameter works on the storage API? For example, the list_blobs (link) takes a retry param. Does this only retry in certain scenarios?

list_blobs is a straightforward 1:1 mapping between backend API and client function, so it doesn't have these problems.

The BigQuery client behavior is consistent with long-running operations in other APIs such as batch translation (batch_translate_text) with the Google Cloud Translate API, where the retry behavior will retry submitting the job if the submit request fails, but it does not resubmit the request if the translation job fails

[tswast]

As discussed in #809, after brainstorming with @jimfulton, we can make the query job save the original request so that we don't have to worry about anything except for resetting the job ID before retrying.