Fixes a regression related to BQ read transform upgrade via the TransformService

[chamikaramj]

This fixes a regression introduced by #31128.

---

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.

[chamikaramj]

R: @ahmedabu98 or @johnjcasey

[github-actions]

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control

[ahmedabu98]

LGTM

[kennknowles]

This flag is for people to set the version back into the past. I don't think it works for this use case of making sure it is far enough into the future.

[kennknowles]

The problem case will be when this flag is not set at all, which will be the default case. So can we make things work by default when the flag is not set?

[ahmedabu98]

In the default case, the flag should be set by TransformUpgrader here

beam/sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/TransformUpgrader.java

Lines 251 to 259 in 1037ede

	String updateCompatibilityVersion =
	optionsClone.as(StreamingOptions.class).getUpdateCompatibilityVersion();
	if (updateCompatibilityVersion == null || updateCompatibilityVersion.isEmpty()) {
	// Setting the option 'updateCompatibilityVersion' to the current SDK version so that the
	// TransformService uses a compatible schema.
	optionsClone
	.as(StreamingOptions.class)
	.setUpdateCompatibilityVersion(ReleaseInfo.getReleaseInfo().getSdkVersion());
	}

[chamikaramj]

Yeah, this flag is set by the client when upgrading via the TransformUpgrader (client) so BQ transform just have to check it when upgrading and skip new fields (and set defaults).

[kennknowles]

Sounds like this null check is probably redundant, if the configRow crashes out? Or does the row have three states? (null, empty, string)

[chamikaramj]

Yeah, in this case the null check does not help since the field simply doesn't exist. I guess if the field has a null state it can still be good to have (unless we explicitly want to set null values).

[kennknowles]

Hmm, when rows expand the nullable fields should be filled with nulls. I know we have some support for this in the data plane in Dataflow. Perhaps there is something different here because it is not data plane?

[chamikaramj]

What do you mean by "rows expand" here ? Creating a POJO from a "from Row" function created using the SchemaRegistry.fromRowFunction util ?

Note we are trying to get specific fields from the Row which map to construction fields here. We don't have a single construction object to directly parse the Row into. So we should probably just add utils to handle schema-errors due to missing fields here.

[chamikaramj]

"pull_licenses_java.py" error is unrelated.