[SPARK-58208][SQL] Deep-copy stateful expressions before optimization

[marcuslin123]

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.

[vinodkc]

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