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 5110aab9c6f03..cd6af4098001a 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 @@ -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)) 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 b4f41fda42f00..4bb93588074d3 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 @@ -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]) 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 5a91d37203754..08e1f46bddeef 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 @@ -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)) 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 631b08b5395f9..0cda94d7b188d 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 37916f5a93be0..160470b722665 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 54bef646739fd..a0011e1dbbb94 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("1"), 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) + } + }