Improve the retry support for nondeterministic expressions

[firestarman]

Contributes to #11649

This PR is trying to address some requirements described in issue #11649, but not all of them.

It introduces two new classes named GpuExpressionRetryable and RetryStateTracker to initially set up a fundamental support for the items 2 and 3 as below, and adds in the relevant unit tests.

2. We need code changes so that if a retry happens on a non-deterministic expression that is outside of a checkpoint/restore, then we fail instead of retrying.
3. We also want a way to detect a non-deterministic expression being run outside of a checkpoint/restore retry block and throw an error from the plan so that when we can have tests validate that we have this covered.

And it also adds the integration tests for the function rand() being used in HashAggregate, Generate, Projection, ArrowEvalPython and Filter. This is for the item 4, and it still does not cover all the cases where a nondeterministic expression can be used, but we are closer than before.

[firestarman]

build

[firestarman]

build

[firestarman]

build

[firestarman]

The failure in CI is a known issue, pls refer to #11790.

[firestarman]

build

[revans2]

Looks like a great first step. I mostly want to understand the 3.5.1 issue around generate and if we need to file a follow on issue.

I would also love to see some rand tests around

• join (we can restrict the range of rand to make it actually match)
• hash aggregate with some distinct operators so expand is called
• (Do we need/want rand in the key for an aggregate?)
• window operations

Then I think we will have covered most of the major cases when this could be called.

[revans2]

Should there be a follow on issue for us to handle this properly?

[firestarman]

Personally we don't need to do anything. It is a pure Spark bug.
Spark 3.5.0 will complain the exception as below when evaluating a rand() in Generate with the codegen disabled and fail this test.

E                   Caused by: java.lang.IllegalArgumentException: requirement failed: Nondeterministic expression org.apache.spark.sql.catalyst.expressions.Rand should be initialized before eval.
E                   	at scala.Predef$.require(Predef.scala:281)
E                   	at org.apache.spark.sql.catalyst.expressions.Nondeterministic.eval(Expression.scala:497)
E                   	at org.apache.spark.sql.catalyst.expressions.Nondeterministic.eval$(Expression.scala:495)
E                   	at org.apache.spark.sql.catalyst.expressions.RDG.eval(randomExpressions.scala:35)
E                   	at org.apache.spark.sql.catalyst.expressions.CreateArray.$anonfun$eval$1(complexTypeCreator.scala:95)
E                   	at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:286)
...
E                   	at org.apache.spark.sql.catalyst.expressions.CreateArray.eval(complexTypeCreator.scala:95)
E                   	at org.apache.spark.sql.catalyst.expressions.ExplodeBase.eval(generators.scala:375)
E                   	at org.apache.spark.sql.execution.GenerateExec.$anonfun$doExecute$8(GenerateExec.scala:108)

And Spark before 3.5.0 will throw the following exception and fail the test too.

E               pyspark.errors.exceptions.captured.AnalysisException: nondeterministic expressions are only allowed in Project, Filter, Aggregate or Window, found:
E               explode(array(rand(42L))),col
E               in operator Generate explode(array(rand(42))), false, [col#2].;
E               Project [col#2]
E               +- Generate explode(array(rand(42))), false, [col#2]
E                  +- LogicalRDD [a#0], false

While GPU supports Generate with rand for all the versions. So to make this test work we have to ignore it for Spark before 3.5.1.

[firestarman]

I would also love to see some rand tests around

• join (we can restrict the range of rand to make it actually match)
• hash aggregate with some distinct operators so expand is called
• (Do we need/want rand in the key for an aggregate?)
• window operations

Thx for review. Would it be ok to add them by a following PR ? And merging this PR will not close the issue #11649.
Here is some investigation on Window and Expand. #11649 (comment)

[firestarman]

@revans2 Could you help review this again? I'v addressed your comments. Thx in advance.