Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

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

Merged
merged 3 commits into from
Dec 2, 2022
Merged

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

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
5d0a6f2
always attempt to match tables with final destination
ahmedabu98 Dec 1, 2022
e7fecfa
do not match when schema update options exist
ahmedabu98 Dec 1, 2022
6f7a9d0
spotless
ahmedabu98 Dec 1, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 13 additions & 1 deletion ...o/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchLoads.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -716,6 +716,18 @@ private PCollection<KV<DestinationT, WriteTables.Result>> writeTempTables(
ShardedKeyCoder.of(NullableCoder.of(destinationCoder)),
WritePartition.ResultCoder.INSTANCE);

// If the final destination table exists already (and we're appending to it), then the temp
// tables must exactly match schema, partitioning, etc. Wrap the DynamicDestinations object
// with one that makes this happen.
// In the case schemaUpdateOptions are specified by the user, matching does not occur in order
// to respect those options.
DynamicDestinations<?, DestinationT> destinations = dynamicDestinations;
if (schemaUpdateOptions.isEmpty()) {
destinations =
DynamicDestinationsHelpers.matchTableDynamicDestinations(
dynamicDestinations, bigQueryServices);
}

// If WriteBundlesToFiles produced more than DEFAULT_MAX_FILES_PER_PARTITION files or
// DEFAULT_MAX_BYTES_PER_PARTITION bytes, then
// the import needs to be split into multiple partitions, and those partitions will be
Expand All @@ -734,7 +746,7 @@ private PCollection<KV<DestinationT, WriteTables.Result>> writeTempTables(
WriteDisposition.WRITE_EMPTY,
CreateDisposition.CREATE_IF_NEEDED,
sideInputs,
dynamicDestinations,
destinations,
loadJobProjectId,
maxRetryJobs,
ignoreUnknownValues,
Expand Down