claudevdm commented on code in PR #39236:
URL: https://github.com/apache/beam/pull/39236#discussion_r3960207193
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java:
##########
@@ -4011,11 +4011,13 @@ private <DestinationT> WriteResult expandTyped(
// TODO: If the user provided a schema, we should use that. There are
things that can be
// specified in a
// BQ schema that don't have exact matches in a Beam schema (e.g.
GEOGRAPHY types).
- TableSchema tableSchema =
BigQueryUtils.toTableSchema(input.getSchema());
- dynamicDestinations =
- new ConstantSchemaDestinations<>(
- dynamicDestinations,
-
StaticValueProvider.of(BigQueryHelpers.toJsonString(tableSchema)));
+ if (!hasSchema) {
Review Comment:
The definition of hasSchema conflates "the user supplied a
DynamicDestinations" with "the user supplied a schema". The javadoc of
useBeamSchema promises the BQ schema is inferred from the input schema.
This change means that user supplied schema overrules the inferred schema,
which could potentially be a breaking change if someone just passed an invalid
schema while silently relying on input schema?
Also I think this will fail pipelines that have DynamicDestination +
getSchema returns null?
```
@DefaultSchema(JavaFieldSchema.class)
class Event { String type; String name; int number; }
p.apply(Create.of(new Event("click", "a", 1), new Event("view", "b", 2)))
.apply(BigQueryIO.<Event>write()
.to(new DynamicDestinations<Event, String>() {
public String getDestination(ValueInSingleWindow<Event> e) {
return "proj:ds.events_" + e.getValue().type;
}
public TableDestination getTable(String dest) {
return new TableDestination(dest, null);
}
public @Nullable TableSchema getSchema(String dest) {
return null; // inferred from the Beam schema, per useBeamSchema()
}
})
.useBeamSchema()
.withMethod(STORAGE_WRITE_API)
.withCreateDisposition(CREATE_IF_NEEDED));
```
because
https://github.com/jrmccluskey/beam/blob/3353e1ba3380ab65ac9457556f44bd66ceb508db/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java#L3988-L3991
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiDynamicDestinationsTableRow.java:
##########
@@ -94,8 +93,29 @@ public MessageConverter<T> getMessageConverter(
}
};
+ TableSchema destSchema = getSchema(destination);
Review Comment:
Everything here runs for every Storage Write API use.
Can we scope these changes to only the portable provider?
The declared schema is no longer authoritative for the proto descriptor. If
there was a mismatch writing to BQ would raise an error.
Is this a regression for users who relied on the declared schema being
authoritative and on disagreement being loud?
Some potentially changed behaviors
| Case | Master | Branch | Verdict |
|---|---|---|---|
| D equals T, R has an extra key with a null value | Per-row failure to
failed rows | Silently dropped, row written | Changed by the null skip alone.
The documented default "unknown values are errors" no longer holds for nulls |
| D wider than T, R's extra column is null | Stream-level INVALID_ARGUMENT,
job stalls or fails, zero rows | T substituted, column dropped, row written,
job succeeds | Changed |
| D wider than T, R's extra column is non-null | Same stream-level failure |
T substituted, per-row failure to failed rows, job succeeds | Changed |
| D wider than T, user set `ignoreUnknownValues()` | Still the stream-level
failure, since that option only governs row versus descriptor | T substituted,
extra column dropped for every row | Changed, and arguably what that user
wanted |
| D wider than T, then someone adds the column to T while the job runs |
Next append succeeds, the job heals on its own | Cached T is never refreshed in
strict mode, so the column keeps being dropped or failed until the workers
restart | Changed, an operability regression |
D = declared schema, T = existing table schema, R = a row.
D = declared schema, T = existing table schema, R = a row.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
