[damccorm]

Description

Optimizes two per-element hot paths in BigQueryWriteFn (sdks/python/apache_beam/io/gcp/bigquery.py):

1. Fast-path row size estimation (_fast_row_and_insert_id_size):
   • BigQueryWriteFn._check_row_size previously called objsize.get_deep_size(row_and_insert_id) on every incoming streaming insert row. objsize.get_deep_size performs a general-purpose gc.get_referents object graph traversal, costing ~170 µs per row.
   • Adds _fast_row_and_insert_id_size to sum sys.getsizeof directly when row_and_insert_id is a flat (dict, insert_id) tuple containing primitive scalar keys and values (str, int, float, bool, bytes, None), falling back to get_deep_size whenever nested containers or non-primitive objects are present (121.09x faster on 50,000 8-field streaming rows: 8.47s -> 0.070s).
2. Skip redundant parse_table_reference in BigQueryWriteFn.process:
   • Skips resolving self.schema and calling bigquery_tools.parse_table_reference(destination) on every element when self.create_disposition == BigQueryDisposition.CREATE_NEVER or destination is already present in _KNOWN_TABLES (matching the existing early-return guard inside _create_table_if_needed).

Testing

• Verified with pytest sdks/python/apache_beam/io/gcp/bigquery_test.py sdks/python/apache_beam/io/gcp/bigquery_tools_test.py.

[damccorm]

/gemini review

[github-actions]

📋 Review Summary

This PR introduces two targeted optimizations in BigQueryWriteFn to reduce per-element overhead during BigQuery streaming inserts: a fast-path row size estimator (_fast_row_and_insert_id_size) that avoids objsize.get_deep_size (gc.get_referents) on flat primitive dicts, and an early guard in BigQueryWriteFn.process to skip schema resolution and parse_table_reference when tables are already in _KNOWN_TABLES or CREATE_NEVER is set.

🔍 General Feedback

• In _fast_row_and_insert_id_size, objects referenced multiple times within the same row (such as shared string values or interned/cached integers) are summed on every occurrence without tracking seen object IDs, unlike objsize.get_deep_size. For rows containing repeated references to the same large string or bytes object, this can overestimate row size and prematurely trigger batch flushes or DLQ routing.
• Adding unit tests specifically covering _fast_row_and_insert_id_size (including nested fallback cases and shared object references) is recommended.

---

💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.

[github-actions]

🟡 Medium: objsize.get_deep_size tracks visited object IDs so that objects referenced multiple times within the same row (e.g., multiple columns referencing the same string/bytes object, or cached/interned scalars like True/False, small ints, or "") are only counted once. Here, v is only deduplicated against insert_id, so if multiple keys in row share the same value reference (especially a large str or bytes object), total will count sys.getsizeof(v) multiple times. This could cause a row to be overestimated relative to get_deep_size and falsely exceed _max_insert_payload_size (routing valid rows to FAILED_ROWS).

Consider either falling back to get_deep_size when total >= max_insert_payload_size in _check_row_size before rejecting a row, or tracking seen object IDs when needed.

[github-actions]

🟡 Medium: If _fast_row_and_insert_id_size overestimates a row due to shared object references across fields (which get_deep_size deduplicates), a row near the limit could be routed to the DLQ even though get_deep_size(row_and_insert_id) < self._max_insert_payload_size. Re-checking with get_deep_size(row_and_insert_id) inside the if row_byte_size >= self._max_insert_payload_size: branch would preserve exact backwards-compatible DLQ behavior at zero cost on the normal path.