sundeep8967 commented on code in PR #72286:
URL: https://github.com/apache/airflow/pull/72286#discussion_r4040501826
##########
providers/google/src/airflow/providers/google/cloud/hooks/bigquery.py:
##########
@@ -2459,7 +2459,9 @@ def get_records(
fields = [field for field in fields if not selected_fields or
field["name"] in selected_fields]
fields_names = [field["name"] for field in fields]
col_types = [field["type"] for field in fields]
- for dict_row in rows:
+ for i, dict_row in enumerate(rows):
+ if i > 0 and i % 1000 == 0:
+ await asyncio.sleep(0)
Review Comment:
Added unit tests for this in `test_bigquery.py` using `AsyncMock` to verify
`asyncio.sleep(0)` is called the expected number of times, and verified it
doesn't sleep when set to 0.
Regarding the 1000 threshold:
In the single-threaded Triggerer event loop, processing rows with `bq_cast`
takes roughly ~10-20µs per row. Uninterrupted processing of large result sets
(e.g. 50k-100k rows) blocks the loop for 1-2 seconds, delaying I/O polling and
heartbeats.
Yielding every 1000 rows caps uninterrupted CPU execution to ~10-20ms (well
below heartbeat thresholds) while keeping `asyncio.sleep(0)` task-scheduling
overhead down to <0.02%.
Exposing it as `yield_frequency: int = 1000` avoids the hardcoded magic
number and gives callers full control to adjust the chunk size or disable
yielding entirely with 0.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
