Conversation
…us rows can be deleted
There was a problem hiding this comment.
Pull request overview
This PR updates the BigQuery load path to avoid the streaming buffer (which blocks immediate DELETE/UPDATE) by using BigQuery load jobs for production runs, while retaining streaming inserts for the BigQuery emulator (which doesn’t support load jobs).
Changes:
- Added
_insert_rows_to_table()to centralize writing rows via either streaming inserts (emulator) or load jobs (prod). - Updated
load_data()to acceptuse_streaming_insertand route writes through the new helper. - Wired
_main()to enable streaming inserts automatically whenBIGQUERY_EMULATOR_HOSTis set.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| if load_job.errors: | ||
| error_msg = f"BigQuery load errors for table {table}: {load_job.errors}" | ||
| logger.error(error_msg) | ||
| raise Exception(error_msg) |
There was a problem hiding this comment.
Same as the streaming-insert branch: raising a generic Exception here bypasses main()'s RuntimeError handler and will likely result in an uncaught exception traceback. Prefer raising RuntimeError (or another exception type that main() handles) for consistent failure behavior.
| error_msg = f"BigQuery insert errors for table {table}: {errors}" | ||
| logger.error(error_msg) | ||
| raise Exception(error_msg) |
There was a problem hiding this comment.
These branches raise a generic Exception, but main() only catches RuntimeError. If an insert/load fails, this will escape the top-level handler and produce an uncaught exception traceback instead of returning a clean exit code via main(). Consider raising RuntimeError (or a dedicated custom exception) here so failures are handled consistently.
When trying to delete rows right after inserting them, BigQuery returns a 400 Bad Request error.
The root cause is insert_rows_json(), which uses the Streaming Insert API. Rows inserted this way go into a streaming buffer and are immutable for up to 90 minutes.
This pull request fixes the issue by replacing the use of Strreaming API with a load job, which writes directly to BigQuery storage. Rows inserted via load jobs are immediately mutable. This means, if the script fails, running again the DELETE in delete_existing_snapshot() will work right away.
The trade-off: load jobs have slightly higher latency to start (~a few seconds overhead per job) but that's negligible for a batch ETL use case. The streaming API was designed for real-time, high-frequency appends where you need sub-second availability so not necessary for a batch ETL pipeline like this.