Repository navigation
add WRITE_TRUNCATE_DATA write_disposition to the bigquery python sdk - #40622
GaryPWhite wants to merge 114 commits into
Conversation
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
📋 Review Summary
This PR adds BigQueryDisposition.WRITE_TRUNCATE_DATA to the Python BigQuery I/O connector. However, BigQueryBatchFileLoads in bigquery_file_loads.py was not updated to handle WRITE_TRUNCATE_DATA when copying from temporary tables or updating destination schemas, which can lead to data loss on multi-partition loads and broken schema_update_options behavior.
🔍 General Feedback
- Critical (
bigquery_file_loads.py): InBigQueryBatchFileLoads._load_data(sdks/python/apache_beam/io/gcp/bigquery_file_loads.py:970),if self.write_disposition in ('WRITE_EMPTY', 'WRITE_TRUNCATE'):groups temporary table load job IDs by destination table soTriggerCopyJobstruncates the destination table only once on the first copy job and usesWRITE_APPENDfor the rest. Because'WRITE_TRUNCATE_DATA'is missing from this check, copy jobs from multiple temp tables will run in parallel as single-element batches across workers—each issuingWRITE_TRUNCATE_DATAand overwriting data copied by previous jobs (see #24535). - Schema updates (
bigquery_file_loads.py): InUpdateDestinationSchema.process(sdks/python/apache_beam/io/gcp/bigquery_file_loads.py:368),self._write_disposition not in ('WRITE_TRUNCATE', 'WRITE_APPEND')will causeUpdateDestinationSchemato return early forWRITE_TRUNCATE_DATA, skippingschemaUpdateOptionson the destination table prior to copying temp tables. - Storage Write API (
bigquery.py):WriteToBigQuerywithmethod=STORAGE_WRITE_APIdelegates to the Java cross-language expansion service, whereBigQueryIO.Write.WriteDispositiononly definesWRITE_TRUNCATE,WRITE_APPEND, andWRITE_EMPTY. PassingWRITE_TRUNCATE_DATAwithSTORAGE_WRITE_APIwill fail with a JavaIllegalArgumentExceptionduring expansion; consider validating this upfront in Python. - Testing: Please add unit tests covering
WRITE_TRUNCATE_DATA(e.g., addingparam(write_disposition=BigQueryDisposition.WRITE_TRUNCATE_DATA)toTestBigQueryFileLoads.test_multiple_partition_files_write_dispositionsinbigquery_file_loads_test.pyand testingBigQueryDisposition.validate_write/BigQueryWriteFninbigquery_test.py).
💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.
| @staticmethod | ||
| def validate_write(disposition): | ||
| values = ( | ||
| BigQueryDisposition.WRITE_TRUNCATE_DATA, |
There was a problem hiding this comment.
🔴 Critical: Allowing WRITE_TRUNCATE_DATA here enables it for BigQueryBatchFileLoads, but sdks/python/apache_beam/io/gcp/bigquery_file_loads.py only checks for ('WRITE_EMPTY', 'WRITE_TRUNCATE') in BigQueryBatchFileLoads._load_data (line 970) and ('WRITE_TRUNCATE', 'WRITE_APPEND') in UpdateDestinationSchema.process (line 368).
Without including 'WRITE_TRUNCATE_DATA' in BigQueryBatchFileLoads._load_data, loads that spill into multiple temporary tables will not be grouped by destination before TriggerCopyJobs. As a result, parallel copy jobs will each execute with WRITE_TRUNCATE_DATA and overwrite one another's data. Please update both checks in bigquery_file_loads.py.
| and append actions occur as one atomic update upon job completion. | ||
| table. * WRITE_TRUNCATE_DATA: If the table already exists, BigQuery | ||
| overwrites the table data, but does not use schema and table constraints | ||
| from the source table.* WRITE_APPEND: If the table already exists, |
There was a problem hiding this comment.
🟡 Medium: bigquery_v2_messages.py is an autogenerated Apitools client file (# NOTE: This file is autogenerated and should not be edited by hand.), and writeDisposition is an unvalidated _messages.StringField(8), so editing this docstring has no runtime effect. In addition, lines 5445–5450 introduce trailing whitespace that will fail lint checks (table.* WRITE_APPEND: ). Consider reverting manual edits to this file or removing the trailing whitespace.
| - BigQueryDisposition.CREATE_NEVER: fail the write if does not exist. | ||
| write_disposition: A string describing what happens if the table has | ||
| already some data. Possible values are: | ||
| - BigQueryDisposition.WRITE_TRUNCATE_DATA: delete table data but preserve table metadata. |
There was a problem hiding this comment.
🟢 Low: This line is 98 characters long (and line 2133 is 105 characters), exceeding the 80-character line length limit enforced by lint checks.
|
Assigning reviewers: R: @damccorm for label python. Note: If you would like to opt out of this review, comment Available commands:
The PR bot will only process comments in the main thread (not review comments). |
Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
|
still coming back to this with some of gem/copilots comments folks, putting it down for the night :) |
|
Sounds good, converting to draft for now - please comment when it is ready |
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #40622 +/- ##
=============================================
+ Coverage 56.16% 79.92% +23.75%
=============================================
Files 1124 488 -636
Lines 178336 82539 -95797
Branches 1489 0 -1489
=============================================
- Hits 100170 65972 -34198
+ Misses 75645 16567 -59078
+ Partials 2521 0 -2521
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
…0627) Bumps [google.golang.org/api](https://github.com/googleapis/google-api-go-client) from 0.301.0 to 0.302.0. - [Release notes](https://github.com/googleapis/google-api-go-client/releases) - [Changelog](https://github.com/googleapis/google-api-go-client/blob/main/CHANGES.md) - [Commits](googleapis/google-api-go-client@v0.301.0...v0.302.0) --- updated-dependencies: - dependency-name: google.golang.org/api dependency-version: 0.302.0 dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
…pache#40626) * [Prism] Guard ClearOrderedListState against uninitialized state maps Problem: In TentativeData.ClearOrderedListState (sdks/go/pkg/beam/runners/prism/internal/engine/data.go), Prism indexed directly into d.state[stateID][w] and wrote the cleared slice back to kMap[string(uKey)] without checking whether d.state, d.state[stateID], or winMap[w] had been initialized: winMap := d.state[stateID] w := d.toWindow(wKey) kMap := winMap[w] data := kMap[string(uKey)] ... kMap[string(uKey)] = StateData{Bag: cleared} // panic: assignment to entry in nil map SDK harnesses (such as the Python SDK's SynchronousOrderedListRuntimeState.commit and the Rust SDK's OrderedListCell) buffer ordered-list mutations in memory during a bundle and flush pending StateRequest_Clear range requests before StateRequest_Append requests at bundle commit. When a bundle clears an ordered-list state cell (or range) in a window for which no prior bundle has persisted state on the runner, TentativeData.state has no map for (stateID, window), kMap is nil, and Prism panics on line 336. Fix: Use the existing d.clearState(stateID, wKey) helper in ClearOrderedListState and return early when kMap is nil, matching ClearBagState, ClearMultimapState, and ClearMultimapKeysState. Reproduction (Python SDK on PrismRunner): ```python import apache_beam as beam from apache_beam.coders import VarIntCoder from apache_beam.options.pipeline_options import PipelineOptions from apache_beam.testing.util import assert_that, equal_to from apache_beam.transforms.userstate import OrderedListStateSpec from apache_beam.utils.timestamp import Timestamp class ClearOrderedListDoFn(beam.DoFn): EVENTS = OrderedListStateSpec("events", VarIntCoder()) def process(self, element, events=beam.DoFn.StateParam(EVENTS)): key, (ts, op) = element if op == "CLEAR": events.clear() yield f"cleared:{key}" else: events.add((Timestamp(micros=ts), op)) entries = [f"{int(t.micros)}={v}" for t, v in events.read()] yield f"entry:{key}:{','.join(entries)}" if __name__ == "__main__": options = PipelineOptions(["--runner=PrismRunner", "--environment_type=LOOPBACK"]) with beam.Pipeline(options=options) as p: out = ( p | beam.Create([("k", (30, 300)), ("k", (10, 100)), ("k", (0, "CLEAR"))]) | beam.ParDo(ClearOrderedListDoFn()) ) assert_that( out, equal_to(["entry:k:30=300", "entry:k:10=100,30=300", "cleared:k"]), ) ``` * Refactor comments in data_test.go Updated comments for clarity in data_test.go.
…l-python-connector 1.23.0 (apache#40623)
* Restore Java 11 support in setJavaVerOptions the '11' branch. Workflows that still pass -PtestJavaVersion=11 now fail while Gradle configures the projects: Unknown Java Version 11 for setting additional java options * Trigger test workflow
* Enable throttler. * Preserve while loop instead of if.
… to readAsJson (apache#40702) * [DebeziumIO] Add opt-in full metadata (op and Debezium source fields) to readAsJson (apache#40701)
… transformWithState (apache#40703) * [apache#36841] Spark 4 streaming: windowed GroupByKey and Combine via transformWithState * Address review: move taskRunner to the ParDo processor, one SparkStateInternals per visit
…_phase2 Merge ValidatesRunner tests into small number of Dataflow jobs, each job hosting multiple tests.
…che#40695) * Expand Beam SQL TPC-DS query support and add Python TPC-DS suite - Add `BeamCrossJoinRel` and `BeamCrossJoinRule` to support bounded-by-bounded pure cross joins in Beam Calcite SQL. - Extend `BeamCoGBKJoinRule` to split mixed equi/non-equi inner join predicates into an equi-join followed by a post-join `BeamCalcRel` filter. - Support `LAG`, `LEAD`, `FIRST_VALUE`, `LAST_VALUE`, `NTH_VALUE` (`RESPECT`/`IGNORE NULLS`), and `CUME_DIST` analytic window functions in `BeamWindowRel` and `BeamBuiltinAnalyticFunctions`, and convert window inputs to `BeamLogicalConvention` in `BeamWindowRule`. - Fix `NULLS FIRST` and `NULLS LAST` comparator ordering in `BeamSortRel` when combined with `ASC`/`DESC`. - Normalize set operator (`UNION`, `INTERSECT`, `EXCEPT`) input schemas in `BeamSetOperatorRelBase` when inputs differ only by field names or nullability. - Return `NULL` instead of `0.0` from `VarianceFn` for empty or single-element sample variance/stddev groups to match SQL standard semantics. - Remove workarounds in TPC-DS table schemas used to cast `DECIMAL` to `DOUBLE` for fractional columns Having these improvements, 55 out of 103 queries now passing (up from 24). Then for test suite, - add `--queries=supported` and `10M`/`10MB` dataset size support to `TpcdsParametersReader`. - Add the Python TPC-DS benchmark and regression test suite (`sdks/python/apache_beam/testing/benchmarks/tpcds`) running on `PrismRunner` and cross-language `SqlTransform`. - Update Java TPC-DS GitHub Actions workflows to use `--queries=supported` and add the `beam_PostCommit_Python_Tpcds_Prism.yml` workflow. - Move query result file to a temp bucket with lifecycle policy, prevent resource leak
…0001) * [Python] Keep BigQuery load jobs distinct for table aliases * Document BigQuery destination-alias fix * [Python] Disable gRPC fork support before tox test startup
* Use GcsUtil V2 by default, add disable_gcsutil_v2 fallback GcsUtil now uses GcsUtilV2 (java-storage) by default. The disable_gcsutil_v2 experiment falls back to GcsUtilV1 (gcsio). use_gcsutil_v2 is still accepted but does nothing. * Fallback to GcsUtil V1 when use_grpc_for_gcs is set * Select GcsUtil version explicitly in ParquetIOLT * Edited CHANGES.md * Fix GcsUtilIT.testRewriteMultiPart to keep the V1 experiment * Trigger Java PostCommits for GcsUtil V2 default * Preserve kmsKeyName in GcsUtil V2 getObject. Fix test.
…apache#40671) * Scope RunInference model_identifier annotation to BeamML_RunInference * Address review: drop VertexModelMonitoringV2 model_identifier shadow, add TODO(apache#40715) for threshold transforms
…transient network errors during response body read (apache#40420)
Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
|
/gemini review |
There was a problem hiding this comment.
📋 Review Summary
This PR aims to add support for the WRITE_TRUNCATE_DATA write disposition in BigQuery IO across Java and Python SDKs. However, the Java UpdateSchemaDestination change introduces a syntax error that breaks compilation, and the PR branch appears to have been rebased/force-pushed onto an older base without updating the PR's target branch (master), pulling 110+ unrelated commits and 387 changed files into the diff.
🔍 General Feedback
- Git / Branch State: Please rebase the 4
WRITE_TRUNCATE_DATAcommits cleanly onto the latestapache:masterbranch so the PR only contains the relevant changes. - Java Compilation & Logic (
UpdateSchemaDestination.java): There is a missing closing parenthesis in theifcondition, and the condition itself is inverted (setting the schema only whenwriteDispositionisWRITE_TRUNCATE_DATArather than skipping it whenWRITE_TRUNCATE_DATAis used or whenschemaUpdateOptionsare not present). - Test & Fake Service Coverage:
FakeJobServicein Java andBigQueryWrapper.get_or_create_tablein Python do not yet handleWRITE_TRUNCATE_DATA, which will cause tests or table creation checks to not reflect the truncation behavior.
💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.
💬 Inline Findings
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/UpdateSchemaDestination.java(line 173): 🛑 Critical: There are two issues here:
- Syntax Error: The
ifstatement is missing a closing parenthesis)before{, which prevents the Java module from compiling. - Inverted Logic:
loadConfig.setSchema(destinationSchema)is currently only called whenwriteDispositionequalsWRITE_TRUNCATE_DATA, breaking all other write dispositions (WRITE_APPEND,WRITE_EMPTY,WRITE_TRUNCATE) by omitting the destination schema on load jobs. Furthermore, BigQuery allowsschema+schemaUpdateOptionswithWRITE_TRUNCATE_DATA, so if the goal is to avoid overwriting the existing table's schema whenWRITE_TRUNCATE_DATAis used without schema update options, the condition should be negated (or gated on!BigQueryIO.Write.WriteDisposition.WRITE_TRUNCATE_DATA.name().equals(writeDisposition)).
.equals(BigQueryIO.Write.WriteDisposition.WRITE_TRUNCATE_DATA.toString())) {
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiLoads.java(line 270): 🟠 High:BigQueryIO.Write.validate()rejectsWRITE_TRUNCATE_DATAfor Storage Write API methods (STORAGE_WRITE_APIandSTORAGE_API_AT_LEAST_ONCE), so this condition is currently unreachable. Additionally,CreateTableHelpers.createTableWrapperdoes not truncate existing table data—treatingWRITE_TRUNCATE_DATAlikeCREATE_IF_NEEDEDhere would silently append data instead of truncating.
if (createDisposition == CreateDisposition.CREATE_IF_NEEDED) {
-
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/providers/BigQueryStorageWriteApiSchemaTransformProvider.java(line 288): 🟠 High: MappingWRITE_TRUNCATE_DATAinBigQueryStorageWriteApiSchemaTransformProviderwill fail validation inBigQueryIO.Write.validate(), which explicitly disallowsWRITE_TRUNCATE_DATAwithSTORAGE_WRITE_APIandSTORAGE_API_AT_LEAST_ONCE. Unless Storage Write API support forWRITE_TRUNCATE_DATAis implemented, this mapping should be removed. -
sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOWriteTest.java(line 1007): 🟡 Medium: To properly verifyWRITE_TRUNCATE_DATAbehavior (that pre-existing rows are truncated while the existing table schema/constraints are preserved), this test should pre-populate the destination table with initial rows before running the pipeline and assert that only the newly written rows remain afterward. Note thatFakeJobService.validateDispositionsalso needs to be updated to handleWriteDisposition.WRITE_TRUNCATE_DATAso the fake service actually clears existing rows during the test.
|
going to take more doing than I have the time for right just now -- specifically seems like digging into the java BigQueryIOWrite utils to supplement the python sdk. |
I've noticed when using the python SDK for Dataflow pipelines, we can't use the
WRITE_TRUNCATE_DATAwrite disposition. I believe this change is as simple as adding the disposition for use, adding it tovalidate_writeand the streaming insert prevention list.I've also added the change to
CHANGES.md. AFAIK there's no issue open for this right now.LMK if there's anything missing.