[BigQueryIO] Use final destination table schema and metadata when creating temp tables

[ahmedabu98]

Fixes #22372

When we create temp tables, the jobs we send will naturally have CREATE_IF_NEEDED disposition. Users are facing a problem when writing with CREATE_NEVER, which makes providing a schema optional. When no schema is provided, the job that creates temp tables is sent with a null schema and an error is thrown.

The changes in this PR uses a DynamicDestinations object that returns the schema of the table when the user-provided schema is null.

Note: this code was removed in #17365 because SchemaUpdateOptions were introduced for temp tables. This PR matches the destination schema only when no SchemaUpdateOptions are provided

[ahmedabu98]

R: @MarcoRob
R: @Abacn

[github-actions]

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

[Abacn]

The test failure looks relevant: https://ci-beam.apache.org/job/beam_PreCommit_Java_GCP_IO_Direct_Commit/988/testReport/junit/org.apache.beam.sdk.io.gcp.bigquery/BigQuerySchemaUpdateOptionsIT/runWriteTestTempTables/

[ahmedabu98]

Run Java PreCommit

[ahmedabu98]

Run Java PreCommit

[Abacn]

Thanks for the quick fix. Just wondering why we do not need the condition of if (createDisposition.equals(CreateDisposition.CREATE_IF_NEEDED) || createDisposition.equals(CreateDisposition.CREATE_NEVER)) that is removed in https://github.com/apache/beam/pull/17365/files#diff-3edb3abd8909e7075e679e8b93420d4119858d6a9418b6ec10a204a592f062abL682-L692 ?

[ahmedabu98]

I believe it's an unnecessary check, since those are the only two existing CreateDispositions:

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

Line 2170 in 46919b7

public enum CreateDisposition {

that condition will always be true

[Abacn]

Thanks! This LGTM

[Abacn]

java precommit test failure irrelevant flake: org.apache.beam.runners.dataflow.worker.StreamingDataflowWorkerTest.testLatencyAttributionToQueuedState[0: [streamingEngine=false]]