This is an automated email from the ASF dual-hosted git repository.
peter-toth pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new e39918d47128 [SPARK-58209][SQL] Mark CallMethodViaReflection and
XPathExtract stateful to prevent data races
e39918d47128 is described below
commit e39918d47128d8d82b7404cfb152b528afe721e2
Author: Anshul Baliga <[email protected]>
AuthorDate: Thu Oct 1 16:05:08 2026 +0200
[SPARK-58209][SQL] Mark CallMethodViaReflection and XPathExtract stateful
to prevent data races
### What changes were proposed in this pull request?
Mark `CallMethodViaReflection` and `XPathExtract` as `stateful` so that
`freshCopyIfContainsStatefulExpression()` gives each evaluation its own
instance, with a comment on each naming the protected state:
- `CallMethodViaReflection`: the reusable `buffer` array that
`evalInternal` fills with the argument values before handing it to the
reflective call.
- `XPathExtract`: the `evaluator`, whose `UDFXPathUtil` reuses one parser
and reader across calls and is documented as one-per-instance. The expression
is `RuntimeReplaceable`, but the fresh-copy site added by SPARK-58208
(`QueryExecution.cloneWithFreshStatefulExpressions`) runs on the analyzed plan,
before `ReplaceExpressions`, and `replacement` reads the copy's own lazy
`evaluator`, so a fresh copy really does get a fresh `UDFXPathUtil`.
Tests added:
- `CallMethodViaReflectionSuite`: the expression reports `stateful` and a
fresh copy is a distinct instance that still evaluates; and a two-thread test
in which the first evaluation blocks in its second argument, after writing its
first argument into the buffer, while a second evaluation runs to completion.
On a shared instance the first call returns `m20` instead of `m10`; with fresh
copies both results are correct.
- `XPathExpressionSuite`: every `XPathExtract` subclass reports `stateful`,
a fresh copy is a distinct instance, and the copy's `replacement` carries a
different evaluator object than the original's; the copy still evaluates
correctly.
This supersedes #57371 by huanghanyu-vungle. Of the three expressions named
in the JIRA, `UserDefinedGenerator` is left alone: the fresh-copy site does map
`Generate.generator`, but `ConvertToLocalRelation` does not handle `Generate`
and `GenerateExec` evaluates a per-task deserialized copy, so no two threads
ever share one instance.
### Why are the changes needed?
Sub-task of SPARK-58154. Both expressions keep per-instance mutable state
but were not marked stateful, so the fresh-copy mechanism skipped them. When
one instance is evaluated from two threads at once, for example sibling
DataFrames derived from a common parent whose projections over a
`LocalRelation` are evaluated on the driver by `ConvertToLocalRelation`, the
shared state is corrupted: reflection arguments are swapped, and the shared XML
parser throws or returns wrong values. On th [...]
### Does this PR introduce _any_ user-facing change?
No. Concurrent evaluation of the same `reflect` / `java_method` or
`xpath_*` call now returns the correct result instead of racing;
single-threaded behavior is unchanged.
### How was this patch tested?
```
build/sbt "catalyst/testOnly
org.apache.spark.sql.catalyst.expressions.CallMethodViaReflectionSuite
org.apache.spark.sql.catalyst.expressions.xml.*"
```
`CallMethodViaReflectionSuite`: 13 tests pass (11 existing, 2 new);
removing the `stateful` override makes the two-thread test fail
deterministically with `m20 did not equal m10`.
`XPathExpressionSuite`: 10 tests pass (9 existing, 1 new); removing the
override makes the new test fail on the shared-evaluator assertion.
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #58999 from
anshulbaliga7/SPARK-58209-call-method-via-reflection-stateful.
Authored-by: Anshul Baliga <[email protected]>
Signed-off-by: Peter Toth <[email protected]>
---
.../expressions/CallMethodViaReflection.scala | 5 ++
.../spark/sql/catalyst/expressions/xml/xpath.scala | 6 ++
.../expressions/CallMethodViaReflectionSuite.scala | 75 +++++++++++++++++++++-
.../expressions/xml/XPathExpressionSuite.scala | 23 +++++++
4 files changed, 108 insertions(+), 1 deletion(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/CallMethodViaReflection.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/CallMethodViaReflection.scala
index b9a51a8f4025..cd6425d6393d 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/CallMethodViaReflection.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/CallMethodViaReflection.scala
@@ -216,6 +216,11 @@ case class CallMethodViaReflection(
/** A temporary buffer used to hold intermediate results returned by
children. */
@transient private lazy val buffer = new Array[Object](argExprs.length)
+ // Stateful because of the reusable `buffer` holding the reflection
arguments: it is filled by
+ // `evalInternal` and then handed to the call, so two evaluations sharing
one instance would
+ // overwrite each other's arguments. Fresh copies give every evaluation its
own buffer.
+ override def stateful: Boolean = true
+
override protected def withNewChildrenInternal(
newChildren: IndexedSeq[Expression]): CallMethodViaReflection =
copy(children = newChildren)
}
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/xml/xpath.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/xml/xpath.scala
index 0e1cafbb24c0..9c97e4599d2d 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/xml/xpath.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/xml/xpath.scala
@@ -67,6 +67,12 @@ abstract class XPathExtract
protected def evaluator: XPathEvaluator
+ // Stateful because of the `evaluator`, whose `UDFXPathUtil` reuses one
parser and reader
+ // across calls and is documented as one-per-instance. The fresh-copy site
on the analyzed plan
+ // sees this expression before `ReplaceExpressions` runs, and `replacement`
reads the copy's own
+ // lazy `evaluator`, so a fresh copy gives each concurrent evaluation its
own `UDFXPathUtil`.
+ override def stateful: Boolean = true
+
override def replacement: Expression = Invoke(
Literal.create(evaluator, ObjectType(classOf[XPathEvaluator])),
"evaluate",
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CallMethodViaReflectionSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CallMethodViaReflectionSuite.scala
index bd7aed831ae0..9d56fe514fad 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CallMethodViaReflectionSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CallMethodViaReflectionSuite.scala
@@ -18,16 +18,23 @@
package org.apache.spark.sql.catalyst.expressions
import java.sql.Timestamp
+import java.util.concurrent.{CountDownLatch, TimeUnit}
+
+import scala.concurrent.{ExecutionContext, Future}
+import scala.concurrent.duration._
import org.apache.spark.{SPARK_DOC_ROOT, SparkFunSuite,
SparkIllegalArgumentException}
import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.DataTypeMismatch
import org.apache.spark.sql.catalyst.expressions.Cast.toSQLType
-import
org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
+import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenFallback,
GenerateUnsafeProjection}
import org.apache.spark.sql.catalyst.util.QuotingUtils.toSQLConf
import org.apache.spark.sql.catalyst.util.TypeUtils.ordinalNumber
import org.apache.spark.sql.internal.{SQLConf, StaticSQLConf}
import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+import org.apache.spark.util.ThreadUtils
/** A static class for testing purpose. */
object ReflectStaticClass {
@@ -35,6 +42,29 @@ object ReflectStaticClass {
def method2(v1: Int): String = "m" + v1
def method3(v1: java.lang.Integer): String = "m" + v1
def method4(v1: Int, v2: String): String = "m" + v1 + v2
+ def method5(v1: Int, v2: Int): String = "m" + v1 + v2
+}
+
+/**
+ * An argument whose evaluation for the row with id 1 blocks until the row
with id 2 has been
+ * evaluated. Placed after another argument, it holds a
`CallMethodViaReflection` evaluation
+ * between filling its argument buffer and making the call, so a buffer shared
with a concurrent
+ * evaluation is observable.
+ */
+case class ReflectBlockingArgument(
+ firstEvaluationStarted: CountDownLatch,
+ secondEvaluationStarted: CountDownLatch) extends LeafExpression with
CodegenFallback {
+ override def nullable: Boolean = false
+ override def dataType: DataType = IntegerType
+ override def eval(input: InternalRow): Any = {
+ if (input.getInt(0) == 1) {
+ firstEvaluationStarted.countDown()
+ assert(secondEvaluationStarted.await(10, TimeUnit.SECONDS))
+ } else {
+ secondEvaluationStarted.countDown()
+ }
+ 0
+ }
}
/** A non-static class for testing purpose. */
@@ -215,6 +245,49 @@ class CallMethodViaReflectionSuite extends SparkFunSuite
with ExpressionEvalHelp
CallMethodViaReflection(Seq(Literal("\"quote"), Literal("\"quote"),
Literal(null))) :: Nil)
}
+ test("SPARK-58209: CallMethodViaReflection is stateful and produces fresh
copies") {
+ val expr = createExpr(staticClassName, "method2", 5)
+ assert(expr.stateful, "CallMethodViaReflection.stateful should be true")
+ val copy = expr.freshCopyIfContainsStatefulExpression()
+ assert(copy ne expr,
+ "freshCopyIfContainsStatefulExpression should return a new instance " +
+ "for CallMethodViaReflection")
+ copy.asInstanceOf[Nondeterministic].initialize(0)
+ assert(copy.eval(InternalRow.empty) === UTF8String.fromString("m5"))
+ }
+
+ test("SPARK-58209: fresh copies do not share the reflection argument
buffer") {
+ val firstEvaluationStarted = new CountDownLatch(1)
+ val secondEvaluationStarted = new CountDownLatch(1)
+ // The first argument is written into the buffer before the blocking
second argument is
+ // evaluated, so on a shared buffer the second evaluation overwrites the
first one's value.
+ val expr = CallMethodViaReflection(Seq(
+ Literal(staticClassName),
+ Literal("method5"),
+ BoundReference(0, IntegerType, nullable = false),
+ ReflectBlockingArgument(firstEvaluationStarted,
secondEvaluationStarted)))
+ def freshEvaluator(): Expression = {
+ val copy = expr.freshCopyIfContainsStatefulExpression()
+ copy.asInstanceOf[Nondeterministic].initialize(0)
+ copy
+ }
+ val firstEvaluator = freshEvaluator()
+ val secondEvaluator = freshEvaluator()
+
+ val executor = ThreadUtils.newDaemonFixedThreadPool(2,
"call-method-via-reflection-test")
+ val executionContext = ExecutionContext.fromExecutorService(executor)
+ try {
+ val firstResult =
Future(firstEvaluator.eval(InternalRow(1)))(executionContext)
+ assert(firstEvaluationStarted.await(10, TimeUnit.SECONDS))
+ val secondResult =
Future(secondEvaluator.eval(InternalRow(2)))(executionContext)
+
+ assert(ThreadUtils.awaitResult(secondResult, 10.seconds) ===
UTF8String.fromString("m20"))
+ assert(ThreadUtils.awaitResult(firstResult, 10.seconds) ===
UTF8String.fromString("m10"))
+ } finally {
+ executor.shutdownNow()
+ }
+ }
+
private def createExpr(className: String, methodName: String, args: Any*) = {
CallMethodViaReflection(
Literal.create(className, StringType) +:
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/xml/XPathExpressionSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/xml/XPathExpressionSuite.scala
index 252bcea76007..2fee5a65c3ac 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/xml/XPathExpressionSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/xml/XPathExpressionSuite.scala
@@ -20,6 +20,7 @@ package org.apache.spark.sql.catalyst.expressions.xml
import org.apache.spark.SparkFunSuite
import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.DataTypeMismatch
import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.expressions.objects.Invoke
import org.apache.spark.sql.types.StringType
/**
@@ -218,4 +219,26 @@ class XPathExpressionSuite extends SparkFunSuite with
ExpressionEvalHelper {
testExpr(XPathDouble)
testExpr(XPathString)
}
+
+ test("SPARK-58209: XPathExtract is stateful and fresh copies get their own
evaluator") {
+ // The evaluator owns a `UDFXPathUtil` whose parser and reader are reused
across calls, so two
+ // evaluations sharing one expression instance race on them. A fresh copy
must therefore carry
+ // its own evaluator in its replacement, not the one captured by the
original.
+ def evaluatorOf(e: Expression): AnyRef =
+ e.asInstanceOf[XPathExtract].replacement.asInstanceOf[Invoke]
+ .targetObject.asInstanceOf[Literal].value.asInstanceOf[AnyRef]
+
+ Seq(XPathBoolean, XPathShort, XPathInt, XPathLong, XPathFloat,
XPathDouble, XPathString)
+ .foreach { ctor =>
+ val expr = ctor(Literal("<a><b>1</b></a>"), Literal("a/b"))
+ assert(expr.stateful, s"${expr.prettyName}.stateful should be true")
+ val copy = expr.freshCopyIfContainsStatefulExpression()
+ assert(copy ne expr, s"${expr.prettyName} should be freshly copied")
+ assert(evaluatorOf(copy) ne evaluatorOf(expr),
+ s"a fresh copy of ${expr.prettyName} should not share the original's
evaluator")
+ }
+ val copy = XPathString(Literal("<a><b>b1</b></a>"), Literal("a/b"))
+ .freshCopyIfContainsStatefulExpression()
+ checkEvaluation(copy, "b1")
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
