[Prototype] Arrow support

[ohmayr]

Thank you for opening a Pull Request! Before submitting your PR, there are a few things you can do to make sure it goes smoothly:

• Make sure to open an issue as a bug/issue before writing your code! That way we can discuss the change, evaluate designs, and agree on the general idea
• Ensure the tests and linter pass
• Code coverage does not decrease (if any source code was changed)
• Appropriate docs were updated (if necessary)

Fixes #<issue_number_goes_here> 🦕

[gemini-code-assist]

Code Review

This pull request adds support for Apache Arrow serialization options and compression codecs (such as LZ4 and ZSTD) across the BigQuery and BigQuery Storage clients, enabling faster multi-page query results and introducing a performance benchmark. Feedback on these changes highlights a critical bug in table.py where a schema-applying elif block is unreachable, code duplication in reader.py when calling read_rows, a security risk from a hardcoded sandbox project ID in the benchmark script, and the use of a private _transport attribute instead of the public close() method for client cleanup.

[gemini-code-assist]

CRITICAL BUG: The elif block checking self._arrow_schema is not None is dead code and will never be executed.

Because the preceding if condition checks self._query_results_format == QueryResultsFormat.ARROW.value without checking self._arrow_schema, any time the format is ARROW, the if block will match and return pyarrow.Table.from_batches(record_batches) immediately, completely bypassing the elif block that applies the schema.

To fix this, we should modify the if condition to only match when self._arrow_schema is None.

Suggested change

	or self._query_results_format == QueryResultsFormat.ARROW.value
	):
	return pyarrow.Table.from_batches(record_batches)
	elif (
	self._query_results_format == QueryResultsFormat.ARROW.value
	and self._arrow_schema is not None
	):
	return pyarrow.Table.from_batches(record_batches, schema=self._arrow_schema)
	or (
	self._query_results_format == QueryResultsFormat.ARROW.value
	and self._arrow_schema is None
	)
	):
	return pyarrow.Table.from_batches(record_batches)
	elif (
	self._query_results_format == QueryResultsFormat.ARROW.value
	and self._arrow_schema is not None
	):
	return pyarrow.Table.from_batches(record_batches, schema=self._arrow_schema)

[gemini-code-assist]

The if/else block duplicates the call to self._client.read_rows. We can simplify this by dynamically constructing the keyword arguments dictionary, which improves readability and maintainability.

                kwargs = {
                    "read_stream": self._name,
                    "offset": self._offset,
                    **self._read_rows_kwargs,
                }
                if self._arrow_serialization_options is not None:
                    kwargs["arrow_serialization_options"] = self._arrow_serialization_options
                self._wrapped = self._client.read_rows(**kwargs)

[gemini-code-assist]

Avoid committing hardcoded sandbox project IDs or credentials to the repository. This is a security and privacy risk, and makes the benchmark script brittle for other users who do not have access to this project.

Instead, let the user specify the project ID via the GCLOUD_PROJECT environment variable. Ensure you validate the environment variable and fail fast if it is missing, empty, or contains only whitespace, rather than silently falling back to a default value.

Suggested change

	os.environ.setdefault("GCLOUD_PROJECT", "omairn-project-sandbox-174760")
	project_id = os.environ.get("GCLOUD_PROJECT")
	if not project_id or not project_id.strip():
	raise ValueError("GCLOUD_PROJECT environment variable must be set to a valid, non-empty value.")

References

1. When parsing environment variables, if a user explicitly sets an invalid value (such as whitespace-only), fail fast and raise an error to notify them of the invalid configuration rather than silently falling back to a default value.

[gemini-code-assist]

Avoid accessing the private _transport attribute to close the client. Instead, use the public close() method on the client itself, which is cleaner and more robust against future changes in google-api-core.

Additionally, when closing resources during cleanup, wrap the close call in a try-except block to handle exceptions individually, log any failures as warnings, and use a finally block to clear the resource reference. This ensures that a failure in closing one resource does not prevent other resources from being closed.

            if owns_bqstorage_client and bqstorage_client is not None:
                try:
                    bqstorage_client.close()
                except Exception as e:
                    import warnings
                    warnings.warn(f"Failed to close bqstorage_client: {e}")
                finally:
                    bqstorage_client = None

References

1. When closing multiple resources during cleanup, wrap each close call in a try-except block to handle exceptions individually, log any failures as warnings, and use a finally block to clear the resource references. This ensures that a failure in closing one resource does not prevent other resources from being closed.