Abacn commented on code in PR #25410:
URL: https://github.com/apache/beam/pull/25410#discussion_r1102105132
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/UpdateSchemaDestination.java:
##########
@@ -127,25 +147,24 @@ public void processElement(
ProcessContext context,
BoundedWindow window)
throws IOException {
- DestinationT destination = null;
- BigQueryOptions options =
context.getPipelineOptions().as(BigQueryOptions.class);
+ dynamicDestinations.setSideInputAccessorFromProcessContext(context);
+ List<KV<TableDestination, WriteTables.Result>> outputs =
Lists.newArrayList();
for (KV<DestinationT, WriteTables.Result> entry : element) {
- destination = entry.getKey();
- if (destination != null) {
Review Comment:
I neglected the handling of null destination in the original implementation.
I do not see what kind of scenario would have null destination (unless user
provided DynamicDestination.getDestination returns a null, but the
documentation says it may not return null). Nevertheless, the upstream
WriteTempTables essentially has this processElement calling
`dynamicDestinations.getTable(destination)` for all incoming elements:
https://github.com/apache/beam/blob/78c15648b514f5e61b83c593715114516ea639fb/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/WriteTables.java#L221
and asking it return a nonNull tableDestination. (same call used in
UpdateSchemaDestination). So make the behavior consistent to WriteTables here
--
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]
