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/11/07 15:34:02 UTC
[GitHub] [beam] brachi-wernick opened a new pull request #15915: [beam-12737] add API to handle failed rows in the collection due to sql query runtime error
brachi-wernick opened a new pull request #15915:
URL: https://github.com/apache/beam/pull/15915
### The problem
When calling to
```
collection.apply(SqlTransform.query(query))
```
There are cases that a query can have some errors for a small amount of events in the collection, for example UDF can produce runtime error when it gets some invalid input data.
Current behaviour is that query is failed and records are retried forever, and huge pipeline can have backlog just because 1-2 invalid events.
We need an option to send failed rows to some DLQ and continue processing the rest of the data.
### The solution:
Added an option to send transformer that get the failed rows with their error and can handler them, sending to PubSub DLQ, failed error table, logs etc..
```
input.apply(SqlTransform.query(query)
.withErrorsTransformer(new BeamSqlErrorTransformer())
.registerUdf("func1", Func1.class)
.registerUdf("func2", Func2.class)
```
`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>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Samza/lastCompletedBuild/badge/icon">
</a>
</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>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Samza/lastCompletedBuild/badge/icon">
</a>
</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 fStatus" 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>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Samza/lastCompletedBuild/badge/icon">
</a>
</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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] apilloud commented on pull request #15915: [beam-12737] add API to handle failed rows in the collection due to sql query runtime error
Posted by GitBox <gi...@apache.org>.
apilloud commented on pull request #15915:
URL: https://github.com/apache/beam/pull/15915#issuecomment-969675045
Oops, I intended to squash commits. Looks like I may have hit the wrong button.
--
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] apilloud commented on a change in pull request #15915: [beam-12737] add API to handle failed rows in the collection due to sql query runtime error
Posted by GitBox <gi...@apache.org>.
apilloud commented on a change in pull request #15915:
URL: https://github.com/apache/beam/pull/15915#discussion_r745129204
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRel.java
##########
@@ -256,18 +283,25 @@ public void setup() {
}
@ProcessElement
- public void processElement(@FieldAccess("row") Row row, OutputReceiver<Row> r) {
+ public void processElement(
+ @FieldAccess("row") Row row, OutputReceiver<Row> r, ProcessContext c) {
Review comment:
Haven't done a full review yet. Using `ProcessContext` creates an implicit access to the raw input row, which breaks our ability to pushdown projects outside of SQL. (Tests are going in shortly with #15886.) You'll need to use `MultiOutputReceiver` instead.
--
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] apilloud commented on a change in pull request #15915: [beam-12737] add API to handle failed rows in the collection due to sql query runtime error
Posted by GitBox <gi...@apache.org>.
apilloud commented on a change in pull request #15915:
URL: https://github.com/apache/beam/pull/15915#discussion_r746834918
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRel.java
##########
@@ -112,11 +119,19 @@
private static final long MILLIS_PER_DAY = 86400000L;
private static final ParameterExpression rowParam = Expressions.parameter(Row.class, "row");
+ private PTransform<PCollection<BeamCalcRelError>, POutput> errorsTransformer;
+ private static final TupleTag<Row> rows = new TupleTag<Row>() {};
+ private static final TupleTag<BeamCalcRelError> errors = new TupleTag<BeamCalcRelError>() {};
public BeamCalcRel(RelOptCluster cluster, RelTraitSet traits, RelNode input, RexProgram program) {
super(cluster, traits, input, program);
}
+ @Override
+ public void withErrorsTransformer(PTransform<PCollection<BeamCalcRelError>, POutput> ptransform) {
Review comment:
You could also add a default for the new `buildPTransform` that calls the old one, this would allow you to avoid storing state and also not make a breaking change. There are other places the pipeline can crash, we'll eventually want every RelNode to catch and return errors. I'm fine with leaving it as is, this is an internal API so easy for someone to change again if it becomes a problem.
--
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] brachi-wernick commented on a change in pull request #15915: [beam-12737] add API to handle failed rows in the collection due to sql query runtime error
Posted by GitBox <gi...@apache.org>.
brachi-wernick commented on a change in pull request #15915:
URL: https://github.com/apache/beam/pull/15915#discussion_r748900363
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRel.java
##########
@@ -192,9 +207,17 @@ public Calc copy(RelTraitSet traitSet, RelNode input, RexProgram program) {
outputSchema,
options.getVerifyRowValues(),
getJarPaths(program),
- inputGetter.getFieldAccess());
-
- return upstream.apply(ParDo.of(calcFn)).setRowSchema(outputSchema);
+ inputGetter.getFieldAccess(),
+ errorsTransformer != null);
+
+ PCollectionTuple tuple =
+ upstream.apply(ParDo.of(calcFn).withOutputTags(rows, TupleTagList.of(errors)));
+ PCollection<BeamCalcRelError> errorPCollection =
+ tuple.get(errors).setCoder(BeamCalcRelErrorCoder.of(RowCoder.of(upstream.getSchema())));
Review comment:
I added a complex test, with 2 udaf, each one produced a different error on a different schema (one before group by and one after group by)
I assert in the tests that I get 2 error rows each one with its matching schema.
from logs: (you can see that `CastUdf` and `CalculatePrice` exist in 2 diff steps.
```
INFO: SQL:
SELECT `PCOLLECTION`.`id`, `PCOLLECTION`.`country_code`, `CalculatePrice`(SUM(`CastUdf`(`PCOLLECTION`.`amount`)), `PCOLLECTION`.`currency`) AS `sum_amount`
FROM `beam`.`PCOLLECTION` AS `PCOLLECTION`
GROUP BY `PCOLLECTION`.`id`, `PCOLLECTION`.`country_code`, `PCOLLECTION`.`currency`
Nov 14, 2021 9:37:51 PM org.apache.beam.sdk.extensions.sql.impl.CalciteQueryPlanner convertToBeamRel
INFO: SQLPlan>
LogicalProject(id=[$0], country_code=[$1], sum_amount=[CalculatePrice($3, $2)])
LogicalAggregate(group=[{0, 1, 2}], agg#0=[SUM($3)])
LogicalProject(id=[$0], country_code=[$2], currency=[$3], $f3=[CastUdf($1)])
BeamIOSourceRel(table=[[beam, PCOLLECTION]])
```
--
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] apilloud commented on a change in pull request #15915: [beam-12737] add API to handle failed rows in the collection due to sql query runtime error
Posted by GitBox <gi...@apache.org>.
apilloud commented on a change in pull request #15915:
URL: https://github.com/apache/beam/pull/15915#discussion_r746196223
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRel.java
##########
@@ -192,9 +207,17 @@ public Calc copy(RelTraitSet traitSet, RelNode input, RexProgram program) {
outputSchema,
options.getVerifyRowValues(),
getJarPaths(program),
- inputGetter.getFieldAccess());
-
- return upstream.apply(ParDo.of(calcFn)).setRowSchema(outputSchema);
+ inputGetter.getFieldAccess(),
+ errorsTransformer != null);
+
+ PCollectionTuple tuple =
+ upstream.apply(ParDo.of(calcFn).withOutputTags(rows, TupleTagList.of(errors)));
+ PCollection<BeamCalcRelError> errorPCollection =
+ tuple.get(errors).setCoder(BeamCalcRelErrorCoder.of(RowCoder.of(upstream.getSchema())));
Review comment:
This won't work, there is no guarantee that your error row matches the input row.
If your input row is `1 as key, "a" as value`. If your SqlTransform is `SELECT CustomUdafWithError(key)`, you'd end up with a row of `1 as key` in the error. On the overall SQL transform you can end up with multiple different coders, for example if you have a join.
I think you'll actually need a coder that encodes the coder (or at least the schema), it's going to get complex quick. We might just want to return the string (or possibly JSON) representation of the error row instead? Can you add tests where you have errors from the same SqlTransform with different rows schemas? We are trying to move to Schema coder for everything, but I think this use case (variable row structure) isn't something we support.
cc: @reuvenlax
--
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] apilloud commented on a change in pull request #15915: [beam-12737] add API to handle failed rows in the collection due to sql query runtime error
Posted by GitBox <gi...@apache.org>.
apilloud commented on a change in pull request #15915:
URL: https://github.com/apache/beam/pull/15915#discussion_r746207788
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRel.java
##########
@@ -256,18 +283,24 @@ public void setup() {
}
@ProcessElement
- public void processElement(@FieldAccess("row") Row row, OutputReceiver<Row> r) {
+ public void processElement(@FieldAccess("row") Row row, MultiOutputReceiver r) {
assert se != null;
- final Object[] v;
try {
- v = (Object[]) se.evaluate(new Object[] {row, CONTEXT_INSTANCE});
+ Object[] v = (Object[]) se.evaluate(new Object[] {row, CONTEXT_INSTANCE});
+ if (v != null) {
+ final Row output = toBeamRow(Arrays.asList(v), outputSchema, verifyRowValues);
+ r.get(rows).output(output);
Review comment:
nit: This is a somewhat performance sensitive path, the extra `get` might slow things down. Is it possible to use `OutputReciever<Row>` for this case still?
--
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] brachi-wernick commented on a change in pull request #15915: [beam-12737] add API to handle failed rows in the collection due to sql query runtime error
Posted by GitBox <gi...@apache.org>.
brachi-wernick commented on a change in pull request #15915:
URL: https://github.com/apache/beam/pull/15915#discussion_r746887637
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRel.java
##########
@@ -112,11 +119,19 @@
private static final long MILLIS_PER_DAY = 86400000L;
private static final ParameterExpression rowParam = Expressions.parameter(Row.class, "row");
+ private PTransform<PCollection<BeamCalcRelError>, POutput> errorsTransformer;
+ private static final TupleTag<Row> rows = new TupleTag<Row>() {};
+ private static final TupleTag<BeamCalcRelError> errors = new TupleTag<BeamCalcRelError>() {};
public BeamCalcRel(RelOptCluster cluster, RelTraitSet traits, RelNode input, RexProgram program) {
super(cluster, traits, input, program);
}
+ @Override
+ public void withErrorsTransformer(PTransform<PCollection<BeamCalcRelError>, POutput> ptransform) {
Review comment:
good idea, this is easy fix 6bf54e2
--
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] apilloud merged pull request #15915: [beam-12737] add API to handle failed rows in the collection due to sql query runtime error
Posted by GitBox <gi...@apache.org>.
apilloud merged pull request #15915:
URL: https://github.com/apache/beam/pull/15915
--
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] TheNeuralBit commented on pull request #15915: [beam-12737] add API to handle failed rows in the collection due to sql query runtime error
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on pull request #15915:
URL: https://github.com/apache/beam/pull/15915#issuecomment-965578619
I haven't looked at the actual implementation but I do have a high-level question about the API:
```
input.apply(SqlTransform.query(query)
.withErrorsTransformer(new BeamSqlErrorTransformer())
.registerUdf("func1", Func1.class)
.registerUdf("func2", Func2.class)
```
This API looks odd to me. I think it could be more familiar to Beam users if there were an option to make SqlTransform return a tagged PCollection, with one of the PCollections representing error information for failed records.
--
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] brachi-wernick commented on a change in pull request #15915: [beam-12737] add API to handle failed rows in the collection due to sql query runtime error
Posted by GitBox <gi...@apache.org>.
brachi-wernick commented on a change in pull request #15915:
URL: https://github.com/apache/beam/pull/15915#discussion_r748900559
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRelError.java
##########
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.extensions.sql.impl.rel;
+
+import com.google.auto.value.AutoValue;
+import org.apache.beam.sdk.values.Row;
+
+@AutoValue
+public abstract class BeamCalcRelError {
+
+ public static BeamCalcRelError create(Row row, String error) {
Review comment:
applied this change in https://github.com/apache/beam/pull/15915/commits/df8912dca3a84b372fc27aa49971504d02835dfc
--
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] apilloud commented on a change in pull request #15915: [beam-12737] add API to handle failed rows in the collection due to sql query runtime error
Posted by GitBox <gi...@apache.org>.
apilloud commented on a change in pull request #15915:
URL: https://github.com/apache/beam/pull/15915#discussion_r746208690
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRel.java
##########
@@ -112,11 +119,19 @@
private static final long MILLIS_PER_DAY = 86400000L;
private static final ParameterExpression rowParam = Expressions.parameter(Row.class, "row");
+ private PTransform<PCollection<BeamCalcRelError>, POutput> errorsTransformer;
+ private static final TupleTag<Row> rows = new TupleTag<Row>() {};
+ private static final TupleTag<BeamCalcRelError> errors = new TupleTag<BeamCalcRelError>() {};
public BeamCalcRel(RelOptCluster cluster, RelTraitSet traits, RelNode input, RexProgram program) {
super(cluster, traits, input, program);
}
+ @Override
+ public void withErrorsTransformer(PTransform<PCollection<BeamCalcRelError>, POutput> ptransform) {
Review comment:
Should this be an argument to `buildPTransform` instead? It seems kind of roundabout to be doing it this way.
--
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] brachi-wernick commented on pull request #15915: [beam-12737] add API to handle failed rows in the collection due to sql query runtime error
Posted by GitBox <gi...@apache.org>.
brachi-wernick commented on pull request #15915:
URL: https://github.com/apache/beam/pull/15915#issuecomment-965640978
@TheNeuralBit
>This API looks odd to me. I think it could be more familiar to Beam users if there were an option to make SqlTransform return a tagged PCollection, with one of the PCollections representing error information for failed records.
Agree, that's was my first intention, but since we already return rows from this transform, we will need to break the API and return 2 tuple tags, rows and errors, or we can create another SqlTransform with error handling (which I less prefer because I think it will be hard to maintain 2 sql transforms )
--
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] apilloud commented on a change in pull request #15915: [beam-12737] add API to handle failed rows in the collection due to sql query runtime error
Posted by GitBox <gi...@apache.org>.
apilloud commented on a change in pull request #15915:
URL: https://github.com/apache/beam/pull/15915#discussion_r746191174
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRelError.java
##########
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.extensions.sql.impl.rel;
+
+import java.util.Objects;
+import org.apache.beam.sdk.values.Row;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+public class BeamCalcRelError {
+
+ private Row row;
+ private String error;
+
+ public BeamCalcRelError(Row row, String error) {
+ this.row = row;
+ this.error = error == null ? "empty error msg" : error;
+ }
+
+ public Row getRow() {
+ return row;
+ }
+
+ public void setRow(Row row) {
+ this.row = row;
+ }
+
+ public String getError() {
+ return error;
+ }
+
+ public void setError(String error) {
+ this.error = error;
+ }
+
+ @Override
+ public boolean equals(@Nullable Object o) {
Review comment:
We prefer autovalue to this boilerplate. For an example: https://github.com/apache/beam/blob/243128a8fc52798e1b58b0cf1a271d95ee7aa241/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/planner/NodeStats.java
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamSqlRelUtils.java
##########
@@ -71,15 +77,20 @@
* A {@link BeamRelNode} is a recursive structure, the {@code BeamQueryPlanner} visits it with a
* DFS(Depth-First-Search) algorithm.
*/
- static PCollection<Row> toPCollection(
- Pipeline pipeline, BeamRelNode node, Map<Integer, PCollection<Row>> cache) {
+ public static PCollection<Row> toPCollection(
+ Pipeline pipeline,
+ BeamRelNode node,
+ Map<Integer, PCollection<Row>> cache,
+ @Nullable PTransform<PCollection<BeamCalcRelError>, POutput> errorTransformer) {
Review comment:
nit: this should go before the cache.
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamSqlRelUtils.java
##########
@@ -39,12 +41,15 @@
public class BeamSqlRelUtils {
public static PCollection<Row> toPCollection(Pipeline pipeline, BeamRelNode node) {
- return toPCollection(pipeline, node, new HashMap());
+ return toPCollection(pipeline, node, new HashMap(), null);
}
/** Transforms the inputs into a PInput. */
private static PCollectionList<Row> buildPCollectionList(
- List<RelNode> inputRels, Pipeline pipeline, Map<Integer, PCollection<Row>> cache) {
+ List<RelNode> inputRels,
+ Pipeline pipeline,
+ Map<Integer, PCollection<Row>> cache,
+ @Nullable PTransform<PCollection<BeamCalcRelError>, POutput> errorTransformer) {
Review comment:
nit: you should add this before the `cache` field.
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRel.java
##########
@@ -192,9 +207,17 @@ public Calc copy(RelTraitSet traitSet, RelNode input, RexProgram program) {
outputSchema,
options.getVerifyRowValues(),
getJarPaths(program),
- inputGetter.getFieldAccess());
-
- return upstream.apply(ParDo.of(calcFn)).setRowSchema(outputSchema);
+ inputGetter.getFieldAccess(),
+ errorsTransformer != null);
+
+ PCollectionTuple tuple =
+ upstream.apply(ParDo.of(calcFn).withOutputTags(rows, TupleTagList.of(errors)));
+ PCollection<BeamCalcRelError> errorPCollection =
+ tuple.get(errors).setCoder(BeamCalcRelErrorCoder.of(RowCoder.of(upstream.getSchema())));
Review comment:
This won't work, there is no guarantee that your input row matches the input row.
If your input row is `1 as key, "a" as value`. If your SqlTransform is `SELECT CustomUdafWithError(key)`, you'd end up with a row of `1 as key` in the error. On the overall SQL transform you can end up with multiple different coders, for example if you have a join.
I think you'll actually need a coder that to encodes the coder, it's going to get complex quick. We might just want to return the string (or possibly JSON) representation of the error row instead? Can you add tests where you have errors from the same SqlTransform with different rows schemas? We are trying to move to Schema coder for everything, but I think this use case (variable row structure) isn't something we support.
cc: @reuvenlax
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamSqlRelUtils.java
##########
@@ -39,12 +41,15 @@
public class BeamSqlRelUtils {
public static PCollection<Row> toPCollection(Pipeline pipeline, BeamRelNode node) {
Review comment:
You should add a new `public static PCollection<Row> toPCollection(Pipeline pipeline, BeamRelNode node, @Nullable PTransform<PCollection<BeamCalcRelError>, POutput> errorTransformer)` right after this one.
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/SqlTransform.java
##########
@@ -164,7 +168,10 @@
BeamSqlEnv sqlEnv = sqlEnvBuilder.build();
ddlStrings().forEach(sqlEnv::executeDdl);
return BeamSqlRelUtils.toPCollection(
- input.getPipeline(), sqlEnv.parseQuery(queryString(), queryParameters()));
+ input.getPipeline(),
+ sqlEnv.parseQuery(queryString(), queryParameters()),
+ new HashMap<>(),
Review comment:
This shouldn't be here.
##########
File path: sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlErrorTest.java
##########
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.extensions.sql;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.when;
+
+import java.util.Objects;
+import org.apache.beam.sdk.extensions.sql.impl.rel.BeamCalcRelError;
+import org.apache.beam.sdk.extensions.sql.impl.udf.CustomUdafWithError;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.schemas.SchemaCoder;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.Row;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+
+public class BeamSqlErrorTest extends org.apache.beam.sdk.extensions.sql.BeamSqlDslBase {
Review comment:
nit: drop the `org.apache.beam.sdk.extensions.sql.` it isn't need.
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamSqlRelUtils.java
##########
@@ -71,15 +77,20 @@
* A {@link BeamRelNode} is a recursive structure, the {@code BeamQueryPlanner} visits it with a
* DFS(Depth-First-Search) algorithm.
*/
- static PCollection<Row> toPCollection(
- Pipeline pipeline, BeamRelNode node, Map<Integer, PCollection<Row>> cache) {
+ public static PCollection<Row> toPCollection(
Review comment:
This should remain package private.
--
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] brachi-wernick commented on a change in pull request #15915: [beam-12737] add API to handle failed rows in the collection due to sql query runtime error
Posted by GitBox <gi...@apache.org>.
brachi-wernick commented on a change in pull request #15915:
URL: https://github.com/apache/beam/pull/15915#discussion_r746953427
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/SqlTransform.java
##########
@@ -298,6 +304,11 @@ public SqlTransform registerUdaf(String functionName, Combine.CombineFn combineF
return toBuilder().setUdafDefinitions(newUdafs).build();
}
+ public SqlTransform withErrorsTransformer(
+ PTransform<PCollection<BeamCalcRelError>, POutput> errorsTransformer) {
Review comment:
sure thing: https://github.com/apache/beam/pull/15915/commits/c0abbcee7f3c34e631599cc44007dacceece3e46
--
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] brachi-wernick commented on a change in pull request #15915: [beam-12737] add API to handle failed rows in the collection due to sql query runtime error
Posted by GitBox <gi...@apache.org>.
brachi-wernick commented on a change in pull request #15915:
URL: https://github.com/apache/beam/pull/15915#discussion_r746956498
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRelError.java
##########
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.extensions.sql.impl.rel;
+
+import com.google.auto.value.AutoValue;
+import org.apache.beam.sdk.values.Row;
+
+@AutoValue
+public abstract class BeamCalcRelError {
+
+ public static BeamCalcRelError create(Row row, String error) {
Review comment:
Correct, But I think it is less clear to "hide" the error inside `row`.
Also, there is a chance that the row already contains an 'error' field and adding our 'error' can override the existing value. (we can escape it with `__error__` or similar)
I prefer to keep the error message field explicit, does it sound good for you? am I missing something?
--
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] brachi-wernick commented on a change in pull request #15915: [beam-12737] add API to handle failed rows in the collection due to sql query runtime error
Posted by GitBox <gi...@apache.org>.
brachi-wernick commented on a change in pull request #15915:
URL: https://github.com/apache/beam/pull/15915#discussion_r746445391
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRel.java
##########
@@ -192,9 +207,17 @@ public Calc copy(RelTraitSet traitSet, RelNode input, RexProgram program) {
outputSchema,
options.getVerifyRowValues(),
getJarPaths(program),
- inputGetter.getFieldAccess());
-
- return upstream.apply(ParDo.of(calcFn)).setRowSchema(outputSchema);
+ inputGetter.getFieldAccess(),
+ errorsTransformer != null);
+
+ PCollectionTuple tuple =
+ upstream.apply(ParDo.of(calcFn).withOutputTags(rows, TupleTagList.of(errors)));
+ PCollection<BeamCalcRelError> errorPCollection =
+ tuple.get(errors).setCoder(BeamCalcRelErrorCoder.of(RowCoder.of(upstream.getSchema())));
Review comment:
sure, I can change it to be a json string instead of row, but before I change it, I want to make sure we don't miss here something, because I prefer to send the row itself to the error transformer.
The error row is not the row *after* running the expression and getting the error, it is the origin row, as it was *before* running the DoFn, that's why I set the schema to be `upstream.getSchema()`,
The row *after* the expression execution is a different one, and has `outputSchema`.
Also each node in the same SqlTransform create this chaining, and will assign a diff input and output schema, these schemas are per node and not per all the sql transform.
This is how will it look like when expanding it in DF (ignore the failed icons, this is because I abort the execution but it worked well)
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRel.java
##########
@@ -112,11 +119,19 @@
private static final long MILLIS_PER_DAY = 86400000L;
private static final ParameterExpression rowParam = Expressions.parameter(Row.class, "row");
+ private PTransform<PCollection<BeamCalcRelError>, POutput> errorsTransformer;
+ private static final TupleTag<Row> rows = new TupleTag<Row>() {};
+ private static final TupleTag<BeamCalcRelError> errors = new TupleTag<BeamCalcRelError>() {};
public BeamCalcRel(RelOptCluster cluster, RelTraitSet traits, RelNode input, RexProgram program) {
super(cluster, traits, input, program);
}
+ @Override
+ public void withErrorsTransformer(PTransform<PCollection<BeamCalcRelError>, POutput> ptransform) {
Review comment:
Yes, I can add to `buildPTransform`, but it will break the API for all `BeamRelNode` implementations.
This way, when I add it with a default function, any implementation that doesn't need to handle errors can continue as it was before, and the one that wants to add error handling will implement this method and use this variable.
Does it make sense? or you still prefer to break the api
--
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] apilloud commented on a change in pull request #15915: [beam-12737] add API to handle failed rows in the collection due to sql query runtime error
Posted by GitBox <gi...@apache.org>.
apilloud commented on a change in pull request #15915:
URL: https://github.com/apache/beam/pull/15915#discussion_r746898199
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/SqlTransform.java
##########
@@ -298,6 +304,11 @@ public SqlTransform registerUdaf(String functionName, Combine.CombineFn combineF
return toBuilder().setUdafDefinitions(newUdafs).build();
}
+ public SqlTransform withErrorsTransformer(
+ PTransform<PCollection<BeamCalcRelError>, POutput> errorsTransformer) {
Review comment:
Can you make this `PTransform<PCollection<BeamCalcRelError>, ? extends POutput>` instead? (Or really `PTransform<PCollection<Row>, ? extends POutput>`)
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRelError.java
##########
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.extensions.sql.impl.rel;
+
+import com.google.auto.value.AutoValue;
+import org.apache.beam.sdk.values.Row;
+
+@AutoValue
+public abstract class BeamCalcRelError {
+
+ public static BeamCalcRelError create(Row row, String error) {
Review comment:
It appears this class could just be replaced with `Row`?
```
schema = Schema.builder().addRowField("row", row.getSchema()).addStringField("error").build()
Row.withSchema(schema).attachValues(row, error).build()
```
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRel.java
##########
@@ -192,9 +207,17 @@ public Calc copy(RelTraitSet traitSet, RelNode input, RexProgram program) {
outputSchema,
options.getVerifyRowValues(),
getJarPaths(program),
- inputGetter.getFieldAccess());
-
- return upstream.apply(ParDo.of(calcFn)).setRowSchema(outputSchema);
+ inputGetter.getFieldAccess(),
+ errorsTransformer != null);
+
+ PCollectionTuple tuple =
+ upstream.apply(ParDo.of(calcFn).withOutputTags(rows, TupleTagList.of(errors)));
+ PCollection<BeamCalcRelError> errorPCollection =
+ tuple.get(errors).setCoder(BeamCalcRelErrorCoder.of(RowCoder.of(upstream.getSchema())));
Review comment:
Thanks for the explanation, do keep this as Rows. Your schema is still not quite right, I believe this is correct: `RowCoder.of(SelectHelpers.getOutputSchema(upstream.getSchema(), inputGetter.getFieldAccess()))`
You should add tests that cover more complex cases, like the graph above. It would also be good to add a test that feeds multiple error transformers with different rows from a single SqlTransform into a union PCollection with some useful output (to replicate the typical use case of having a single dead letter queue).
This works because we know the row at expand time and you have a different error transformer per transform. If you went to the tagged output approach, you'd need a tag per internal transform, so that seems like a non-starter.
--
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] apilloud commented on a change in pull request #15915: [beam-12737] add API to handle failed rows in the collection due to sql query runtime error
Posted by GitBox <gi...@apache.org>.
apilloud commented on a change in pull request #15915:
URL: https://github.com/apache/beam/pull/15915#discussion_r746997515
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRelError.java
##########
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.extensions.sql.impl.rel;
+
+import com.google.auto.value.AutoValue;
+import org.apache.beam.sdk.values.Row;
+
+@AutoValue
+public abstract class BeamCalcRelError {
+
+ public static BeamCalcRelError create(Row row, String error) {
Review comment:
I'm not sure what you mean by hiding the error. This creates a new row with a nested row and an error string. You'll probably want to add some constants for the column names somewhere, but you can pick the names. From a usage prospective, if you want to extract the 'original' row and error string:
```
String error = errorRow.getString("error")
Row row = errorRow.getRow("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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] brachipa commented on a change in pull request #15915: [beam-12737] add API to handle failed rows in the collection due to sql query runtime error
Posted by GitBox <gi...@apache.org>.
brachipa commented on a change in pull request #15915:
URL: https://github.com/apache/beam/pull/15915#discussion_r746354961
##########
File path: sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRel.java
##########
@@ -112,11 +119,19 @@
private static final long MILLIS_PER_DAY = 86400000L;
private static final ParameterExpression rowParam = Expressions.parameter(Row.class, "row");
+ private PTransform<PCollection<BeamCalcRelError>, POutput> errorsTransformer;
+ private static final TupleTag<Row> rows = new TupleTag<Row>() {};
+ private static final TupleTag<BeamCalcRelError> errors = new TupleTag<BeamCalcRelError>() {};
public BeamCalcRel(RelOptCluster cluster, RelTraitSet traits, RelNode input, RexProgram program) {
super(cluster, traits, input, program);
}
+ @Override
+ public void withErrorsTransformer(PTransform<PCollection<BeamCalcRelError>, POutput> ptransform) {
Review comment:
Yes, I can add to `buildPTransform`, but it will break the API for all `BeamRelNode` implementations.
When I add it with a default function, any implementation that don't need to handle errors can continue as it was before, and the one that want to do error handling will implement this method and use this variable.
--
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