[Feature Request]: Specify output_type in ReadFromBigQuery Beam YAML transform

[zycietocalareszta]

What would you like to happen?

It would be awesome to have the ability to specify output_type for ReadFromBigQuery Apache Beam YAML transform when using query.
Currently attempt to query the BigQuery table with this transform ends ups with "ValueError: Invalid transform specification at "Read from BigQuery" at line 3: Both a query and an output type of 'BEAM_ROW' were specified. 'BEAM_ROW' is not currently supported with queries." exception.

beam/sdks/python/apache_beam/io/gcp/bigquery.py

Lines 2973 to 2977 in c0a5895

	if self.output_type == 'BEAM_ROW' and self._kwargs.get('query',
	None) is not None:
	raise ValueError(
	"Both a query and an output type of 'BEAM_ROW' were specified. "
	"'BEAM_ROW' is not currently supported with queries.")

The workaround is to use combination of table, fields and row_restriction config parameters, but this does not allow for any aggregation, meaning that in some cases users must read a lot of data into memory instead of having BigQuery take care of it.

Issue Priority

Priority: 2 (default / most feature requests should be filed as P2)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Infrastructure
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[claudevdm]

@svetakvsundhar is this feasible? I see you added the beam_row option.

I am guessing since the table doesnt exist for a given query, we cant derive beam schema from the tableschema. Can we add the ability for a user to pass a schema?

[github-actions]

This issue has been marked as stale due to 150 days of inactivity. It will be closed in 30 days if no further activity occurs. If you think that’s incorrect or this issue still needs to be addressed, please simply write any comment. If closed, you can reopen the issue at any time. Thank you for your contributions.

[nikitagrover19]

Hey, looked into this more. Turns out the fix needs two files, not one.

yaml_io.py's read_from_bigquery() hardcodes output_type='BEAM_ROW' for every YAML read, and there's no schema param exposed at all. So even if we fix ReadFromBigQuery itself, YAML users still couldn't pass a schema through without touching this file too.

Here's what I'm thinking:

• In bigquery.py, add a query_output_schema param to ReadFromBigQuery. When someone sets query + BEAM_ROW + this schema, skip the get_table() lookup and build the row type directly from what they passed in. convert_to_usertype() already accepts any TableSchema-shaped object, so that part needs no changes. No schema still throws the existing error.
• In yaml_io.py, add a schema param to read_from_bigquery() (same pattern read_from_pubsub already uses), require it when query is set, and pass it through.

One thing I'm not 100% sure on - should the YAML schema be a BigQuery-style field list, or something more generic like JSON schema? Leaning toward BigQuery-style since it's a BigQuery transform, but open to being told otherwise.

I'd like to take this one if it sounds reasonable - happy to put up a PR.

[nikitagrover19]

.take-issue