[SPARK-29777][SparkR] SparkR::cleanClosure aggressively removes a function required by user function

R/pkg/R/utils.R

@@ -546,8 +546,11 @@ processClosure <- function(node, oldEnv, defVars, checkedFuncs, newEnv) {
                 ifelse(identical(func, obj), TRUE, FALSE)
               })
               if (sum(found) > 0) {
-                # If function has been examined, ignore.
-                break
+                # If function has been examined
+                if (identical(parent.env(environment(funcList[found][[1]])), func.env)) {
+                  # If the parent environment is identical to current parent
+                  break
+                }
               }
               # Function has not been examined, record it and recursively clean its closure.
               assign(nodeChar,

R/pkg/tests/fulltests/test_utils.R

@@ -110,6 +110,15 @@ test_that("cleanClosure on R functions", {
   actual <- get("y", envir = env, inherits = FALSE)
   expect_equal(actual, y)
 
+  # Test for combination for nested and sequenctial functions in a closure
+  f1 <- function(x) x + 1
+  f2 <- function(x) f1(x) + 2
+  user_func <- function(x) { f1(x); f2(x) }
+  c_user_func_env <- environment(cleanClosure(user_func))
+  expect_equal(length(c_user_func_env), 2)
+  inner_c_user_func_env <- environment(c_user_func_env$f2)
+  expect_equal(length(inner_c_user_func_env), 1)
+
   # Test for function (and variable) definitions.
   f <- function(x) {
     g <- function(y) { y * 2 }

core/src/main/scala/org/apache/spark/api/r/RRunner.scala

@@ -127,7 +127,7 @@ private[spark] class RRunner[IN, OUT](
           }
         } catch {
           case eof: EOFException =>
-            throw new SparkException("R worker exited unexpectedly (cranshed)", eof)
+            throw new SparkException("R worker exited unexpectedly", eof)
         }
       }
     }