This is an automated email from the ASF dual-hosted git repository.
HyukjinKwon pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 6539411d1ce9 [SPARK-58204][SQL] Fix unprotected mutable state in
RegExpReplace, RegExpExtractBase, StringTranslate, FormatNumber, and
NamedLambdaVariable by marking them stateful
6539411d1ce9 is described below
commit 6539411d1ce9f6ebc6a05a07c064c4b8bdee0a41
Author: Vinod KC <[email protected]>
AuthorDate: Wed Aug 5 08:01:07 2026 +0900
[SPARK-58204][SQL] Fix unprotected mutable state in RegExpReplace,
RegExpExtractBase, StringTranslate, FormatNumber, and NamedLambdaVariable by
marking them stateful
### What changes were proposed in this pull request?
Override `stateful: Boolean = true` on five Catalyst expressions that
maintain mutable
`transient` state but were not marked stateful:
- `RegExpReplace`
- `RegExpExtractBase` (`RegExpExtract`, `RegExpExtractAll`)
- `StringTranslate`
- `FormatNumber`
- `NamedLambdaVariable`
For `NamedLambdaVariable`, also override `withNewChildrenInternal` to reset
the `AtomicReference` on copy while preserving `exprId`.
`LeafLike.withNewChildrenInternal` returns `this` by default, which would make
`freshCopyIfContainsStatefulExpression()` a no-op even with the `stateful` flag
set.
### Why are the changes needed?
`freshCopyIfContainsStatefulExpression()` only copies subtrees containing a
`stateful = true` node. Without the override, these expressions are shared
across projections. When two projections evaluate concurrently — e.g. via
`ConvertToLocalRelation` on the driver or pipelined execution on executors —
threads race on the shared `transient` cache fields and silently corrupt
results.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Added one test per affected group in the existing suites
(`HigherOrderFunctionsSuite`, `RegexpExpressionsSuite`,
`StringExpressionsSuite`), each asserting `stateful == true` and that
`freshCopyIfContainsStatefulExpression()` returns a distinct instance with
independent state.
### Was this patch authored or co-authored using generative AI tooling?
No
Closes #57353 from vinodkc/fix-spark58154-stateful-expressions.
Authored-by: Vinod KC <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>
(cherry picked from commit 60794df86d97eda23038135264297d44d0fb5afe)
Signed-off-by: Hyukjin Kwon <[email protected]>
---
.../expressions/higherOrderFunctions.scala | 6 ++++++
.../catalyst/expressions/regexpExpressions.scala | 2 ++
.../catalyst/expressions/stringExpressions.scala | 2 ++
.../expressions/HigherOrderFunctionsSuite.scala | 10 +++++++++
.../expressions/RegexpExpressionsSuite.scala | 24 ++++++++++++++++++++++
.../expressions/StringExpressionsSuite.scala | 19 +++++++++++++++++
6 files changed, 63 insertions(+)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/higherOrderFunctions.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/higherOrderFunctions.scala
index cc34421b6230..3877de93a834 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/higherOrderFunctions.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/higherOrderFunctions.scala
@@ -86,9 +86,15 @@ case class NamedLambdaVariable(
override def qualifier: Seq[String] = Seq.empty
+ override def stateful: Boolean = true
+
override def newInstance(): NamedExpression =
copy(exprId = NamedExpression.newExprId, value = new AtomicReference())
+ override def withNewChildrenInternal(
+ newChildren: IndexedSeq[Expression]): NamedLambdaVariable =
+ copy(value = new AtomicReference())
+
override def toAttribute: Attribute = {
AttributeReference(name, dataType, nullable, Metadata.empty)(exprId,
Seq.empty)
}
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/regexpExpressions.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/regexpExpressions.scala
index 0b439c44239f..07d1d00ec756 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/regexpExpressions.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/regexpExpressions.scala
@@ -748,6 +748,7 @@ case class RegExpReplace(subject: Expression, regexp:
Expression, rep: Expressio
// last replacement string, we don't want to convert a UTF8String =>
java.langString every time.
@transient private var lastReplacement: String = _
@transient private var lastReplacementInUTF8: UTF8String = _
+ override def stateful: Boolean = true
final override val nodePatterns: Seq[TreePattern] = Seq(REGEXP_REPLACE)
override def nullSafeEval(s: Any, p: Any, r: Any, i: Any): Any = {
@@ -865,6 +866,7 @@ abstract class RegExpExtractBase
@transient private var lastRegex: UTF8String = _
// last regex pattern, we cache it for performance concern
@transient private var pattern: Pattern = _
+ override def stateful: Boolean = true
final override val nodePatterns: Seq[TreePattern] =
Seq(REGEXP_EXTRACT_FAMILY)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/stringExpressions.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/stringExpressions.scala
index b10f941e2d66..5bd8148103bd 100755
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/stringExpressions.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/stringExpressions.scala
@@ -1255,6 +1255,7 @@ case class StringTranslate(srcExpr: Expression,
matchingExpr: Expression, replac
@transient private var lastMatching: UTF8String = _
@transient private var lastReplace: UTF8String = _
@transient private var dict: JMap[String, String] = _
+ override def stateful: Boolean = true
final lazy val collationId: Int =
first.dataType.asInstanceOf[StringType].collationId
@@ -3743,6 +3744,7 @@ case class FormatNumber(x: Expression, d: Expression)
// as a decimal separator.
@transient
private lazy val numberFormat = new DecimalFormat("", new
DecimalFormatSymbols(Locale.US))
+ override def stateful: Boolean = true
override protected def nullSafeEval(xObject: Any, dObject: Any): Any = {
right.dataType match {
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/HigherOrderFunctionsSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/HigherOrderFunctionsSuite.scala
index c33d258ac4de..b9c870f31090 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/HigherOrderFunctionsSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/HigherOrderFunctionsSuite.scala
@@ -940,6 +940,16 @@ class HigherOrderFunctionsSuite extends SparkFunSuite with
ExpressionEvalHelper
"actualType" -> toSQLType(StringType)
)))
}
+
+ test("NamedLambdaVariable is stateful and produces a fresh copy") {
+ val lv = NamedLambdaVariable("x", IntegerType, nullable = false)
+ assert(lv.stateful, "NamedLambdaVariable.stateful should be true")
+ val copy = lv.freshCopyIfContainsStatefulExpression()
+ assert(copy ne lv,
+ "freshCopyIfContainsStatefulExpression should return a new instance for
NamedLambdaVariable")
+ assert(copy.asInstanceOf[NamedLambdaVariable].value ne lv.value,
+ "fresh copy should have an independent AtomicReference value")
+ }
}
case class CodegenFallbackExpr(child: Expression) extends UnaryExpression with
CodegenFallback {
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/RegexpExpressionsSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/RegexpExpressionsSuite.scala
index 0bf29553ea33..5377b9c7fc0b 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/RegexpExpressionsSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/RegexpExpressionsSuite.scala
@@ -711,4 +711,28 @@ class RegexpExpressionsSuite extends SparkFunSuite with
ExpressionEvalHelper {
)
)
}
+
+ test("RegExpReplace and RegExpExtractBase are stateful and produce fresh
copies") {
+ val s = Literal("hello world")
+ val p = Literal("(\\w+)")
+ val r = Literal("X")
+
+ val replace = RegExpReplace(s, p, r)
+ assert(replace.stateful, "RegExpReplace.stateful should be true")
+ val replaceCopy = replace.freshCopyIfContainsStatefulExpression()
+ assert(replaceCopy ne replace,
+ "freshCopyIfContainsStatefulExpression should return a new instance for
RegExpReplace")
+
+ val extract = RegExpExtract(s, p, Literal(1))
+ assert(extract.stateful, "RegExpExtract.stateful should be true")
+ val extractCopy = extract.freshCopyIfContainsStatefulExpression()
+ assert(extractCopy ne extract,
+ "freshCopyIfContainsStatefulExpression should return a new instance for
RegExpExtract")
+
+ val extractAll = RegExpExtractAll(s, p, Literal(1))
+ assert(extractAll.stateful, "RegExpExtractAll.stateful should be true")
+ val extractAllCopy = extractAll.freshCopyIfContainsStatefulExpression()
+ assert(extractAllCopy ne extractAll,
+ "freshCopyIfContainsStatefulExpression should return a new instance for
RegExpExtractAll")
+ }
}
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/StringExpressionsSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/StringExpressionsSuite.scala
index 711b5edd72ad..050a0245343f 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/StringExpressionsSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/StringExpressionsSuite.scala
@@ -2268,4 +2268,23 @@ class StringExpressionsSuite extends SparkFunSuite with
ExpressionEvalHelper {
}
}
}
+
+ test("StringTranslate and FormatNumber are stateful and produce fresh
copies") {
+ val src = Literal("aeiou")
+ val matching = Literal("aeiou")
+ val replace = Literal("12345")
+ val translate = StringTranslate(src, matching, replace)
+ assert(translate.stateful, "StringTranslate.stateful should be true")
+ val translateCopy = translate.freshCopyIfContainsStatefulExpression()
+ assert(translateCopy ne translate,
+ "freshCopyIfContainsStatefulExpression should return a new instance for
StringTranslate")
+
+ val num = Literal(1234567.89)
+ val fmt = Literal(2)
+ val formatNumber = FormatNumber(num, fmt)
+ assert(formatNumber.stateful, "FormatNumber.stateful should be true")
+ val formatNumberCopy = formatNumber.freshCopyIfContainsStatefulExpression()
+ assert(formatNumberCopy ne formatNumber,
+ "freshCopyIfContainsStatefulExpression should return a new instance for
FormatNumber")
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
