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 bfb95152ba99 [SPARK-58205][SQL] Mark JSON/CSV/XML expressions stateful
to prevent shared-evaluator data races
bfb95152ba99 is described below
commit bfb95152ba996746578177007b654da8879f9888
Author: Vinod KC <[email protected]>
AuthorDate: Wed Aug 5 07:59:52 2026 +0900
[SPARK-58205][SQL] Mark JSON/CSV/XML expressions stateful to prevent
shared-evaluator data races
### 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 [...]
### 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
Closes #57354 from vinodkc/fix-spark58204-json-csv-xml-stateful.
Authored-by: Vinod KC <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>
(cherry picked from commit c2685ecf98e167be991683db7d6a8b9b33e1297b)
Signed-off-by: Hyukjin Kwon <[email protected]>
---
.../sql/catalyst/expressions/csvExpressions.scala | 2 ++
.../sql/catalyst/expressions/jsonExpressions.scala | 5 +++++
.../sql/catalyst/expressions/xmlExpressions.scala | 2 ++
.../catalyst/expressions/CsvExpressionsSuite.scala | 14 ++++++++++++++
.../catalyst/expressions/JsonExpressionsSuite.scala | 20 ++++++++++++++++++++
.../catalyst/expressions/XmlExpressionsSuite.scala | 13 +++++++++++++
6 files changed, 56 insertions(+)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/csvExpressions.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/csvExpressions.scala
index ae10f0245141..15d4c15dcbde 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/csvExpressions.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/csvExpressions.scala
@@ -110,6 +110,7 @@ case class CsvToStructs(
@transient
private lazy val evaluator: CsvToStructsEvaluator = CsvToStructsEvaluator(
options, nullableSchema, nameOfCorruptRecord, timeZoneId, requiredSchema)
+ override def stateful: Boolean = true
override def nullSafeEval(input: Any): Any = {
evaluator.evaluate(input.asInstanceOf[UTF8String])
@@ -296,6 +297,7 @@ case class StructsToCsv(
lazy val converter: Any => UTF8String = {
(row: Any) =>
UTF8String.fromString(gen.writeToString(row.asInstanceOf[InternalRow]))
}
+ override def stateful: Boolean = true
override def withTimeZone(timeZoneId: String): TimeZoneAwareExpression =
copy(timeZoneId = Option(timeZoneId))
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
index 349ac1300296..f01d34777bac 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
@@ -84,6 +84,7 @@ case class GetJsonObject(json: Expression, path: Expression)
} else {
new GetJsonObjectEvaluator()
}
+ override def stateful: Boolean = true
override def eval(input: InternalRow): Any = {
evaluator.setJson(json.eval(input).asInstanceOf[UTF8String])
@@ -215,6 +216,8 @@ case class MultiGetJsonObject(
}
}
+ override def stateful: Boolean = true
+
@transient
private lazy val evaluator = MultiGetJsonObjectEvaluator(
fallbackPaths.map(UTF8String.fromString),
@@ -308,6 +311,7 @@ case class JsonTuple(children: Seq[Expression])
@transient
private lazy val evaluator: JsonTupleEvaluator =
JsonTupleEvaluator(foldableFieldNames)
+ override def stateful: Boolean = true
override def eval(input: InternalRow): IterableOnce[InternalRow] = {
val json = jsonExpr.eval(input).asInstanceOf[UTF8String]
@@ -729,6 +733,7 @@ case class JsonToStructs(
@transient
private lazy val evaluator = new JsonToStructsEvaluator(
options, nullableSchema, nameOfCorruptRecord, timeZoneId,
variantAllowDuplicateKeys)
+ override def stateful: Boolean = true
override def nullSafeEval(json: Any): Any =
evaluator.evaluate(json.asInstanceOf[UTF8String])
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/xmlExpressions.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/xmlExpressions.scala
index f75536c5475a..31aea4910543 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/xmlExpressions.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/xmlExpressions.scala
@@ -103,6 +103,7 @@ case class XmlToStructs(
@transient
private lazy val evaluator: XmlToStructsEvaluator =
XmlToStructsEvaluator(options, nullableSchema, nameOfCorruptRecord,
timeZoneId, child)
+ override def stateful: Boolean = true
private val nameOfCorruptRecord =
SQLConf.get.getConf(SQLConf.COLUMN_NAME_OF_CORRUPT_RECORD)
@@ -295,6 +296,7 @@ case class StructsToXml(
@transient
private lazy val evaluator = StructsToXmlEvaluator(options, child.dataType,
timeZoneId)
+ override def stateful: Boolean = true
override def withTimeZone(timeZoneId: String): TimeZoneAwareExpression =
copy(timeZoneId = Option(timeZoneId))
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CsvExpressionsSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CsvExpressionsSuite.scala
index 631b08b5395f..0cda94d7b188 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CsvExpressionsSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CsvExpressionsSuite.scala
@@ -314,4 +314,18 @@ class CsvExpressionsSuite extends SparkFunSuite with
ExpressionEvalHelper {
CsvToStructs(schema, Map.empty, Literal.create(null, StringType),
UTC_OPT),
null)
}
+
+ test("CsvToStructs and StructsToCsv are stateful and produce fresh copies") {
+ val schema = StructType(StructField("a", IntegerType) :: Nil)
+
+ val csvToStructs = CsvToStructs(schema, Map.empty, Literal("1"), UTC_OPT)
+ assert(csvToStructs.stateful)
+ assert(csvToStructs.freshCopyIfContainsStatefulExpression() ne
csvToStructs)
+
+ val struct = Literal.create(create_row(1), schema)
+ val structsToCsv = StructsToCsv(Map.empty, struct, UTC_OPT)
+ assert(structsToCsv.stateful)
+ assert(structsToCsv.freshCopyIfContainsStatefulExpression() ne
structsToCsv)
+ }
+
}
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala
index 37916f5a93be..160470b72266 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/JsonExpressionsSuite.scala
@@ -1041,4 +1041,24 @@ class JsonExpressionsSuite extends SparkFunSuite with
ExpressionEvalHelper {
input)
}
+ test("JsonToStructs, GetJsonObject, JsonTuple, MultiGetJsonObject are
stateful " +
+ "and produce fresh copies") {
+ val schema = StructType(StructField("a", IntegerType) :: Nil)
+ val jsonToStructs = JsonToStructs(schema, Map.empty, Literal("{}"),
UTC_OPT)
+ assert(jsonToStructs.stateful)
+ assert(jsonToStructs.freshCopyIfContainsStatefulExpression() ne
jsonToStructs)
+
+ val getJsonObject = GetJsonObject(Literal("{}"), Literal("$.a"))
+ assert(getJsonObject.stateful)
+ assert(getJsonObject.freshCopyIfContainsStatefulExpression() ne
getJsonObject)
+
+ val jsonTuple = JsonTuple(Literal("{}") :: Literal("a") :: Nil)
+ assert(jsonTuple.stateful)
+ assert(jsonTuple.freshCopyIfContainsStatefulExpression() ne jsonTuple)
+
+ val multiGetJsonObject = MultiGetJsonObject(Literal("{}"), Seq("$.a",
"$.b"))
+ assert(multiGetJsonObject.stateful)
+ assert(multiGetJsonObject.freshCopyIfContainsStatefulExpression() ne
multiGetJsonObject)
+ }
+
}
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/XmlExpressionsSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/XmlExpressionsSuite.scala
index 54bef646739f..a0011e1dbbb9 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/XmlExpressionsSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/XmlExpressionsSuite.scala
@@ -522,4 +522,17 @@ class XmlExpressionsSuite extends SparkFunSuite with
ExpressionEvalHelper {
)
}
+ test("XmlToStructs and StructsToXml are stateful and produce fresh copies") {
+ val schema = StructType(StructField("a", IntegerType) :: Nil)
+
+ val xmlToStructs = XmlToStructs(schema, Map.empty, Literal("<a>1</a>"),
UTC_OPT)
+ assert(xmlToStructs.stateful)
+ assert(xmlToStructs.freshCopyIfContainsStatefulExpression() ne
xmlToStructs)
+
+ val struct = Literal.create(InternalRow(1), schema)
+ val structsToXml = StructsToXml(Map.empty, struct, UTC_OPT)
+ assert(structsToXml.stateful)
+ assert(structsToXml.freshCopyIfContainsStatefulExpression() ne
structsToXml)
+ }
+
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
