Add feature to update table schemas for GCStoBQ load jobs #134
Conversation
| } | ||
|
|
||
| @Test | ||
| public void schemaUpdateSkippedWhenEnabled() throws Exception { |
There was a problem hiding this comment.
This name makes no sense to me. shcemaUpdateSkipped, but the test verifies that it was not skipped, right?
Is there a better name for this?
There was a problem hiding this comment.
My bad, corrected 👍
| GcsToBqWriter writer = | ||
| new GcsToBqWriter( | ||
| storage, bigQuery, schemaManager, retries, retryWaitMs, autoCreate, mockTime); | ||
| storage, bigQuery, schemaManager, retries, retryWaitMs, autoCreate,false, mockTime); |
There was a problem hiding this comment.
should be spaces around the "false" in the constructor.
| @@ -0,0 +1,242 @@ | |||
| package com.wepay.kafka.connect.bigquery.integration; | |||
There was a problem hiding this comment.
This file needs the standard header for the build to complete.
|
@Claudenw thanks for the heads up. |
Overview
Closes #96
Github Issue : #96
Schema evolution was not applied on the GCS batch path. Even with "allowNewBigQueryFields" and/or "allowSchemaUnionization" enabled, new columns were not added and required fields were not relaxed, causing LOAD jobs to fail with “No such field” errors and rejected rows.
This change enables the GCS → BigQuery batch path to update destination table schemas before issuing LOAD jobs.
When schema evolution is enabled, the batch writer derives a BigQuery schema from the batch’s records and calls SchemaManager.updateSchema(...) within the existing retry/timeout budget. If all schema-evolution flags(
allowNewBigQueryFields,allowBigQueryRequiredFieldRelaxation,allowSchemaUnionization) are disabled, the writer bypasses this step, preserving current behavior.Implementation Details
BigQuerySinkTask
Capture and persist all three schema-evolution flags at start-up:
allowNewBigQueryFieldsallowBigQueryRequiredFieldRelaxationallowSchemaUnionizationSupply these flags to the GCS writer so it uses
SchemaManageronly when pre-patching is required (at least one flag enabled).GcsToBqWriter
After ensuring the table exists , convert the batch’s schema to a BigQuery schema and invoke:
inside the existing retry/timeout wrapper, before uploading the batch payload to GCS.
If retries are exhausted, raise a clear error.
No-op when all schema-evolution flags are disabled = strict backward compatibility.
Testing
The result of the integration test:
GcsBatchSchemaEvolutionIT sends two batches through the GCS path. The first uses the original schema; the second relaxes a previously required field and adds a new nullable column. The table schema is updated before the second load, so both fields are NULLABLE and the new column appears as expected.
The load completes as expected and the data aligns with the evolved schema. This behavior confirms that the batch path now respects schema-relaxation and field-addition settings.
For local testing, I added a short pause (about 10 seconds) between the update and the second load to watch the schema change show up in Bigquery before proceeding. You can see the schema of the table before and after the second record inserted into the table.
Unit Tests