[BEAM-10925] Load user-provided jars in CalcFn. #13835

ibzib · 2021-01-28T21:47:08Z

Note that it is possible for one CalcFn to require multiple jars, so for
now it is the user's responsibility to ensure there are no conflicts.
This is no different from user-provided jars in the Java SDK.

This adds an optional jar path field to ScalarFunctionImpl. This field
will be populated in a later PR.

R: @amaliujia @apilloud

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

Choose reviewer(s) 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.

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

Post-Commit Tests Status (on master branch)

Lang	Dataflow	Samza	Twister2
Go	---	---	---
Java
Python		---	---
XLang		---	---

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website	Whitespace	Typescript
Non-portable
Portable	---		---	---	---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

Note that it is possible for one CalcFn to require multiple jars, so for now it is the user's responsibility to ensure there are no conflicts. This is no different from user-provided jars in the Java SDK. This adds an optional jar path field to ScalarFunctionImpl. This field will be populated in a later PR.

amaliujia

To me it is ok to have test cases when the jar path field is populated in another PR.

amaliujia · 2021-01-29T00:14:16Z

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

+  private static final Map<String, FunctionDefinitions> functionCache = new HashMap<>();
+
+  /** Maps potentially remote jar paths to their local file copies. */
+  private static final Map<String, File> jarCache = new HashMap<>();


To confirm: for the created random files, as I recall they will live while the JVM is alive?

Just to confirm that there will not be the case the there is a entry in this jarCache but the jar file does not exist.

I can only think of a couple ways the cache could contain handles to nonexistent files:

If the file was deleted somehow.

If the cache was serialized and shipped to workers. But we don't include a JavaUdfLoader instance as part of CalcFn, instead we construct a new JavaUdfLoader instance in CalcFn::setup. Besides, the cache is a static variable, and static variables are never serialized.

I see.

I see there are IOException thrown when loading jars. As long as there are meaningful exception are caught, at least we will have exceptions in log.

amaliujia · 2021-01-29T00:21:57Z

...va/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRel.java

+      if (!jarPaths.isEmpty()) {
+        JavaUdfLoader udfLoader = new JavaUdfLoader();
+        ClassLoader classLoader = udfLoader.createClassLoader(jarPaths);
+        se.setParentClassLoader(classLoader);


From the java doc I cannot really tell what this setParentClassLoader behave: https://janino-compiler.github.io/janino/apidocs/org/codehaus/janino/ScriptEvaluator.html#setParentClassLoader(java.lang.ClassLoader)

Does it use classLoader to replace the default one or combine with the default one? Basically I guess to make ScriptEvaluator work functionally, it needs to know both Beam and UDF implementation.

The new classloader replaces the default one, but that is okay. The classloader we construct does not only contain the UDF jars. It also has a parent classloader, which is the same as the default one.

ibzib · 2021-01-29T01:35:55Z

run sql postcommit

ibzib · 2021-01-29T01:36:22Z

Run SQL PostCommit

ibzib · 2021-01-29T01:36:43Z

There we go, Jenkins tests didn't trigger for some reason.

amaliujia · 2021-01-29T04:00:03Z

...va/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRel.java

+    List<String> jarPaths = new ArrayList<>();
+    for (RexNode node : program.getExprList()) {
+      if (node instanceof RexCall) {
+        SqlOperator op = ((RexCall) node).op;


Nit: I am not sure whether there could be nested call case here (i.e. one of the operands are actually a RexCall). If there is, maybe add a TODO or a check for such case.

We don't need to handle nested call in this PR. The solution for that will be complicated.

A RexNode can only refer to an input or previous RexNode in the list returned by getExprList, so that is handled here.
https://github.com/apache/calcite/blob/3d13846a13398a1ba6c1fa84a7d0c0cc543f23d4/core/src/main/java/org/apache/calcite/rex/RexProgram.java#L80-L85

apilloud

LGTM

apilloud · 2021-01-29T17:24:37Z

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

+  public ClassLoader createClassLoader(List<String> inputJarPaths) throws IOException {
+    List<File> localJars = new ArrayList<>();
+    for (String inputJar : inputJarPaths) {
+      localJars.add(getLocalJar(inputJar));


Couldn't you merge these into a single loop?
urls.add(getLocalJar(inputJar).toURI().toURL())

Yep, done. (I originally tried to do this the functional way using streams, but Java doesn't really let you throw checked exceptions from lambdas. 😢 )

apilloud · 2021-01-29T17:39:40Z

...va/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRel.java

+    List<String> jarPaths = new ArrayList<>();
+    for (RexNode node : program.getExprList()) {
+      if (node instanceof RexCall) {
+        SqlOperator op = ((RexCall) node).op;


A RexNode can only refer to an input or previous RexNode in the list returned by getExprList, so that is handled here.
https://github.com/apache/calcite/blob/3d13846a13398a1ba6c1fa84a7d0c0cc543f23d4/core/src/main/java/org/apache/calcite/rex/RexProgram.java#L80-L85

apilloud · 2021-01-29T17:41:41Z

...va/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRel.java

-      calcFn.compile();
+      try {
+        calcFn.compile();
+      } catch (IOException e) {


nit: possibly handle this inside compile() so it doesn't expose an IOException?

apilloud · 2021-01-29T17:44:46Z

...va/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRel.java

@@ -274,6 +298,25 @@ public void processElement(ProcessContext c) {
    }
  }

+  private static List<String> getJarPaths(RexProgram program) {
+    List<String> jarPaths = new ArrayList<>();


Use ImmutableList.Builder instead.

I don't think this code will ever actually be invoked by unprivileged code, but this makes spotbugs happy.

This is required to pass the jar path to BeamCalcRel (see apache#13835).

amaliujia self-requested a review January 28, 2021 23:55

amaliujia reviewed Jan 29, 2021

View reviewed changes

amaliujia approved these changes Jan 29, 2021

View reviewed changes

amaliujia reviewed Jan 29, 2021

View reviewed changes

apilloud approved these changes Jan 29, 2021

View reviewed changes

ibzib added 2 commits January 29, 2021 10:17

[BEAM-10925] Fix code style

989c317

Wrap classloader creation in doPrivileged block.

6f4c664

I don't think this code will ever actually be invoked by unprivileged code, but this makes spotbugs happy.

ibzib merged commit 96280f7 into apache:master Jan 29, 2021

ibzib added a commit to ibzib/beam that referenced this pull request Jan 29, 2021

[BEAM-10925] Add optional jar path field to ZetaSqlScalarFunctionImpl.

de66889

This is required to pass the jar path to BeamCalcRel (see apache#13835).

ibzib added a commit to ibzib/beam that referenced this pull request Jan 29, 2021

[BEAM-10925] Add optional jar path field to ZetaSqlScalarFunctionImpl.

477fc28

This is required to pass the jar path to BeamCalcRel (see apache#13835).

ibzib mentioned this pull request Jan 29, 2021

[BEAM-10925] Add optional jar path field to ZetaSqlScalarFunctionImpl. #13844

Merged

4 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[BEAM-10925] Load user-provided jars in CalcFn. #13835

[BEAM-10925] Load user-provided jars in CalcFn. #13835

ibzib commented Jan 28, 2021

amaliujia left a comment

amaliujia Jan 29, 2021

ibzib Jan 29, 2021

amaliujia Jan 29, 2021

amaliujia Jan 29, 2021

ibzib Jan 29, 2021

ibzib commented Jan 29, 2021

ibzib commented Jan 29, 2021

ibzib commented Jan 29, 2021

amaliujia Jan 29, 2021 •

edited

apilloud Jan 29, 2021

apilloud left a comment

apilloud Jan 29, 2021

ibzib Jan 29, 2021

apilloud Jan 29, 2021

apilloud Jan 29, 2021

ibzib Jan 29, 2021

apilloud Jan 29, 2021

ibzib Jan 29, 2021

[BEAM-10925] Load user-provided jars in CalcFn. #13835

[BEAM-10925] Load user-provided jars in CalcFn. #13835

Conversation

ibzib commented Jan 28, 2021

Post-Commit Tests Status (on master branch)

Pre-Commit Tests Status (on master branch)

GitHub Actions Tests Status (on master branch)

amaliujia 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

ibzib commented Jan 29, 2021

ibzib commented Jan 29, 2021

ibzib commented Jan 29, 2021

amaliujia Jan 29, 2021 • edited

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

Choose a reason for hiding this comment

amaliujia Jan 29, 2021 •

edited