Pass original message down through conversion for storage write api

[johnjcasey]

Enable users to specify an alternate way to generate the table row for the error output for BQIO's storage write api.

The user passes in a function of ElementT -> TableRow, and we maintain an index of the original elements passed in to BQIO. If the function exists, we use it to generate the error row, instead of the default behavior of emitting the failure directly.

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• [ ] Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
• [ ] Update CHANGES.md with noteworthy changes.
• [ ] If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

[github-actions[bot]]

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

[github-actions[bot]]

Assigning reviewers. If you would like to opt out of this review, comment assign to next reviewer:

R: @robertwb for label java. R: @Abacn for label io.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[johnjcasey]

@Abacn @ahmedabu98 could you take a look at this?

[Abacn]

Also going to run some load test to see if it has performance implications

update:

"AvgInputThroughputElementsPerSec": 51674.9203125,

identical to 2.55.0 (51205), 2.56.0 (47579)

[reuvenlax]

If I understand this correctly, we are now propagating both ElementT and StorageApiWritePayload - correct? Doesn't this double the amount of data being processed?

[github-actions[bot]]

Reminder, please take a look at this pr: @robertwb @Abacn

[github-actions[bot]]

Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment assign to next reviewer:

R: @damondouglas for label java. R: @chamikaramj for label io.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

[robertwb]

Looks like this was approved but has conflicts that need to be resolved.

[chamikaramj]

There was an unresolved discussion about maintaining update compatibility without duplicating a lot of code: https://github.com/apache/beam/pull/31106#discussion_r1587792477

[reuvenlax]

I'm also wanting to know whether there was something motivating this change

• i.e. is their a Beam user that currently needs this? In addition to being careful about perf, this PR adds quite a bit of complexity to code that is already fairly complex.

On Mon, Jun 10, 2024 at 9:51 AM Robert Bradshaw @.***> wrote:

@.**** commented on this pull request.

In sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiLoads.java https://github.com/apache/beam/pull/31106#discussion_r1633552968:

@@ -52,16 +52,18 @@ /** This @.*** PTransform} manages loads into BigQuery using the Storage API. */ public class StorageApiLoads<DestinationT, ElementT> extends PTransform<PCollection<KV<DestinationT, ElementT>>, WriteResult> {

• final TupleTag<KV<DestinationT, StorageApiWritePayload>> successfulConvertedRowsTag =

•  new TupleTag<>("successfulRows");
  

• final TupleTag<KV<DestinationT, KV<ElementT, StorageApiWritePayload>>>

As a side comment, this is another motivation to use schema coders more ubiquitously--adding another field is update compatible.

On another note, anything that involves shuffling more data in the main data path should be looked at carefully from a perf standpoint. We've gone to a lot of effort (e.g. with dynamic destinations) to ensure shuffling metadata doesn't become a perf impediment.

— Reply to this email directly, view it on GitHub https://github.com/apache/beam/pull/31106#discussion_r1633552968, or unsubscribe https://github.com/notifications/unsubscribe-auth/AFAYJVNQ6T5NBLVCJN5CWDDZGXKQ5AVCNFSM6AAAAABGZJZPSWVHI2DSMVQWIX3LMV43YUDVNRWFEZLROVSXG5CSMV3GSZLXHMZDCMBYGMYDAOBRGI . You are receiving this because you were mentioned.Message ID: <apache/beam/pull/31106/review/2108300812 @.***>

[github-actions[bot]]

Reminder, please take a look at this pr: @damondouglas @chamikaramj

[github-actions[bot]]

Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment assign to next reviewer:

R: @kennknowles for label java. R: @ahmedabu98 for label io.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

[github-actions[bot]]

Reminder, please take a look at this pr: @kennknowles @ahmedabu98

[github-actions[bot]]

Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment assign to next reviewer:

R: @damondouglas for label java. R: @Abacn for label io.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

[github-actions[bot]]

Reminder, please take a look at this pr: @damondouglas @Abacn

[github-actions[bot]]

Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment assign to next reviewer:

R: @kennknowles for label java. R: @ahmedabu98 for label io.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)