[BEAM-11277] Respect schemaUpdateOptions during BigQuery load with temporary tables #14113

chunyang · 2021-02-27T09:38:44Z

The problem: Schema update options are not respected when using temporary tables to load data into BigQuery (e.g., when the number of files to load necessitate using multiple load jobs). Unlike when using query or load jobs, BigQuery does not allow field addition or relaxation when using copy jobs to append data to an existing table.

The solution: Before starting the copy jobs to move data from temporary tables to the final destination table, run zero row load jobs against the destination table(s) using the temporary table schemas and user provided schema update options. These zero row load jobs will update (if needed) the schema of the destination table(s) to accept the data from the forthcoming copy jobs.

If no schema update options are configured, then no zero-row load jobs will be run--UpdateDestinationSchema and WaitForSchemaModJobs become no-ops.

Summary of changes:

(This PR is stacked on top of [BEAM-11884] Pass destination as str rather than TableReference within BigQueryBatchFileLoads #14112)
Trigger zero-row load jobs against destination tables after successfully loading data into temporary tables and before triggering copy jobs.
Update BigQueryWrapper.perform_load_jobs to allow triggering load jobs using either a list of source URIs or a single file like object (for this PR the file like object is an empty file, or zero-row CSV file).

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	Dataflow	Samza	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.

When using temporary tables to append data to an existing table, first update the schema of the destination table if schema field addition or relaxation are allowed in schemaUpdateOptions. This needs to be done as a separate step because BQ copy jobs do not support schema update when appending to an existing table. WIP because empty files list does not work when submitting load jobs. Pointing to an empty file in GCS does work but that means this empty file needs to be created.

codecov · 2021-02-27T09:56:58Z

Codecov Report

Merging #14113 (b9ff38e) into master (40eef35) will decrease coverage by 0.01%.
The diff coverage is n/a.

@@            Coverage Diff             @@
##           master   #14113      +/-   ##
==========================================
- Coverage   83.39%   83.38%   -0.02%     
==========================================
  Files         469      469              
  Lines       58711    58783      +72     
==========================================
+ Hits        48962    49016      +54     
- Misses       9749     9767      +18

Impacted Files	Coverage Δ
...amples/snippets/transforms/aggregation/__init__.py
...dks/python/apache_beam/examples/wordcount_xlang.py
.../sdks/python/apache_beam/options/value_provider.py
.../examples/snippets/transforms/elementwise/regex.py
...ers/interactive/testing/integration/screen_diff.py
...python/apache_beam/examples/wordcount_xlang_sql.py
...d/srcs/sdks/python/apache_beam/io/filebasedsink.py
...che_beam/portability/api/beam_expansion_api_pb2.py
...he_beam/io/flink/flink_streaming_impulse_source.py
...ython/apache_beam/io/gcp/experimental/spannerio.py
... and 928 more

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 40eef35...b9ff38e. Read the comment docs.

chunyang · 2021-03-01T17:07:53Z

R: @pabloem
R: @angoenka

sdks/python/apache_beam/io/gcp/bigquery_file_loads.py

pabloem · 2021-03-10T23:50:20Z

Run Python 3.8 PostCommit

pabloem · 2021-03-11T01:09:43Z

sdks/python/apache_beam/io/gcp/bigquery_file_loads.py

+        location=temp_table_load_job_reference.location)
+    temp_table_schema = temp_table_load_job.configuration.load.schema
+


Does it make sense to compare the schema of the destination table with the schema of the temp table job? We'd save one load, right?

Added a simple comparison of destination_table.schema == temp_table_schema. It will work for trivial cases but doesn't catch the cases where the order of fields in a record differs. E.g., the following schemas are different according to == even though the temp table can be directly appended to the destination table without error.

<TableSchema fields: [ <TableFieldSchema fields: [], name: 'bytes', type: 'BYTES'>, <TableFieldSchema fields: [], name: 'date', type: 'DATE'>, <TableFieldSchema fields: [], name: 'time', type: 'TIME'> ]>

<TableSchema fields: [ <TableFieldSchema fields: [], name: 'date', type: 'DATE'>, <TableFieldSchema fields: [], name: 'time', type: 'TIME'>, <TableFieldSchema fields: [], name: 'bytes', type: 'BYTES'> ]>

I can probably write a function to check the schema recursively but do you know if one already exists?

…elaxation

apache#14113 (comment) Reusing one single PCollection concentrates the different paths into a single stage which complicates firing of triggers for the stage.

..when destination table schema matches temp table schema.

pabloem · 2021-03-23T18:32:47Z

Run Python 3.8 PostCommit

pabloem · 2021-03-23T22:28:58Z

@kmjung @vachan-shetty are you aware of a function in Beam or in BQ APIs that can help us compare two schemas?

kmjung · 2021-03-23T22:43:58Z

I'm not aware of anything like this in open source, no, although I will admit to not being very familiar with the Python SDK.

pabloem · 2021-03-31T19:19:23Z

Merging this for now. @chunyang feel free to add the schema matching functionality, or please create a JIRA issue if you can so we won't forget about it.

Thanks a lot for implementing this. It's great to have it.

chunyang added 3 commits February 27, 2021 01:13

Iterate over temp_tables_pc only once

d98f1d1

Consistency: Use pvalue rather than beam.pvalue

de0d90c

chunyang force-pushed the cyang/copy-schema-relaxation branch from fc1f910 to 9d55947 Compare March 1, 2021 17:00

chunyang marked this pull request as ready for review March 1, 2021 17:04

chunyang changed the title ~~[BEAM-11277] WIP: Respect schemaUpdateOptions even when using temporary tables~~ [BEAM-11277] Respect schemaUpdateOptions during BigQuery load with temporary tables Mar 1, 2021

chunyang force-pushed the cyang/copy-schema-relaxation branch from 9d55947 to 9af141d Compare March 1, 2021 17:09

chunyang added 3 commits March 1, 2021 13:42

Trigger schema update load job using empty file

91e9a15

Add tests for BigQueryWrapper.perform_load_job

1e25380

Add integration test

911945c

chunyang force-pushed the cyang/copy-schema-relaxation branch from 9af141d to 911945c Compare March 1, 2021 21:43

chunyang commented Mar 3, 2021

View reviewed changes

sdks/python/apache_beam/io/gcp/bigquery_file_loads.py Outdated Show resolved Hide resolved

pabloem reviewed Mar 11, 2021

View reviewed changes

chunyang added 4 commits March 19, 2021 11:51

Merge remote-tracking branch 'origin/master' into cyang/copy-schema-r…

b20c2cb

…elaxation

Use separate singletons in _load_data

0da0532

apache#14113 (comment) Reusing one single PCollection concentrates the different paths into a single stage which complicates firing of triggers for the stage.

Avoid running a schema modification load job...

6acd291

..when destination table schema matches temp table schema.

Fix spacing

b9ff38e

pabloem merged commit dcd8778 into apache:master Mar 31, 2021

chunyang deleted the cyang/copy-schema-relaxation branch March 31, 2021 20:20

chunyang mentioned this pull request Mar 31, 2021

[BEAM-11277] Add method to check if two TableSchema are equal #14394

Merged

4 tasks

chamikaramj mentioned this pull request Aug 17, 2021

[BEAM-12482] Pass null for schemaUpdateOptions #15281

Closed

4 tasks

chamikaramj mentioned this pull request Oct 6, 2021

[BEAM-12482] Ensure that we ignore schema update options when loading temporary tables that don't exist. #15635

Merged

4 tasks

MiguelAnzoWizeline mentioned this pull request Dec 22, 2021

[BEAM-12482] Respect schemaUpdateOptions during BigQuery load with temporary tables #16303

Closed

4 tasks

MarcoRob mentioned this pull request Apr 13, 2022

[BEAM-12482] Update Schema Destination during Bigquery load job when using temporary tables using zeroloadjob #17365

Merged

4 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

[BEAM-11277] Respect schemaUpdateOptions during BigQuery load with temporary tables #14113

[BEAM-11277] Respect schemaUpdateOptions during BigQuery load with temporary tables #14113

Uh oh!

chunyang commented Feb 27, 2021 •

edited

Loading

Uh oh!

codecov bot commented Feb 27, 2021 •

edited

Loading

Uh oh!

chunyang commented Mar 1, 2021

Uh oh!

Uh oh!

pabloem commented Mar 10, 2021

Uh oh!

pabloem Mar 11, 2021

Uh oh!

chunyang Mar 19, 2021

Uh oh!

pabloem commented Mar 23, 2021

Uh oh!

pabloem commented Mar 23, 2021

Uh oh!

kmjung commented Mar 23, 2021

Uh oh!

pabloem commented Mar 31, 2021

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

		location=temp_table_load_job_reference.location)
		temp_table_schema = temp_table_load_job.configuration.load.schema

[BEAM-11277] Respect schemaUpdateOptions during BigQuery load with temporary tables #14113

[BEAM-11277] Respect schemaUpdateOptions during BigQuery load with temporary tables #14113

Uh oh!

Conversation

chunyang commented Feb 27, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Post-Commit Tests Status (on master branch)

Pre-Commit Tests Status (on master branch)

GitHub Actions Tests Status (on master branch)

Uh oh!

codecov bot commented Feb 27, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

chunyang commented Mar 1, 2021

Uh oh!

Uh oh!

pabloem commented Mar 10, 2021

Uh oh!

pabloem Mar 11, 2021

Choose a reason for hiding this comment

Uh oh!

chunyang Mar 19, 2021

Choose a reason for hiding this comment

Uh oh!

pabloem commented Mar 23, 2021

Uh oh!

pabloem commented Mar 23, 2021

Uh oh!

kmjung commented Mar 23, 2021

Uh oh!

pabloem commented Mar 31, 2021

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

chunyang commented Feb 27, 2021 •

edited

Loading

codecov bot commented Feb 27, 2021 •

edited

Loading