spark-test commented on code in PR #59881: URL: https://github.com/apache/spark/pull/59881#discussion_r4406295078
########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/misc.scala: ########## @@ -318,6 +318,11 @@ case class Uuid(randomSeed: Option[Long] = None) extends LeafExpression with Non override def stateful: Boolean = true + // Stateful fresh copies use this hook even for leaves. The default leaf hook returns this, + // which would share randomGenerator across evaluators; copy preserves the seed without + // sharing the initialized generator. + override def withNewChildrenInternal(newChildren: IndexedSeq[Expression]): Expression = copy() Review Comment: **Blocking (P1):** Returning `copy()` here means `freshCopyIfContainsStatefulExpression` now gives each caller a new `Uuid` whose `initialized` flag is false. That works for evaluators that initialize their own prepared copies. It breaks the two `CodegenFallback` generators that evaluate their children through a nested `InterpretedProjection` they never initialize: `UserDefinedGenerator`, which classic `Dataset.explode(Column*)` builds, and `HiveGenericUDTF`. `GenerateExec.doExecute` initializes the Nondeterministic nodes of `boundGenerator` in place and then calls `eval` (GenerateExec.scala:81-85, 104). `UserDefinedGenerator.eval` then lazily builds `new InterpretedProjection(children)` (generators.scala:111-126), and `HiveGenericUDTF` does the same through `inputProjection` (hiveUDFs.scala:263-268). That projection fresh-copies stateful children in its constructor. Until now the "copy" of a `Uuid` was the same instance `GenerateExec` had just initialized. With this override it is a fresh, uninitialized instance, so `Nondeterministic.eval` fails its `require(initialized)` check. For example, on classic Spark: ```scala spark.range(3).explode(uuid())(r => Seq(Tuple1(r.getString(0)))).collect() ``` On master this returns three rows. With this change, every task fails with `IllegalArgumentException: requirement failed: Nondeterministic expression org.apache.spark.sql.catalyst.expressions.Uuid should be initialized before eval.` A Hive GenericUDTF whose argument contains `uuid()`, such as `array(uuid())` in a SELECT list or LATERAL VIEW, fails the same way. No configuration restores the old behavior; users would have to compute `uuid()` in an earlier projection. The same gap already breaks `rand()` and `monotonically_increasing_id()` arguments to these generators on master, because their fresh copies were already new instances. This change extends the failure to `uuid()`, which used to work. Ordinary projections and filters, generated code, and codegen-capable generators such as `explode()` are unaffected. The new `MiscExpressionsSuite` cases call `freshCopyIfContainsStatefulExpression`, `initialize`, and `eval` directly, and no test evaluates `uuid()` through a consumer of this hook, so the suite does not catch the regression. **Recommended change:** Make UserDefinedGenerator and HiveGenericUDTF evaluate the child expressions that GenerateExec has already initialized for the partition, instead of fresh copies held by a nested InterpretedProjection that is never initialized. Build the input row by evaluating the generator's children directly, or through an equivalent helper that does not copy stateful expressions. Keep the new Uuid copy hook as it is. Add Dataset explode tests and Hive UDTF tests that pass uuid() into each generator. **Why this works:** GenerateExec already initializes every Nondeterministic node of the bound generator with the partition index before calling eval. When the generator evaluates those same nodes, every uuid() argument is initialized before use and produces the same seed-plus-partition-index sequence the shared instance produced before this PR. rand() and monotonically_increasing_id() arguments, which already fail inside these generators before this PR because their fresh copies are new uninitialized instances, are likewise initialized before use. Uuid copies keep independent generators wherever Spark builds a separate evaluator. **Scope:** Change how the two CodegenFallback generators evaluate their children, and add end-to-end regression coverage for a uuid() argument through each generator. **Compatibility:** Uuid's per-copy semantics from this PR, and the deterministic seed-plus-partition-index sequence each initialized evaluator produces. **Risks:** The input row the generators build must stay identical, including null child values and the Scala and UDT conversion UserDefinedGenerator applies, so the user function and the Hive UDTF see exactly the inputs they saw before. Direct evaluation relies on each task owning its own deserialized bound generator. If a generator instance were ever evaluated concurrently from more than one partition, its children would share state, so this must stay true. **Constraints:** Keep Uuid's per-copy withNewChildrenInternal behavior from this PR and the seed-plus-partition-index sequence each initialized evaluator produces. Leave GenerateExec's whole-stage codegen path and the codegen-capable generators unchanged. **Exclusions:** - Pre-existing initialization gaps in evaluators that never initialize nondeterministic inputs are out of scope: script transformation, imperative aggregate input projections, and the right-side evaluators of InterpretedOrdering. They behave the same before and after this PR. **Success:** A classic Dataset.explode call whose input contains uuid() returns rows instead of failing, with the values the merge base produced for the same seed and partition. A Hive GenericUDTF whose argument contains uuid() returns rows when used in a SELECT list or a LATERAL VIEW. Fresh copies of Uuid built for separate evaluators still keep the seed and own independent random generators. **Verification:** - **Regression:** Passing uuid() as a classic Dataset.explode input returns the expected rows instead of failing the requirement that a nondeterministic expression be initialized before eval. - **Regression:** A Hive GenericUDTF whose argument contains uuid() returns rows instead of failing the requirement that a nondeterministic expression be initialized before eval. - **Compatibility:** For a fixed seed and partition, the uuid() values a generator receives equal the values of a Uuid with that seed initialized for that partition, which is what the merge base produced. - **Behavior:** Fresh copies of Uuid still have distinct identities, keep the seed, and advance and re-initialize independently of one another. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
