[SPARK-21499] [SQL] Support creating persistent function for Spark UDAF(UserDefinedAggregateFunction) #18700

gatorsmile · 2017-07-21T08:05:40Z

What changes were proposed in this pull request?

This PR is to enable users to create persistent Scala UDAF (that extends UserDefinedAggregateFunction).

CREATE FUNCTION myDoubleAvg AS 'test.org.apache.spark.sql.MyDoubleAvg'

Before this PR, Spark UDAF only can be registered through the API spark.udf.register(...)

How was this patch tested?

Added test cases

SparkQA · 2017-07-21T09:34:16Z

Test build #79829 has finished for PR 18700 at commit 4028155.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
throw new AnalysisException(s\"Can not load class '$className' when registering \" +

gatorsmile · 2017-08-11T21:04:44Z

retest this please

SparkQA · 2017-08-11T22:31:58Z

Test build #80553 has finished for PR 18700 at commit 4028155.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
throw new AnalysisException(s\"Can not load class '$className' when registering \" +

SparkQA · 2017-08-13T03:50:04Z

Test build #80575 has finished for PR 18700 at commit a65607c.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
throw new AnalysisException(s\"Can not load class '$className' when registering \" +

SparkQA · 2017-08-13T03:54:48Z

Test build #80577 has finished for PR 18700 at commit 12cefc2.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

gatorsmile · 2017-08-13T06:16:26Z

cc @cloud-fan @ueshin

cloud-fan · 2017-08-14T14:15:34Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala

+    (children: Seq[Expression]) => {
+      try {
+        val clsForUDAF =
+          Utils.classForName("org.apache.spark.sql.expressions.UserDefinedAggregateFunction")


shall we move the UDAF interface to catalyst?

/** * The base class for implementing user-defined aggregate functions (UDAF). * * @since 1.5.0 */ @InterfaceStability.Stable abstract class UserDefinedAggregateFunction

This interface has been marked as stable. Can we still move it? or make a trait in Catalyst?

cloud-fan · 2017-08-14T14:18:19Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala

@@ -1096,8 +1099,42 @@ class SessionCatalog(
   * This performs reflection to decide what type of [[Expression]] to return in the builder.
   */
  protected def makeFunctionBuilder(name: String, functionClassName: String): FunctionBuilder = {


this will be overwritten by HiveSessionCatalog, does it mean we can not register spark UDAF if hive support is enabled?

The changes here are for HiveSessionCatalog. Also, we have a test case in HiveUDAFSuite.scala to verify it.

SparkQA · 2017-08-21T10:14:01Z

Test build #80924 has finished for PR 18700 at commit bd5ae26.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2017-08-21T14:01:09Z

sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveSessionCatalog.scala

-    // When we instantiate hive UDF wrapper class, we may throw exception if the input
-    // expressions don't satisfy the hive UDF, such as type mismatch, input number
-    // mismatch, etc. Here we catch the exception and throw AnalysisException instead.
+  override def makeFunctionBuilder(name: String, functionClassName: String): FunctionBuilder = {


do we still need to overwrite this?

This is for issuing different exceptions.

SparkQA · 2017-08-21T20:27:22Z

Test build #80933 has finished for PR 18700 at commit 7251be9.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2017-08-22T06:48:35Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala

+    val clazz = Utils.classForName(functionClassName)
+    (children: Seq[Expression]) => {
+      try {
+        makeFunctionExpression(name, Utils.classForName(functionClassName), children).getOrElse {


Utils.classForName(functionClassName) -> clazz

cloud-fan · 2017-08-22T06:51:19Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala

-  protected def makeFunctionBuilder(name: String, functionClassName: String): FunctionBuilder = {
-    // TODO: at least support UDAFs here
-    throw new UnsupportedOperationException("Use sqlContext.udf.register(...) instead.")
+  protected def makeFunctionExpression(


seems we need to catch exception for this method anyway, how about we just make this method return Expression and document that it can throw exception if the given class is not supported? Then HiveSessionCatalog can define its own exception message.

SparkQA · 2017-08-22T09:52:47Z

Test build #80974 has finished for PR 18700 at commit d3fbdc5.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2017-08-22T16:20:37Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala

+  /**
+   * Checks whether the Hive metastore is being used
+   */
+  private def isUsingHiveMetastore: Boolean = {


do we still need this?

cloud-fan · 2017-08-22T16:21:33Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala

+  }
+
+  /**
+   * Constructs a [[FunctionBuilder]] based on the provided class that represents a function.


this method returns Expression, not FunctionBuilder

cloud-fan · 2017-08-22T16:30:03Z

LGTM

SparkQA · 2017-08-22T18:41:11Z

Test build #80992 has finished for PR 18700 at commit 57607b5.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-08-22T18:46:15Z

Test build #80995 has finished for PR 18700 at commit 05e8168.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-08-22T19:17:34Z

Test build #80997 has finished for PR 18700 at commit 50224a7.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-08-22T19:51:42Z

Test build #80996 has finished for PR 18700 at commit aff8f9e.

This patch fails SparkR unit tests.
This patch does not merge cleanly.
This patch adds no public classes.

gatorsmile · 2017-08-22T19:59:38Z

The latest build #80997 passed.

gatorsmile · 2017-08-22T20:01:54Z

Thanks! Merging to master.

support UDAF

4028155

gatorsmile added 3 commits August 11, 2017 23:20

Merge remote-tracking branch 'upstream/master' into javaUDFinScala

f634043

fix.

a65607c

fix.

12cefc2

cloud-fan reviewed Aug 14, 2017

View reviewed changes

fix.

bd5ae26

cloud-fan reviewed Aug 21, 2017

View reviewed changes

fix.

7251be9

cloud-fan reviewed Aug 22, 2017

View reviewed changes

typo

d3fbdc5

gatorsmile added 3 commits August 22, 2017 08:03

fix.

57607b5

fix.

05e8168

fix.

aff8f9e

cloud-fan reviewed Aug 22, 2017

View reviewed changes

fix.

7d9aabd

cloud-fan reviewed Aug 22, 2017

View reviewed changes

gatorsmile added 2 commits August 22, 2017 09:22

Merge remote-tracking branch 'upstream/master' into pr-18700

8ea4ad1

fix.

50224a7

asfgit closed this in 43d71d9 Aug 22, 2017

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-21499] [SQL] Support creating persistent function for Spark UDAF(UserDefinedAggregateFunction) #18700

[SPARK-21499] [SQL] Support creating persistent function for Spark UDAF(UserDefinedAggregateFunction) #18700

gatorsmile commented Jul 21, 2017 •

edited

Loading

SparkQA commented Jul 21, 2017

gatorsmile commented Aug 11, 2017

SparkQA commented Aug 11, 2017

SparkQA commented Aug 13, 2017

SparkQA commented Aug 13, 2017

gatorsmile commented Aug 13, 2017

cloud-fan Aug 14, 2017

gatorsmile Aug 14, 2017

cloud-fan Aug 14, 2017 •

edited

Loading

gatorsmile Aug 14, 2017

SparkQA commented Aug 21, 2017

cloud-fan Aug 21, 2017

gatorsmile Aug 21, 2017

SparkQA commented Aug 21, 2017

cloud-fan Aug 22, 2017

cloud-fan Aug 22, 2017

SparkQA commented Aug 22, 2017

cloud-fan Aug 22, 2017

cloud-fan Aug 22, 2017

cloud-fan commented Aug 22, 2017

SparkQA commented Aug 22, 2017

SparkQA commented Aug 22, 2017

SparkQA commented Aug 22, 2017

SparkQA commented Aug 22, 2017

gatorsmile commented Aug 22, 2017

gatorsmile commented Aug 22, 2017

[SPARK-21499] [SQL] Support creating persistent function for Spark UDAF(UserDefinedAggregateFunction) #18700

[SPARK-21499] [SQL] Support creating persistent function for Spark UDAF(UserDefinedAggregateFunction) #18700

Conversation

gatorsmile commented Jul 21, 2017 • edited Loading

What changes were proposed in this pull request?

How was this patch tested?

SparkQA commented Jul 21, 2017

gatorsmile commented Aug 11, 2017

SparkQA commented Aug 11, 2017

SparkQA commented Aug 13, 2017

SparkQA commented Aug 13, 2017

gatorsmile commented Aug 13, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cloud-fan Aug 14, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Aug 21, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Aug 21, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Aug 22, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cloud-fan commented Aug 22, 2017

SparkQA commented Aug 22, 2017

SparkQA commented Aug 22, 2017

SparkQA commented Aug 22, 2017

SparkQA commented Aug 22, 2017

gatorsmile commented Aug 22, 2017

gatorsmile commented Aug 22, 2017

gatorsmile commented Jul 21, 2017 •

edited

Loading

cloud-fan Aug 14, 2017 •

edited

Loading