You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2021/06/14 22:41:45 UTC
[GitHub] [beam] pabloem opened a new pull request #15013: BigQueryIO returns successful rows for streaming inserts
pabloem opened a new pull request #15013:
URL: https://github.com/apache/beam/pull/15013
**Please** add a meaningful description for your change here
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and mention them in a comment (`R: @username`).
- [ ] Format the pull request title like `[BEAM-XXX] Fixes bug in ApproximateQuantiles`, where you replace `BEAM-XXX` with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://beam.apache.org/contribute/#make-reviewers-job-easier).
`ValidatesRunner` compliance status (on master branch)
--------------------------------------------------------
<table>
<thead>
<tr>
<th>Lang</th>
<th>ULR</th>
<th>Dataflow</th>
<th>Flink</th>
<th>Samza</th>
<th>Spark</th>
<th>Twister2</th>
</tr>
</thead>
<tbody>
<tr>
<td>Go</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/badge/icon">
</a>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
</tr>
<tr>
<td>Java</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_ULR/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_ULR/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/badge/icon?subject=V1">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Streaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Streaming/lastCompletedBuild/badge/icon?subject=V1+Streaming">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/badge/icon?subject=V1+Java+11">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2/lastCompletedBuild/badge/icon?subject=V2">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2_Streaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2_Streaming/lastCompletedBuild/badge/icon?subject=V2+Streaming">
</a><br>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/badge/icon?subject=Java+8">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/badge/icon?subject=Java+11">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/badge/icon?subject=Portable">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/badge/icon?subject=Portable+Streaming">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Samza/lastCompletedBuild/badge/icon?subject=Portable">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/badge/icon?subject=Portable">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/badge/icon?subject=Structured+Streaming">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Twister2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Twister2/lastCompletedBuild/badge/icon">
</a>
</td>
</tr>
<tr>
<td>Python</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/badge/icon?subject=V1">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/badge/icon?subject=V2">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/badge/icon?subject=ValCont">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Python_PVR_Flink_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Python_PVR_Flink_Cron/lastCompletedBuild/badge/icon?subject=Portable">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Flink/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
</tr>
<tr>
<td>XLang</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Direct/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Direct/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Dataflow/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Dataflow/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
</tr>
</tbody>
</table>
Examples testing status on various runners
--------------------------------------------------------
<table>
<thead>
<tr>
<th>Lang</th>
<th>ULR</th>
<th>Dataflow</th>
<th>Flink</th>
<th>Samza</th>
<th>Spark</th>
<th>Twister2</th>
</tr>
</thead>
<tbody>
<tr>
<td>Go</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
<tr>
<td>Java</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Cron/lastCompletedBuild/badge/icon?subject=V1">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Java11_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Java11_Cron/lastCompletedBuild/badge/icon?subject=V1+Java11">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_Examples_Dataflow_V2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_Examples_Dataflow_V2/lastCompletedBuild/badge/icon?subject=V2">
</a><br>
</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
<tr>
<td>Python</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
<tr>
<td>XLang</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
</tbody>
</table>
Post-Commit SDK/Transform Integration Tests Status (on master branch)
------------------------------------------------------------------------------------------------
<table>
<thead>
<tr>
<th>Go</th>
<th>Java</th>
<th>Python</th>
</tr>
</thead>
<tbody>
<tr>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/badge/icon?subject=3.6">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/badge/icon?subject=3.7">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python38/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python38/lastCompletedBuild/badge/icon?subject=3.8">
</a>
</td>
</tr>
</tbody>
</table>
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
<table>
<thead>
<tr>
<th>---</th>
<th>Java</th>
<th>Python</th>
<th>Go</th>
<th>Website</th>
<th>Whitespace</th>
<th>Typescript</th>
</tr>
</thead>
<tbody>
<tr>
<td>Non-portable</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/badge/icon">
</a><br>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/badge/icon?subject=Tests">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/badge/icon?subject=Lint">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/badge/icon?subject=Docker">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/badge/icon?subject=Docs">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Whitespace_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Whitespace_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Typescript_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Typescript_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
</tr>
<tr>
<td>Portable</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_GoPortable_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_GoPortable_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
</tbody>
</table>
See [.test-infra/jenkins/README](https://github.com/apache/beam/blob/master/.test-infra/jenkins/README.md) for trigger phrase, status and link of all Jenkins jobs.
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15013: [BEAM-12497] BigQueryIO returns successful rows for streaming inserts
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15013:
URL: https://github.com/apache/beam/pull/15013#issuecomment-867983955
@ihji
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15013: BigQueryIO returns successful rows for streaming inserts
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15013:
URL: https://github.com/apache/beam/pull/15013#issuecomment-861051146
Run Java PostCommit
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15013: [BEAM-12497] BigQueryIO returns successful rows for streaming inserts
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15013:
URL: https://github.com/apache/beam/pull/15013#issuecomment-867221825
PTALLLL
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15013: BigQueryIO returns successful rows for streaming inserts
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15013:
URL: https://github.com/apache/beam/pull/15013#issuecomment-861646937
Run Java PostCommit
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15013: [BEAM-12497] BigQueryIO returns successful rows for streaming inserts
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15013:
URL: https://github.com/apache/beam/pull/15013#issuecomment-873485258
PTAL : ) I'll be on vacation, but I'm still happy to merge this if it looks good : )
--
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
[GitHub] [beam] aaltay commented on pull request #15013: [BEAM-12497] BigQueryIO returns successful rows for streaming inserts
Posted by GitBox <gi...@apache.org>.
aaltay commented on pull request #15013:
URL: https://github.com/apache/beam/pull/15013#issuecomment-881152870
@ihji - Could you review the latest changes?
--
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
[GitHub] [beam] ihji commented on a change in pull request #15013: [BEAM-12497] BigQueryIO returns successful rows for streaming inserts
Posted by GitBox <gi...@apache.org>.
ihji commented on a change in pull request #15013:
URL: https://github.com/apache/beam/pull/15013#discussion_r659015077
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchedStreamingWrite.java
##########
@@ -306,10 +311,13 @@ public void finishBundle(FinishBundleContext context) throws Exception {
ShardedKey.Coder.of(StringUtf8Coder.of()), IterableCoder.of(valueCoder)))
.apply(
ParDo.of(new InsertBatchedElements())
- .withOutputTags(mainOutputTag, TupleTagList.of(failedOutputTag)));
+ .withOutputTags(
+ mainOutputTag,
+ TupleTagList.of(failedOutputTag).and(SUCCESSFUL_ROWS_TAG)));
PCollection<ErrorT> failedInserts = result.get(failedOutputTag);
failedInserts.setCoder(failedOutputCoder);
Review comment:
ditto.
##########
File path: sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImplTest.java.bak
##########
@@ -0,0 +1,1333 @@
+/*
Review comment:
please remove a backup file.
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchedStreamingWrite.java
##########
@@ -191,23 +192,25 @@ private BatchedStreamingWrite(
}
@Override
- public PCollection<ErrorT> expand(PCollection<KV<String, TableRowInfo<ElementT>>> input) {
+ public PCollectionTuple expand(PCollection<KV<String, TableRowInfo<ElementT>>> input) {
return batchViaStateful
? input.apply(new ViaStateful())
: input.apply(new ViaBundleFinalization());
}
private class ViaBundleFinalization
- extends PTransform<PCollection<KV<String, TableRowInfo<ElementT>>>, PCollection<ErrorT>> {
+ extends PTransform<PCollection<KV<String, TableRowInfo<ElementT>>>, PCollectionTuple> {
@Override
- public PCollection<ErrorT> expand(PCollection<KV<String, TableRowInfo<ElementT>>> input) {
+ public PCollectionTuple expand(PCollection<KV<String, TableRowInfo<ElementT>>> input) {
PCollectionTuple result =
input.apply(
ParDo.of(new BatchAndInsertElements())
- .withOutputTags(mainOutputTag, TupleTagList.of(failedOutputTag)));
+ .withOutputTags(
+ mainOutputTag, TupleTagList.of(failedOutputTag).and(SUCCESSFUL_ROWS_TAG)));
Review comment:
It's interesting that we already produced multiple pcollections but just returned failed inserts.
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StreamingWriteTables.java
##########
@@ -334,44 +347,47 @@ public WriteResult expand(PCollection<KV<TableDestination, ElementT>> input) {
ShardedKeyCoder.of(StringUtf8Coder.of()),
TableRowInfoCoder.of(elementCoder)));
- return shardedTagged
- .apply(Reshuffle.of())
- // Put in the global window to ensure that DynamicDestinations side inputs are accessed
- // correctly.
- .apply(
- "GlobalWindow",
- Window.<KV<ShardedKey<String>, TableRowInfo<ElementT>>>into(new GlobalWindows())
- .triggering(DefaultTrigger.of())
- .discardingFiredPanes())
- .apply(
- "StripShardId",
- MapElements.via(
- new SimpleFunction<
- KV<ShardedKey<String>, TableRowInfo<ElementT>>,
- KV<String, TableRowInfo<ElementT>>>() {
- @Override
- public KV<String, TableRowInfo<ElementT>> apply(
- KV<ShardedKey<String>, TableRowInfo<ElementT>> input) {
- return KV.of(input.getKey().getKey(), input.getValue());
- }
- }))
- .setCoder(KvCoder.of(StringUtf8Coder.of(), TableRowInfoCoder.of(elementCoder)))
- // Also batch the TableRows in a best effort manner via bundle finalization before
- // inserting to BigQuery.
- .apply(
- "StreamingWrite",
- new BatchedStreamingWrite<>(
- bigQueryServices,
- retryPolicy,
- failedInsertsTag,
- coder,
- errorContainer,
- skipInvalidRows,
- ignoreUnknownValues,
- ignoreInsertIds,
- toTableRow,
- toFailsafeTableRow)
- .viaDoFnFinalization());
+ PCollectionTuple result =
Review comment:
ditto.
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StreamingWriteTables.java
##########
@@ -305,20 +316,22 @@ public WriteResult expand(PCollection<KV<TableDestination, ElementT>> input) {
// Auto-sharding is achieved via GroupIntoBatches.WithShardedKey transform which groups and at
// the same time batches the TableRows to be inserted to BigQuery.
- return unshardedTagged.apply(
- "StreamingWrite",
- new BatchedStreamingWrite<>(
- bigQueryServices,
- retryPolicy,
- failedInsertsTag,
- coder,
- errorContainer,
- skipInvalidRows,
- ignoreUnknownValues,
- ignoreInsertIds,
- toTableRow,
- toFailsafeTableRow)
- .viaStateful());
+ PCollectionTuple result =
Review comment:
looks like there's no change except introducing a temporary variable `result`.
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImpl.java
##########
@@ -828,9 +855,41 @@ public void deleteDataset(String projectId, String datasetId)
List<ValueInSingleWindow<T>> failedInserts,
ErrorContainer<T> errorContainer,
boolean skipInvalidRows,
- boolean ignoreUnkownValues,
+ boolean ignoreUnknownValues,
boolean ignoreInsertIds)
throws IOException, InterruptedException {
+ return insertAll(
Review comment:
Looks like now we have four variants of `insertAll`:
```
a) public insertAll (with successfulRows)
b) public insertAll (without successfulRows)
c) protected insertAll (with successfulRows)
d) protected insertAll (without successfulRows)
// b calls a, d calls c, a calls c
```
If these methods are mostly for internal use, I would prefer to have just:
```
a) public insertAll (with successfulRows)
b) protected insertAll (with successfulRows)
// a calls b
```
and explicitly put `null` for all existing call-sites of `a` and `b`.
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchedStreamingWrite.java
##########
@@ -191,23 +192,25 @@ private BatchedStreamingWrite(
}
@Override
- public PCollection<ErrorT> expand(PCollection<KV<String, TableRowInfo<ElementT>>> input) {
+ public PCollectionTuple expand(PCollection<KV<String, TableRowInfo<ElementT>>> input) {
return batchViaStateful
? input.apply(new ViaStateful())
: input.apply(new ViaBundleFinalization());
}
private class ViaBundleFinalization
- extends PTransform<PCollection<KV<String, TableRowInfo<ElementT>>>, PCollection<ErrorT>> {
+ extends PTransform<PCollection<KV<String, TableRowInfo<ElementT>>>, PCollectionTuple> {
@Override
- public PCollection<ErrorT> expand(PCollection<KV<String, TableRowInfo<ElementT>>> input) {
+ public PCollectionTuple expand(PCollection<KV<String, TableRowInfo<ElementT>>> input) {
PCollectionTuple result =
input.apply(
ParDo.of(new BatchAndInsertElements())
- .withOutputTags(mainOutputTag, TupleTagList.of(failedOutputTag)));
+ .withOutputTags(
+ mainOutputTag, TupleTagList.of(failedOutputTag).and(SUCCESSFUL_ROWS_TAG)));
PCollection<ErrorT> failedInserts = result.get(failedOutputTag);
failedInserts.setCoder(failedOutputCoder);
Review comment:
Looks like this can also be a one liner like just below.
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImpl.java
##########
@@ -844,6 +903,7 @@ public void deleteDataset(String projectId, String datasetId)
+ "as many elements as rowList");
}
+ final Set<Integer> failedIndices = new HashSet<>();
Review comment:
How big `failedIndices` set could be? Any implication on performance impact? Probably okay since `insertAll` is only for streaming insert.
--
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
[GitHub] [beam] aaltay commented on pull request #15013: [BEAM-12497] BigQueryIO returns successful rows for streaming inserts
Posted by GitBox <gi...@apache.org>.
aaltay commented on pull request #15013:
URL: https://github.com/apache/beam/pull/15013#issuecomment-915474731
@dgorkovets - This commit is in the 2.32.0 branch (https://github.com/apache/beam/commit/7edca9e4e72d9d31fff6e0e1dcfb712bb35980e8). We can probably close the jira?
--
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
[GitHub] [beam] pabloem commented on pull request #15013: BigQueryIO returns successful rows for streaming inserts
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15013:
URL: https://github.com/apache/beam/pull/15013#issuecomment-861739573
Run Java PostCommit
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on a change in pull request #15013: [BEAM-12497] BigQueryIO returns successful rows for streaming inserts
Posted by GitBox <gi...@apache.org>.
pabloem commented on a change in pull request #15013:
URL: https://github.com/apache/beam/pull/15013#discussion_r660976142
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImpl.java
##########
@@ -828,9 +855,41 @@ public void deleteDataset(String projectId, String datasetId)
List<ValueInSingleWindow<T>> failedInserts,
ErrorContainer<T> errorContainer,
boolean skipInvalidRows,
- boolean ignoreUnkownValues,
+ boolean ignoreUnknownValues,
boolean ignoreInsertIds)
throws IOException, InterruptedException {
+ return insertAll(
Review comment:
Done
##########
File path: sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImplTest.java.bak
##########
@@ -0,0 +1,1333 @@
+/*
Review comment:
Done.
--
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
[GitHub] [beam] pabloem commented on pull request #15013: BigQueryIO returns successful rows for streaming inserts
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15013:
URL: https://github.com/apache/beam/pull/15013#issuecomment-861685668
Run Python PostCommit
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on a change in pull request #15013: [BEAM-12497] BigQueryIO returns successful rows for streaming inserts
Posted by GitBox <gi...@apache.org>.
pabloem commented on a change in pull request #15013:
URL: https://github.com/apache/beam/pull/15013#discussion_r660922229
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchedStreamingWrite.java
##########
@@ -191,23 +192,25 @@ private BatchedStreamingWrite(
}
@Override
- public PCollection<ErrorT> expand(PCollection<KV<String, TableRowInfo<ElementT>>> input) {
+ public PCollectionTuple expand(PCollection<KV<String, TableRowInfo<ElementT>>> input) {
return batchViaStateful
? input.apply(new ViaStateful())
: input.apply(new ViaBundleFinalization());
}
private class ViaBundleFinalization
- extends PTransform<PCollection<KV<String, TableRowInfo<ElementT>>>, PCollection<ErrorT>> {
+ extends PTransform<PCollection<KV<String, TableRowInfo<ElementT>>>, PCollectionTuple> {
@Override
- public PCollection<ErrorT> expand(PCollection<KV<String, TableRowInfo<ElementT>>> input) {
+ public PCollectionTuple expand(PCollection<KV<String, TableRowInfo<ElementT>>> input) {
PCollectionTuple result =
input.apply(
ParDo.of(new BatchAndInsertElements())
- .withOutputTags(mainOutputTag, TupleTagList.of(failedOutputTag)));
+ .withOutputTags(
+ mainOutputTag, TupleTagList.of(failedOutputTag).and(SUCCESSFUL_ROWS_TAG)));
PCollection<ErrorT> failedInserts = result.get(failedOutputTag);
failedInserts.setCoder(failedOutputCoder);
Review comment:
Done
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchedStreamingWrite.java
##########
@@ -191,23 +192,25 @@ private BatchedStreamingWrite(
}
@Override
- public PCollection<ErrorT> expand(PCollection<KV<String, TableRowInfo<ElementT>>> input) {
+ public PCollectionTuple expand(PCollection<KV<String, TableRowInfo<ElementT>>> input) {
return batchViaStateful
? input.apply(new ViaStateful())
: input.apply(new ViaBundleFinalization());
}
private class ViaBundleFinalization
- extends PTransform<PCollection<KV<String, TableRowInfo<ElementT>>>, PCollection<ErrorT>> {
+ extends PTransform<PCollection<KV<String, TableRowInfo<ElementT>>>, PCollectionTuple> {
@Override
- public PCollection<ErrorT> expand(PCollection<KV<String, TableRowInfo<ElementT>>> input) {
+ public PCollectionTuple expand(PCollection<KV<String, TableRowInfo<ElementT>>> input) {
PCollectionTuple result =
input.apply(
ParDo.of(new BatchAndInsertElements())
- .withOutputTags(mainOutputTag, TupleTagList.of(failedOutputTag)));
+ .withOutputTags(
+ mainOutputTag, TupleTagList.of(failedOutputTag).and(SUCCESSFUL_ROWS_TAG)));
Review comment:
Yeah, it's curious.
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchedStreamingWrite.java
##########
@@ -306,10 +311,13 @@ public void finishBundle(FinishBundleContext context) throws Exception {
ShardedKey.Coder.of(StringUtf8Coder.of()), IterableCoder.of(valueCoder)))
.apply(
ParDo.of(new InsertBatchedElements())
- .withOutputTags(mainOutputTag, TupleTagList.of(failedOutputTag)));
+ .withOutputTags(
+ mainOutputTag,
+ TupleTagList.of(failedOutputTag).and(SUCCESSFUL_ROWS_TAG)));
PCollection<ErrorT> failedInserts = result.get(failedOutputTag);
failedInserts.setCoder(failedOutputCoder);
Review comment:
Done.
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StreamingWriteTables.java
##########
@@ -305,20 +316,22 @@ public WriteResult expand(PCollection<KV<TableDestination, ElementT>> input) {
// Auto-sharding is achieved via GroupIntoBatches.WithShardedKey transform which groups and at
// the same time batches the TableRows to be inserted to BigQuery.
- return unshardedTagged.apply(
- "StreamingWrite",
- new BatchedStreamingWrite<>(
- bigQueryServices,
- retryPolicy,
- failedInsertsTag,
- coder,
- errorContainer,
- skipInvalidRows,
- ignoreUnknownValues,
- ignoreInsertIds,
- toTableRow,
- toFailsafeTableRow)
- .viaStateful());
+ PCollectionTuple result =
Review comment:
Removed. Thanks!
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StreamingWriteTables.java
##########
@@ -334,44 +347,47 @@ public WriteResult expand(PCollection<KV<TableDestination, ElementT>> input) {
ShardedKeyCoder.of(StringUtf8Coder.of()),
TableRowInfoCoder.of(elementCoder)));
- return shardedTagged
- .apply(Reshuffle.of())
- // Put in the global window to ensure that DynamicDestinations side inputs are accessed
- // correctly.
- .apply(
- "GlobalWindow",
- Window.<KV<ShardedKey<String>, TableRowInfo<ElementT>>>into(new GlobalWindows())
- .triggering(DefaultTrigger.of())
- .discardingFiredPanes())
- .apply(
- "StripShardId",
- MapElements.via(
- new SimpleFunction<
- KV<ShardedKey<String>, TableRowInfo<ElementT>>,
- KV<String, TableRowInfo<ElementT>>>() {
- @Override
- public KV<String, TableRowInfo<ElementT>> apply(
- KV<ShardedKey<String>, TableRowInfo<ElementT>> input) {
- return KV.of(input.getKey().getKey(), input.getValue());
- }
- }))
- .setCoder(KvCoder.of(StringUtf8Coder.of(), TableRowInfoCoder.of(elementCoder)))
- // Also batch the TableRows in a best effort manner via bundle finalization before
- // inserting to BigQuery.
- .apply(
- "StreamingWrite",
- new BatchedStreamingWrite<>(
- bigQueryServices,
- retryPolicy,
- failedInsertsTag,
- coder,
- errorContainer,
- skipInvalidRows,
- ignoreUnknownValues,
- ignoreInsertIds,
- toTableRow,
- toFailsafeTableRow)
- .viaDoFnFinalization());
+ PCollectionTuple result =
Review comment:
undone. Thanks!
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImpl.java
##########
@@ -844,6 +903,7 @@ public void deleteDataset(String projectId, String datasetId)
+ "as many elements as rowList");
}
+ final Set<Integer> failedIndices = new HashSet<>();
Review comment:
if all elements failed in a single request, we would have 500 failed indices by default (this setting can be tweaked by the user, but it would usually be 500 or less) - so I think it's an acceptable size. WDYT?
--
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
[GitHub] [beam] steveniemitz commented on pull request #15013: [BEAM-12497] BigQueryIO returns successful rows for streaming inserts
Posted by GitBox <gi...@apache.org>.
steveniemitz commented on pull request #15013:
URL: https://github.com/apache/beam/pull/15013#issuecomment-875230351
oh this is cool, I actually have something similar in [our fork](https://github.com/twitter-forks/beam/commit/566dc38e2bc22951ae59e97c15e1b1445db8edba) for batch inserts, I wonder how hard this would be to unify.
--
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
[GitHub] [beam] pabloem commented on pull request #15013: [BEAM-12497] BigQueryIO returns successful rows for streaming inserts
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15013:
URL: https://github.com/apache/beam/pull/15013#issuecomment-861901948
r: @dhercher
r: @ihji
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji merged pull request #15013: [BEAM-12497] BigQueryIO returns successful rows for streaming inserts
Posted by GitBox <gi...@apache.org>.
ihji merged pull request #15013:
URL: https://github.com/apache/beam/pull/15013
--
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
[GitHub] [beam] pabloem commented on pull request #15013: [BEAM-12497] BigQueryIO returns successful rows for streaming inserts
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15013:
URL: https://github.com/apache/beam/pull/15013#issuecomment-865293796
@ihji PTAL
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] dgorkovets commented on pull request #15013: [BEAM-12497] BigQueryIO returns successful rows for streaming inserts
Posted by GitBox <gi...@apache.org>.
dgorkovets commented on pull request #15013:
URL: https://github.com/apache/beam/pull/15013#issuecomment-913035221
@pabloem , BEAM-12497 is still open but feature is merged. It is not included into 2.32.0 release notes. Does this feature officially supported?
--
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
[GitHub] [beam] pabloem commented on pull request #15013: [BEAM-12497] BigQueryIO returns successful rows for streaming inserts
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15013:
URL: https://github.com/apache/beam/pull/15013#issuecomment-865293796
@ihji PTAL
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ihji commented on a change in pull request #15013: [BEAM-12497] BigQueryIO returns successful rows for streaming inserts
Posted by GitBox <gi...@apache.org>.
ihji commented on a change in pull request #15013:
URL: https://github.com/apache/beam/pull/15013#discussion_r670972421
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImpl.java
##########
@@ -844,6 +903,7 @@ public void deleteDataset(String projectId, String datasetId)
+ "as many elements as rowList");
}
+ final Set<Integer> failedIndices = new HashSet<>();
Review comment:
500 elements should be fine. Thanks for the explanation.
--
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