Fix regression from 21.12 where udfs defined in repl no longer worked

[abellina]

Signed-off-by: Alessandro Bellina [email protected]

Closes #5019.

Reverts a localized change to the udf-compiler introduced with this PR (#3726).

With ShimLoader.loadClass, we couldn't resolve a UDF defined in the repl, as documented here: #5019.

[abellina]

build

[wjxiz1992]

Should we add some tests for such case?

[abellina]

Should we add some tests for such case?

I am not sure this is worth it, and honestly I do not know how to replicate the repl in a test. I guess I could define a lambda in a different class loader, but then again I am not sure if it is wortwhile.

[gerashegalov]

LGTM
Would be great if we can also verify a non-REPL use of Scala UDF in a Spark Job manually. We don't have Scala integration tests.

[abellina]

@gerashegalov

Would be great if we can also verify a non-REPL use of Scala UDF in a Spark Job manually. We don't have Scala integration tests.

Sorry it took me a little while. We do have the OpcodeSuite unit tests that do cover all the translations available (a spark session is created there). I also create a simple app, and submitted it using spark-submit against a standalone cluster:

package com.nvidia.spark.udf

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.udf

object SimpleApp {
  def main(args: Array[String]) {
    val spark = SparkSession.builder.appName("Simple Application").getOrCreate()
    import spark.implicits._
    val dataset = List(true, false, true, false).toDS().repartition(1)
    val myudf: Boolean => Boolean = { x => !x }
    val u = udf(myudf)

    var result = dataset.withColumn("new", u('value))
    result.explain(true)
    result.show()

    spark.conf.set("spark.rapids.sql.udfCompiler.enabled", "true")
    result = dataset.withColumn("new", u('value))
    result.explain(true)

    result.show()
    spark.stop()
  }
}

With the plugin disabled:

== Physical Plan ==
*(1) Project [value#1, UDF(value#1) AS new#8]
+- Exchange RoundRobinPartitioning(1), REPARTITION_WITH_NUM, [id=#9]
   +- LocalTableScan [value#1]

With the plugin enabled:

== Physical Plan ==
*(1) Project [value#1, if (NOT value#1) true else false AS new#23]
+- Exchange RoundRobinPartitioning(1), REPARTITION_WITH_NUM, [id=#41]
   +- LocalTableScan [value#1]

[abellina]

build