You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by gu...@apache.org on 2025/04/20 23:28:26 UTC
(spark) branch master updated: [SPARK-51848][SQL] Fix parsing XML records with defined schema of array/structs/map of Variant
This is an automated email from the ASF dual-hosted git repository.
gurwls223 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 35b003b39cf8 [SPARK-51848][SQL] Fix parsing XML records with defined schema of array/structs/map of Variant
35b003b39cf8 is described below
commit 35b003b39cf847a476b9e9ae7cc431540329fa32
Author: Xiaonan Yang <xi...@databricks.com>
AuthorDate: Mon Apr 21 08:28:08 2025 +0900
[SPARK-51848][SQL] Fix parsing XML records with defined schema of array/structs/map of Variant
### What changes were proposed in this pull request?
When parsing XML records with a specified schema, which has variant columns inside complex structures like `Array`, `Struct`, or `Map`, those records can't be parsed correctly.
This PR fixes the bug and adds more tests to cover such scenarios.
### Why are the changes needed?
Bug fixes.
### Does this PR introduce _any_ user-facing change?
N/A
### How was this patch tested?
New UTs
### Was this patch authored or co-authored using generative AI tooling?
N/A
Closes #50646 from xiaonanyang-db/SPARK-51848.
Authored-by: Xiaonan Yang <xi...@databricks.com>
Signed-off-by: Hyukjin Kwon <gu...@apache.org>
---
.../spark/sql/catalyst/xml/StaxXmlParser.scala | 22 +++++--
.../datasources/xml/XmlVariantSuite.scala | 71 ++++++++++++++--------
2 files changed, 64 insertions(+), 29 deletions(-)
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 c82571d1f37f..b17e89b53610 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
@@ -202,6 +202,8 @@ class StaxXmlParser(
case st: StructType => convertObject(parser, st)
case MapType(StringType, vt, _) => convertMap(parser, vt, attributes)
case ArrayType(st, _) => convertField(parser, st, startElementName)
+ case VariantType =>
+ StaxXmlParser.convertVariant(parser, attributes, options)
case _: StringType =>
convertTo(
StaxXmlParserUtils.currentStructureAsString(
@@ -231,6 +233,8 @@ class StaxXmlParser(
value
case (_: Characters, st: StructType) =>
convertObject(parser, st)
+ case (_: Characters, VariantType) =>
+ StaxXmlParser.convertVariant(parser, Array.empty, options)
case (_: Characters, _: StringType) =>
convertTo(
StaxXmlParserUtils.currentStructureAsString(
@@ -387,14 +391,15 @@ class StaxXmlParser(
val newValue = dt match {
case st: StructType =>
convertObjectWithAttributes(parser, st, field, attributes)
+ case VariantType =>
+ StaxXmlParser.convertVariant(parser, attributes, options)
case dt: DataType =>
convertField(parser, dt, field)
}
row(index) = values :+ newValue
case VariantType =>
- val v = StaxXmlParser.convertVariant(parser, attributes, options)
- row(index) = new VariantVal(v.getValue, v.getMetadata)
+ row(index) = StaxXmlParser.convertVariant(parser, attributes, options)
case dt: DataType =>
row(index) = convertField(parser, dt, field, attributes)
@@ -921,8 +926,7 @@ object StaxXmlParser {
def parseVariant(xml: String, options: XmlOptions): VariantVal = {
val parser = StaxXmlParserUtils.filteredReader(xml)
val rootAttributes = StaxXmlParserUtils.gatherRootAttributes(parser)
- val variant = convertVariant(parser, rootAttributes, options)
- val v = new VariantVal(variant.getValue, variant.getMetadata)
+ val v = convertVariant(parser, rootAttributes, options)
parser.close()
v
}
@@ -938,6 +942,14 @@ object StaxXmlParser {
* @return A Variant representing the XML element with its attributes and child content
*/
def convertVariant(
+ parser: XMLEventReader,
+ attributes: Array[Attribute],
+ options: XmlOptions): VariantVal = {
+ val v = convertVariantInternal(parser, attributes, options)
+ new VariantVal(v.getValue, v.getMetadata)
+ }
+
+ private def convertVariantInternal(
parser: XMLEventReader,
attributes: Array[Attribute],
options: XmlOptions): Variant = {
@@ -975,7 +987,7 @@ object StaxXmlParser {
val attributes = s.getAttributes.asScala.map(_.asInstanceOf[Attribute]).toArray
val field = StaxXmlParserUtils.getName(s.asStartElement.getName, options)
val variants = fieldToVariants.getOrElseUpdate(field, new java.util.ArrayList[Variant]())
- variants.add(convertVariant(parser, attributes, options))
+ variants.add(convertVariantInternal(parser, attributes, options))
case c: Characters if !c.isWhiteSpace =>
// Treat the character as a value tag field, where we use the [[XMLOptions.valueTag]] as
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/XmlVariantSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/XmlVariantSuite.scala
index ad21a6931387..505aa15b3487 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/XmlVariantSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/XmlVariantSuite.scala
@@ -418,14 +418,20 @@ class XmlVariantSuite extends QueryTest with SharedSparkSession with TestXmlData
val df = createDSLDataFrame(
fileName = "books-complicated.xml",
schemaDDL = Some(
- "_id string, author string, title string, genre variant, price double, " +
- "publish_dates variant"
+ "_id string, author string, title string, " +
+ "genre struct<genreid int, name variant>, " + // Struct with variant
+ "price variant, " + // Scalar as variant
+ "publish_dates struct<publish_date array<variant>>" // Array with variant
),
extraOptions = Map("rowTag" -> "book")
)
checkAnswer(
- df.select(variant_get(col("genre"), "$.name", "string")),
- Seq(Row("Computer"), Row("Fantasy"), Row("Fantasy"))
+ df.select(
+ variant_get(col("genre.name"), "$", "string"),
+ variant_get(col("price"), "$", "double"),
+ variant_get(col("publish_dates.publish_date").getItem(0), "$.month", "int")
+ ),
+ Seq(Row("Computer", 44.95, 10), Row("Fantasy", 5.95, 12), Row("Fantasy", null, 11))
)
}
@@ -557,32 +563,49 @@ class XmlVariantSuite extends QueryTest with SharedSparkSession with TestXmlData
test("SQL: read partial XML record as variant using from_xml with a defined schema") {
val xmlStr =
"""
- |<book>
- | <author>Gambardella</author>
- | <title>Hello</title>
- | <genre>
- | <genreid>1</genreid>
- | <name>Computer</name>
- | </genre>
- | <price>44.95</price>
- | <publish_dates>
- | <publish_date>
- | <day>1</day>
- | <month>10</month>
- | <year>2000</year>
- | </publish_date>
- | </publish_dates>
- | </book>
+ |<ROW>
+ | <scalar>Hello</scalar>
+ | <arrayField><a>1</a></arrayField>
+ | <arrayField><a>2</a></arrayField>
+ | <structField>
+ | <a><b>3</b></a>
+ | <c><d>4</d></c>
+ | </structField>
+ | <nestedVariantField>
+ | <a>5</a>
+ | <b>
+ | <c>6</c>
+ | </b>
+ | </nestedVariantField>
+ | <mapField>
+ | <a><b>7</b></a>
+ | <c>8</c>
+ | </mapField>
+ | </ROW>
| """.stripMargin.replaceAll("\\s+", "")
// Read specific elements in the XML record as variant
val schemaDDL =
- "author string, title string, genre variant, price double, publish_dates variant"
+ "scalar variant, " +
+ "arrayField array<variant>, " + // Array with variants
+ "structField struct<a variant, c struct<d variant>>, " + // Struct with variants
+ "nestedVariantField variant, " +
+ "mapField map<string, variant>"
// Verify we can extract fields from the variant type
checkAnswer(
spark
- .sql(s"""SELECT from_xml('$xmlStr', '$schemaDDL') as book""".stripMargin)
- .select(variant_get(col("book.publish_dates"), "$.publish_date.year", "int")),
- Seq(Row(2000))
+ .sql(s"""SELECT from_xml('$xmlStr', '$schemaDDL') as row""".stripMargin)
+ .select(
+ variant_get(col("row.scalar"), "$", "string"),
+ variant_get(col("row.arrayField").getItem(0), "$.a", "int"),
+ variant_get(col("row.arrayField").getItem(1), "$.a", "int"),
+ variant_get(col("row.structField.a"), "$.b", "int"),
+ variant_get(col("row.structField.c.d"), "$", "int"),
+ variant_get(col("row.nestedVariantField"), "$.a", "int"),
+ variant_get(col("row.nestedVariantField"), "$.b.c", "int"),
+ variant_get(col("row.mapField.a"), "$.b", "int"),
+ variant_get(col("row.mapField.c"), "$", "int")
+ ),
+ Seq(Row("Hello", 1, 2, 3, 4, 5, 6, 7, 8))
)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org