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

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/csvExpressions.scala

@@ -102,6 +102,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])
@@ -275,6 +276,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))

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala

@@ -75,6 +75,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])
@@ -206,6 +207,8 @@ case class MultiGetJsonObject(
     }
   }
 
+  override def stateful: Boolean = true
+
   @transient
   private lazy val evaluator = MultiGetJsonObjectEvaluator(
     fallbackPaths.map(UTF8String.fromString),
@@ -293,6 +296,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]
@@ -420,6 +424,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])
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/xmlExpressions.scala

@@ -96,6 +96,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)
 
@@ -278,6 +279,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))

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)
+  }
+
 }

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)
+  }
+
 }

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)
+  }
+
 }