fix(bigquery): keep the Arrow first page and wait without rows

[ohmayr]

Keeps the jobs.query first page of Arrow results and streams only the rest, instead of dropping it and rereading every row.

• Uses pageRowCount and totalRows to decide whether more rows remain.
• Waits for the job with maxResults=0, so no unused JSON rows are fetched.
• QueryJob.result() keeps an Arrow first page, and reading the results twice gives the same rows.
• Adds unit tests and system tests comparing Arrow with JSON results, including other locations, all data types, extreme values, cached results, failed queries, and the existing query parameter cases.

[gemini-code-assist]

Code Review

This pull request enhances the handling of ARROW query results in query_and_wait by requesting zero rows during wait operations (since getQueryResults only returns STRUCT_ENCODING rows) and preserving the first page response for ARROW queries. It also introduces a helper function has_more_rows to check if rows remain by comparing pageRowCount and totalRows. The review feedback recommends making has_more_rows more robust by wrapping the integer conversions in a try-except block to safely handle potential TypeError or ValueError exceptions from malformed API responses.

[gemini-code-assist]

The current implementation of has_more_rows directly calls int() on the result of response.get(). If pageRowCount or totalRows is present in the response but has a None value (or is an empty/invalid string), this will raise a TypeError or ValueError. Using a safer parsing approach with a try-except block ensures the function is robust against unexpected or malformed API responses.

Suggested change

	def has_more_rows(response: Dict[str, Any]) -> bool:
	"""Return True if rows remain after an ``ARROW`` ``jobs.query`` response.
	``ARROW`` responses say how many rows the first page holds in
	``pageRowCount``, so compare that with ``totalRows`` instead of checking
	for a ``pageToken``.
	"""
	return int(response.get("pageRowCount", 0)) < int(response.get("totalRows", 0))
	def has_more_rows(response: Dict[str, Any]) -> bool:
	"""Return True if rows remain after an ``ARROW`` ``jobs.query`` response.
	``ARROW`` responses say how many rows the first page holds in
	``pageRowCount``, so compare that with ``totalRows`` instead of checking
	for a ``pageToken``.
	"""
	try:
	page_row_count = int(response.get("pageRowCount") or 0)
	total_rows = int(response.get("totalRows") or 0)
	except (ValueError, TypeError):
	return False
	return page_row_count < total_rows

[ohmayr]

pageRowCount and totalRows come from the jobs.query response, where they're always integer strings when present, and the existing code in this module already reads totalRows with int(). I'd rather not add handling for responses the API doesn't send.