This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 1f80fa0fe491 [SPARK-57624][SQL] from_xml to variant should honor the
parse mode
1f80fa0fe491 is described below
commit 1f80fa0fe491c897a6e26252799986e2e195b5ae
Author: Matt Zhang <[email protected]>
AuthorDate: Thu Jul 2 11:18:33 2026 +0800
[SPARK-57624][SQL] from_xml to variant should honor the parse mode
### What changes were proposed in this pull request?
`from_xml(..., 'variant', map('mode', ...))` ignored the parse `mode`.
`XmlToStructsEvaluator.evaluate` called `StaxXmlParser.parseVariant` directly
for variant output, bypassing the `FailureSafeParser` the struct path uses, so
a malformed record aborted the whole job regardless of `mode`.
This routes variant output through `FailureSafeParser` like the struct
path. For variant output the evaluator builds a single-field `value: variant`
parser schema and sets an internal `XmlOptions.rootVariantType` flag;
`StaxXmlParser.doParseColumn` then parses the whole record as a root Variant
(as it already does for `singleVariantColumn`) and wraps it in a one-field row,
which the evaluator unwraps back to the variant. Reusing `doParseColumn` keeps
parse failures wrapped in `BadReco [...]
It also fixes `from_xml` to variant under whole-stage codegen:
`XmlToStructs.doGenCode` hardcast the result to `InternalRow` (a compile error
for variant output) and did not null-check the result (NPE on a
PERMISSIVE-rescued `null`). It now casts to the actual output type and
null-checks.
### Why are the changes needed?
`from_xml` to variant is unusable on input that contains any malformed
record: the documented `mode` option has no effect for variant output, so
PERMISSIVE cannot rescue bad records and a single bad row fails the whole query
(including streaming jobs). Separately, `from_xml` to variant fails to compile
under whole-stage codegen.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
New unit tests in `XmlFunctionsSuite`, both run under interpreted execution
and whole-stage codegen:
- "from_xml variant output honors the parse mode": a malformed record (an
illegal XML control char) plus a well-formed record, asserting PERMISSIVE
rescues the bad record to `null` and parses the good one, and FAILFAST throws.
- "from_xml variant output validates each record against
rowValidationXSDPath": an XSD-invalid record plus a valid one, asserting the
invalid record is rescued to `null` and the valid one parses.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
Closes #56681 from mzhang/oss-from-xml-variant-mode.
Lead-authored-by: Matt Zhang <[email protected]>
Co-authored-by: Matt Zhang <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../expressions/xml/XmlExpressionEvalUtils.scala | 33 +++++++++----
.../sql/catalyst/expressions/xmlExpressions.scala | 17 ++++++-
.../spark/sql/catalyst/xml/StaxXmlParser.scala | 24 +++++-----
.../apache/spark/sql/catalyst/xml/XmlOptions.scala | 15 ++++--
.../org/apache/spark/sql/XmlFunctionsSuite.scala | 54 ++++++++++++++++++++++
5 files changed, 115 insertions(+), 28 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/xml/XmlExpressionEvalUtils.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/xml/XmlExpressionEvalUtils.scala
index 438110d7acc4..b1fbccab532e 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/xml/XmlExpressionEvalUtils.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/xml/XmlExpressionEvalUtils.scala
@@ -136,8 +136,14 @@ case class XmlToStructsEvaluator(
// This converts parsed rows to the desired output by the given schema.
@transient
- private lazy val converter =
- (rows: Iterator[InternalRow]) => if (rows.hasNext) rows.next() else null
+ private lazy val converter: Iterator[InternalRow] => Any = nullableSchema
match {
+ // For variant output the parser wraps the value in a single-field row
(see `parser`),
+ // so unwrap it back to the variant.
+ case _: VariantType =>
+ (rows: Iterator[InternalRow]) => if (rows.hasNext)
rows.next().getVariant(0) else null
+ case _ =>
+ (rows: Iterator[InternalRow]) => if (rows.hasNext) rows.next() else null
+ }
// Parser that parse XML strings as internal rows
@transient
@@ -147,11 +153,21 @@ case class XmlToStructsEvaluator(
throw QueryCompilationErrors.parseModeUnsupportedError("from_xml", mode)
}
- // The parser is only used when the input schema is StructType
- val schema = nullableSchema.asInstanceOf[StructType]
- ExprUtils.verifyColumnNameOfCorruptRecord(schema,
parsedOptions.columnNameOfCorruptRecord)
- val rawParser = new StaxXmlParser(schema, parsedOptions)
+ // For Variant output, wrap the value in a single-field struct and have
the parser emit a
+ // root Variant via `rootVariantType`; the struct path uses the schema
as-is. Routing both
+ // through StaxXmlParser.doParseColumn keeps the parse `mode` honored
(PERMISSIVE -> null,
+ // FAILFAST -> rethrow) and gives Variant output XSD validation as well.
+ val (schema, rawOptions) = nullableSchema match {
+ case _: VariantType =>
+ (StructType(Array(StructField("value", VariantType))),
+ new XmlOptions(options, timeZoneId.get, nameOfCorruptRecord,
rootVariantType = true))
+ case _ =>
+ val s = nullableSchema.asInstanceOf[StructType]
+ ExprUtils.verifyColumnNameOfCorruptRecord(s,
parsedOptions.columnNameOfCorruptRecord)
+ (s, parsedOptions)
+ }
+ val rawParser = new StaxXmlParser(schema, rawOptions)
val xsdSchema =
Option(parsedOptions.rowValidationXSDPath).map(ValidatorUtil.getSchema)
new FailureSafeParser[String](
@@ -163,10 +179,7 @@ case class XmlToStructsEvaluator(
final def evaluate(xml: UTF8String): Any = {
if (xml == null) return null
- nullableSchema match {
- case _: VariantType => StaxXmlParser.parseVariant(xml.toString,
parsedOptions)
- case _: StructType => converter(parser.parse(xml.toString))
- }
+ converter(parser.parse(xml.toString))
}
}
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 1eb32791955a..5a91d3720375 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
@@ -18,7 +18,7 @@ package org.apache.spark.sql.catalyst.expressions
import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
import
org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{DataTypeMismatch,
TypeCheckSuccess}
-import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext,
ExprCode}
+import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext,
CodeGenerator, ExprCode}
import org.apache.spark.sql.catalyst.expressions.objects.StaticInvoke
import org.apache.spark.sql.catalyst.expressions.xml.{StructsToXmlEvaluator,
XmlExpressionEvalUtils, XmlToStructsEvaluator}
import org.apache.spark.sql.catalyst.util.DropMalformedMode
@@ -109,7 +109,20 @@ case class XmlToStructs(
override protected def doGenCode(ctx: CodegenContext, ev: ExprCode):
ExprCode = {
val expr = ctx.addReferenceObj("this", this)
- defineCodeGen(ctx, ev, input => s"(InternalRow)
$expr.nullSafeEval($input)")
+ // nullSafeEval returns an InternalRow for struct output and a VariantVal
for variant output.
+ // The variant result can be null (e.g. a malformed record rescued under
PERMISSIVE mode), so
+ // cast to the actual output type and null-check the result.
+ val resultType = CodeGenerator.javaType(dataType)
+ val result = ctx.freshName("xmlResult")
+ nullSafeCodeGen(ctx, ev, input =>
+ s"""
+ |$resultType $result = ($resultType) $expr.nullSafeEval($input);
+ |if ($result == null) {
+ | ${ev.isNull} = true;
+ |} else {
+ | ${ev.value} = $result;
+ |}
+ """.stripMargin)
}
override def inputTypes: Seq[AbstractDataType] =
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlParser.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlParser.scala
index f4eb443d95f5..82e127382b74 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlParser.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlParser.scala
@@ -148,18 +148,18 @@ class StaxXmlParser(
xsdSchema.foreach { schema =>
schema.newValidator().validate(new StreamSource(new StringReader(xml)))
}
- options.singleVariantColumn match {
- case Some(_) =>
- // If the singleVariantColumn is specified, parse the entire xml
string as a Variant
- val v = StaxXmlParser.parseVariant(xml, options)
- Some(InternalRow(v))
- case _ =>
- // Otherwise, parse the xml string as Structs
- val parser = StaxXmlParserUtils.filteredReader(xml)
- val rootAttributes = StaxXmlParserUtils.gatherRootAttributes(parser)
- val result = Some(convertObject(parser, schema, rootAttributes))
- parser.close()
- result
+ if (options.singleVariantColumn.isDefined || options.rootVariantType) {
+ // If the singleVariantColumn is specified or the requested output is
a root Variant,
+ // parse the entire xml string as a Variant.
+ val v = StaxXmlParser.parseVariant(xml, options)
+ Some(InternalRow(v))
+ } else {
+ // Otherwise, parse the xml string as Structs
+ val parser = StaxXmlParserUtils.filteredReader(xml)
+ val rootAttributes = StaxXmlParserUtils.gatherRootAttributes(parser)
+ val result = Some(convertObject(parser, schema, rootAttributes))
+ parser.close()
+ result
}
} catch {
case e: SparkUpgradeException => throw e
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/XmlOptions.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/XmlOptions.scala
index 35e77189adb7..bd4da3b27c86 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/XmlOptions.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/XmlOptions.scala
@@ -34,7 +34,10 @@ class XmlOptions(
val parameters: CaseInsensitiveMap[String],
private val defaultTimeZoneId: String,
private val defaultColumnNameOfCorruptRecord: String,
- private val rowTagRequired: Boolean)
+ private val rowTagRequired: Boolean,
+ // Set internally (e.g. by `from_xml` with a Variant schema) to parse each
record as a
+ // root Variant value. Not a user-facing option.
+ val rootVariantType: Boolean)
extends FileSourceOptions(parameters) with Logging {
import XmlOptions._
@@ -43,12 +46,14 @@ class XmlOptions(
parameters: Map[String, String] = Map.empty,
defaultTimeZoneId: String = SQLConf.get.sessionLocalTimeZone,
defaultColumnNameOfCorruptRecord: String =
SQLConf.get.columnNameOfCorruptRecord,
- rowTagRequired: Boolean = false) = {
+ rowTagRequired: Boolean = false,
+ rootVariantType: Boolean = false) = {
this(
CaseInsensitiveMap(parameters),
defaultTimeZoneId,
defaultColumnNameOfCorruptRecord,
- rowTagRequired)
+ rowTagRequired,
+ rootVariantType)
}
@@ -58,7 +63,8 @@ class XmlOptions(
parameters != null && parameters == other.parameters) &&
defaultTimeZoneId == other.defaultTimeZoneId &&
defaultColumnNameOfCorruptRecord ==
other.defaultColumnNameOfCorruptRecord &&
- rowTagRequired == other.rowTagRequired
+ rowTagRequired == other.rowTagRequired &&
+ rootVariantType == other.rootVariantType
case _ => false
}
@@ -67,6 +73,7 @@ class XmlOptions(
result = 31 * result + defaultTimeZoneId.hashCode()
result = 31 * result + defaultColumnNameOfCorruptRecord.hashCode()
result = 31 * result + (if (rowTagRequired) 1 else 0)
+ result = 31 * result + (if (rootVariantType) 1 else 0)
result
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/XmlFunctionsSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/XmlFunctionsSuite.scala
index dd6bc024dd8e..73cdc7df139a 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/XmlFunctionsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/XmlFunctionsSuite.scala
@@ -101,6 +101,60 @@ class XmlFunctionsSuite extends SharedSparkSession {
}
}
+ test("from_xml variant output honors the parse mode") {
+ // A raw control char (code 5, ENQ) in XML text is illegal in XML 1.0 and
the parser rejects
+ // it. Before the fix the variant path bypassed FailureSafeParser, so the
parse `mode` had no
+ // effect and a malformed record aborted the whole query.
+ val badRec = "<Event>ab" + 5.toChar + "cd</Event>"
+ val goodRec = "<Event><a>1</a></Event>"
+ val df = Seq(badRec, goodRec).toDF("value")
+ df.createOrReplaceTempView("from_xml_variant_mode")
+
+ // Exercise both the interpreted path and whole-stage codegen.
+ Seq("true", "false").foreach { wholeStage =>
+ withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> wholeStage) {
+ // PERMISSIVE: the good record parses, the malformed record is rescued
to null. Compare
+ // to_json of the variant so the expected value is concrete; key off
`value` since row
+ // order is not guaranteed.
+ checkAnswer(
+ spark.sql("SELECT value, to_json(from_xml(value, 'variant', " +
+ "map('rowTag','Event','mode','PERMISSIVE'))) AS json FROM
from_xml_variant_mode"),
+ Seq(Row(goodRec, "{\"a\":1}"), Row(badRec, null)))
+
+ // FAILFAST: the malformed record aborts the query.
+ checkError(
+ exception = intercept[SparkException] {
+ spark.sql("SELECT from_xml(value, 'variant', " +
+ "map('rowTag','Event','mode','FAILFAST')) AS d FROM
from_xml_variant_mode").collect()
+ },
+ condition = "MALFORMED_RECORD_IN_PARSING.WITHOUT_SUGGESTION",
+ parameters = Map("badRecord" -> "[null]", "failFastMode" ->
"FAILFAST"))
+ }
+ }
+ }
+
+ test("from_xml variant output validates each record against
rowValidationXSDPath") {
+ val xsdPath =
+
getTestResourcePath("test-data/xml-resources/basket.xsd").replace("file:/", "/")
+ // The first record satisfies basket.xsd; the second has an element not
allowed by the schema.
+ val valid =
"<basket><entry><key>1</key><value>fork</value></entry></basket>"
+ val invalid = "<basket><unexpected>x</unexpected></basket>"
+ val df = Seq(valid, invalid).toDF("value")
+ df.createOrReplaceTempView("from_xml_xsd")
+
+ Seq("true", "false").foreach { wholeStage =>
+ withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> wholeStage) {
+ val result = spark.sql("SELECT value, to_json(from_xml(value,
'variant', " +
+ s"map('rowTag','basket','rowValidationXSDPath','$xsdPath'))) AS json
FROM from_xml_xsd")
+ .collect().map(r => r.getString(0) -> r.getString(1)).toMap
+ assert(result(invalid) == null,
+ s"XSD-invalid record should be rescued to null
(wholeStageCodegen=$wholeStage)")
+ assert(result(valid) != null && result(valid).contains("fork"),
+ s"XSD-valid record should parse (wholeStageCodegen=$wholeStage),
got: ${result(valid)}")
+ }
+ }
+ }
+
test("from_xml with option (timestampFormat)") {
val df = Seq("""<ROW><time>26/08/2015 18:00</time></ROW>""").toDS()
val schema = new StructType().add("time", TimestampType)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
