[BEAM-5202] register UDF/UDAF with ServiceLoader #6580

mingmxu · 2018-10-05T06:29:02Z

add @AutoService(UdfUdafRegister.class) to register all UDF/UDAF in one call,

avoid multiple registerUdf()/registerUdaf() methods in SqlTransform;
enable UDF/UDAF load in BeamSqlCli;

R: @akedin @apilloud @amaliujia @xumingming

Follow this checklist to help us incorporate your contribution quickly and easily:

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.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

It will help us expedite review of your Pull Request if you tag someone (e.g. @username) to look at it.

Post-Commit Tests Status (on master branch)

Lang	Apex	Dataflow	Flink	Gearpump	Samza	Spark
Go	---	---	---	---	---	---
Java
Python	---			---	---	---

amaliujia · 2018-10-05T16:07:07Z

sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/SqlTransform.java

+    List<UdafDefinition> newUdafs = new ArrayList<>();
+
+    for (UdfUdafRegister ins :
+        Lists.newArrayList(ServiceLoader.<UdfUdafRegister>load(UdfUdafRegister.class))) {


ServiceLoader implements Iterable interface so you don't need to wrap it into a arraylist to for each.

amaliujia · 2018-10-05T16:11:21Z

...a/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslUdfUdafTest.java

+
+  /** Auto register for test. */
+  @AutoService(UdfUdafRegister.class)
+  public static class UdfUdafRegisterTest implements UdfUdafRegister {


why UdfUdafRegisterTest is a static class without a static variable/function? Why to make it static?

Static nested classes have nothing to do with static members, it's about access to the outer class' non-static variables (obviously): https://docs.oracle.com/javase/tutorial/java/javaOO/nested.html

Static in this context means that it doesn't reference the the parent class. If it wasn't static it would only be accessible on an instantiation of BeamSqlDslUdfUdafTest, which means the service loader couldn't find it.

apilloud

I was thinking about doing something like so we can support UDFs in our JDBC based SQL shell and this gets us much closer. I still need to make some internal refactors and hook this up to BeamCalciteSchemaFactory once it goes in.

apilloud · 2018-10-05T16:10:07Z

sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/UdfUdafRegister.java

+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.extensions.sql;


can you move this to meta/provider/UdfUdafProvider.java

apilloud · 2018-10-05T16:20:03Z

...a/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslUdfUdafTest.java

+    @Override
+    public Map<String, Class<? extends BeamSqlUdf>> getEvalUdfs() {
+      return new HashMap<String, Class<? extends BeamSqlUdf>>(
+          ImmutableMap.of("cubic", CubicInteger.class));


These udfs are registered as long as the test jar is in the classpath. Can you make the name of these test functions a little more unique so we don't have collisions?

apilloud · 2018-10-05T16:21:57Z

sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/UdfUdafRegister.java

+/** Register for UDF and UDAF. */
+public interface UdfUdafRegister {
+  default Map<String, Class<? extends BeamSqlUdf>> getEvalUdfs() {
+    return new HashMap<>();


nit: use Collections.emptyMap() for these.

apilloud · 2018-10-05T16:22:51Z

sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/BeamSqlEnv.java

+  /** Load all UDF/UDAF from {@link UdfUdafRegister}. */
+  public void autoRegisterUdfUdaf() {
+    for (UdfUdafRegister ins :
+        Lists.newArrayList(ServiceLoader.<UdfUdafRegister>load(UdfUdafRegister.class))) {


nit: Lists.newArrayList is unneeded.

apilloud · 2018-10-05T16:27:08Z

sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/SqlTransform.java

@@ -151,6 +156,35 @@ public static SqlTransform query(String queryString) {
        .build();
  }

+  /** Load all UDF/UDAF from {@link UdfUdafRegister}. */
+  public SqlTransform autoRegisterUdfUdaf() {
+    List<UdfDefinition> newUdfs = new ArrayList<>();


Instead of copying everything, could you just set a bool, then call env.autoRegisterUdfUdaf() in registerFunctions?

apilloud · 2018-10-05T16:34:05Z

...a/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslUdfUdafTest.java

+
+  /** Auto register for test. */
+  @AutoService(UdfUdafRegister.class)
+  public static class UdfUdafRegisterTest implements UdfUdafRegister {


Static in this context means that it doesn't reference the the parent class. If it wasn't static it would only be accessible on an instantiation of BeamSqlDslUdfUdafTest, which means the service loader couldn't find it.

akedin

Nice feature! Few comments:

akedin · 2018-10-05T16:24:15Z

...a/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslUdfUdafTest.java

+
+  /** Auto register for test. */
+  @AutoService(UdfUdafRegister.class)
+  public static class UdfUdafRegisterTest implements UdfUdafRegister {


Static nested classes have nothing to do with static members, it's about access to the outer class' non-static variables (obviously): https://docs.oracle.com/javase/tutorial/java/javaOO/nested.html

akedin · 2018-10-05T16:36:09Z

sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/UdfUdafRegister.java

+
+/** Register for UDF and UDAF. */
+public interface UdfUdafRegister {
+  default Map<String, Class<? extends BeamSqlUdf>> getEvalUdfs() {


By reading this it's unclear what are eval udfs vs serialize udfs, add comments. And I would probably just call these methods getBeamSqlUdfs() and getSerializableFunctionUdfs() to avoid inventing another name for these things.

Another approach that I can think of (don't know if it's better or not) to make it a bit more extensible is to use TypeDescriptor-style type specification:

udfs = ImmutableMap.builder() .put(BeamSqlUdf.class, new HashMap<String, BeamSqlUdf>()) .put(SerializableFunction.class, new HashMap<String, SerializableFunction<?, ?>>) .build(); public <T> Map<String, T> getUdf(TypeDescriptor<T> udfTypeDescriptor) { udfType = // get udf class from udfTypeDescriptor return udfs.get(udfType); }

I imagine the usage will be like: register.getUdfs(new TypeDescriptor<BeamSqlUdf>> {}).

And I wonder wether it will be useful to have methods like register.getUdf(String name, TypeDescriptor udfType);

akedin · 2018-10-05T16:38:04Z

sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/UdfUdafRegister.java

+import org.apache.beam.sdk.transforms.SerializableFunction;
+
+/** Register for UDF and UDAF. */
+public interface UdfUdafRegister {


I think we call these either Registrar or Registry (pipeline options registrar, coder registry)

This is basically the same as TableProvider but for Udf, so what about UdfProivder? Or should we rename TableProvider to TableRegistry?

UdfUdafProvider sounds good to me.

akedin · 2018-10-05T16:39:05Z

sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/UdfUdafRegister.java

+  }
+
+  default Map<String, SerializableFunction<?, ?>> getSerializeUdfs() {
+    return new HashMap<>();


I think it should return an immutable empty map in this case (Collections.emptyMap())

amaliujia · 2018-10-05T17:42:00Z

A further thought is, as a following step, we could allow user set jar path to point to user defined functions. Both SQL shell and SqlTransform can benefit from it.

mingmxu · 2018-10-05T18:48:09Z

Thanks @ALL, PLNK

Updates:

move to meta/provider/UdfUdafProvider;
keep AutoService code in BeamSqlEnv only;

amaliujia · 2018-10-05T18:50:39Z

LGTM

akedin

LGTM

apilloud

This is great! Thanks! LGTM

apilloud · 2018-10-08T16:24:06Z

...a/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslUdfUdafTest.java

+
+    @Override
+    public Map<String, Class<? extends BeamSqlUdf>> getBeamSqlUdfs() {
+      return new HashMap<String, Class<? extends BeamSqlUdf>>(


Sorry I missed this the first time around. The new HashMap is unnecessary here and below.

let me fix it and will merge after CI pass

move to `meta/provider/UdfUdafProvider` and remove duplicated AutoService code. fixup fix

amaliujia reviewed Oct 5, 2018

View reviewed changes

apilloud reviewed Oct 5, 2018

View reviewed changes

akedin reviewed Oct 5, 2018

View reviewed changes

mingmxu force-pushed the BEAM-5202 branch 2 times, most recently from d1b1fb9 to aa37bb1 Compare October 5, 2018 18:46

akedin approved these changes Oct 5, 2018

View reviewed changes

apilloud approved these changes Oct 8, 2018

View reviewed changes

mingmxu force-pushed the BEAM-5202 branch from 2216933 to ffd5c85 Compare October 8, 2018 21:00

add UdfUdafRegister to auto load UDF/UDAF.

20f710b

move to `meta/provider/UdfUdafProvider` and remove duplicated AutoService code. fixup fix

mingmxu force-pushed the BEAM-5202 branch from ffd5c85 to 20f710b Compare October 8, 2018 21:02

mingmxu merged commit 8a88ffd into apache:master Oct 8, 2018

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[BEAM-5202] register UDF/UDAF with ServiceLoader #6580

[BEAM-5202] register UDF/UDAF with ServiceLoader #6580

mingmxu commented Oct 5, 2018

amaliujia Oct 5, 2018 •

edited

amaliujia Oct 5, 2018

akedin Oct 5, 2018

apilloud Oct 5, 2018

amaliujia Oct 5, 2018

apilloud left a comment

apilloud Oct 5, 2018

apilloud Oct 5, 2018

apilloud Oct 5, 2018

apilloud Oct 5, 2018

apilloud Oct 5, 2018

apilloud Oct 5, 2018

akedin left a comment

akedin Oct 5, 2018

akedin Oct 5, 2018

akedin Oct 5, 2018

apilloud Oct 5, 2018

mingmxu Oct 5, 2018

akedin Oct 5, 2018

amaliujia commented Oct 5, 2018

mingmxu commented Oct 5, 2018

amaliujia commented Oct 5, 2018

akedin left a comment

apilloud left a comment

apilloud Oct 8, 2018

mingmxu Oct 8, 2018

[BEAM-5202] register UDF/UDAF with ServiceLoader #6580

[BEAM-5202] register UDF/UDAF with ServiceLoader #6580

Conversation

mingmxu commented Oct 5, 2018

R: @akedin @apilloud @amaliujia @xumingming

Post-Commit Tests Status (on master branch)

amaliujia Oct 5, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

apilloud left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

akedin left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

amaliujia commented Oct 5, 2018

mingmxu commented Oct 5, 2018

amaliujia commented Oct 5, 2018

akedin left a comment

Choose a reason for hiding this comment

apilloud left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

amaliujia Oct 5, 2018 •

edited