BigQuery execute_sql blocks the event loop for the whole query

[yumizu-da]

🔴 Required Information

Describe the Bug:
execute_sql in the BigQuery toolset is a plain def, and the BigQuery client it calls blocks (src/google/adk/integrations/bigquery/query_tool.py, L341 and the _execute_sql helper it delegates to at L169). ADK awaits a sync tool inline on the running loop, so nothing else in the process runs until the query comes back. On a server handling more than one session, the others just wait.

Spanner hit the same thing and was fixed in 1dbcecc. Bigtable's execute_sql is async too. BigQuery still has no asyncio.to_thread anywhere.

Steps to Reproduce:

1. uv pip install "google-adk[gcp]==2.9.0"
2. Save the script below (also under Minimal Reproduction Code) as repro.py. It replaces the BigQuery client with one that takes 2 seconds, so no credentials or real dataset are needed, and counts how many times a 100 ms heartbeat task gets to run during the query.
3. python repro.py

Expected Behavior:
The heartbeat keeps ticking every 100 ms while the query is in flight.

Observed Behavior:
It doesn't tick at all until the query returns:

heartbeat ran 3 times
longest gap between heartbeats: 2.108s

The gap is the query duration.

Environment Details:

• ADK Library Version (pip show google-adk): 2.9.0. Also checked main at 7ae1c9b, still sync there.
• Desktop OS: macOS 25.6.0 (arm64)
• Python Version (python -V): 3.11.12, in a uv venv set up per the contribution guide (uv venv --python "python3.11" + uv sync --all-extras)

Model Information:

• Are you using LiteLLM: No
• Which model is being used: N/A, no model involved

---

🟡 Optional Information

Regression:
No, it has always been sync.

Additional Context:
Same fix as Spanner: make execute_sql a coroutine and push _execute_sql into a thread. _execute_sql has to stay sync because forecast, analyze_contribution and detect_anomalies call it directly. Those three block the loop as well, but that's a separate change.

I have a patch and tests ready, so please assign this to me.

Minimal Reproduction Code:

import asyncio
import time
from unittest import mock

from google.adk.integrations.bigquery import query_tool
from google.adk.integrations.bigquery.config import BigQueryToolConfig
from google.adk.tools.tool_context import ToolContext
from google.auth.credentials import Credentials
from google.cloud import bigquery

QUERY_SECONDS = 2.0


async def heartbeat(stop):
    ticks = []
    start = time.monotonic()
    while not stop.is_set():
        await asyncio.sleep(0.1)
        ticks.append(time.monotonic() - start)
    return ticks


def slow_query_and_wait(*args, **kwargs):
    time.sleep(QUERY_SECONDS)
    return [{"num": 123}]


async def main():
    credentials = mock.create_autospec(Credentials, instance=True)
    tool_context = mock.create_autospec(ToolContext, instance=True)
    settings = BigQueryToolConfig()

    with mock.patch.object(bigquery, "Client", autospec=True) as client:
        bq_client = client.return_value
        query_job = mock.create_autospec(bigquery.QueryJob)
        query_job.statement_type = "SELECT"
        bq_client.query.return_value = query_job
        bq_client.query_and_wait.side_effect = slow_query_and_wait

        stop = asyncio.Event()
        beat = asyncio.create_task(heartbeat(stop))
        await asyncio.sleep(0.3)

        result = query_tool.execute_sql(
            "my_project", "SELECT 123 AS num", credentials, settings,
            tool_context,
        )
        if asyncio.iscoroutine(result):
            result = await result

        stop.set()
        ticks = await beat

    gaps = [b - a for a, b in zip(ticks, ticks[1:])]
    print(f"heartbeat ran {len(ticks)} times")
    print(f"longest gap between heartbeats: {max(gaps):.3f}s")
    print(f"result: {result}")


asyncio.run(main())

How often has this issue occurred?:

• Always (100%)

[surajksharma07]

@yumizu-da Loop is fully blocked for the query duration, 3 heartbeats vs 22 after the patch. Wrapping execute_sql in asyncio.to_thread the same way Spanner/Bigtable do fixes it locally too, matches what #7161 does. Could you confirm it holds against a real dataset/session and not just the mocked client before that PR merges?

[yumizu-da]

@surajksharma07 Thanks for checking. Confirmed against real BigQuery, no mocked client this time.

The script below runs two sessions concurrently through BigQueryToolset: session A issues an expensive self join over bigquery-public-data.samples.natality, and session B asks for a cheap query 0.2s later.

Three runs each, by how late B's query was sent:

run	main	this PR
1	1.376s	0.001s
2	1.145s	0.001s
3	1.223s	0.001s

On main B's query isn't issued until A comes back, and the delay tracks A's completion exactly each run.

gcloud auth application-default login
export BQ_PROJECT_ID=<a project you can run BigQuery jobs in>

git checkout main && uv run --extra gcp verify_bigquery_async.py
gh pr checkout 7161 && uv run --extra gcp verify_bigquery_async.py

verify_bigquery_async.py

"""Two concurrent sessions against real BigQuery, nothing mocked.

Session A runs a self join over `natality` that takes a few seconds. Session B
asks for a cheap query DELAY later. The number that matters is how late B's
query was *sent*: that is time B sat waiting for the loop, not for BigQuery.

Finish order is not a signal either way, since the cheap query carries about a
second of fixed client overhead of its own.
"""python

import asyncio
import os
import time
from unittest import mock

import google.auth
from google.adk.integrations import bigquery as bq
from google.adk.integrations.bigquery.config import BigQueryToolConfig
from google.adk.tools.tool_context import ToolContext

DELAY = 0.2
SLOW = """
SELECT COUNT(*) AS c
FROM `bigquery-public-data.samples.natality` AS a
JOIN `bigquery-public-data.samples.natality` AS b ON a.state = b.state
WHERE a.year = 2000 AND b.year = 2000 AND a.month = 1 AND b.month = 1
"""
FAST = """
SELECT island, COUNT(*) AS population
FROM `bigquery-public-data`.`ml_datasets`.`penguins` GROUP BY island
"""


async def main():
  project_id = os.environ["BQ_PROJECT_ID"]
  credentials, _ = google.auth.default(
      scopes=["https://www.googleapis.com/auth/bigquery"]
  )
  toolset = bq.BigQueryToolset(
      tool_filter=["execute_sql"],
      credentials_config=bq.BigQueryCredentialsConfig(credentials=credentials),
      bigquery_tool_config=BigQueryToolConfig(),
  )
  (execute_sql,) = await toolset.get_tools()

  start = time.monotonic()
  marks = {}

  async def session(name, query, delay):
    await asyncio.sleep(delay)
    marks[f"{name} sent"] = time.monotonic() - start
    tool_context = mock.create_autospec(ToolContext, instance=True)
    tool_context.state = {}
    result = await execute_sql.run_async(
        args={"project_id": project_id, "query": query},
        tool_context=tool_context,
    )
    marks[f"{name} back"] = time.monotonic() - start
    return result["status"]

  try:
    statuses = await asyncio.gather(
        session("A slow", SLOW, 0.0), session("B fast", FAST, DELAY)
    )
  finally:
    await toolset.close()

  for label, at in sorted(marks.items(), key=lambda kv: kv[1]):
    print(f"  {label}: {at:6.3f}s")

  late = marks["B fast sent"] - DELAY
  overlapped = marks["B fast sent"] < marks["A slow back"]
  print(f"  statuses: {statuses}")
  print(f"  B sent {late:.3f}s later than asked")
  print(f"  B ran while A was in flight: {overlapped}")


asyncio.run(main())