[SPARK-45171][SQL] Initialize non-deterministic expressions in `Gener…

…ateExec`

### What changes were proposed in this pull request?

Before evaluating the generator function in `GenerateExec`, initialize non-deterministic expressions.

### Why are the changes needed?

The following query fails:
```
select *
from explode(
  transform(sequence(0, cast(rand()*1000 as int) + 1), x -> x * 22)
);

23/09/14 09:27:25 ERROR Executor: Exception in task 0.0 in stage 3.0 (TID 3)
java.lang.IllegalArgumentException: requirement failed: Nondeterministic expression org.apache.spark.sql.catalyst.expressions.Rand should be initialized before eval.
	at scala.Predef$.require(Predef.scala:281)
	at org.apache.spark.sql.catalyst.expressions.Nondeterministic.eval(Expression.scala:497)
	at org.apache.spark.sql.catalyst.expressions.Nondeterministic.eval$(Expression.scala:495)
	at org.apache.spark.sql.catalyst.expressions.RDG.eval(randomExpressions.scala:35)
	at org.apache.spark.sql.catalyst.expressions.BinaryArithmetic.eval(arithmetic.scala:384)
	at org.apache.spark.sql.catalyst.expressions.UnaryExpression.eval(Expression.scala:543)
	at org.apache.spark.sql.catalyst.expressions.BinaryArithmetic.eval(arithmetic.scala:384)
	at org.apache.spark.sql.catalyst.expressions.Sequence.eval(collectionOperations.scala:3062)
	at org.apache.spark.sql.catalyst.expressions.SimpleHigherOrderFunction.eval(higherOrderFunctions.scala:275)
	at org.apache.spark.sql.catalyst.expressions.SimpleHigherOrderFunction.eval$(higherOrderFunctions.scala:274)
	at org.apache.spark.sql.catalyst.expressions.ArrayTransform.eval(higherOrderFunctions.scala:308)
	at org.apache.spark.sql.catalyst.expressions.ExplodeBase.eval(generators.scala:375)
	at org.apache.spark.sql.execution.GenerateExec.$anonfun$doExecute$8(GenerateExec.scala:108)
...
```
However, this query succeeds:
```
select *
from explode(
  sequence(0, cast(rand()*1000 as int) + 1)
);

0
1
2
3
...
801
802
803
```
The difference is that `transform` turns off whole-stage codegen, which exposes a bug in `GenerateExec` in which the non-deterministic expression passed to the generator function is not initialized before being used.

This PR fixes the bug in `GenerateExec`.

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

New unit test.

### Was this patch authored or co-authored using generative AI tooling?

No.

Closes #42933 from bersprockets/nondeterm_issue.

Lead-authored-by: Bruce Robbins <[email protected]>
Co-authored-by: Hyukjin Kwon <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>

sql/core/src/main/scala/org/apache/spark/sql/execution/GenerateExec.scala

@@ -78,6 +78,10 @@ case class GenerateExec(
     // boundGenerator.terminate() should be triggered after all of the rows in the partition
     val numOutputRows = longMetric("numOutputRows")
     child.execute().mapPartitionsWithIndexInternal { (index, iter) =>
+      boundGenerator.foreach {
+        case n: Nondeterministic => n.initialize(index)
+        case _ =>
+      }
       val generatorNullRow = new GenericInternalRow(generator.elementSchema.length)
       val rows = if (requiredChildOutput.nonEmpty) {
 

sql/core/src/test/scala/org/apache/spark/sql/GeneratorFunctionSuite.scala

@@ -536,6 +536,13 @@ class GeneratorFunctionSuite extends QueryTest with SharedSparkSession {
     checkAnswer(df,
       Row(1, 1) :: Row(1, 2) :: Row(2, 2) :: Row(2, 3) :: Row(3, null) :: Nil)
   }
+
+  test("SPARK-45171: Handle evaluated nondeterministic expression") {
+    withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "false") {
+      val df = sql("select explode(array(rand(0)))")
+      checkAnswer(df, Row(0.7604953758285915d))
+    }
+  }
 }
 
 case class EmptyGenerator() extends Generator with LeafLike[Expression] {