feat(bigquery): read Arrow results from explicit destinations

[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)