You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "marcuslin123 (via GitHub)" <gi...@apache.org> on 2026/08/13 03:13:09 UTC
[PR] [SPARK-58208][SQL] Deep-copy stateful expressions before optimization [spark]
marcuslin123 opened a new pull request, #57980:
URL: https://github.com/apache/spark/pull/57980
### What changes were proposed in this pull request?
This PR deep-copies stateful expressions before they are evaluated by the optimizer.
Specifically, it:
- clones the logical plan and replaces stateful expressions with fresh copies before `QueryExecution.optimizedPlan` invokes the optimizer;
- uses fresh stateful expressions when `ConvertToLocalRelation` evaluates project and filter expressions; and
- adds regression coverage for both paths using `NamedLambdaVariable`.
### Why are the changes needed?
Logical plans can be shared by DataFrames derived from the same base DataFrame. Although `QueryExecution` cloned the plan before optimization, stateful expressions such as `NamedLambdaVariable` could retain shared mutable state across those clones. Concurrent optimization could therefore mutate the same expression state from multiple threads and silently corrupt query results.
`ConvertToLocalRelation` had the same issue because it evaluated expressions from the shared plan directly.
### Does this PR introduce _any_ user-facing change?
Yes. Concurrent actions on DataFrames derived from a shared logical plan no longer risk silent result corruption caused by shared mutable expression state.
### How was this patch tested?
Added regression tests to `ConvertToLocalRelationSuite` and `QueryExecutionSuite`.
```
build/sbt 'catalyst/testOnly org.apache.spark.sql.catalyst.optimizer.ConvertToLocalRelationSuite'
build/sbt 'sql/testOnly org.apache.spark.sql.execution.QueryExecutionSuite'
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Codex (GPT-5). Codex was used for code assistance and review; the patch was not entirely generated by Codex.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58208][SQL] Deep-copy stateful expressions before optimization [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #57980:
URL: https://github.com/apache/spark/pull/57980#issuecomment-5363162731
cc @cloud-fan
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58208][SQL] Deep-copy stateful expressions before optimization [spark]
Posted by "vinodkc (via GitHub)" <gi...@apache.org>.
vinodkc commented on PR #57980:
URL: https://github.com/apache/spark/pull/57980#issuecomment-5296866063
@marcuslin123 , Can you please check test failures ?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58208][SQL] Deep-copy stateful expressions before optimization [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #57980:
URL: https://github.com/apache/spark/pull/57980#issuecomment-5373856691
**Merge Summary:**
- merged into master https://github.com/apache/spark/commit/73d3dbcf71acbb6fecb15b4480462543eb1bff20
- merged into branch-4.x https://github.com/apache/spark/commit/55a13c1dd76752ad0a0d76233e932e756eb62f8c
*Posted by `merge_spark_pr.py`*
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58208][SQL] Deep-copy stateful expressions before optimization [spark]
Posted by "vinodkc (via GitHub)" <gi...@apache.org>.
vinodkc commented on code in PR #57980:
URL: https://github.com/apache/spark/pull/57980#discussion_r3772376574
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala:
##########
@@ -308,15 +308,21 @@ class QueryExecution(
def assertCommandExecuted(): Unit = commandExecuted
+ private def cloneWithFreshStatefulExpressions(plan: LogicalPlan): LogicalPlan = {
+ plan.clone().transformWithSubqueries {
+ case node => node.mapExpressions(_.freshCopyIfContainsStatefulExpression())
Review Comment:
mapExpressions uses fastEquals to decide whether an expression actually changed:
```
val newE = f(e)
if (newE.fastEquals(e)) e // structural equality — not reference equality
else { changed = true; newE }
```
This works for NamedLambdaVariable because value: AtomicReference is a constructor argument — two NLVs with different AtomicReference instances are not structurally equal, so fastEquals returns false and the fresh copy sticks.
But for expressions like RegExpReplace, StringTranslate, FormatNumber (SPARK-58204), the mutable state lives in a @transient var — not in the constructor. A fresh copy of RegExpReplace has identical constructor args (subject, regexp, replacement), so fastEquals returns true and the fresh copy is silently discarded — the deep-copy becomes a no-op for those expressions.
In practice this is not a correctness issue today — `ConvertToLocalRelation` has its own direct `freshCopyIfContainsStatefulExpression()` call that covers driver-side evaluation, and non-local plans serialize fresh copies to each executor task. But it is a latent gap for future stateful expressions.
A simple fix is to use reference equality (ne) instead:
```
// change detection with ne instead of fastEquals
val newE = e.freshCopyIfContainsStatefulExpression()
if (newE ne e) { changed = true; newE } else e
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58208][SQL] Deep-copy stateful expressions before optimization [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan closed pull request #57980: [SPARK-58208][SQL] Deep-copy stateful expressions before optimization
URL: https://github.com/apache/spark/pull/57980
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58208][SQL] Deep-copy stateful expressions before optimization [spark]
Posted by "vinodkc (via GitHub)" <gi...@apache.org>.
vinodkc commented on PR #57980:
URL: https://github.com/apache/spark/pull/57980#issuecomment-5358249275
cc @HyukjinKwon
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-58208][SQL] Deep-copy stateful expressions before optimization [spark]
Posted by "vinodkc (via GitHub)" <gi...@apache.org>.
vinodkc commented on code in PR #57980:
URL: https://github.com/apache/spark/pull/57980#discussion_r3806794244
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/QueryPlan.scala:
##########
@@ -258,13 +258,28 @@ abstract class QueryPlan[PlanType <: QueryPlan[PlanType]]
* query operator based on the mapped expressions.
*/
def mapExpressions(f: Expression => Expression): this.type = {
+ mapExpressions(f, useFastEquals = true)
+ }
+
+ /**
+ * A variant of [[mapExpressions]] that retains structurally equal replacement expressions.
+ */
+ private[sql] def mapExpressionsWithReferenceEquality(
+ f: Expression => Expression): this.type = {
+ mapExpressions(f, useFastEquals = false)
+ }
+
+ private def mapExpressions(
+ f: Expression => Expression,
+ useFastEquals: Boolean): this.type = {
var changed = false
@inline def transformExpression(e: Expression): Expression = {
val newE = CurrentOrigin.withOrigin(e.origin) {
f(e)
}
- if (newE.fastEquals(e)) {
+ val unchanged = if (useFastEquals) newE.fastEquals(e) else newE.eq(e)
Review Comment:
Could you add a short comment here explaining why this path needs eq instead of fastEquals? It's a subtle point (a fresh stateful copy is structurally equal to the original), and a quick note would stop someone from later "simplifying" it back to fastEquals and quietly reintroducing the bug.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/randomExpressions.scala:
##########
@@ -308,7 +308,7 @@ case class Uniform(
override def withNewChildrenInternal(
newFirst: Expression, newSecond: Expression, newThird: Expression): Expression =
- Uniform(newFirst, newSecond, newThird, hideSeed)
+ copy(min = newFirst, max = newSecond, seedExpression = newThird)
Review Comment:
The two methods just above, withNewSeed (line 303) and withShiftedSeed (line 306), have the same problem, they rebuild Uniform without passing timeZoneId. Might be worth fixing those two as well so they're all consistent.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org