peter-toth commented on code in PR #57371:
URL: https://github.com/apache/spark/pull/57371#discussion_r3771253877
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/generators.scala:
##########
@@ -105,6 +105,8 @@ case class UserDefinedGenerator(
children: Seq[Expression])
extends Generator with CodegenFallback {
+ override def stateful: Boolean = true
Review Comment:
**Finding 3.** Nothing ever fresh-copies a `Generator`, so this override is
inert too.
`GenerateExec` binds the generator once
(`sql/core/src/main/scala/org/apache/spark/sql/execution/GenerateExec.scala:76`)
and calls `boundGenerator.eval(row)` directly (`:104`, `:117`) — no
`freshCopyIfContainsStatefulExpression()` on that path. A generator cannot
reach the other three call sites either: `ExtractGenerator` lifts generators
out of `Project` into `Generate`, so they never sit in a projection or an
ordering, and `Generator.foldable` is hard-coded `false`
(`generators.scala:58`), so `ConstantFolding.tryFold` never sees one.
`inputRow` / `convertToScala` are `@transient var`s that deserialization
resets, so they can only be shared if one instance is evaluated by two threads
in the same JVM. Unlike the `Project` case in `ConvertToLocalRelation`,
`Generate` is not evaluated on the driver, so I don't see the path that would
need protecting here.
`JsonTuple` (also a `Generator`) was marked in SPARK-58205 with the same
inertness, so this is at least consistent with the family. But that makes it
defensive-only: either drop it, or keep it with a comment saying so — and
either way drop the claim that it prevents a race today.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/xml/xpath.scala:
##########
@@ -39,6 +39,8 @@ abstract class XPathExtract
/** XPath expressions are always nullable, e.g. if the xml string is empty.
*/
override def nullable: Boolean = true
+ override def stateful: Boolean = true
Review Comment:
**Finding 1.** This override cannot take effect, and the sharing it targets
survives it.
`XPathExtract` is `RuntimeReplaceable`. `ReplaceExpressions` lives in
`FinishAnalysis`, which is the optimizer's *first* batch
(`Optimizer.scala:177`) and is non-excludable (`Optimizer.scala:307`), so every
`XPathExtract` is rewritten into its `replacement` before anything else runs:
```scala
Invoke(Literal.create(evaluator, ObjectType(classOf[XPathEvaluator])),
"evaluate", ...)
```
All four readers of `stateful` run after that batch —
`ConstantFolding.tryFold`
(`sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala:65`),
`ExpressionsEvaluator.prepareExpressions` (`ExpressionsEvaluator.scala:31`),
`InterpretedOrdering` (`ordering.scala:42`) and
`CodeGenerator.generateExpressions` (`codegen/CodeGenerator.scala:1376`). So no
`XPathExtract` instance is ever passed to
`freshCopyIfContainsStatefulExpression()`.
The state is out of reach even if one were. The mutable object is
`XPathEvaluator.xpathUtil` (`xml/XmlExpressionEvalUtils.scala:53`), wrapping
`UDFXPathUtil` with its shared `builder` / `inputSource` / `reader`.
`replacement` captures that evaluator *by value* into a `Literal`. `Invoke` is
already `stateful = true` (`objects/objects.scala:72`), but a fresh copy keeps
its non-stateful children, so both copies still point at the same evaluator.
This is the same shape as `StructsToJson`, which the already-merged sibling
SPARK-58205 (#57354) deliberately **excluded**, with this note in its
description:
> `StructsToJson` is also `RuntimeReplaceable`; `ReplaceExpressions`
rewrites it to `Invoke(Literal(evaluator), ...)` before any
`freshCopyIfContainsStatefulExpression` runs, making the override dead at
execution. The sharing via `Literal(evaluator)` will be addressed by
SPARK-58208's deep-copy in `QueryExecution.optimizedPlan`.
I'd drop this hunk and add the same exclusion note to the PR description, so
`XPathExtract` is handled by SPARK-58208 alongside `StructsToJson`. If you want
to close it here instead, it has to be fixed where the state lives — e.g. make
`xpathUtil` a `ThreadLocal` in `XPathEvaluator` — not with a flag on the
expression.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CallMethodViaReflectionSuite.scala:
##########
@@ -64,6 +64,17 @@ class CallMethodViaReflectionSuite extends SparkFunSuite
with ExpressionEvalHelp
SQLConf.withExistingConf(conf)(f)
}
+ test("SPARK-58209: reflect expressions are copied before evaluation") {
Review Comment:
**Finding 4.** This is the one hunk that really does something, so it's
worth a test that can tell the difference.
`buffer` (`CallMethodViaReflection.scala:206`) is filled from the children
and then handed to `method.invoke` inside the same `evalInternal`, so two
threads sharing one instance can have thread A invoke with thread B's arguments
— silently wrong results, no exception. And the sharing is reachable:
`ConvertToLocalRelation` (`Optimizer.scala:2698`) builds an
`InterpretedMutableProjection` and evaluates it on the driver during each
derived query's optimization, and `prepareExpressions` is exactly where the
copies get made.
The three assertions here hold for any change that sets the flag, so they
don't separate a fix from a no-op. `ApplyFunctionExpressionSuite` from
SPARK-58578 is a good template for this same buffer-sharing shape: take two
`freshCopyIfContainsStatefulExpression()` copies, evaluate them from two
threads with a `CountDownLatch` so the interleaving is deterministic, and
assert each one sees its own argument. That one fails when only the `stateful`
line is reverted.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/CallMethodViaReflection.scala:
##########
@@ -71,6 +71,8 @@ case class CallMethodViaReflection(
// This could be pretty much anything.
override def expensive: Boolean = true
+ override def stateful: Boolean = true
Review Comment:
**Finding 6.** Worth naming the state, the way the other overrides in
catalyst do (`objects/objects.scala:71` — `// InvokeLike is stateful because of
the evaluatedArgs Array`; `ScalaUDF.scala:62-63`):
```suggestion
// Stateful because of the reusable `buffer` array holding the reflection
arguments.
override def stateful: Boolean = true
```
--
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]
