diff --git a/packages/google-cloud-bigquery/google/cloud/bigquery/__init__.py b/packages/google-cloud-bigquery/google/cloud/bigquery/__init__.py index b9a71add39b2..2b447aab702f 100644 --- a/packages/google-cloud-bigquery/google/cloud/bigquery/__init__.py +++ b/packages/google-cloud-bigquery/google/cloud/bigquery/__init__.py @@ -96,6 +96,7 @@ from google.cloud.bigquery.query import StructQueryParameter from google.cloud.bigquery.query import StructQueryParameterType from google.cloud.bigquery.query import UDFResource +from google.cloud.bigquery.retry import DEFAULT_INSERT_ROWS_RETRY from google.cloud.bigquery.retry import DEFAULT_RETRY from google.cloud.bigquery.routine import DeterminismLevel from google.cloud.bigquery.routine import Routine @@ -206,6 +207,7 @@ "ParquetOptions", "ScriptOptions", "TransactionInfo", + "DEFAULT_INSERT_ROWS_RETRY", "DEFAULT_RETRY", # Standard SQL types "StandardSqlDataType", diff --git a/packages/google-cloud-bigquery/google/cloud/bigquery/client.py b/packages/google-cloud-bigquery/google/cloud/bigquery/client.py index 4967bcdb1134..6c66e68465bd 100644 --- a/packages/google-cloud-bigquery/google/cloud/bigquery/client.py +++ b/packages/google-cloud-bigquery/google/cloud/bigquery/client.py @@ -115,6 +115,7 @@ from google.cloud.bigquery.query import _QueryResults from google.cloud.bigquery.retry import ( DEFAULT_GET_JOB_TIMEOUT, + DEFAULT_INSERT_ROWS_RETRY, DEFAULT_JOB_RETRY, DEFAULT_RETRY, DEFAULT_TIMEOUT, @@ -4066,7 +4067,7 @@ def insert_rows_json( skip_invalid_rows: Optional[bool] = None, ignore_unknown_values: Optional[bool] = None, template_suffix: Optional[str] = None, - retry: retries.Retry = DEFAULT_RETRY, + retry: retries.Retry = DEFAULT_INSERT_ROWS_RETRY, timeout: TimeoutType = DEFAULT_TIMEOUT, ) -> Sequence[dict]: """Insert rows into a table without applying local type conversions. @@ -4120,7 +4121,12 @@ def insert_rows_json( based on the schema of the template table. See https://cloud.google.com/bigquery/streaming-data-into-bigquery#template-tables retry (Optional[google.api_core.retry.Retry]): - How to retry the RPC. + How to retry the RPC. Defaults to + :data:`~google.cloud.bigquery.retry.DEFAULT_INSERT_ROWS_RETRY`, + which, unlike :data:`~google.cloud.bigquery.retry.DEFAULT_RETRY`, + does not retry :class:`requests.exceptions.SSLError`, since + for streaming inserts that usually indicates an invalid request + (e.g. a schema mismatch) rather than a transient failure. timeout (Optional[float]): The number of seconds to wait for the underlying HTTP transport before using ``retry``. @@ -4190,6 +4196,7 @@ def insert_rows_json( path = "%s/insertAll" % table.path # We can always retry, because every row has an insert ID. span_attributes = {"path": path} + try: response = self._call_api( retry, @@ -4200,12 +4207,15 @@ def insert_rows_json( data=data, timeout=timeout, ) - except requests.exceptions.SSLError as exc: - msg = ( - "An SSL/Connection error occurred while streaming rows. This " - "could be due to an invalid request (e.g., invalid table schema)." - ) - raise requests.exceptions.SSLError(msg) from exc + except (requests.exceptions.SSLError, core_exceptions.RetryError) as exc: + cause = exc.cause if isinstance(exc, core_exceptions.RetryError) else exc + if isinstance(cause, requests.exceptions.SSLError): + msg = ( + "An SSL/Connection error occurred while streaming rows. This " + "could be due to an invalid request (e.g., invalid table schema)." + ) + raise requests.exceptions.SSLError(msg) from exc + raise errors = [] for error in response.get("insertErrors", ()): diff --git a/packages/google-cloud-bigquery/google/cloud/bigquery/retry.py b/packages/google-cloud-bigquery/google/cloud/bigquery/retry.py index 4e78e7d28dcb..89384fc3fa65 100644 --- a/packages/google-cloud-bigquery/google/cloud/bigquery/retry.py +++ b/packages/google-cloud-bigquery/google/cloud/bigquery/retry.py @@ -40,11 +40,6 @@ _DEFAULT_RETRY_DEADLINE = 10.0 * 60.0 # 10 minutes -# Exceptions that are subclasses of types in _UNSTRUCTURED_RETRYABLE_TYPES -# but should not be retried because they typically indicate persistent -# configuration or security issues. -_UNSTRUCTURED_NON_RETRYABLE_TYPES = (requests.exceptions.SSLError,) - # Ambiguous errors (e.g. internalError, backendError, rateLimitExceeded) retry # until the full `_DEFAULT_RETRY_DEADLINE`. This is because the # `jobs.getQueryResults` REST API translates a job failure into an HTTP error. @@ -69,13 +64,9 @@ def _should_retry(exc): """Predicate for determining when to retry. - We retry if the 'reason' is in _RETRYABLE_REASONS or if the exception - is an instance of one of the _UNSTRUCTURED_RETRYABLE_TYPES, unless it - is explicitly excluded by being in _UNSTRUCTURED_NON_RETRYABLE_TYPES. + We retry if and only if the 'reason' is in _RETRYABLE_REASONS or is + in _UNSTRUCTURED_RETRYABLE_TYPES. """ - if isinstance(exc, _UNSTRUCTURED_NON_RETRYABLE_TYPES): - return False - try: reason = exc.errors[0]["reason"] except (AttributeError, IndexError, TypeError, KeyError): @@ -95,6 +86,34 @@ def _should_retry(exc): To modify the default retry behavior, call a ``with_XXX`` method on ``DEFAULT_RETRY``. For example, to change the deadline to 30 seconds, pass ``retry=bigquery.DEFAULT_RETRY.with_deadline(30)``. + +Note: streaming insert methods (``Client.insert_rows``, +``Client.insert_rows_json``, ``Client.insert_rows_from_dataframe``) default to +:data:`DEFAULT_INSERT_ROWS_RETRY` instead. +""" + + +def _should_retry_insert_rows(exc): + """Predicate for determining when to retry streaming inserts (insertAll). + + Unlike standard API calls, tabledata.insertAll failures due to schema + mismatches often manifest as an SSLError because the server abruptly terminates + the connection. These errors will not resolve on retry and should fail + immediately with descriptive guidance. + """ + if isinstance(exc, requests.exceptions.SSLError): + return False + return _should_retry(exc) + + +DEFAULT_INSERT_ROWS_RETRY = DEFAULT_RETRY.with_predicate(_should_retry_insert_rows) +"""The default retry object for streaming inserts. + +Used by ``Client.insert_rows``, ``Client.insert_rows_json`` and +``Client.insert_rows_from_dataframe``. Identical to :data:`DEFAULT_RETRY`, +except that :class:`requests.exceptions.SSLError` is *not* retried, because +for ``tabledata.insertAll`` it typically indicates an invalid request (e.g. a +schema mismatch) rather than a transient failure. """ diff --git a/packages/google-cloud-bigquery/tests/system/test_ssl_retry.py b/packages/google-cloud-bigquery/tests/system/test_ssl_retry.py index 0c90e039f9fd..fa0eb0be13ce 100644 --- a/packages/google-cloud-bigquery/tests/system/test_ssl_retry.py +++ b/packages/google-cloud-bigquery/tests/system/test_ssl_retry.py @@ -22,8 +22,8 @@ def test_insert_rows_json_ssl_error_no_retry(bigquery_client, dataset_id, project_id): """ - Verify that SSLError during insert_rows_json is NOT retried and - propagates a descriptive error message immediately. + Verify that, with the default retry, an SSLError during insert_rows_json + is NOT retried and propagates a descriptive error message immediately. """ table_id = f"{project_id}.{dataset_id}.test_ssl_retry_{int(time.time())}" schema = [bigquery.SchemaField("name", "STRING")] @@ -32,7 +32,6 @@ def test_insert_rows_json_ssl_error_no_retry(bigquery_client, dataset_id, projec try: # We mock the api_request to simulate the GFE abruptly closing the connection # which manifests as a requests.exceptions.SSLError. - bigquery_client._connection.api_request call_count = 0 def mock_api_request(*args, **kwargs): @@ -43,19 +42,18 @@ def mock_api_request(*args, **kwargs): with mock.patch.object( bigquery_client._connection, "api_request", side_effect=mock_api_request ): - # Use a reasonably short deadline for the test, although it should fail on the first attempt anyway. - retry = bigquery.DEFAULT_RETRY.with_deadline(5.0) - + # Intentionally do not pass ``retry`` so that the method's default + # (DEFAULT_INSERT_ROWS_RETRY) is exercised. start_time = time.time() with pytest.raises(requests.exceptions.SSLError) as excinfo: - bigquery_client.insert_rows_json(table, [{"name": "test"}], retry=retry) + bigquery_client.insert_rows_json(table, [{"name": "test"}]) duration = time.time() - start_time # Verification: # 1. It should NOT have retried (total calls should be 1) assert call_count == 1 - # 2. It should have failed quickly (much less than the 5s deadline) + # 2. It should have failed quickly (the default deadline is 10 minutes) assert duration < 2.0 # 3. The error message should contain our descriptive wrapping diff --git a/packages/google-cloud-bigquery/tests/unit/test_client.py b/packages/google-cloud-bigquery/tests/unit/test_client.py index a6f6b921af07..1aaebea97108 100644 --- a/packages/google-cloud-bigquery/tests/unit/test_client.py +++ b/packages/google-cloud-bigquery/tests/unit/test_client.py @@ -7102,9 +7102,76 @@ def test_insert_rows_json_w_ssl_error(self): with self.assertRaises(requests.exceptions.SSLError) as context: client.insert_rows_json(table, ROWS) + # The default retry for streaming inserts does not retry SSLErrors. + self.assertEqual(conn.api_request.call_count, 1) self.assertIn("invalid table schema", str(context.exception)) self.assertIn("SSL/Connection error occurred", str(context.exception)) + def test_insert_rows_json_w_ssl_error_w_default_retry(self): + """When a caller opts into DEFAULT_RETRY, SSLErrors are retried until + the deadline, and the resulting RetryError is still surfaced as a + descriptive SSLError.""" + import requests.exceptions + from google.api_core import exceptions as core_exceptions + from google.cloud.bigquery.dataset import DatasetReference + from google.cloud.bigquery.retry import DEFAULT_RETRY + from google.cloud.bigquery.table import Table + + PROJECT = "PROJECT" + DS_ID = "DS_ID" + TABLE_ID = "TABLE_ID" + ROWS = [{"full_name": "Bhettye Rhubble", "age": "27", "joined": None}] + + creds = _make_credentials() + client = self._make_one(project=PROJECT, credentials=creds, _http=object()) + conn = client._connection = make_connection({}) + conn.api_request.side_effect = requests.exceptions.SSLError("EOF occurred") + + table_ref = DatasetReference(PROJECT, DS_ID).table(TABLE_ID) + table = Table(table_ref) + + # Keep the test fast: negligible backoff and a very short deadline. + retry = DEFAULT_RETRY.with_delay(initial=0.001, maximum=0.001).with_deadline( + 0.05 + ) + with self.assertRaises(requests.exceptions.SSLError) as context: + client.insert_rows_json(table, ROWS, retry=retry) + + self.assertGreater(conn.api_request.call_count, 1) + self.assertIn("invalid table schema", str(context.exception)) + self.assertIn("SSL/Connection error occurred", str(context.exception)) + self.assertIsInstance(context.exception.__cause__, core_exceptions.RetryError) + + def test_insert_rows_json_w_retry_error_non_ssl_cause(self): + """A RetryError caused by something other than an SSLError is re-raised + unchanged.""" + import requests.exceptions + from google.api_core import exceptions as core_exceptions + from google.cloud.bigquery.dataset import DatasetReference + from google.cloud.bigquery.table import Table + + PROJECT = "PROJECT" + DS_ID = "DS_ID" + TABLE_ID = "TABLE_ID" + ROWS = [{"full_name": "Bhettye Rhubble", "age": "27", "joined": None}] + + creds = _make_credentials() + client = self._make_one(project=PROJECT, credentials=creds, _http=object()) + conn = client._connection = make_connection({}) + retry_error = core_exceptions.RetryError( + "Deadline exceeded", + cause=requests.exceptions.ConnectionError("connection reset"), + ) + conn.api_request.side_effect = retry_error + + table_ref = DatasetReference(PROJECT, DS_ID).table(TABLE_ID) + table = Table(table_ref) + + with self.assertRaises(core_exceptions.RetryError) as context: + client.insert_rows_json(table, ROWS) + + self.assertIs(context.exception, retry_error) + def test_list_partitions(self): from google.cloud.bigquery.table import Table diff --git a/packages/google-cloud-bigquery/tests/unit/test_retry.py b/packages/google-cloud-bigquery/tests/unit/test_retry.py index a249d1909909..72402d3736c9 100644 --- a/packages/google-cloud-bigquery/tests/unit/test_retry.py +++ b/packages/google-cloud-bigquery/tests/unit/test_retry.py @@ -53,7 +53,7 @@ def test_w_unstructured_requests_connectionerror(self): def test_w_unstructured_requests_sslerror(self): exc = requests.exceptions.SSLError() - self.assertFalse(self._call_fut(exc)) + self.assertTrue(self._call_fut(exc)) def test_w_unstructured_requests_chunked_encoding_error(self): exc = requests.exceptions.ChunkedEncodingError() @@ -160,3 +160,32 @@ def test_DEFAULT_JOB_RETRY_job_rate_limit_exceeded_retry_predicate(): assert DEFAULT_JOB_RETRY._predicate( ClientError("fail", errors=[dict(reason="backendError")]) ) + + +class Test_should_retry_insert_rows(unittest.TestCase): + def _call_fut(self, exc): + from google.cloud.bigquery.retry import _should_retry_insert_rows + + return _should_retry_insert_rows(exc) + + def test_w_unstructured_requests_sslerror(self): + exc = requests.exceptions.SSLError() + self.assertFalse(self._call_fut(exc)) + + def test_w_unstructured_requests_connectionerror(self): + exc = requests.exceptions.ConnectionError() + self.assertTrue(self._call_fut(exc)) + + def test_w_backendError(self): + exc = mock.Mock(errors=[{"reason": "backendError"}], spec=["errors"]) + self.assertTrue(self._call_fut(exc)) + + +def test_DEFAULT_INSERT_ROWS_RETRY_predicate(): + from google.cloud.bigquery.retry import DEFAULT_INSERT_ROWS_RETRY + + exc_ssl = requests.exceptions.SSLError() + assert not DEFAULT_INSERT_ROWS_RETRY._predicate(exc_ssl) + + exc_conn = requests.exceptions.ConnectionError() + assert DEFAULT_INSERT_ROWS_RETRY._predicate(exc_conn)