[1.3.0] Error when calling UDF that throws an exception

Open

#8,724 opened on Feb 13, 2025 5 comments 1 reaction 0 assignees View on GitHub

bug triage

Dominant language: Scala
Stars: 1.6k
Forks: 657
Avg merge: 2d 14h
Merged PRs (30d): 80

Description

### Backend

VL (Velox)

### Bug description

[Expected behavior] and [actual behavior].
When we took Gluten 1.2.1, Gluten could call a Java UDF that (correctly) threw an exception, and handle it gracefully. We tried to take 1.3.0, and it gives an error `org.apache.spark.SparkException: [FAILED_EXECUTE_UDF] Failed to execute user defined function`

I ran into this while running integration tests in a more complex setup and that stacktrace included the error below. But I believe it wasn't actually a memory leak, because it would hang for a few seconds before that error. The expected behavior is that after the exception is thrown, the code to catch the exception executes successfully (and the test passes). In 1.3.0, it stops after the exception is thrown.

`project/ep/build-velox/build/velox_ep/velox/exec/Driver.cpp:601, Function:operator(), Expression: Operator::getOutput failed for [operator: ValueStream, plan node ID: 0]: Error during calling Java code from native code: org.apache.spark.SparkException: [FAILED_EXECUTE_UDF] Failed to execute user defined function (`functions$$$Lambda/0x00007f0049900ba0`: (string, string) => boolean).`
...

`, Source: RUNTIME, ErrorCode: INVALID_STATE
E20250211 22:55:13.386138 19912 VeloxMemoryManager.cc:401] Failed to release Velox memory manager after 43350ms as there are still outstanding memory resources.
E20250211 22:55:13.386179 19912 MemoryPool.cpp:442] [MEM] Memory leak (Used memory): Memory Pool[default_leaf LEAF root[root] parent[root] MALLOC track-usage thread-safe]
E20250211 22:55:13.386235 19912 Exceptions.h:66] Line: /home/circleci/project/cpp/velox/memory/VeloxMemoryManager.cc:102, Function:removePool, Expression: pool->reservedBytes() == 0 (1048576 vs. 0), Source: RUNTIME, ErrorCode: INVALID_STATE
terminate called after throwing an instance of 'facebook::velox::VeloxRuntimeError'`

I think this is a minimal repro. It's TPCDS q9.

```
export gluten_jar=~/gluten-velox-bundle-spark3.5_2.12-centos_7_x86_64-1.3.0.jar
export SPARK_HOME=~/spark-3.5.4-bin-hadoop3
export PATH=$PATH:/$SPARK_HOME/bin

spark-shell --conf spark.plugins=org.apache.gluten.GlutenPlugin --conf spark.memory.offHeap.enabled=true --conf spark.memory.offHeap.size=20g --conf spark.driver.extraClassPath=${gluten_jar} --conf spark.executor.extraClassPath=${gluten_jar} --conf spark.shuffle.manager=org.apache.spark.shuffle.sort.ColumnarShuffleManager --jars ${gluten_jar} --conf spark.gluten.enabled=true --conf spark.driver.extraJavaOptions="-Dio.netty.tryReflectionSetAccessible=true"

scala> val storeSalesDF = spark.read.format("csv").option("header", "true").load("/home/coder/store_sales.csv")
scala> val reasonsDF = spark.read.format("csv").option("header", "true").load("/home/coder/reason.csv")
scala> storeSalesDF.createOrReplaceTempView("store_sales")
scala> reasonsDF.createOrReplaceTempView("reason")

scala> val resultDF = spark.sql("""

| | SELECT CASE

| | WHEN (SELECT Count(*)

| | FROM store_sales

| | WHERE ss_quantity BETWEEN 1 AND 20) > 3672 THEN

| | (SELECT Avg(ss_ext_list_price)

| | FROM store_sales

| | WHERE ss_quantity BETWEEN 1 AND 20)

| | ELSE (SELECT Avg(ss_net_profit)

| | FROM store_sales

| | WHERE ss_quantity BETWEEN 1 AND 20)

| | END bucket1,

| | CASE

| | WHEN (SELECT Count(*)

| | FROM store_sales

| | WHERE ss_quantity BETWEEN 21 AND 40) > 3392 THEN

| | (SELECT Avg(ss_ext_list_price)

| | FROM store_sales

| | WHERE ss_quantity BETWEEN 21 AND 40)

| | ELSE (SELECT Avg(ss_net_profit)

| | FROM store_sales

| | WHERE ss_quantity BETWEEN 21 AND 40)

| | END bucket2,

| | CASE

| | WHEN (SELECT Count(*)

| | FROM store_sales

| | WHERE ss_quantity BETWEEN 41 AND 60) > 32784 THEN

| | (SELECT Avg(ss_ext_list_price)

| | FROM store_sales

| | WHERE ss_quantity BETWEEN 41 AND 60)

| | ELSE (SELECT Avg(ss_net_profit)

| | FROM store_sales

| | WHERE ss_quantity BETWEEN 41 AND 60)

| | END bucket3,

| | CASE

| | WHEN (SELECT Count(*)

| | FROM store_sales

| | WHERE ss_quantity BETWEEN 61 AND 80) > 26032 THEN

| | (SELECT Avg(ss_ext_list_price)

| | FROM store_sales

| | WHERE ss_quantity BETWEEN 61 AND 80)

| | ELSE (SELECT Avg(ss_net_profit)

| | FROM store_sales

| | WHERE ss_quantity BETWEEN 61 AND 80)

| | END bucket4,

| | CASE

| | WHEN (SELECT Count(*)

| | FROM store_sales

| | WHERE ss_quantity BETWEEN 81 AND 100) > 23982 THEN

| | (SELECT Avg(ss_ext_list_price)

| | FROM store_sales

| | WHERE ss_quantity BETWEEN 81 AND 100)

| | ELSE (SELECT Avg(ss_net_profit)

| | FROM store_sales

| | WHERE ss_quantity BETWEEN 81 AND 100)

| | END bucket5

| | FROM reason

| | WHERE r_reason_sk = 1

| | """)

scala> def throwException() = {throw new IllegalStateException("test error")}
throwException: ()Nothing

scala> val throwExceptionUdf = udf(throwException _)
throwExceptionUdf: org.apache.spark.sql.expressions.UserDefinedFunction = SparkUserDefinedFunction($Lambda$6066/0x00007fa6ed56a3f8@1c650d81,NullType,List(),Some(class[value[0]: void]),None,true,true)

scala> val resultDF2 = resultDF.withColumn("exception", throwExceptionUdf())
resultDF2: org.apache.spark.sql.DataFrame = [bucket1: double, bucket2: double ... 4 more fields]

scala> resultDF2.show()

```

### Spark version

None

### Spark configurations

Spark 3.5.4

I downloaded Spark and Gluten from https://gluten.apache.org/#31-use-a-pre-built-jar

`spark-shell --conf spark.plugins=org.apache.gluten.GlutenPlugin --conf spark.memory.offHeap.enabled=true --conf spark.memory.offHeap.size=20g --conf spark.driver.extraClassPath=${gluten_jar} --conf spark.executor.extraClassPath=${gluten_jar} --conf spark.shuffle.manager=org.apache.spark.shuffle.sort.ColumnarShuffleManager --jars ${gluten_jar} --conf spark.gluten.enabled=true --conf spark.driver.extraJavaOptions="-Dio.netty.tryReflectionSetAccessible=true"`

### System information

Ubuntu 20.04

### Relevant logs

```bash
25/02/13 02:46:51 WARN ProjectExecTransformer: Validation failed with exception for plan: ProjectExecTransformer, due to: Not supported to map spark function name to substrait function name: toprettystring(CASE WHEN (Subquery subquery#86, [id=#1775].count(1) > 3672) THEN ReusedSubquery Subquery subquery#86, [id=#1775].avg(ss_ext_list_price) ELSE ReusedSubquery Subquery subquery#86, [id=#1775].avg(ss_net_profit) END, Some(Etc/UTC)), class name: ToPrettyString.

25/02/13 02:46:51 WARN ProjectExecTransformer: Validation failed with exception for plan: ProjectExecTransformer, due to: Not supported to map spark function name to substrait function name: toprettystring(CASE WHEN (Subquery subquery#86, [id=#1775].count(1) > 3672) THEN ReusedSubquery Subquery subquery#86, [id=#1775].avg(ss_ext_list_price) ELSE ReusedSubquery Subquery subquery#86, [id=#1775].avg(ss_net_profit) END, Some(Etc/UTC)), class name: ToPrettyString.

25/02/13 02:46:51 WARN GlutenFallbackReporter: Validation failed for plan: Scan csv [QueryId=5], due to: Unsupported file format TextReadFormat..

25/02/13 02:46:51 WARN GlutenFallbackReporter: Validation failed for plan: Project[QueryId=5], due to: Not supported to map spark function name to substrait function name: toprettystring(CASE WHEN (Subquery subquery#86, [id=#1775].count(1) > 3672) THEN ReusedSubquery Subquery subquery#86, [id=#1775].avg(ss_ext_list_price) ELSE ReusedSubquery Subquery subquery#86, [id=#1775].avg(ss_net_profit) END, Some(Etc/UTC)), class name: ToPrettyString..

25/02/13 02:46:51 WARN GlutenFallbackReporter: Validation failed for plan: Scan csv [QueryId=5], due to: Unsupported file format TextReadFormat..

25/02/13 02:46:52 ERROR TaskResources: Task 23 failed by error:

org.apache.spark.SparkException: [FAILED_EXECUTE_UDF] Failed to execute user defined function (`$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$Lambda$6066/0x00007fa6ed56a3f8`: () => void).

at org.apache.spark.sql.errors.QueryExecutionErrors$.failedExecuteUserDefinedFunctionError(QueryExecutionErrors.scala:198)

at org.apache.spark.sql.errors.QueryExecutionErrors.failedExecuteUserDefinedFunctionError(QueryExecutionErrors.scala)

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.WholeStageCodegenEvaluatorFactory$WholeStageCodegenPartitionEvaluator$$anon$1.hasNext(WholeStageCodegenEvaluatorFactory.scala:43)

at org.apache.spark.sql.execution.SparkPlan.$anonfun$getByteArrayRdd$1(SparkPlan.scala:388)

at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsInternal$2(RDD.scala:893)

at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsInternal$2$adapted(RDD.scala:893)

at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)

at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:367)

at org.apache.spark.rdd.RDD.iterator(RDD.scala:331)

at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:93)

at org.apache.spark.TaskContext.runTaskWithListeners(TaskContext.scala:166)

at org.apache.spark.scheduler.Task.run(Task.scala:141)

at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$4(Executor.scala:620)

at org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally(SparkErrorUtils.scala:64)

at org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally$(SparkErrorUtils.scala:61)

at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:94)

at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:623)

at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)

at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)

at java.base/java.lang.Thread.run(Thread.java:840)

Caused by: java.lang.IllegalStateException: test error

at $line24.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.throwException(:22)

at $line25.$read$$iw$$iw$$iw$$iw$$iw$$iw$$iw$$iw.$anonfun$throwExceptionUdf$1(:23)

... 20 more

E20250213 02:46:52.092774 97905 Exceptions.h:66] Line: /root/src/weiting/gluten/ep/build-velox/build/velox_ep/velox/exec/Task.cpp:2039, Function:terminate, Expression: Cancelled, Source: RUNTIME, ErrorCode: INVALID_STATE

25/02/13 02:46:52 ERROR Executor: Exception in task 0.0 in stage 33.0 (TID 23)