fix(bigquery): apply max_results to the whole Arrow result

[ohmayr]

Applies max_results to the whole Arrow result. Before, it only limited the jobs.query first page and every remaining row was streamed.

• Slices the batch that reaches the limit and cancels the stream (feat(bigquery-storage): add Arrow options and cancel to read_rows #18643).
• Doesn't open a stream when the first page already reaches the limit.
• to_arrow() and to_dataframe() no longer warn about a REST fallback for Arrow results, and to_arrow() keeps the caller's bqstorage_client.
• Adds unit tests and a system test for limits within the first page, across the stream, and past the end.

[gemini-code-assist]

Code Review

This pull request adds support for limiting query results via max_results when using the ARROW format, including slicing Arrow record batches and canceling the BigQuery Storage API stream early if the limit is reached. The review feedback highlights a critical bug in table.py where disabling create_bqstorage_client for ARROW format prevents automatic client creation when streaming is required. Additionally, the reviewer recommends wrapping reader.cancel() in a try-except block to prevent masking other exceptions and suggests adding a unit test to cover the scenario where bqstorage_client is not provided.

[gemini-code-assist]

Setting create_bqstorage_client = False when self._query_results_format == QueryResultsFormat.ARROW.value prevents the creation of a bqstorage_client when the user hasn't provided one. If the query results exceed the first page, to_arrow() will fail with an AttributeError because it needs a bqstorage_client to stream the remaining pages. Instead, we should preserve create_bqstorage_client so that a client can be automatically created if needed.

        if self._query_results_format != QueryResultsFormat.ARROW.value:
            self._maybe_warn_max_results(bqstorage_client)
            if not self._should_use_bqstorage(
                bqstorage_client, create_bqstorage_client
            ):
                create_bqstorage_client = False
                bqstorage_client = None

[ohmayr]

This doesn't fail: when to_arrow() passes bqstorage_client=None for ARROW results, the download creates the client itself before it opens a stream (test_download_arrow_from_job_id_constructs_stream_and_reads covers that path). Creating one in to_arrow() would also make it create and close a client even when every row is in the first page. The next PR (#18652) replaces that with one shared client per Client.

[gemini-code-assist]

If reader.cancel() raises an exception (for example, if the stream is in a state where calling cancel raises an error, or if it's a mock/wrapper that doesn't fully implement the interface), it will mask any original exception raised inside the try block. To prevent masking the original exception and making debugging difficult, wrap reader.cancel() in a try...except block and log the exception as a warning instead of silently passing.

Suggested change

	finally:
	# Stop the stream if reading ends before it does: at max_results, on
	# an error, or when the caller closes the iterator.
	if not finished:
	reader.cancel()
	finally:
	# Stop the stream if reading ends before it does: at max_results, on
	# an error, or when the caller closes the iterator.
	if not finished:
	try:
	reader.cancel()
	except Exception as exc:
	logger.warning("Failed to cancel reader: %s", exc)

References

1. Avoid broad except Exception: blocks that silently return None. Instead, log the exception (e.g., using logger.warning) to aid in debugging and prevent masking underlying issues.

[ohmayr]

The reader here is our ReadRowsStream, whose cancel() just sets a flag and cancels the gRPC call, which doesn't raise even if the call already finished. I'd rather not add a try/except for an error that can't happen.

[gemini-code-assist]

This test only verifies the behavior when a bqstorage_client is explicitly provided. It would be highly beneficial to also add a test case where bqstorage_client=None (the default) to ensure that to_arrow() correctly creates a temporary bqstorage_client when streaming is required, preventing regressions like the one identified in table.py.

[ohmayr]

Good idea. #18652 (the next PR) adds test_to_arrow_with_arrow_format_reads_with_the_shared_client, which covers to_arrow() with bqstorage_client=None and checks that the stream gets the shared client.