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/03/17 02:43:39 UTC
[GitHub] [beam] ibzib opened a new pull request #14263: [BEAM-10943] Enable SqlTransform::registerUdf in ZetaSQL.
ibzib opened a new pull request #14263:
URL: https://github.com/apache/beam/pull/14263
**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).
Post-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
Lang | SDK | Dataflow | Flink | Samza | Spark | Twister2
--- | --- | --- | --- | --- | --- | ---
Go | [](https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/) | ---
Java | [](https://ci-beam.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/) | [](https://ci-beam
.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Twister2/lastCompletedBuild/)
Python | [](https://ci-beam.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python38/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Python_PVR_Flink_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/) | ---
XLang | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Direct/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Dataflow/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/) | ---
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
--- |Java | Python | Go | Website | Whitespace | Typescript
--- | --- | --- | --- | --- | --- | ---
Non-portable | [](https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/lastCompletedBuild/) <br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Whitespace_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Typescript_Cron/lastCompletedBuild/)
Portable | --- | [](https://ci-beam.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/) | --- | --- | --- | ---
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] ibzib commented on pull request #14263: [BEAM-10943] Enable SqlTransform::registerUdf in ZetaSQL.
Posted by GitBox <gi...@apache.org>.
ibzib commented on pull request #14263:
URL: https://github.com/apache/beam/pull/14263#issuecomment-809757188
R: @apilloud
I wrote three tests with overlapping coverage, since the test with the most complete coverage can't pass yet.
--
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] ibzib merged pull request #14263: [BEAM-10943] Enable SqlTransform::registerUdf in ZetaSQL.
Posted by GitBox <gi...@apache.org>.
ibzib merged pull request #14263:
URL: https://github.com/apache/beam/pull/14263
--
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] ibzib commented on a change in pull request #14263: [BEAM-10943] Enable SqlTransform::registerUdf in ZetaSQL.
Posted by GitBox <gi...@apache.org>.
ibzib commented on a change in pull request #14263:
URL: https://github.com/apache/beam/pull/14263#discussion_r604343668
##########
File path: sdks/java/extensions/sql/zetasql/src/main/java/org/apache/beam/sdk/extensions/sql/zetasql/BeamZetaSqlCatalog.java
##########
@@ -276,6 +285,49 @@ private void addWindowTvfs() {
null));
}
+ private void addUdfsFromSchema() {
+ for (String functionName : calciteSchema.getFunctionNames()) {
+ Collection<org.apache.beam.vendor.calcite.v1_20_0.org.apache.calcite.schema.Function>
+ functions = calciteSchema.getFunctions(functionName);
+ if (functions.size() != 1) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Expected exactly 1 definition for function '%s', but found %d.",
Review comment:
It's a Beam ZetaSQL limitation. I filed a JIRA for it: BEAM-12073
--
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] ibzib commented on a change in pull request #14263: [BEAM-10943] Enable SqlTransform::registerUdf in ZetaSQL.
Posted by GitBox <gi...@apache.org>.
ibzib commented on a change in pull request #14263:
URL: https://github.com/apache/beam/pull/14263#discussion_r604343897
##########
File path: sdks/java/extensions/sql/zetasql/src/main/java/org/apache/beam/sdk/extensions/sql/zetasql/BeamZetaSqlCatalog.java
##########
@@ -276,6 +285,49 @@ private void addWindowTvfs() {
null));
}
+ private void addUdfsFromSchema() {
+ for (String functionName : calciteSchema.getFunctionNames()) {
+ Collection<org.apache.beam.vendor.calcite.v1_20_0.org.apache.calcite.schema.Function>
+ functions = calciteSchema.getFunctions(functionName);
+ if (functions.size() != 1) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Expected exactly 1 definition for function '%s', but found %d.",
+ functionName, functions.size()));
+ }
+ for (org.apache.beam.vendor.calcite.v1_20_0.org.apache.calcite.schema.Function function :
+ functions) {
+ if (function instanceof ScalarFunctionImpl) {
Review comment:
There are no other possible cases I'm aware of. I'll add that.
--
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] apilloud commented on a change in pull request #14263: [BEAM-10943] Enable SqlTransform::registerUdf in ZetaSQL.
Posted by GitBox <gi...@apache.org>.
apilloud commented on a change in pull request #14263:
URL: https://github.com/apache/beam/pull/14263#discussion_r604324748
##########
File path: sdks/java/extensions/sql/zetasql/src/main/java/org/apache/beam/sdk/extensions/sql/zetasql/BeamZetaSqlCatalog.java
##########
@@ -276,6 +285,49 @@ private void addWindowTvfs() {
null));
}
+ private void addUdfsFromSchema() {
+ for (String functionName : calciteSchema.getFunctionNames()) {
+ Collection<org.apache.beam.vendor.calcite.v1_20_0.org.apache.calcite.schema.Function>
+ functions = calciteSchema.getFunctions(functionName);
+ if (functions.size() != 1) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Expected exactly 1 definition for function '%s', but found %d.",
Review comment:
Is this an assert or a Beam ZetaSQL limitation? Is the case where we get multiple values back is when the user supplied multiple variants of the same function? If so: `Beam ZetaSQL supports exactly 1...`
Users are going to eventually ask for multiple variants of the same UDF for different types, it appears that is already supported by Calcite and ZetaSQL.
##########
File path: sdks/java/extensions/sql/zetasql/src/main/java/org/apache/beam/sdk/extensions/sql/zetasql/BeamZetaSqlCatalog.java
##########
@@ -276,6 +285,49 @@ private void addWindowTvfs() {
null));
}
+ private void addUdfsFromSchema() {
+ for (String functionName : calciteSchema.getFunctionNames()) {
+ Collection<org.apache.beam.vendor.calcite.v1_20_0.org.apache.calcite.schema.Function>
+ functions = calciteSchema.getFunctions(functionName);
+ if (functions.size() != 1) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Expected exactly 1 definition for function '%s', but found %d.",
+ functionName, functions.size()));
+ }
+ for (org.apache.beam.vendor.calcite.v1_20_0.org.apache.calcite.schema.Function function :
+ functions) {
+ if (function instanceof ScalarFunctionImpl) {
Review comment:
Are there other cases we should be concerned with? If there is nothing, should we add an `else` case that throws an exception? (I'm worried about cases where UDFs are silently dropped.)
--
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] ibzib commented on a change in pull request #14263: [BEAM-10943] Enable SqlTransform::registerUdf in ZetaSQL.
Posted by GitBox <gi...@apache.org>.
ibzib commented on a change in pull request #14263:
URL: https://github.com/apache/beam/pull/14263#discussion_r603623438
##########
File path: sdks/java/extensions/sql/zetasql/src/test/java/org/apache/beam/sdk/extensions/sql/zetasql/ZetaSqlJavaUdfTest.java
##########
@@ -186,6 +188,25 @@ public void testExpectedNullArgument() {
pipeline.run().waitUntilFinish(Duration.standardMinutes(PIPELINE_EXECUTION_WAITTIME_MINUTES));
}
+ public static class IncrementFn implements BeamSqlUdf {
+ public Long eval(Long i) {
+ return i + 1;
+ }
+ }
+
+ @Test
+ public void testSqlTransformRegisterUdf() {
+ String sql = "SELECT increment(0);";
+ PCollection<Row> stream =
+ pipeline.apply(
+ SqlTransform.query(sql)
+ .withQueryPlannerClass(ZetaSQLQueryPlanner.class)
Review comment:
Note to self: this test doesn't work because ZetaSQLQueryPlanner does not have BeamJavaUdfCalcRule enabled by default, and we can't pass a ZetaSQLQueryPlanner instance to SqlTransform (only the class itself).
--
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