[BUG] udf compiler failed to translate UDF in spark-shell

[wjxiz1992]

Describe the bug
udf compiler throws exceptions when trying to compile a UDF in spark-shell, I've tried 3.3.0 and 3.2.0.

 WARN GpuScalaUDFLogical: Unable to translate UDF <unknown>: java.lang.ClassNotFoundException: $line15.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw

But the unit tests of udf compiler works well if I run e.g.
mvn test -DwildcardSuites=com.nvidia.spark.OpcodeSuite -Dbuildver=330

or I run the unit test directly in the IDEA IDE.

Steps/Code to reproduce bug

start a spark-shell with spark-rapids plugin enabled and udf compiler enabled, then run the following code:

scala> val dataset = List(true, false, true, false).toDS().repartition(1)
dataset: org.apache.spark.sql.Dataset[Boolean] = [value: boolean]

scala>  val myudf: Boolean => Boolean = { x => !x }
myudf: Boolean => Boolean = $Lambda$2902/304271588@7746ae18

scala> val u = udf(myudf)
u: org.apache.spark.sql.expressions.UserDefinedFunction = SparkUserDefinedFunction($Lambda$2902/304271588@7746ae18,BooleanType,List(Some(class[value[0]: boolean])),Some(class[value[0]: boolean]),None,false,true)

scala> val result = dataset.withColumn("new", u('value))
22/03/23 19:12:44 WARN GpuScalaUDFLogical: Unable to translate UDF <unknown>: java.lang.ClassNotFoundException: $line15.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw
22/03/23 19:12:44 WARN GpuScalaUDFLogical: Unable to translate UDF <unknown>: java.lang.ClassNotFoundException: $line15.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw
result: org.apache.spark.sql.DataFrame = [value: boolean, new: boolean]

The code above is the same code in https://github.com/NVIDIA/spark-rapids/blob/branch-22.04/udf-compiler/src/test/scala/com/nvidia/spark/OpcodeSuite.scala#L70-L77.

Expected behavior
The UDF compiler should work well in spark-shell

Environment details (please complete the following information)

• Environment location: local mode
• Spark configuration settings:

        --conf spark.plugins=com.nvidia.spark.SQLPlugin \
        --conf spark.rapids.sql.enabled=true \
        --conf spark.rapids.sql.udfCompiler.enabled=true \
        --conf spark.rapids.cudfVersionOverride=true \
        --conf spark.rapids.sql.variableFloatAgg.enabled=true \
        --conf spark.rapids.sql.exec.CollectLimitExec=true \
        --conf spark.rapids.memory.pinnedPool.size=1G \
        --conf spark.rapids.memory.gpu.allocFraction=0.35 \
        --conf spark.rapids.memory.gpu.maxAllocFraction=0.6 \
        --conf spark.rapids.memory.gpu.pool=ASYNC \
        --conf spark.rapids.sql.explain=ALL \

the jars I'm using:

CUDF_JAR="cudf-22.04.0-20220323.003224-55.jar"
PLUGIN_JAR="rapids-4-spark_2.12-22.04.0-20220323.072303-42.jar"

Additional context
I also tried to build a jar for the code and ran it but failed with the same error.

[abellina]

So it looks like this used to work up until 21.10. It broke in 21.12.

Here's a 21.10 output (I also tried prior versions of the plugin and cuDF):

22/03/23 20:42:32 WARN RapidsPluginUtils: RAPIDS Accelerator 21.10.0-SNAPSHOT using cudf 21.10.0-SNAPSHOT. To disable GPU support set `spark.rapids.sql.enabled` to false
22/03/23 20:42:32 WARN RapidsExecutorPlugin: Ignoring error due to spark.rapids.cudfVersionOverride=true: Found cudf version 21.10.0-SNAPSHOT, RAPIDS Accelerator expects 21.10.0
22/03/23 20:42:38 WARN GpuDeviceManager: Initial RMM allocation (1137.53125 MB) is larger than the adjusted maximum allocation (186.890625 MB), lowering initial allocation to the adjusted maximum allocation.
22/03/23 20:42:39 WARN Plugin: Installing rapids UDF compiler extensions to Spark. The compiler is disabled by default. To enable it, set `spark.rapids.sql.udfCompiler.enabled` to true
Spark context Web UI available at http://worker1:4040
Spark context available as 'sc' (master = local[*], app id = local-1648068152365).
Spark session available as 'spark'.
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 3.1.2
      /_/
         
Using Scala version 2.12.10 (OpenJDK 64-Bit Server VM, Java 1.8.0_312)
Type in expressions to have them evaluated.
Type :help for more information.

scala>  val dataset = List(true, false, true, false).toDS().repartition(1)
dataset: org.apache.spark.sql.Dataset[Boolean] = [value: boolean]

scala> val myudf: Boolean => Boolean = { x => !x }
myudf: Boolean => Boolean = $Lambda$2374/1167527656@54f5457e

scala> val u = udf(myudf)
u: org.apache.spark.sql.expressions.UserDefinedFunction = SparkUserDefinedFunction($Lambda$2374/1167527656@54f5457e,BooleanType,List(Some(class[value[0]: boolean])),Some(class[value[0]: boolean]),None,false,true)

scala>  val result = dataset.withColumn("new", u('value))
result: org.apache.spark.sql.DataFrame = [value: boolean, new: boolean]

scala> result.explain
22/03/23 20:42:59 WARN GpuOverrides: 
*Exec <ProjectExec> will run on GPU
  *Expression <Alias> if (NOT value#1) true else false AS new#8 will run on GPU
    *Expression <If> if (NOT value#1) true else false will run on GPU
      *Expression <Not> NOT value#1 will run on GPU
  *Exec <ShuffleExchangeExec> will run on GPU
    *Partitioning <RoundRobinPartitioning> will run on GPU
    !NOT_FOUND <LocalTableScanExec> cannot run on GPU because no GPU enabled version of operator class org.apache.spark.sql.execution.LocalTableScanExec could be found
      @Expression <AttributeReference> value#1 could run on GPU

== Physical Plan ==
GpuColumnarToRow false
+- GpuProject [value#1, if (NOT value#1) true else false AS new#8]
   +- GpuShuffleCoalesce 2147483647
      +- GpuColumnarExchange gpuroundrobinpartitioning(1), REPARTITION_WITH_NUM, [id=#35]
         +- GpuRowToColumnar targetsize(2147483647)
            +- LocalTableScan [value#1]

[abellina]

Filed the PR: #5030 that reverts a change that was put into 21.12 that was causing this.