[SPARK-58205][SQL] Mark JSON/CSV/XML expressions stateful to prevent shared-evaluator data races

[vinodkc]

What changes were proposed in this pull request?

Adds override def stateful: Boolean = true to eight JSON/CSV/XML expressions that own their own eval/nullSafeEval and hold mutable state in @transient lazy val evaluator fields:
GetJsonObject, MultiGetJsonObject, JsonTuple, JsonToStructs, CsvToStructs, StructsToCsv, XmlToStructs, and StructsToXml.

StructsToJson, SchemaOfJson, and SchemaOfCsv are excluded:

• SchemaOfJson and SchemaOfCsv are RuntimeReplaceable with foldable-only children and config-only lazy-vals — constant-folded, so no race is possible.
• 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.

Why are the changes needed?

These expressions hold mutable state in @transient lazy val evaluator fields but did not override stateful (which defaults to false). As a result, freshCopyIfContainsStatefulExpression() never made fresh copies of them. When the same expression node appears more than once in a query plan — e.g. df.select(from_json(col, schema), from_json(col, schema)) — both references share the same underlying evaluator instance. In interpreted execution this causes data-race corruption and silently incorrect results.

Does this PR introduce any user-facing change?

No. Queries that were previously correct continue to produce the same results. Queries that previously silently produced incorrect results due to evaluator sharing now produce correct results.

How was this patch tested?

Added one unit test per suite (JsonExpressionsSuite, CsvExpressionsSuite,XmlExpressionsSuite)

Was this patch authored or co-authored using generative AI tooling?

No

[HyukjinKwon]

3 blocking, 0 non-blocking, 0 nits.
The stateful override is correct and effective for 7 of the 10 expressions. The 3 RuntimeReplaceable expressions and a missed sibling need another look.

Correctness (2)

• jsonExpressions.scala:524 / 593, csvExpressions.scala:191: stateful is dead at execution for the RuntimeReplaceable subset (StructsToJson, SchemaOfJson, SchemaOfCsv) — see inline
• JsonExpressionsSuite.scala:1060: the new test passes but doesn't exercise the executed path for the RuntimeReplaceable claim — see inline

Design / architecture (1)

• MultiGetJsonObject (jsonExpressions.scala:176, not in this diff) holds per-row mutable evaluator state but is not marked stateful — same pattern as the fix, on a node where it would be effective. MultiGetJsonObjectEvaluator resets a shared outputBuffer per call (JsonExpressionEvalUtils.scala:614,761) and holds fallbackEvaluators, a Seq of mutable GetJsonObjectEvaluators that are setJson-mutated (:610-620). It's created by OptimizeCsvJsonExprs (:231) and can appear duplicated in a Project. Consider adding override def stateful: Boolean = true here too.

Verification

Traced how stateful reaches execution for each expression kind. For the 7 that own eval/nullSafeEval and are not RuntimeReplaceable, freshCopyIfContainsStatefulExpression (Expression.scala:176) copies the node at execution and its @transient lazy val evaluator re-initializes fresh per copy — the fix is effective. For the 3 RuntimeReplaceable expressions, ReplaceExpressions (Finish Analysis batch, Optimizer.scala:338) rewrites the node to Invoke/StaticInvoke(Literal(evaluator), ...) before any execution-time freshCopy; RuntimeReplaceable.eval asserts input == null (Expression.scala:462), so the node is never row-evaluated in place and its stateful override no longer exists at execution. After replacement the evaluator is pinned inside a Literal (a non-stateful LeafExpression), and freshCopy on a Literal returns this — so both copies keep the same evaluator. SchemaOfJson/SchemaOfCsv additionally require a foldable child and hold only config lazy-vals (no per-row var), so they are constant-folded and can't race — the overrides are unnecessary and inconsistent with the unmarked SchemaOfXml.

PR description suggestions

• Document: the description lists the ten expressions but does not distinguish the RuntimeReplaceable subset (StructsToJson, SchemaOfJson, SchemaOfCsv), where the override behaves differently — worth noting the mechanism/scope and justifying or dropping them.

[vinodkc]

Added override def stateful: Boolean = true to MultiGetJsonObject.