You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "reuvenlax (via GitHub)" <gi...@apache.org> on 2026/04/29 19:23:58 UTC
[PR] only expand update graph if needed [beam]
reuvenlax opened a new pull request, #38332:
URL: https://github.com/apache/beam/pull/38332
(no comment)
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] only expand update graph if needed [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on code in PR #38332:
URL: https://github.com/apache/beam/pull/38332#discussion_r3163621377
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiConvertMessages.java:
##########
@@ -147,109 +147,117 @@ public PCollectionTuple expand(PCollection<KV<DestinationT, ElementT>> input) {
.get(patchTableSchemaTag)
.setCoder(KvCoder.of(destinationCoder, ProtoCoder.of(TableSchema.class)));
result.get(elementsWaitingForSchemaTag).setCoder(KvCoder.of(destinationCoder, elementCoder));
+ if (!hasSchemaUpdateOptions) {
+ // Don't expand the update graph if it's not needed.
+ return result;
+ } else {
+ final int numShards =
+ input
+ .getPipeline()
+ .getOptions()
+ .as(BigQueryOptions.class)
+ .getSchemaUpgradeBufferingShards();
- final int numShards =
- input
- .getPipeline()
- .getOptions()
- .as(BigQueryOptions.class)
- .getSchemaUpgradeBufferingShards();
+ // Throttle the stream to the patch-table function so that only a single update per table per
+ // two seconds gets processed (to match quotas). The combiner merges incremental schemas, so
+ // we
+ // won't miss any updates.
+ PCollection<KV<ShardedKey<DestinationT>, ElementT>> tablesPatched =
+ result
+ .get(patchTableSchemaTag)
+ .apply(
+ "rewindow",
+ Window.<KV<DestinationT, TableSchema>>configure()
+ .triggering(
+ Repeatedly.forever(
+ AfterProcessingTime.pastFirstElementInPane()
+ .plusDelayOf(Duration.standardSeconds(2))))
+ .discardingFiredPanes())
+ .apply("merge schemas", Combine.fewKeys(new MergeSchemaCombineFn()))
+ .setCoder(KvCoder.of(destinationCoder, ProtoCoder.of(TableSchema.class)))
+ .apply(
+ "Patch table schema",
+ ParDo.of(
+ new PatchTableSchemaDoFn<>(operationName, bqServices, dynamicDestinations)))
+ .setCoder(KvCoder.of(destinationCoder, NullableCoder.of(elementCoder)))
+ // We need to make sure that all shards of the buffering transform are notified.
+ .apply(
+ "fanout to all shards",
+ FlatMapElements.via(
+ new SimpleFunction<
+ KV<DestinationT, ElementT>,
+ Iterable<KV<ShardedKey<DestinationT>, ElementT>>>() {
+ @Override
+ public Iterable<KV<ShardedKey<DestinationT>, ElementT>> apply(
+ KV<DestinationT, ElementT> elem) {
+ return IntStream.range(0, numShards)
+ .mapToObj(
+ i ->
+ KV.of(
+ StorageApiConvertMessages.AssignShardFn.getShardedKey(
+ elem.getKey(), i, numShards),
+ elem.getValue()))
+ .collect(Collectors.toList());
+ }
+ }))
+ .setCoder(
+ KvCoder.of(ShardedKey.Coder.of(destinationCoder), NullableCoder.of(elementCoder)))
+ .apply(
+ Window.<KV<ShardedKey<DestinationT>, ElementT>>configure()
+ .triggering(DefaultTrigger.of()));
- // Throttle the stream to the patch-table function so that only a single update per table per
- // two seconds gets processed (to match quotas). The combiner merges incremental schemas, so we
- // won't miss any updates.
- PCollection<KV<ShardedKey<DestinationT>, ElementT>> tablesPatched =
- result
- .get(patchTableSchemaTag)
- .apply(
- "rewindow",
- Window.<KV<DestinationT, TableSchema>>configure()
- .triggering(
- Repeatedly.forever(
- AfterProcessingTime.pastFirstElementInPane()
- .plusDelayOf(Duration.standardSeconds(2))))
- .discardingFiredPanes())
- .apply("merge schemas", Combine.fewKeys(new MergeSchemaCombineFn()))
- .setCoder(KvCoder.of(destinationCoder, ProtoCoder.of(TableSchema.class)))
- .apply(
- "Patch table schema",
- ParDo.of(
- new PatchTableSchemaDoFn<>(operationName, bqServices, dynamicDestinations)))
- .setCoder(KvCoder.of(destinationCoder, NullableCoder.of(elementCoder)))
- // We need to make sure that all shards of the buffering transform are notified.
- .apply(
- "fanout to all shards",
- FlatMapElements.via(
- new SimpleFunction<
- KV<DestinationT, ElementT>,
- Iterable<KV<ShardedKey<DestinationT>, ElementT>>>() {
- @Override
- public Iterable<KV<ShardedKey<DestinationT>, ElementT>> apply(
- KV<DestinationT, ElementT> elem) {
- return IntStream.range(0, numShards)
- .mapToObj(
- i ->
- KV.of(
- StorageApiConvertMessages.AssignShardFn.getShardedKey(
- elem.getKey(), i, numShards),
- elem.getValue()))
- .collect(Collectors.toList());
- }
- }))
- .setCoder(
- KvCoder.of(ShardedKey.Coder.of(destinationCoder), NullableCoder.of(elementCoder)))
- .apply(
- Window.<KV<ShardedKey<DestinationT>, ElementT>>configure()
- .triggering(DefaultTrigger.of()));
+ // Any elements that are waiting for a schema update are sent to this stateful DoFn to be
+ // buffered.
+ // Note: we currently do not provide the DynamicDestinations object access to the side input
+ // in
+ // this path.
+ // This is because side inputs are not currently available from timer callbacks. Since side
+ // inputs are generally
+ // used for getSchema and in this case we read the schema from the table, this is unlikely to
+ // be
+ // a problem.
+ PCollection<KV<ShardedKey<DestinationT>, ElementT>> shardedWaitingElements =
+ result
+ .get(elementsWaitingForSchemaTag)
+ // TODO: Consider using GroupIntoBatchs.withShardingKey to get auto sharding here
+ // instead of fixed sharding.
+ .apply("assignShard", ParDo.of(new AssignShardFn<>(numShards)))
+ .setCoder(
+ KvCoder.of(
+ ShardedKey.Coder.of(destinationCoder), NullableCoder.of(elementCoder)));
- // Any elements that are waiting for a schema update are sent to this stateful DoFn to be
- // buffered.
- // Note: we currently do not provide the DynamicDestinations object access to the side input in
- // this path.
- // This is because side inputs are not currently available from timer callbacks. Since side
- // inputs are generally
- // used for getSchema and in this case we read the schema from the table, this is unlikely to be
- // a problem.
- PCollection<KV<ShardedKey<DestinationT>, ElementT>> shardedWaitingElements =
- result
- .get(elementsWaitingForSchemaTag)
- // TODO: Consider using GroupIntoBatchs.withShardingKey to get auto sharding here
- // instead of fixed sharding.
- .apply("assignShard", ParDo.of(new AssignShardFn<>(numShards)))
- .setCoder(
- KvCoder.of(ShardedKey.Coder.of(destinationCoder), NullableCoder.of(elementCoder)));
+ PCollectionList<KV<ShardedKey<DestinationT>, ElementT>> waitingElementsList =
+ PCollectionList.of(shardedWaitingElements).and(tablesPatched);
+ PCollectionTuple retryResult =
+ waitingElementsList
+ .apply("Buffered flatten", Flatten.pCollections())
+ .apply(
+ "bufferElements",
+ ParDo.of(new SchemaUpdateHoldingFn<>(elementCoder, convertMessagesDoFn))
+ .withOutputTags(
+ successfulWritesTag,
+ TupleTagList.of(ImmutableList.of(failedWritesTag, BAD_RECORD_TAG))));
+ retryResult.get(successfulWritesTag).setCoder(successCoder);
+ retryResult.get(failedWritesTag).setCoder(errorCoder);
+ retryResult.get(BAD_RECORD_TAG).setCoder(BadRecord.getCoder(input.getPipeline()));
- PCollectionList<KV<ShardedKey<DestinationT>, ElementT>> waitingElementsList =
- PCollectionList.of(shardedWaitingElements).and(tablesPatched);
- PCollectionTuple retryResult =
- waitingElementsList
- .apply("Buffered flatten", Flatten.pCollections())
- .apply(
- "bufferElements",
- ParDo.of(new SchemaUpdateHoldingFn<>(elementCoder, convertMessagesDoFn))
- .withOutputTags(
- successfulWritesTag,
- TupleTagList.of(ImmutableList.of(failedWritesTag, BAD_RECORD_TAG))));
- retryResult.get(successfulWritesTag).setCoder(successCoder);
- retryResult.get(failedWritesTag).setCoder(errorCoder);
- retryResult.get(BAD_RECORD_TAG).setCoder(BadRecord.getCoder(input.getPipeline()));
-
- // Flatten successes and failures from both the regular transform and the retry transform.
- PCollection<KV<DestinationT, StorageApiWritePayload>> allSuccesses =
- PCollectionList.of(result.get(successfulWritesTag))
- .and(retryResult.get(successfulWritesTag))
- .apply("flattenSuccesses", Flatten.pCollections());
- PCollection<BigQueryStorageApiInsertError> allFailures =
- PCollectionList.of(result.get(failedWritesTag))
- .and(retryResult.get(failedWritesTag))
- .apply("flattenFailures", Flatten.pCollections());
- PCollection<BadRecord> allBadRecords =
- PCollectionList.of(result.get(BAD_RECORD_TAG))
- .and(retryResult.get(BAD_RECORD_TAG))
- .apply("flattenBadRecords", Flatten.pCollections());
- return PCollectionTuple.of(successfulWritesTag, allSuccesses)
- .and(failedWritesTag, allFailures)
- .and(BAD_RECORD_TAG, allBadRecords);
+ // Flatten successes and failures from both the regular transform and the retry transform.
+ PCollection<KV<DestinationT, StorageApiWritePayload>> allSuccesses =
+ PCollectionList.of(result.get(successfulWritesTag))
+ .and(retryResult.get(successfulWritesTag))
+ .apply("flattenSuccesses", Flatten.pCollections());
+ PCollection<BigQueryStorageApiInsertError> allFailures =
+ PCollectionList.of(result.get(failedWritesTag))
+ .and(retryResult.get(failedWritesTag))
+ .apply("flattenFailures", Flatten.pCollections());
+ PCollection<BadRecord> allBadRecords =
+ PCollectionList.of(result.get(BAD_RECORD_TAG))
+ .and(retryResult.get(BAD_RECORD_TAG))
+ .apply("flattenBadRecords", Flatten.pCollections());
+ return PCollectionTuple.of(successfulWritesTag, allSuccesses)
+ .and(failedWritesTag, allFailures)
+ .and(BAD_RECORD_TAG, allBadRecords);
+ }
Review Comment:
To improve readability and reduce nesting, you can use a guard clause. Instead of wrapping the rest of the method in an `else` block, you can return early and then have the main logic at the top level of the method.
```java
final int numShards =
input
.getPipeline()
.getOptions()
.as(BigQueryOptions.class)
.getSchemaUpgradeBufferingShards();
// Throttle the stream to the patch-table function so that only a single update per table per
// two seconds gets processed (to match quotas). The combiner merges incremental schemas, so
// we
// won't miss any updates.
PCollection<KV<ShardedKey<DestinationT>, ElementT>> tablesPatched =
result
.get(patchTableSchemaTag)
.apply(
"rewindow",
Window.<KV<DestinationT, TableSchema>>configure()
.triggering(
Repeatedly.forever(
AfterProcessingTime.pastFirstElementInPane()
.plusDelayOf(Duration.standardSeconds(2))))
.discardingFiredPanes())
.apply("merge schemas", Combine.fewKeys(new MergeSchemaCombineFn()))
.setCoder(KvCoder.of(destinationCoder, ProtoCoder.of(TableSchema.class)))
.apply(
"Patch table schema",
ParDo.of(
new PatchTableSchemaDoFn<>(operationName, bqServices, dynamicDestinations)))
.setCoder(KvCoder.of(destinationCoder, NullableCoder.of(elementCoder)))
// We need to make sure that all shards of the buffering transform are notified.
.apply(
"fanout to all shards",
FlatMapElements.via(
new SimpleFunction<
KV<DestinationT, ElementT>,
Iterable<KV<ShardedKey<DestinationT>, ElementT>>>() {
@Override
public Iterable<KV<ShardedKey<DestinationT>, ElementT>> apply(
KV<DestinationT, ElementT> elem) {
return IntStream.range(0, numShards)
.mapToObj(
i ->
KV.of(
StorageApiConvertMessages.AssignShardFn.getShardedKey(
elem.getKey(), i, numShards),
elem.getValue()))
.collect(Collectors.toList());
}
}))
.setCoder(
KvCoder.of(ShardedKey.Coder.of(destinationCoder), NullableCoder.of(elementCoder)))
.apply(
Window.<KV<ShardedKey<DestinationT>, ElementT>>configure()
.triggering(DefaultTrigger.of()));
// Any elements that are waiting for a schema update are sent to this stateful DoFn to be
// buffered.
// Note: we currently do not provide the DynamicDestinations object access to the side input
// in
// this path.
// This is because side inputs are not currently available from timer callbacks. Since side
// inputs are generally
// used for getSchema and in this case we read the schema from the table, this is unlikely to
// be
// a problem.
PCollection<KV<ShardedKey<DestinationT>, ElementT>> shardedWaitingElements =
result
.get(elementsWaitingForSchemaTag)
// TODO: Consider using GroupIntoBatchs.withShardingKey to get auto sharding here
// instead of fixed sharding.
.apply("assignShard", ParDo.of(new AssignShardFn<>(numShards)))
.setCoder(
KvCoder.of(
ShardedKey.Coder.of(destinationCoder), NullableCoder.of(elementCoder)));
PCollectionList<KV<ShardedKey<DestinationT>, ElementT>> waitingElementsList =
PCollectionList.of(shardedWaitingElements).and(tablesPatched);
PCollectionTuple retryResult =
waitingElementsList
.apply("Buffered flatten", Flatten.pCollections())
.apply(
"bufferElements",
ParDo.of(new SchemaUpdateHoldingFn<>(elementCoder, convertMessagesDoFn))
.withOutputTags(
successfulWritesTag,
TupleTagList.of(ImmutableList.of(failedWritesTag, BAD_RECORD_TAG))));
retryResult.get(successfulWritesTag).setCoder(successCoder);
retryResult.get(failedWritesTag).setCoder(errorCoder);
retryResult.get(BAD_RECORD_TAG).setCoder(BadRecord.getCoder(input.getPipeline()));
// Flatten successes and failures from both the regular transform and the retry transform.
PCollection<KV<DestinationT, StorageApiWritePayload>> allSuccesses =
PCollectionList.of(result.get(successfulWritesTag))
.and(retryResult.get(successfulWritesTag))
.apply("flattenSuccesses", Flatten.pCollections());
PCollection<BigQueryStorageApiInsertError> allFailures =
PCollectionList.of(result.get(failedWritesTag))
.and(retryResult.get(failedWritesTag))
.apply("flattenFailures", Flatten.pCollections());
PCollection<BadRecord> allBadRecords =
PCollectionList.of(result.get(BAD_RECORD_TAG))
.and(retryResult.get(BAD_RECORD_TAG))
.apply("flattenBadRecords", Flatten.pCollections());
return PCollectionTuple.of(successfulWritesTag, allSuccesses)
.and(failedWritesTag, allFailures)
.and(BAD_RECORD_TAG, allBadRecords);
```
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] only expand update graph if needed [beam]
Posted by "reuvenlax (via GitHub)" <gi...@apache.org>.
reuvenlax commented on PR #38332:
URL: https://github.com/apache/beam/pull/38332#issuecomment-4346847592
FYI @ahmedabu98
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] only expand update graph if needed [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on PR #38332:
URL: https://github.com/apache/beam/pull/38332#issuecomment-4346846849
## Summary of Changes
Hello, I'm Gemini Code Assist[^1]! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!
This pull request introduces a performance optimization by conditionally expanding the update graph in the BigQuery Storage API conversion logic. By checking for the presence of schema update options before building the graph, the pipeline avoids unnecessary overhead when these features are not in use.
### Highlights
* **Conditional Graph Expansion**: Added a conditional check to skip the expansion of the update graph if schema update options are not required, optimizing pipeline construction.
🧠**New Feature in Public Preview:** You can now enable **Memory** to help **Gemini Code Assist** learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. **Click [here](https://codeassist.google/code-review/login) to enable Memory in your admin console.**
<details>
<summary><b>Using Gemini Code Assist</b></summary>
<br>
The full guide for Gemini Code Assist can be found on our [documentation page](https://developers.google.com/gemini-code-assist/docs/review-github-code), here are some quick tips.
<b>Invoking Gemini</b>
You can request assistance from Gemini at any point by creating a comment using either `/gemini <command>` or `@gemini-code-assist <command>`. Below is a summary of the supported commands on the current page.
Feature | Command | Description
--- | --- | ---
Code Review | `/gemini review` | Performs a code review for the current pull request in its current state.
Pull Request Summary | `/gemini summary` | Provides a summary of the current pull request in its current state.
Comment | @gemini-code-assist | Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help | `/gemini help` | Displays a list of available commands.
<b>Customization</b>
To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a `.gemini/` folder in the base of the repository. Detailed instructions can be found [here](https://developers.google.com/gemini-code-assist/docs/customize-gemini-behavior-github).
<b>Limitations & Feedback</b>
Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with :thumbsup: and :thumbsdown: on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up [here](https://google.qualtrics.com/jfe/form/SV_2cyuGuTWsEw84yG).
</details>
[^1]: Review the [Privacy Notices](https://policies.google.com/privacy), [Generative AI Prohibited Use Policy](https://policies.google.com/terms/generative-ai/use-policy), [Terms of Service](https://policies.google.com/terms), and learn how to configure Gemini Code Assist in GitHub [here](https://developers.google.com/gemini-code-assist/docs/customize-gemini-behavior-github). Gemini can make mistakes, so double check it and [use code with caution](https://support.google.com/legal/answer/13505487).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] only expand update graph if needed [beam]
Posted by "reuvenlax (via GitHub)" <gi...@apache.org>.
reuvenlax commented on PR #38332:
URL: https://github.com/apache/beam/pull/38332#issuecomment-4348463642
Adding @ahmedabu98 as a reviewer, since @Abacn will be out of office.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] only expand update graph if needed [beam]
Posted by "reuvenlax (via GitHub)" <gi...@apache.org>.
reuvenlax merged PR #38332:
URL: https://github.com/apache/beam/pull/38332
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org