Skip to content

[SPARK-10671] [SQL] Throws an analysis exception if we cannot find Hive UDFs #8941

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
wants to merge 2 commits into from
Closed

[SPARK-10671] [SQL] Throws an analysis exception if we cannot find Hive UDFs #8941

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
780ebb7
Throws an analysis exception if we cannot find Hive UDFs
cloud-fan Sep 29, 2015
666dd63
address comment
cloud-fan Oct 1, 2015
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
68 changes: 46 additions & 22 deletions sql/hive/src/main/scala/org/apache/spark/sql/hive/hiveUDFs.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,20 +60,36 @@ private[hive] class HiveFunctionRegistry(underlying: analysis.FunctionRegistry)

val functionClassName = functionInfo.getFunctionClass.getName

if (classOf[UDF].isAssignableFrom(functionInfo.getFunctionClass)) {
HiveSimpleUDF(new HiveFunctionWrapper(functionClassName), children)
} else if (classOf[GenericUDF].isAssignableFrom(functionInfo.getFunctionClass)) {
HiveGenericUDF(new HiveFunctionWrapper(functionClassName), children)
} else if (
classOf[AbstractGenericUDAFResolver].isAssignableFrom(functionInfo.getFunctionClass)) {
HiveUDAFFunction(new HiveFunctionWrapper(functionClassName), children)
} else if (classOf[UDAF].isAssignableFrom(functionInfo.getFunctionClass)) {
HiveUDAFFunction(
new HiveFunctionWrapper(functionClassName), children, isUDAFBridgeRequired = true)
} else if (classOf[GenericUDTF].isAssignableFrom(functionInfo.getFunctionClass)) {
HiveGenericUDTF(new HiveFunctionWrapper(functionClassName), children)
} else {
sys.error(s"No handler for udf ${functionInfo.getFunctionClass}")
// 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.
try {
if (classOf[UDF].isAssignableFrom(functionInfo.getFunctionClass)) {
HiveSimpleUDF(new HiveFunctionWrapper(functionClassName), children)
} else if (classOf[GenericUDF].isAssignableFrom(functionInfo.getFunctionClass)) {
HiveGenericUDF(new HiveFunctionWrapper(functionClassName), children)
} else if (
classOf[AbstractGenericUDAFResolver].isAssignableFrom(functionInfo.getFunctionClass)) {
HiveUDAFFunction(new HiveFunctionWrapper(functionClassName), children)
} else if (classOf[UDAF].isAssignableFrom(functionInfo.getFunctionClass)) {
HiveUDAFFunction(
new HiveFunctionWrapper(functionClassName), children, isUDAFBridgeRequired = true)
} else if (classOf[GenericUDTF].isAssignableFrom(functionInfo.getFunctionClass)) {
val udtf = HiveGenericUDTF(new HiveFunctionWrapper(functionClassName), children)
udtf.elementTypes // Force it to check input data types.
udtf
} else {
throw new AnalysisException(s"No handler for udf ${functionInfo.getFunctionClass}")
}
} catch {
case analysisException: AnalysisException =>
// If the exception is an AnalysisException, just throw it.
throw analysisException
case throwable: Throwable =>
// If there is any other error, we throw an AnalysisException.
val errorMessage = s"No handler for Hive udf ${functionInfo.getFunctionClass} " +
s"because: ${throwable.getMessage}."
throw new AnalysisException(errorMessage)
}
}
}
Expand Down Expand Up @@ -134,7 +150,7 @@ private[hive] case class HiveSimpleUDF(funcWrapper: HiveFunctionWrapper, childre
@transient
private lazy val conversionHelper = new ConversionHelper(method, arguments)

val dataType = javaClassToDataType(method.getReturnType)
override val dataType = javaClassToDataType(method.getReturnType)

@transient
lazy val returnInspector = ObjectInspectorFactory.getReflectionObjectInspector(
Expand Down Expand Up @@ -205,7 +221,7 @@ private[hive] case class HiveGenericUDF(funcWrapper: HiveFunctionWrapper, childr
new DeferredObjectAdapter(inspect, child.dataType)
}.toArray[DeferredObject]

lazy val dataType: DataType = inspectorToDataType(returnInspector)
override val dataType: DataType = inspectorToDataType(returnInspector)

override def eval(input: InternalRow): Any = {
returnInspector // Make sure initialized.
Expand All @@ -231,16 +247,24 @@ private[hive] case class HiveGenericUDF(funcWrapper: HiveFunctionWrapper, childr
* Resolves [[UnresolvedWindowFunction]] to [[HiveWindowFunction]].
*/
private[spark] object ResolveHiveWindowFunction extends Rule[LogicalPlan] {
private def shouldResolveFunction(
unresolvedWindowFunction: UnresolvedWindowFunction,
windowSpec: WindowSpecDefinition): Boolean = {
unresolvedWindowFunction.childrenResolved && windowSpec.childrenResolved
}

def apply(plan: LogicalPlan): LogicalPlan = plan transformUp {
case p: LogicalPlan if !p.childrenResolved => p

// We are resolving WindowExpressions at here. When we get here, we have already
// replaced those WindowSpecReferences.
case p: LogicalPlan =>
p transformExpressions {
// We will not start to resolve the function unless all arguments are resolved
// and all expressions in window spec are fixed.
case WindowExpression(
UnresolvedWindowFunction(name, children),
windowSpec: WindowSpecDefinition) =>
u @ UnresolvedWindowFunction(name, children),
windowSpec: WindowSpecDefinition) if shouldResolveFunction(u, windowSpec) =>
// First, let's find the window function info.
val windowFunctionInfo: WindowFunctionInfo =
Option(FunctionRegistry.getWindowFunctionInfo(name.toLowerCase)).getOrElse(
Expand All @@ -256,7 +280,7 @@ private[spark] object ResolveHiveWindowFunction extends Rule[LogicalPlan] {
// are expressions in Order By clause.
if (classOf[GenericUDAFRank].isAssignableFrom(functionClass)) {
if (children.nonEmpty) {
throw new AnalysisException(s"$name does not take input parameters.")
throw new AnalysisException(s"$name does not take input parameters.")
}
windowSpec.orderSpec.map(_.child)
} else {
Expand Down Expand Up @@ -358,7 +382,7 @@ private[hive] case class HiveWindowFunction(
evaluator.init(GenericUDAFEvaluator.Mode.COMPLETE, inputInspectors)
}

override def dataType: DataType =
override val dataType: DataType =
if (!pivotResult) {
inspectorToDataType(returnInspector)
} else {
Expand Down Expand Up @@ -478,7 +502,7 @@ private[hive] case class HiveGenericUDTF(
@transient
protected lazy val collector = new UDTFCollector

lazy val elementTypes = outputInspector.getAllStructFieldRefs.asScala.map {
override lazy val elementTypes = outputInspector.getAllStructFieldRefs.asScala.map {
field => (inspectorToDataType(field.getFieldObjectInspector), true)
}

Expand Down Expand Up @@ -602,6 +626,6 @@ private[hive] case class HiveUDAFFunction(

override def supportsPartial: Boolean = false

override lazy val dataType: DataType = inspectorToDataType(returnInspector)
override val dataType: DataType = inspectorToDataType(returnInspector)
}

59 changes: 58 additions & 1 deletion sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveUDFSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,8 @@ import java.io.{DataInput, DataOutput}
import java.util.{ArrayList, Arrays, Properties}

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hive.ql.udf.generic.{GenericUDAFAverage, GenericUDF}
import org.apache.hadoop.hive.ql.udf.UDAFPercentile
import org.apache.hadoop.hive.ql.udf.generic.{GenericUDFOPAnd, GenericUDTFExplode, GenericUDAFAverage, GenericUDF}
import org.apache.hadoop.hive.ql.udf.generic.GenericUDF.DeferredObject
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory
import org.apache.hadoop.hive.serde2.objectinspector.{ObjectInspector, ObjectInspectorFactory}
Expand Down Expand Up @@ -299,6 +300,62 @@ class HiveUDFSuite extends QueryTest with TestHiveSingleton {

hiveContext.reset()
}

test("Hive UDFs with insufficient number of input arguments should trigger an analysis error") {
Seq((1, 2)).toDF("a", "b").registerTempTable("testUDF")

{
// HiveSimpleUDF
sql(s"CREATE TEMPORARY FUNCTION testUDFTwoListList AS '${classOf[UDFTwoListList].getName}'")
val message = intercept[AnalysisException] {
sql("SELECT testUDFTwoListList() FROM testUDF")
}.getMessage
assert(message.contains("No handler for Hive udf"))
sql("DROP TEMPORARY FUNCTION IF EXISTS testUDFTwoListList")
}

{
// HiveGenericUDF
sql(s"CREATE TEMPORARY FUNCTION testUDFAnd AS '${classOf[GenericUDFOPAnd].getName}'")
val message = intercept[AnalysisException] {
sql("SELECT testUDFAnd() FROM testUDF")
}.getMessage
assert(message.contains("No handler for Hive udf"))
sql("DROP TEMPORARY FUNCTION IF EXISTS testUDFAnd")
}

{
// Hive UDAF
sql(s"CREATE TEMPORARY FUNCTION testUDAFPercentile AS '${classOf[UDAFPercentile].getName}'")
val message = intercept[AnalysisException] {
sql("SELECT testUDAFPercentile(a) FROM testUDF GROUP BY b")
}.getMessage
assert(message.contains("No handler for Hive udf"))
sql("DROP TEMPORARY FUNCTION IF EXISTS testUDAFPercentile")
}

{
// AbstractGenericUDAFResolver
sql(s"CREATE TEMPORARY FUNCTION testUDAFAverage AS '${classOf[GenericUDAFAverage].getName}'")
val message = intercept[AnalysisException] {
sql("SELECT testUDAFAverage() FROM testUDF GROUP BY b")
}.getMessage
assert(message.contains("No handler for Hive udf"))
sql("DROP TEMPORARY FUNCTION IF EXISTS testUDAFAverage")
}

{
// Hive UDTF
sql(s"CREATE TEMPORARY FUNCTION testUDTFExplode AS '${classOf[GenericUDTFExplode].getName}'")
val message = intercept[AnalysisException] {
sql("SELECT testUDTFExplode() FROM testUDF")
}.getMessage
assert(message.contains("No handler for Hive udf"))
sql("DROP TEMPORARY FUNCTION IF EXISTS testUDTFExplode")
}
Copy link
Contributor

@yhuai yhuai Oct 1, 2015

Choose a reason for hiding this comment

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

Can we also add tests for GenericUDF and AbstractGenericUDAFResolver (we have tests of UDAF, UDF, and GenericUDTF at here)?


sqlContext.dropTempTable("testUDF")
}
}

class TestPair(x: Int, y: Int) extends Writable with Serializable {
Expand Down