[ohmayr]

Reads Arrow results for queries with an explicit destination table. The job's default stream rejects those jobs, so this reads the table through a read session instead.

• Opens a single-stream Arrow read session on the destination table, with the same serialization options.
• Marks the RowIterator from query_and_wait() when the job config sets destination.
• Adds unit tests and a system test for explicit destinations.

[gemini-code-assist]

Code Review

This pull request adds support for reading Arrow results from queries with explicit destination tables. It introduces helper functions _open_job_stream and _open_destination_stream to handle streaming from either the default job stream or the destination table. The RowIterator and helper functions are updated to track whether an explicit destination was used. Feedback on the changes includes: 1) Handling cases where row_iterator._billing_project is None in _open_destination_stream to prevent API failures due to invalid project formatting. 2) Raising a clear ValueError when _explicit_destination is true but the destination table is not set, rather than falling back to the default job stream which would fail with a cryptic API error.

[gemini-code-assist]

If row_iterator.client is None (which is possible if bqstorage_client is passed directly to download_arrow_from_job), row_iterator._billing_project will be None. This would cause parent to be formatted as projects/None, leading to an API failure. Falling back to row_iterator._project ensures that a valid project is always used.

Suggested change

	session = bqstorage_client.create_read_session(
	parent=f"projects/{row_iterator._billing_project}",
	read_session=bigquery_storage.types.stream.ReadSession(
	table=row_iterator._table.to_bqstorage(),
	data_format=bigquery_storage.types.stream.DataFormat.ARROW,
	read_options=bigquery_storage.types.stream.ReadSession.TableReadOptions(
	arrow_serialization_options=_read_rows_options(
	row_iterator._arrow_serialization_options
	)
	),
	),
	max_stream_count=1,
	timeout=timeout,
	)
	billing_project = row_iterator._billing_project or row_iterator._project
	if not billing_project:
	raise ValueError("Billing project could not be determined.")
	session = bqstorage_client.create_read_session(
	parent=f"projects/{billing_project}",
	read_session=bigquery_storage.types.stream.ReadSession(
	table=row_iterator._table.to_bqstorage(),
	data_format=bigquery_storage.types.stream.DataFormat.ARROW,
	read_options=bigquery_storage.types.stream.ReadSession.TableReadOptions(
	arrow_serialization_options=_read_rows_options(
	row_iterator._arrow_serialization_options
	)
	),
	),
	max_stream_count=1,
	timeout=timeout,
	)

[gemini-code-assist]

If row_iterator._explicit_destination is True but row_iterator._table is None, the code currently falls back to _open_job_stream. However, the default job stream is guaranteed to reject queries with explicit destination tables, resulting in a cryptic API error. Raising a clear ValueError upfront is much more helpful for debugging.

    if row_iterator._explicit_destination:
        if row_iterator._table is None:
            raise ValueError(
                "The row iterator is marked as having an explicit destination, "
                "but the destination table is not set."
            )
        reader, pa_schema = _open_destination_stream(
            row_iterator, bqstorage_client, timeout, pa_schema
        )
    else:
        reader = _open_job_stream(row_iterator, bqstorage_client, offset, timeout)

[tswast]

It's a bit strange to see this here. I don't see this option in the public docs: https://docs.cloud.google.com/bigquery/docs/reference/storage/rpc/google.cloud.bigquery.storage.v1#readrowsrequest

It does appear here, though; https://github.com/googleapis/googleapis/blob/cec791da64be8ae9fc9d5402515a144e63f673bf/google/cloud/bigquery/storage/v1/storage.proto#L291

I guess the docs will go out when it's officially preview?

[ohmayr]

it's in the proto but marked "This feature is not yet available" and only for job _default streams, so I'm guessing that's why it's not in the docs yet. That's the only place we send it, the explicit destination path passes it through the read session's options instead. I can ask the API team when it goes public.

[tswast]

Not for this PR, but important context: In other code paths, we allow this to be > 1 if we don't find a string like "ORDER BY" in the original query text. See:

google-cloud-python/packages/google-cloud-bigquery/google/cloud/bigquery/job/query.py

Lines 77 to 102 in 31d9365

	_CONTAINS_ORDER_BY = re.compile(r"ORDER\s+BY", re.IGNORECASE)
	_EXCEPTION_FOOTER_TEMPLATE = "{message}\n\nLocation: {location}\nJob ID: {job_id}\n"
	_TIMEOUT_BUFFER_SECS = 0.1
	def _contains_order_by(query):
	"""Do we need to preserve the order of the query results?
	This function has known false positives, such as with ordered window
	functions:
	.. code-block:: sql
	SELECT SUM(x) OVER (
	window_name
	PARTITION BY...
	ORDER BY...
	window_frame_clause)
	FROM ...
	This false positive failure case means the behavior will be correct, but
	downloading results with the BigQuery Storage API may be slower than it
	otherwise would. This is preferable to the false negative case, where
	results are expected to be in order but are not (due to parallel reads).
	"""
	return query and _CONTAINS_ORDER_BY.search(query)

As an alternative, to consider: for the polars connector, I'm requesting the user explicitly request ordered rows or not. See:

https://github.com/pola-rs/polars-bigquery-client/blob/4d025fdb314eef123c8b1ac0e7efeff40a81d653/py-arrow-bigquery/arrow_bigquery/_read_bigquery.py#L58

for table reads. Query code path is a work-in-progress.

I believe this is optional, though I would recommend checking with Alex to see if very large results are in scope for the arrow + python client work. If they are in scope, please create a follow-up PR to make sure the throughput of large unordered results is at least as good as the previous code paths.

[ohmayr]

Good point. The old path only drops to one stream when _contains_order_by() matches, and here we always use one, so for explicit destinations we can do the same check.

The default stream path doesn't open a read session at all, so there's no stream count to set there. Haven't checked yet if SplitReadStream works on it. I'll ask Alex if large results are in scope and file a follow-up if they are.

[ohmayr]

Filed #18804 to track it and to confirm scope with Alex.