Merge branch 'main' into bug-fix-eq-error

Author: chalmerlowe

CHANGELOG.md

@@ -5,6 +5,20 @@
 [1]: https://pypi.org/project/google-cloud-bigquery/#history
 
 
+## [3.36.0](https://github.com/googleapis/python-bigquery/compare/v3.35.1...v3.36.0) (2025-08-20)
+
+
+### Features
+
+* Add created/started/ended properties to RowIterator. ([#2260](https://github.com/googleapis/python-bigquery/issues/2260)) ([0a95b24](https://github.com/googleapis/python-bigquery/commit/0a95b24192395cc3ccf801aa9bc318999873a2bf))
+* Retry query jobs if `jobBackendError` or `jobInternalError` are encountered ([#2256](https://github.com/googleapis/python-bigquery/issues/2256)) ([3deff1d](https://github.com/googleapis/python-bigquery/commit/3deff1d963980800e8b79fa3aaf5b712d4fd5062))
+
+
+### Documentation
+
+* Add a TROUBLESHOOTING.md file with tips for logging ([#2262](https://github.com/googleapis/python-bigquery/issues/2262)) ([b684832](https://github.com/googleapis/python-bigquery/commit/b68483227693ea68f6b12eacca2be1803cffb1d1))
+* Update README to break infinite redirect loop ([#2254](https://github.com/googleapis/python-bigquery/issues/2254)) ([8f03166](https://github.com/googleapis/python-bigquery/commit/8f031666114a826da2ad965f8ecd4727466cb480))
+
 ## [3.35.1](https://github.com/googleapis/python-bigquery/compare/v3.35.0...v3.35.1) (2025-07-21)
 
 

TROUBLESHOOTING.md

@@ -0,0 +1,34 @@
+# Troubleshooting steps
+
+## Enable logging of BQ Storage Read API session creation
+
+It can be helpful to get the BQ Storage Read API session to allow the BigQuery
+backend team to debug cases of API instability. The logs that share the session
+creation are in a module-specific logger. To enable the logs, refer to the
+following code sample:
+
+```python
+import logging
+import google.cloud.bigquery
+
+# Configure the basic logging to show DEBUG level messages
+log_formatter = logging.Formatter(
+    '%(asctime)s - %(levelname)s - %(message)s'
+)
+handler = logging.StreamHandler()
+handler.setFormatter(log_formatter)
+default_logger = logging.getLogger()
+default_logger.setLevel(logging.DEBUG)
+default_logger.addHandler(handler)
+to_dataframe_logger = logging.getLogger("google.cloud.bigquery._pandas_helpers")
+to_dataframe_logger.setLevel(logging.DEBUG)
+to_dataframe_logger.addHandler(handler)
+
+# Example code that touches the BQ Storage Read API.
+bqclient = google.cloud.bigquery.Client()
+results = bqclient.query_and_wait("SELECT * FROM `bigquery-public-data.usa_names.usa_1910_2013`")
+print(results.to_dataframe().head())
+```
+
+In particular, watch for the text "with BQ Storage API session" in the logs
+to get the streaming API session ID to share with your support person.

google/cloud/bigquery/_job_helpers.py

@@ -35,17 +35,22 @@
 predicates where it is safe to generate a new query ID.
 """
 
+from __future__ import annotations
+
 import copy
+import dataclasses
+import datetime
 import functools
 import uuid
 import textwrap
-from typing import Any, Dict, Optional, TYPE_CHECKING, Union
+from typing import Any, Callable, Dict, Optional, TYPE_CHECKING, Union
 import warnings
 
 import google.api_core.exceptions as core_exceptions
 from google.api_core import retry as retries
 
 from google.cloud.bigquery import job
+import google.cloud.bigquery.job.query
 import google.cloud.bigquery.query
 from google.cloud.bigquery import table
 import google.cloud.bigquery.retry
@@ -116,14 +121,21 @@ def query_jobs_insert(
     retry: Optional[retries.Retry],
     timeout: Optional[float],
     job_retry: Optional[retries.Retry],
+    *,
+    callback: Callable = lambda _: None,
 ) -> job.QueryJob:
     """Initiate a query using jobs.insert.
 
     See: https://cloud.google.com/bigquery/docs/reference/rest/v2/jobs/insert
+
+    Args:
+        callback (Callable):
+            A callback function used by bigframes to report query progress.
     """
     job_id_given = job_id is not None
     job_id_save = job_id
     job_config_save = job_config
+    query_sent_factory = QuerySentEventFactory()
 
     def do_query():
         # Make a copy now, so that original doesn't get changed by the process
@@ -136,6 +148,16 @@ def do_query():
 
         try:
             query_job._begin(retry=retry, timeout=timeout)
+            if job_config is not None and not job_config.dry_run:
+                callback(
+                    query_sent_factory(
+                        query=query,
+                        billing_project=query_job.project,
+                        location=query_job.location,
+                        job_id=query_job.job_id,
+                        request_id=None,
+                    )
+                )
         except core_exceptions.Conflict as create_exc:
             # The thought is if someone is providing their own job IDs and they get
             # their job ID generation wrong, this could end up returning results for
@@ -396,6 +418,7 @@ def query_and_wait(
     job_retry: Optional[retries.Retry],
     page_size: Optional[int] = None,
     max_results: Optional[int] = None,
+    callback: Callable = lambda _: None,
 ) -> table.RowIterator:
     """Run the query, wait for it to finish, and return the results.
 
@@ -415,9 +438,8 @@ def query_and_wait(
         location (Optional[str]):
             Location where to run the job. Must match the location of the
             table used in the query as well as the destination table.
-        project (Optional[str]):
-            Project ID of the project of where to run the job. Defaults
-            to the client's project.
+        project (str):
+            Project ID of the project of where to run the job.
         api_timeout (Optional[float]):
             The number of seconds to wait for the underlying HTTP transport
             before using ``retry``.
@@ -441,6 +463,8 @@ def query_and_wait(
             request. Non-positive values are ignored.
         max_results (Optional[int]):
             The maximum total number of rows from this request.
+        callback (Callable):
+            A callback function used by bigframes to report query progress.
 
     Returns:
         google.cloud.bigquery.table.RowIterator:
@@ -479,12 +503,14 @@ def query_and_wait(
                 retry=retry,
                 timeout=api_timeout,
                 job_retry=job_retry,
+                callback=callback,
             ),
             api_timeout=api_timeout,
             wait_timeout=wait_timeout,
             retry=retry,
             page_size=page_size,
             max_results=max_results,
+            callback=callback,
         )
 
     path = _to_query_path(project)
@@ -496,10 +522,24 @@ def query_and_wait(
     if client.default_job_creation_mode:
         request_body["jobCreationMode"] = client.default_job_creation_mode
 
+    query_sent_factory = QuerySentEventFactory()
+
     def do_query():
-        request_body["requestId"] = make_job_id()
+        request_id = make_job_id()
+        request_body["requestId"] = request_id
         span_attributes = {"path": path}
 
+        if "dryRun" not in request_body:
+            callback(
+                query_sent_factory(
+                    query=query,
+                    billing_project=project,
+                    location=location,
+                    job_id=None,
+                    request_id=request_id,
+                )
+            )
+
         # For easier testing, handle the retries ourselves.
         if retry is not None:
             response = retry(client._call_api)(
@@ -542,8 +582,25 @@ def do_query():
                 retry=retry,
                 page_size=page_size,
                 max_results=max_results,
+                callback=callback,
             )
 
+        if "dryRun" not in request_body:
+            callback(
+                QueryFinishedEvent(
+                    billing_project=project,
+                    location=query_results.location,
+                    query_id=query_results.query_id,
+                    job_id=query_results.job_id,
+                    total_rows=query_results.total_rows,
+                    total_bytes_processed=query_results.total_bytes_processed,
+                    slot_millis=query_results.slot_millis,
+                    destination=None,
+                    created=query_results.created,
+                    started=query_results.started,
+                    ended=query_results.ended,
+                )
+            )
         return table.RowIterator(
             client=client,
             api_request=functools.partial(client._call_api, retry, timeout=api_timeout),
@@ -561,6 +618,9 @@ def do_query():
             query=query,
             total_bytes_processed=query_results.total_bytes_processed,
             slot_millis=query_results.slot_millis,
+            created=query_results.created,
+            started=query_results.started,
+            ended=query_results.ended,
         )
 
     if job_retry is not None:
@@ -598,6 +658,9 @@ def _supported_by_jobs_query(request_body: Dict[str, Any]) -> bool:
         "requestId",
         "createSession",
         "writeIncrementalResults",
+        "jobTimeoutMs",
+        "reservation",
+        "maxSlots",
     }
 
     unsupported_keys = request_keys - keys_allowlist
@@ -611,19 +674,52 @@ def _wait_or_cancel(
     retry: Optional[retries.Retry],
     page_size: Optional[int],
     max_results: Optional[int],
+    *,
+    callback: Callable = lambda _: None,
 ) -> table.RowIterator:
     """Wait for a job to complete and return the results.
 
     If we can't return the results within the ``wait_timeout``, try to cancel
     the job.
     """
     try:
-        return job.result(
+        if not job.dry_run:
+            callback(
+                QueryReceivedEvent(
+                    billing_project=job.project,
+                    location=job.location,
+                    job_id=job.job_id,
+                    statement_type=job.statement_type,
+                    state=job.state,
+                    query_plan=job.query_plan,
+                    created=job.created,
+                    started=job.started,
+                    ended=job.ended,
+                )
+            )
+        query_results = job.result(
             page_size=page_size,
             max_results=max_results,
             retry=retry,
             timeout=wait_timeout,
         )
+        if not job.dry_run:
+            callback(
+                QueryFinishedEvent(
+                    billing_project=job.project,
+                    location=query_results.location,
+                    query_id=query_results.query_id,
+                    job_id=query_results.job_id,
+                    total_rows=query_results.total_rows,
+                    total_bytes_processed=query_results.total_bytes_processed,
+                    slot_millis=query_results.slot_millis,
+                    destination=job.destination,
+                    created=job.created,
+                    started=job.started,
+                    ended=job.ended,
+                )
+            )
+        return query_results
     except Exception:
         # Attempt to cancel the job since we can't return the results.
         try:
@@ -632,3 +728,62 @@ def _wait_or_cancel(
             # Don't eat the original exception if cancel fails.
             pass
         raise
+
+
+@dataclasses.dataclass(frozen=True)
+class QueryFinishedEvent:
+    """Query finished successfully."""
+
+    billing_project: Optional[str]
+    location: Optional[str]
+    query_id: Optional[str]
+    job_id: Optional[str]
+    destination: Optional[table.TableReference]
+    total_rows: Optional[int]
+    total_bytes_processed: Optional[int]
+    slot_millis: Optional[int]
+    created: Optional[datetime.datetime]
+    started: Optional[datetime.datetime]
+    ended: Optional[datetime.datetime]
+
+
+@dataclasses.dataclass(frozen=True)
+class QueryReceivedEvent:
+    """Query received and acknowledged by the BigQuery API."""
+
+    billing_project: Optional[str]
+    location: Optional[str]
+    job_id: Optional[str]
+    statement_type: Optional[str]
+    state: Optional[str]
+    query_plan: Optional[list[google.cloud.bigquery.job.query.QueryPlanEntry]]
+    created: Optional[datetime.datetime]
+    started: Optional[datetime.datetime]
+    ended: Optional[datetime.datetime]
+
+
+@dataclasses.dataclass(frozen=True)
+class QuerySentEvent:
+    """Query sent to BigQuery."""
+
+    query: str
+    billing_project: Optional[str]
+    location: Optional[str]
+    job_id: Optional[str]
+    request_id: Optional[str]
+
+
+class QueryRetryEvent(QuerySentEvent):
+    """Query sent another time because the previous attempt failed."""
+
+
+class QuerySentEventFactory:
+    """Creates a QuerySentEvent first, then QueryRetryEvent after that."""
+
+    def __init__(self):
+        self._event_constructor = QuerySentEvent
+
+    def __call__(self, **kwargs):
+        result = self._event_constructor(**kwargs)
+        self._event_constructor = QueryRetryEvent
+        return result