[BEAM-7195] Fix BQ BatchLoads not creating new tables issue

[guangstick]

In WriteTables and WriteRename, the createDisposition is set to CREATE_NEVER when the pane index is great than 0.

beam/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/WriteTables.java

Lines 205 to 210 in ecedd3e

	if (c.pane().getIndex() > 0 && !tempTable) {
	// If writing directly to the destination, then the table is created on the first write
	// and we should change the disposition for subsequent writes.
	writeDisposition = WriteDisposition.WRITE_APPEND;
	createDisposition = CreateDisposition.CREATE_NEVER;
	} else if (tempTable) {

beam/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/WriteRename.java

Lines 145 to 148 in ecedd3e

	WriteDisposition writeDisposition =
	(c.pane().getIndex() == 0) ? firstPaneWriteDisposition : WriteDisposition.WRITE_APPEND;
	CreateDisposition createDisposition =
	(c.pane().getIndex() == 0) ? firstPaneCreateDisposition : CreateDisposition.CREATE_NEVER;

Before actually triggering BQ load, BatchLoads groups all temporary files by a singleton key.

beam/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchLoads.java

Lines 308 to 313 in ecedd3e

	results
	.apply("AttachSingletonKey", WithKeys.of((Void) null))
	.setCoder(
	KvCoder.of(VoidCoder.of(), WriteBundlesToFiles.ResultCoder.of(destinationCoder)))
	.apply("GroupOntoSingleton", GroupByKey.create())
	.apply("ExtractResultValues", Values.create())

This can be problematic when using DynamicDestinations in a streaming pipeline with triggers. Sometimes users would like to create tables in the middle of the pipeline based on the data or time. Because of the GroupOntoSingleton, different BQ destinations are grouped under the same key (always Void) and everything is under a global window, therefore the pane index is always incremental and new tables won't be created even though users have specified CreateDisposition = CREATE_IF_NEEDED

Proposed Solution: Instead of grouping everything into a singleton, grouping by Bigquery destinations so that new destinations can have a pane with index == 0.

---

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

• Choose reviewer(s) and mention them in a comment (R: @username).
• Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
• 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.

Post-Commit Tests Status (on master branch)

Lang	SDK	Dataflow	Flink	Samza	Spark	Twister2
Go		---		---		---
Java
Python				---		---
XLang		---		---		---

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website	Whitespace	Typescript
Non-portable
Portable	---		---	---	---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

[guangstick]

Open to use getter or use @AutoValue here

[guangstick]

R: @reuvenlax @chamikaramj

Question: will the PTransform renaming cause compatibility issues when users update their GCP Dataflow pipelines?

[reuvenlax]

so are you saying that the Reshuffle in writeTempTables is inheriting the pane index from this singleton GBK? I though that Reshuffle should be generating brand-new pane indices starting from 0.

[reuvenlax]

Ping.

[reuvenlax]

FYI - I think that this will fix the bug. Can you add a unit test that exposes the prior bug and is fixed now?

[guangstick]

R: @reuvenlax @chamikaramj

Question: will the PTransform renaming cause compatibility issues when users update their GCP Dataflow pipelines?

Thanks for the review @reuvenlax Yes, I can add a unit test for this. Could you also take a look at this question?

[reuvenlax]

Yes, this will break update. At this point I'm not sure if we have another option here. The old code was based on an incorrect assumption that Reshuffle generates pane indices, when actually Reshuffle simply forwards the pane indices from the previous GBK.

[stale]

This pull request has been marked as stale due to 60 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@beam.apache.org list. Thank you for your contributions.

[aaltay]

Do we want to merge this? Should this be closed?

Or do we not want to do this because it will break updates?

[chamikaramj]

Retest this please

[chamikaramj]

Run Dataflow ValidatesRunner

[chamikaramj]

Run Java PostCommit

[chamikaramj]

Run Java PreCommit

[chamikaramj]

Run Java PostCommit

[chamikaramj]

Run Java PostCommit

[reuvenlax]

Been looking at this, and I don't think that this PR works in the the temp-table code path. Looking to see if there's an easy way to fix this.

[reuvenlax]

I tried to fix these issues in pr/14238