Add schema to SpannerIO read

[Polber]

Following change adds SpannerIO.readWithSchema() method for reading Spanner table schema. This performs a call to the source table to get schema metadata for constructing a RowCoder for the output PCollection without having to manually provide a schema.

Logic follows BigQuerySourceDef very closely.

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
• Update CHANGES.md with noteworthy changes.
• If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

[github-actions]

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

[Abacn]

Format check error message

 Name 'StructTypeWithStruct' must match pattern '^[a-z][a-zA-Z0-9]*$'. [LocalVariableName]

have you been able to find SpannerIO reviewer?

[Abacn]

If this is meant for DataflowTemplates, we should push for GoogleCloudPlatform/DataflowTemplates#1429 again...

[Polber]

@Abacn

If this is meant for DataflowTemplates, we should push for GoogleCloudPlatform/DataflowTemplates#1429 again...

This is meant as a prerequisite to #31987 which adds ReadFromSpanner to Beam YAML.

It allows schema reads from Spanner without needing to specify a schema file which makes using in Beam YAML much more user-friendly and similar to how ReadFromBigQuery works.

Format check error message

 Name 'StructTypeWithStruct' must match pattern '^[a-z][a-zA-Z0-9]*$'. [LocalVariableName]

I can fix format error

have you been able to find SpannerIO reviewer?

No, I was waiting for checks to pass before finding one

[github-actions]

Assigning reviewers. If you would like to opt out of this review, comment assign to next reviewer:

R: @kennknowles for label java.
R: @shunping for label io.
R: @nielm for label spanner.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[damccorm]

@Polber what are next steps here?

[Abacn]

Pin a few potential @nielm @atask-g SpannerIO (owner) reviewer

[damccorm]

Pin a few potential @nielm @atask-g SpannerIO (owner) reviewer

Looks like these folks are out - @manitgupta could you help with a review here (or recommend someone to help)?

[manitgupta]

@bharadwaj-aditya @darshan-sj Can you have a look please?
I will review, but I have not reviewed many SpannerIO PRs in the past..

[nielm]

I am ooo until September but can review it then.

[darshan-sj]

Is it expected that the datatype of the corresponding values in the rowStruct to be of certain datatype? Like what is the datatype expected for DECIMAL? Does Spanner client library give the data in the expected datatype for all these types?

[Polber]

Since data types are not 1:1 for many different sources in Beam (BigQuery has lots of data types that are exclusive to BigQuery and do not exist as Java types) we map to the closest Java type.

For example, FLOAT64 is not a Java type, but Double is implemented as a 64-bit float, so the cast will work when the data is deserialized.

As I said in my other comment, all the data types that were added in this PR were tested and all the data conversions worked successfully

[damccorm]

Had 2 comments, (one just a nit), otherwise LGTM

[damccorm]

My crappy markdown code didn't work

Suggested change

	private SpannerSourceDef createSourceDef() {
	SpannerSourceDef sourceDef;
	if (getReadOperation().getQuery() != null) {
	return SpannerQuerySourceDef.create(getSpannerConfig(), getReadOperation().getQuery());
	}
	return SpannerTableSourceDef.create(
	getSpannerConfig(), getReadOperation().getTable(), getReadOperation().getColumns());
	}
	private SpannerSourceDef createSourceDef() {
	if (getReadOperation().getQuery() != null) {
	return SpannerQuerySourceDef.create(getSpannerConfig(), getReadOperation().getQuery());
	}
	return SpannerTableSourceDef.create(
	getSpannerConfig(), getReadOperation().getTable(), getReadOperation().getColumns());

[damccorm]

Probably worth doing this locally and running ./gradlew :sdks:java:io:google-cloud-platform:spotlessJava' as well to avoid more precommit failures

[damccorm]

Thanks!