[SPARK-32238][SQL] Use Utils.getSimpleName to avoid hitting Malformed class name in ScalaUDF

[Ngone51]

What changes were proposed in this pull request?

This PR proposes to use Utils.getSimpleName(function) instead of function.getClass.getSimpleName to get the class name.

Why are the changes needed?

For some functions(see the demo below), using function.getClass.getSimpleName can hit "Malformed class name" error.

Does this PR introduce any user-facing change?

Yes. For the demo,

 object MalformedClassObject extends Serializable {
    class MalformedNonPrimitiveFunction extends (String => Int) with Serializable {
      override def apply(v1: String): Int = v1.toInt / 0
    }
  }
 OuterScopes.addOuterScope(MalformedClassObject)
 val f = new MalformedClassObject.MalformedNonPrimitiveFunction()
 Seq("20").toDF("col").select(udf(f).apply(Column("col"))).collect()

Before this PR, user can only see the error about "Malformed class name":

An exception or error caused a run to abort: Malformed class name 
java.lang.InternalError: Malformed class name
	at java.lang.Class.getSimpleName(Class.java:1330)
	at org.apache.spark.sql.catalyst.expressions.ScalaUDF.udfErrorMessage$lzycompute(ScalaUDF.scala:1157)
	at org.apache.spark.sql.catalyst.expressions.ScalaUDF.udfErrorMessage(ScalaUDF.scala:1155)
	at org.apache.spark.sql.catalyst.expressions.ScalaUDF.doGenCode(ScalaUDF.scala:1077)
	at org.apache.spark.sql.catalyst.expressions.Expression.$anonfun$genCode$3(Expression.scala:147)
	at scala.Option.getOrElse(Option.scala:189)
	at org.apache.spark.sql.catalyst.expressions.Expression.genCode(Expression.scala:142)
	at org.apache.spark.sql.catalyst.expressions.Alias.genCode(namedExpressions.scala:160)
	at org.apache.spark.sql.execution.ProjectExec.$anonfun$doConsume$1(basicPhysicalOperators.scala:69)
        ...

After this PR, user can see the real root cause of the udf failure:

org.apache.spark.SparkException: Failed to execute user defined function(UDFSuite$MalformedClassObject$MalformedNonPrimitiveFunction: (string) => int)
	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)
	at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
	at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:753)
	at org.apache.spark.sql.execution.SparkPlan.$anonfun$getByteArrayRdd$1(SparkPlan.scala:340)
	at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsInternal$2(RDD.scala:898)
	at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsInternal$2$adapted(RDD.scala:898)
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:337)
	at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
	at org.apache.spark.scheduler.Task.run(Task.scala:127)
	at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:464)
	at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1377)
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:467)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.ArithmeticException: / by zero
	at org.apache.spark.sql.UDFSuite$MalformedClassObject$MalformedNonPrimitiveFunction.apply(UDFSuite.scala:677)
	at org.apache.spark.sql.UDFSuite$MalformedClassObject$MalformedNonPrimitiveFunction.apply(UDFSuite.scala:676)
	... 17 more

How was this patch tested?

Added a test.

[Ngone51]

cc @cloud-fan @maropu @viirya Please take a look, thanks!

[SparkQA]

Test build #125451 has finished for PR 29050 at commit 08749bd.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
• test(\"SPARK-32238: Use Utils.getSimpleName to avoid hitting Malformed class name\")

[maropu]

Nice catch!

[SparkQA]

Test build #125447 has finished for PR 29050 at commit 4117f99.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[maropu]

retest this please

[dongjoon-hyun]

Hi, @Ngone51 . SPARK-32238 is filed as Improvement. If you want to backport this to 3.0.1/2.4.7, please update it to Bug, please.

[Ngone51]

@dongjoon-hyun Ok, I've changed it to Bug. Thanks!

[SparkQA]

Test build #125533 has finished for PR 29050 at commit 08749bd.

• This patch fails due to an unknown error code, -9.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
• test(\"SPARK-32238: Use Utils.getSimpleName to avoid hitting Malformed class name\")

[maropu]

retest this please

[SparkQA]

Test build #125583 has started for PR 29050 at commit 08749bd.

[dongjoon-hyun]

+1, LGTM. Thank you, @Ngone51 .

[dongjoon-hyun]

Retest this please.

[SparkQA]

Test build #125631 has finished for PR 29050 at commit 08749bd.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
• test(\"SPARK-32238: Use Utils.getSimpleName to avoid hitting Malformed class name\")

[dongjoon-hyun]

Retest this please.

[SparkQA]

Test build #125641 has finished for PR 29050 at commit 08749bd.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
• test(\"SPARK-32238: Use Utils.getSimpleName to avoid hitting Malformed class name\")

[dongjoon-hyun]

Thank you, @Ngone51 and all.
Merged to master/3.0/2.4.

[Ngone51]

Thanks all!