Skip to content

[SPARK-32243][SQL]HiveSessionCatalog call super.makeFunctionExpression should throw earlier when got Spark UDAF Invalid arguments number error #29054

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
AngersZhuuuu wants to merge 18 commits into from
Closed

[SPARK-32243][SQL]HiveSessionCatalog call super.makeFunctionExpression should throw earlier when got Spark UDAF Invalid arguments number error #29054

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
18 commits
Select commit Hold shift + click to select a range
5dd3169
save
AngersZhuuuu Jul 10, 2020
4e6b506
Update HiveUDAFSuite.scala
AngersZhuuuu Jul 10, 2020
91ceea0
Update HiveSessionCatalog.scala
AngersZhuuuu Jul 10, 2020
8c3faed
Update HiveUDAFSuite.scala
AngersZhuuuu Jul 10, 2020
b813656
Update HiveSessionCatalog.scala
AngersZhuuuu Jul 11, 2020
1397c66
Update HiveSessionCatalog.scala
AngersZhuuuu Aug 21, 2020
7f9900b
Update HiveSessionCatalog.scala
AngersZhuuuu Aug 21, 2020
9ae1614
Update HiveSessionCatalog.scala
AngersZhuuuu Aug 21, 2020
918aea4
follow commnet
AngersZhuuuu Sep 24, 2020
3df37d2
follow comment
AngersZhuuuu Sep 25, 2020
766c931
Update HiveUDAFSuite.scala
AngersZhuuuu Sep 26, 2020
a8549c4
wrap exception
AngersZhuuuu Sep 26, 2020
95cfebe
Update HiveUDAFSuite.scala
AngersZhuuuu Sep 26, 2020
34a3b98
Update SessionCatalog.scala
AngersZhuuuu Sep 26, 2020
2db41fc
follow comment
AngersZhuuuu Sep 29, 2020
3d9b6e3
follow comment
AngersZhuuuu Sep 29, 2020
42064d7
follow comment
AngersZhuuuu Sep 29, 2020
36229e3
follow comment
AngersZhuuuu Sep 29, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
94 changes: 52 additions & 42 deletions sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveSessionCatalog.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,49 +69,59 @@ private[sql] class HiveSessionCatalog(
// Current thread context classloader may not be the one loaded the class. Need to switch
// context classloader to initialize instance properly.
Utils.withContextClassLoader(clazz.getClassLoader) {
Try(super.makeFunctionExpression(name, clazz, input)).getOrElse {
Copy link
Member

@HyukjinKwon HyukjinKwon Aug 19, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@AngersZhuuuu, can you get rid of this unrelated diffs?

Copy link
Contributor Author

@AngersZhuuuu AngersZhuuuu Aug 21, 2020 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@AngersZhuuuu, can you get rid of this unrelated diffs?

How about current change, it won't change indentation. cc @maropu

var udfExpr: Option[Expression] = None
try {
// 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.
if (classOf[UDF].isAssignableFrom(clazz)) {
udfExpr = Some(HiveSimpleUDF(name, new HiveFunctionWrapper(clazz.getName), input))
udfExpr.get.dataType // Force it to check input data types.
} else if (classOf[GenericUDF].isAssignableFrom(clazz)) {
udfExpr = Some(HiveGenericUDF(name, new HiveFunctionWrapper(clazz.getName), input))
udfExpr.get.dataType // Force it to check input data types.
} else if (classOf[AbstractGenericUDAFResolver].isAssignableFrom(clazz)) {
udfExpr = Some(HiveUDAFFunction(name, new HiveFunctionWrapper(clazz.getName), input))
udfExpr.get.dataType // Force it to check input data types.
} else if (classOf[UDAF].isAssignableFrom(clazz)) {
udfExpr = Some(HiveUDAFFunction(
name,
new HiveFunctionWrapper(clazz.getName),
input,
isUDAFBridgeRequired = true))
udfExpr.get.dataType // Force it to check input data types.
} else if (classOf[GenericUDTF].isAssignableFrom(clazz)) {
udfExpr = Some(HiveGenericUDTF(name, new HiveFunctionWrapper(clazz.getName), input))
udfExpr.get.asInstanceOf[HiveGenericUDTF].elementSchema // Force it to check data types.
try {
super.makeFunctionExpression(name, clazz, input)
} catch {
case NonFatal(exception) =>
var udfExpr: Option[Expression] = None
Copy link
Contributor

@cloud-fan cloud-fan Sep 29, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We can move the code below to a new method.

Copy link
Contributor Author

@AngersZhuuuu AngersZhuuuu Sep 29, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We can move the code below to a new method.

Done

try {
// 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.
if (classOf[UDF].isAssignableFrom(clazz)) {
udfExpr = Some(HiveSimpleUDF(name, new HiveFunctionWrapper(clazz.getName), input))
udfExpr.get.dataType // Force it to check input data types.
} else if (classOf[GenericUDF].isAssignableFrom(clazz)) {
udfExpr = Some(HiveGenericUDF(name, new HiveFunctionWrapper(clazz.getName), input))
udfExpr.get.dataType // Force it to check input data types.
} else if (classOf[AbstractGenericUDAFResolver].isAssignableFrom(clazz)) {
udfExpr = Some(HiveUDAFFunction(name, new HiveFunctionWrapper(clazz.getName), input))
udfExpr.get.dataType // Force it to check input data types.
} else if (classOf[UDAF].isAssignableFrom(clazz)) {
udfExpr = Some(HiveUDAFFunction(
name,
new HiveFunctionWrapper(clazz.getName),
input,
isUDAFBridgeRequired = true))
udfExpr.get.dataType // Force it to check input data types.
} else if (classOf[GenericUDTF].isAssignableFrom(clazz)) {
udfExpr = Some(HiveGenericUDTF(name, new HiveFunctionWrapper(clazz.getName), input))
// Force it to check data types.
udfExpr.get.asInstanceOf[HiveGenericUDTF].elementSchema
}
} catch {
case NonFatal(e) =>
val noHandlerMsg = s"No handler for UDF/UDAF/UDTF '${clazz.getCanonicalName}': $e"
val errorMsg =
if (classOf[GenericUDTF].isAssignableFrom(clazz)) {
s"$noHandlerMsg\nPlease make sure your function overrides " +
"`public StructObjectInspector initialize(ObjectInspector[] args)`."
} else {
noHandlerMsg
}
val analysisException = new AnalysisException(
s"Spark UDAF Error: ${exception.getMessage}\n" +
s"Hive UDF/UDAF/UDTF Error: $errorMsg"
)
analysisException.setStackTrace(e.getStackTrace)
throw analysisException
}
udfExpr.getOrElse {
throw new AnalysisException(
s"Spark UDAF Error: ${exception.getMessage}\n" +
Copy link
Member

@HyukjinKwon HyukjinKwon Aug 22, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What if we load a Hive UDF and it fail? Is it still going to show the the method is missing in Spark?

Copy link
Contributor Author

@AngersZhuuuu AngersZhuuuu Aug 22, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What if we load a Hive UDF and it fail? Is it still going to show the the method is missing in Spark?

Yea, since in hive mode use HiveSessionCatalog, user can use Spark's and Hive's, we don't know user want to use which way(in other word, if we want to know user's intention, it's hard and we need to add a lot code to judge). when all failed show both error message, let user to know why make function failed is enough

s"Hive UDF/UDAF/UDTF Error: " +
s"No handler for UDF/UDAF/UDTF '${clazz.getCanonicalName}'")
}
} catch {
case NonFatal(e) =>
val noHandlerMsg = s"No handler for UDF/UDAF/UDTF '${clazz.getCanonicalName}': $e"
val errorMsg =
if (classOf[GenericUDTF].isAssignableFrom(clazz)) {
s"$noHandlerMsg\nPlease make sure your function overrides " +
"`public StructObjectInspector initialize(ObjectInspector[] args)`."
} else {
noHandlerMsg
}
val analysisException = new AnalysisException(errorMsg)
analysisException.setStackTrace(e.getStackTrace)
throw analysisException
}
udfExpr.getOrElse {
throw new AnalysisException(s"No handler for UDF/UDAF/UDTF '${clazz.getCanonicalName}'")
}
}
}
}
Expand Down
15 changes: 15 additions & 0 deletions sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveUDAFSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -161,6 +161,21 @@ class HiveUDAFSuite extends QueryTest
checkAnswer(sql("select histogram_numeric(a,2) from abc where a=3"), Row(null))
}
}

test("SPARK-32243: Hive mode use spark udaf should show error") {
val functionName = "longProductSum"
val functionClass = "org.apache.spark.sql.hive.execution.LongProductSum"
withUserDefinedFunction(functionName -> true) {
sql(s"CREATE TEMPORARY FUNCTION $functionName AS '$functionClass'")
val e = intercept[AnalysisException] {
sql(s"SELECT $functionName(100)")
}.getMessage
assert(
Seq(s"Invalid number of arguments for function $functionName. Expected: 2; Found: 1;",
"No handler for UDF/UDAF/UDTF 'org.apache.spark.sql.hive.execution.LongProductSum';")
.forall(e.contains))
}
}
}

/**
Expand Down