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 2023/08/22 02:40:03 UTC
[spark] branch master updated: [SPARK-44751][SQL] XML FileFormat Interface implementation
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 206554f1279 [SPARK-44751][SQL] XML FileFormat Interface implementation
206554f1279 is described below
commit 206554f127903f5239b20a7fe1e6e226fcb822ea
Author: Sandip Agarwala <13...@users.noreply.github.com>
AuthorDate: Tue Aug 22 11:39:50 2023 +0900
[SPARK-44751][SQL] XML FileFormat Interface implementation
### What changes were proposed in this pull request?
This is the second PR related to the built-in XML data source implementation ([jira](https://issues.apache.org/jira/browse/SPARK-44751)).
The previous [PR](https://github.com/apache/spark/pull/41832) ported the spark-xml package.
This PR addresses the following:
- Implement FileFormat interface
- Address the review comments in the previous [XML PR](https://github.com/apache/spark/pull/41832)
- Moved from_xml and schema_of_xml to sql/functions
- Moved ".xml" to DataFrameReader/DataFrameWriter
- Removed old APIs like XmlRelation, XmlReader, etc.
- StaxXmlParser changes:
- Use FailureSafeParser
- Convert 'Row' usage to 'InternalRow'
- Convert String to UTF8String
- Handle MapData and ArrayData for MapType and ArrayType respectively
- Use TimestampFormatter to parse timestamp
- Use DateFormatter to parse date
- StaxXmlGenerator changes:
- Convert 'Row' usage to 'InternalRow'
- Handle UTF8String for StringType
- Handle MapData and ArrayData for MapType and ArrayType respectively
- Use TimestampFormatter to format timestamp
- Use DateFormatter to format date
- Update XML tests accordingly because of the above changes
### Why are the changes needed?
These changes are required to bring XML data source capability at par with CSV and JSON and supports features like streaming, which requires FileFormat interface to be implemented.
### Does this PR introduce _any_ user-facing change?
Yes, it adds support for XML data source.
### How was this patch tested?
- Ran all the XML unit tests.
- Github Action
Closes #42462 from sandip-db/xml-file-format-master.
Authored-by: Sandip Agarwala <13...@users.noreply.github.com>
Signed-off-by: Hyukjin Kwon <gu...@apache.org>
---
.../src/main/resources/error/error-classes.json | 5 +
.../org/apache/spark/sql/DataFrameReader.scala | 40 ++
.../org/apache/spark/sql/DataFrameWriter.scala | 28 +
.../scala/org/apache/spark/sql/functions.scala | 106 ++++
.../spark/sql/streaming/DataStreamReader.scala | 18 +
...ror-conditions-datatype-mismatch-error-class.md | 4 +
python/pyspark/sql/tests/test_functions.py | 7 +-
sql/catalyst/pom.xml | 4 +
.../spark/sql/catalyst/expressions/ExprUtils.scala | 23 +
.../sql/catalyst/expressions/xmlExpressions.scala | 207 +++++++
.../spark/sql/catalyst/xml/CreateXmlParser.scala | 121 ++++
.../spark/sql/catalyst/xml}/StaxXmlGenerator.scala | 102 ++--
.../spark/sql/catalyst/xml/StaxXmlParser.scala | 608 +++++++++++++++++++++
.../sql/catalyst/xml}/StaxXmlParserUtils.scala | 18 +-
.../apache/spark/sql/catalyst/xml}/TypeCast.scala | 134 ++---
.../spark/sql/catalyst/xml}/ValidatorUtil.scala | 4 +-
.../spark/sql/catalyst/xml}/XSDToSchema.scala | 3 +-
.../spark/sql/catalyst/xml/XmlInferSchema.scala} | 51 +-
.../apache/spark/sql/catalyst/xml/XmlOptions.scala | 229 ++++++++
.../spark/sql/errors/QueryCompilationErrors.scala | 6 +
...org.apache.spark.sql.sources.DataSourceRegister | 2 +-
.../org/apache/spark/sql/DataFrameReader.scala | 71 +++
.../org/apache/spark/sql/DataFrameWriter.scala | 32 ++
.../sql/execution/datasources/DataSource.scala | 4 +
.../execution/datasources/xml/DefaultSource.scala | 113 ----
.../execution/datasources/xml/XmlDataSource.scala | 207 +++++++
.../datasources/xml/XmlDataToCatalyst.scala | 62 ---
.../execution/datasources/xml/XmlFileFormat.scala | 157 ++++++
.../execution/datasources/xml/XmlInputFormat.scala | 2 +
.../sql/execution/datasources/xml/XmlOptions.scala | 85 ---
.../datasources/xml/XmlOutputWriter.scala | 93 ++++
.../sql/execution/datasources/xml/XmlReader.scala | 205 -------
.../execution/datasources/xml/XmlRelation.scala | 86 ---
.../sql/execution/datasources/xml/XmlUtils.scala | 66 +++
.../sql/execution/datasources/xml/functions.scala | 42 --
.../sql/execution/datasources/xml/package.scala | 162 ------
.../datasources/xml/parsers/StaxXmlParser.scala | 375 -------------
.../xml/util/PartialResultException.scala | 29 -
.../execution/datasources/xml/util/XmlFile.scala | 163 ------
.../scala/org/apache/spark/sql/functions.scala | 97 ++++
.../spark/sql/streaming/DataStreamReader.scala | 21 +
.../execution/datasources/xml/JavaXmlSuite.java | 13 +-
.../test-data/xml-resources/basket_invalid.xml | 10 +
.../apache/spark/sql/DataFrameFunctionsSuite.scala | 5 +-
.../sql/execution/datasources/xml/TestUtils.scala | 9 +-
.../sql/execution/datasources/xml/XmlSuite.scala | 369 ++++++++-----
.../xml/parsers/StaxXmlParserUtilsSuite.scala | 2 +-
.../datasources/xml/util/TypeCastSuite.scala | 202 ++++---
.../datasources/xml/util/XSDToSchemaSuite.scala | 29 +-
.../datasources/xml/util/XmlFileSuite.scala | 69 ---
50 files changed, 2678 insertions(+), 1822 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-classes.json b/common/utils/src/main/resources/error/error-classes.json
index 6ce24bc3b90..632c449b992 100644
--- a/common/utils/src/main/resources/error/error-classes.json
+++ b/common/utils/src/main/resources/error/error-classes.json
@@ -579,6 +579,11 @@
"<errors>"
]
},
+ "INVALID_XML_MAP_KEY_TYPE" : {
+ "message" : [
+ "Input schema <schema> can only contain STRING as a key type for a MAP."
+ ]
+ },
"IN_SUBQUERY_DATA_TYPE_MISMATCH" : {
"message" : [
"The data type of one or more elements in the left hand side of an IN subquery is not compatible with the data type of the output of the subquery. Mismatched columns: [<mismatchedColumns>], left side: [<leftType>], right side: [<rightType>]."
diff --git a/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/DataFrameReader.scala b/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/DataFrameReader.scala
index 10d2af094a0..b01d1479a8c 100644
--- a/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/DataFrameReader.scala
+++ b/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/DataFrameReader.scala
@@ -392,6 +392,46 @@ class DataFrameReader private[sql] (sparkSession: SparkSession) extends Logging
def csv(csvDataset: Dataset[String]): DataFrame =
parse(csvDataset, ParseFormat.PARSE_FORMAT_CSV)
+ /**
+ * Loads a XML file and returns the result as a `DataFrame`. See the documentation on the other
+ * overloaded `xml()` method for more details.
+ *
+ * @since 4.0.0
+ */
+ def xml(path: String): DataFrame = {
+ // This method ensures that calls that explicit need single argument works, see SPARK-16009
+ xml(Seq(path): _*)
+ }
+
+ /**
+ * Loads XML files and returns the result as a `DataFrame`.
+ *
+ * This function will go through the input once to determine the input schema if `inferSchema`
+ * is enabled. To avoid going through the entire data once, disable `inferSchema` option or
+ * specify the schema explicitly using `schema`.
+ *
+ * You can find the XML-specific options for reading XML files in <a
+ * href="https://spark.apache.org/docs/latest/sql-data-sources-xml.html#data-source-option">
+ * Data Source Option</a> in the version you use.
+ *
+ * @since 4.0.0
+ */
+ @scala.annotation.varargs
+ def xml(paths: String*): DataFrame = format("xml").load(paths: _*)
+
+ /**
+ * Loads an `Dataset[String]` storing XML object and returns the result as a `DataFrame`.
+ *
+ * If the schema is not specified using `schema` function and `inferSchema` option is enabled,
+ * this function goes through the input once to determine the input schema.
+ *
+ * @param xmlDataset
+ * input Dataset with one XML object per record
+ * @since 4.0.0
+ */
+ def xml(xmlDataset: Dataset[String]): DataFrame =
+ parse(xmlDataset, ParseFormat.PARSE_FORMAT_UNSPECIFIED)
+
/**
* Loads a Parquet file, returning the result as a `DataFrame`. See the documentation on the
* other overloaded `parquet()` method for more details.
diff --git a/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala b/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala
index b9d1fefb105..0cadcf54af3 100644
--- a/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala
+++ b/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala
@@ -470,6 +470,34 @@ final class DataFrameWriter[T] private[sql] (ds: Dataset[T]) {
format("csv").save(path)
}
+ /**
+ * Saves the content of the `DataFrame` in XML format at the specified path. This is equivalent
+ * to:
+ * {{{
+ * format("xml").save(path)
+ * }}}
+ *
+ * Note that writing a XML file from `DataFrame` having a field `ArrayType` with its element as
+ * `ArrayType` would have an additional nested field for the element. For example, the
+ * `DataFrame` having a field below,
+ *
+ * {@code fieldA [[data1], [data2]]}
+ *
+ * would produce a XML file below. { @code <fieldA> <item>data1</item> </fieldA> <fieldA>
+ * <item>data2</item> </fieldA>}
+ *
+ * Namely, roundtrip in writing and reading can end up in different schema structure.
+ *
+ * You can find the XML-specific options for writing XML files in <a
+ * href="https://spark.apache.org/docs/latest/sql-data-sources-xml.html#data-source-option">
+ * Data Source Option</a> in the version you use.
+ *
+ * @since 4.0.0
+ */
+ def xml(path: String): Unit = {
+ format("xml").save(path)
+ }
+
///////////////////////////////////////////////////////////////////////////////////////
// Builder pattern config options
///////////////////////////////////////////////////////////////////////////////////////
diff --git a/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/functions.scala b/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/functions.scala
index fa8c5782e06..7cd27ecaafb 100644
--- a/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/functions.scala
+++ b/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/functions.scala
@@ -7227,6 +7227,112 @@ object functions {
*/
def to_csv(e: Column): Column = to_csv(e, Collections.emptyMap())
+ // scalastyle:off line.size.limit
+ /**
+ * Parses a column containing a XML string into the data type corresponding to the specified
+ * schema. Returns `null`, in the case of an unparseable string.
+ *
+ * @param e
+ * a string column containing XML data.
+ * @param schema
+ * the schema to use when parsing the XML string
+ * @param options
+ * options to control how the XML is parsed. accepts the same options and the XML data source.
+ * See <a href=
+ * "https://spark.apache.org/docs/latest/sql-data-sources-xml.html#data-source-option"> Data
+ * Source Option</a> in the version you use.
+ * @group collection_funcs
+ *
+ * @since 4.0.0
+ */
+ // scalastyle:on line.size.limit
+ def from_xml(e: Column, schema: StructType, options: Map[String, String]): Column =
+ from_xml(e, lit(schema.toDDL), options.iterator)
+
+ // scalastyle:off line.size.limit
+
+ /**
+ * (Java-specific) Parses a column containing a XML string into the data type corresponding to
+ * the specified schema. Returns `null`, in the case of an unparseable string.
+ *
+ * @param e
+ * a string column containing XML data.
+ * @param schema
+ * the schema to use when parsing the XML string
+ * @param options
+ * options to control how the XML is parsed. accepts the same options and the XML data source.
+ * See <a href=
+ * "https://spark.apache.org/docs/latest/sql-data-sources-xml.html#data-source-option"> Data
+ * Source Option</a> in the version you use.
+ * @group collection_funcs
+ *
+ * @since 4.0.0
+ */
+ // scalastyle:on line.size.limit
+ def from_xml(e: Column, schema: Column, options: java.util.Map[String, String]): Column =
+ from_xml(e, schema, options.asScala.iterator)
+
+ /**
+ * Parses a column containing a XML string into the data type corresponding to the specified
+ * schema. Returns `null`, in the case of an unparseable string.
+ *
+ * @param e
+ * a string column containing XML data.
+ * @param schema
+ * the schema to use when parsing the XML string
+ * @group collection_funcs
+ *
+ * @since 4.0.0
+ */
+ def from_xml(e: Column, schema: StructType): Column =
+ from_xml(e, schema, Map.empty[String, String])
+
+ private def from_xml(e: Column, schema: Column, options: Iterator[(String, String)]): Column = {
+ fnWithOptions("from_xml", options, e, schema)
+ }
+
+ /**
+ * Parses a XML string and infers its schema in DDL format.
+ *
+ * @param xml
+ * a XML string.
+ * @group collection_funcs
+ * @since 4.0.0
+ */
+ def schema_of_xml(xml: String): Column = schema_of_xml(lit(xml))
+
+ /**
+ * Parses a XML string and infers its schema in DDL format.
+ *
+ * @param xml
+ * a foldable string column containing a XML string.
+ * @group collection_funcs
+ * @since 4.0.0
+ */
+ def schema_of_xml(xml: Column): Column = Column.fn("schema_of_xml", xml)
+
+ // scalastyle:off line.size.limit
+
+ /**
+ * Parses a XML string and infers its schema in DDL format using options.
+ *
+ * @param xml
+ * a foldable string column containing XML data.
+ * @param options
+ * options to control how the xml is parsed. accepts the same options and the XML data source.
+ * See <a href=
+ * "https://spark.apache.org/docs/latest/sql-data-sources-xml.html#data-source-option"> Data
+ * Source Option</a> in the version you use.
+ * @return
+ * a column with string literal containing schema in DDL format.
+ * @group collection_funcs
+ * @since 4.0.0
+ */
+ // scalastyle:on line.size.limit
+ def schema_of_xml(xml: Column, options: java.util.Map[String, String]): Column = {
+ fnWithOptions("schema_of_xml", options.asScala.iterator, xml)
+ }
+
/**
* Returns the total number of elements in the array. The function returns null for null input.
*
diff --git a/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/DataStreamReader.scala b/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/DataStreamReader.scala
index 4b0b99dd787..3d2b8c7be25 100644
--- a/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/DataStreamReader.scala
+++ b/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/streaming/DataStreamReader.scala
@@ -189,6 +189,24 @@ final class DataStreamReader private[sql] (sparkSession: SparkSession) extends L
*/
def csv(path: String): DataFrame = format("csv").load(path)
+ /**
+ * Loads a XML file stream and returns the result as a `DataFrame`.
+ *
+ * This function will go through the input once to determine the input schema if `inferSchema`
+ * is enabled. To avoid going through the entire data once, disable `inferSchema` option or
+ * specify the schema explicitly using `schema`.
+ *
+ * You can set the following option(s): <ul> <li>`maxFilesPerTrigger` (default: no max limit):
+ * sets the maximum number of new files to be considered in every trigger.</li> </ul>
+ *
+ * You can find the XML-specific options for reading XML file stream in <a
+ * href="https://spark.apache.org/docs/latest/sql-data-sources-xml.html#data-source-option">
+ * Data Source Option</a> in the version you use.
+ *
+ * @since 4.0.0
+ */
+ def xml(path: String): DataFrame = format("xml").load(path)
+
/**
* Loads a ORC file stream, returning the result as a `DataFrame`.
*
diff --git a/docs/sql-error-conditions-datatype-mismatch-error-class.md b/docs/sql-error-conditions-datatype-mismatch-error-class.md
index ddc3e0c2b1b..3e3c1091baa 100644
--- a/docs/sql-error-conditions-datatype-mismatch-error-class.md
+++ b/docs/sql-error-conditions-datatype-mismatch-error-class.md
@@ -123,6 +123,10 @@ The `<functionName>` does not support ordering on type `<dataType>`.
`<errors>`
+## INVALID_XML_MAP_KEY_TYPE
+
+Input schema `<schema>` can only contain STRING as a key type for a MAP.
+
## IN_SUBQUERY_DATA_TYPE_MISMATCH
The data type of one or more elements in the left hand side of an IN subquery is not compatible with the data type of the output of the subquery. Mismatched columns: [`<mismatchedColumns>`], left side: [`<leftType>`], right side: [`<rightType>`].
diff --git a/python/pyspark/sql/tests/test_functions.py b/python/pyspark/sql/tests/test_functions.py
index d80a3cf52f1..c484e10ec1a 100644
--- a/python/pyspark/sql/tests/test_functions.py
+++ b/python/pyspark/sql/tests/test_functions.py
@@ -81,7 +81,12 @@ class FunctionsTestsMixin:
missing_in_py = jvm_fn_set.difference(py_fn_set)
# Functions that we expect to be missing in python until they are added to pyspark
- expected_missing_in_py = set()
+ expected_missing_in_py = {
+ # TODO: XML functions will soon be added and removed from this list
+ # https://issues.apache.org/jira/browse/SPARK-44788
+ "from_xml",
+ "schema_of_xml",
+ }
self.assertEqual(
expected_missing_in_py, missing_in_py, "Missing functions in pyspark not as expected"
diff --git a/sql/catalyst/pom.xml b/sql/catalyst/pom.xml
index 7a8c0aba8c0..fa456487de5 100644
--- a/sql/catalyst/pom.xml
+++ b/sql/catalyst/pom.xml
@@ -111,6 +111,10 @@
<artifactId>univocity-parsers</artifactId>
<type>jar</type>
</dependency>
+ <dependency>
+ <groupId>org.apache.ws.xmlschema</groupId>
+ <artifactId>xmlschema-core</artifactId>
+ </dependency>
<dependency>
<groupId>org.apache.datasketches</groupId>
<artifactId>datasketches-java</artifactId>
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExprUtils.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExprUtils.scala
index 2fa970bac0c..5a093037e42 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExprUtils.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExprUtils.scala
@@ -117,4 +117,27 @@ object ExprUtils extends QueryErrorsBase {
TypeCheckSuccess
}
}
+
+ /**
+ * Check if the schema is valid for XML
+ *
+ * @param schema The schema to check.
+ * @return
+ * `TypeCheckSuccess` if the schema is valid
+ * `DataTypeMismatch` with an error error if the schema is not valid
+ */
+ def checkXmlSchema(schema: DataType): TypeCheckResult = {
+ val isInvalid = schema.existsRecursively {
+ // XML field names must be StringType
+ case MapType(keyType, _, _) if keyType != StringType => true
+ case _ => false
+ }
+ if (isInvalid) {
+ DataTypeMismatch(
+ errorSubClass = "INVALID_XML_MAP_KEY_TYPE",
+ messageParameters = Map("schema" -> toSQLType(schema)))
+ } else {
+ TypeCheckSuccess
+ }
+ }
}
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
new file mode 100644
index 00000000000..0b664ecb8b8
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/xmlExpressions.scala
@@ -0,0 +1,207 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.catalyst.expressions.xml
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.DataTypeMismatch
+import org.apache.spark.sql.catalyst.expressions.{ExpectsInputTypes, Expression, ExpressionDescription, ExprUtils, NullIntolerant, TimeZoneAwareExpression, UnaryExpression}
+import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
+import org.apache.spark.sql.catalyst.util.{ArrayData, FailFastMode, FailureSafeParser, GenericArrayData, PermissiveMode}
+import org.apache.spark.sql.catalyst.xml.{StaxXmlParser, ValidatorUtil, XmlInferSchema, XmlOptions}
+import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryErrorsBase}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+case class XmlToStructs(
+ schema: DataType,
+ options: Map[String, String],
+ child: Expression,
+ timeZoneId: Option[String] = None)
+ extends UnaryExpression
+ with TimeZoneAwareExpression
+ with CodegenFallback
+ with ExpectsInputTypes
+ with NullIntolerant
+ with QueryErrorsBase {
+
+ def this(child: Expression, schema: Expression, options: Map[String, String]) =
+ this(
+ schema = ExprUtils.evalSchemaExpr(schema),
+ options = options,
+ child = child,
+ timeZoneId = None)
+
+ override def nullable: Boolean = true
+
+ // The XML input data might be missing certain fields. We force the nullability
+ // of the user-provided schema to avoid data corruptions.
+ val nullableSchema = schema.asNullable
+
+ def this(child: Expression, schema: Expression) = this(child, schema, Map.empty[String, String])
+
+ def this(child: Expression, schema: Expression, options: Expression) =
+ this(
+ schema = ExprUtils.evalSchemaExpr(schema),
+ options = ExprUtils.convertToMapData(options),
+ child = child,
+ timeZoneId = None)
+
+ // This converts parsed rows to the desired output by the given schema.
+ @transient
+ lazy val converter = nullableSchema match {
+ case _: StructType =>
+ (rows: Iterator[InternalRow]) => if (rows.hasNext) rows.next() else null
+ case _: ArrayType =>
+ (rows: Iterator[InternalRow]) => if (rows.hasNext) rows.next().getArray(0) else null
+ case _: MapType =>
+ (rows: Iterator[InternalRow]) => if (rows.hasNext) rows.next().getMap(0) else null
+ }
+
+ val nameOfCorruptRecord = SQLConf.get.getConf(SQLConf.COLUMN_NAME_OF_CORRUPT_RECORD)
+
+ @transient lazy val parser = {
+ val parsedOptions = new XmlOptions(options, timeZoneId.get, nameOfCorruptRecord)
+ val mode = parsedOptions.parseMode
+ if (mode != PermissiveMode && mode != FailFastMode) {
+ throw QueryCompilationErrors.parseModeUnsupportedError("from_xml", mode)
+ }
+ val (parserSchema, actualSchema) = nullableSchema match {
+ case s: StructType =>
+ ExprUtils.verifyColumnNameOfCorruptRecord(s, parsedOptions.columnNameOfCorruptRecord)
+ (s, StructType(s.filterNot(_.name == parsedOptions.columnNameOfCorruptRecord)))
+ case other =>
+ (StructType(Array(StructField("value", other))), other)
+ }
+
+ val rowSchema: StructType = schema match {
+ case st: StructType => st
+ case ArrayType(st: StructType, _) => st
+ }
+ val rawParser = new StaxXmlParser(rowSchema, parsedOptions)
+ val xsdSchema = Option(parsedOptions.rowValidationXSDPath).map(ValidatorUtil.getSchema)
+
+ new FailureSafeParser[String](
+ input => rawParser.doParseColumn(input, mode, xsdSchema),
+ mode,
+ parserSchema,
+ parsedOptions.columnNameOfCorruptRecord)
+ }
+
+ override def dataType: DataType = nullableSchema
+
+ override def withTimeZone(timeZoneId: String): TimeZoneAwareExpression = {
+ copy(timeZoneId = Option(timeZoneId))
+ }
+ override def nullSafeEval(xml: Any): Any = xml match {
+ case arr: GenericArrayData =>
+ new GenericArrayData(arr.array.map(s => converter(parser.parse(s.toString))))
+ case arr: ArrayData =>
+ new GenericArrayData(arr.array.map(s => converter(parser.parse(s.toString))))
+ case _ =>
+ val str = xml.asInstanceOf[UTF8String].toString
+ converter(parser.parse(str))
+ }
+
+ override def inputTypes: Seq[AbstractDataType] = StringType :: Nil
+
+ override def sql: String = schema match {
+ case _: MapType => "entries"
+ case _ => super.sql
+ }
+
+ override def prettyName: String = "from_xml"
+
+ protected def withNewChildInternal(newChild: Expression): XmlToStructs =
+ copy(child = newChild)
+}
+
+/**
+ * A function infers schema of XML string.
+ */
+@ExpressionDescription(
+ usage = "_FUNC_(xml[, options]) - Returns schema in the DDL format of XML string.",
+ examples = """
+ Examples:
+ > SELECT _FUNC_('1,abc');
+ STRUCT<_c0: INT, _c1: STRING>
+ """,
+ since = "4.0.0",
+ group = "xml_funcs")
+case class SchemaOfXml(
+ child: Expression,
+ options: Map[String, String])
+ extends UnaryExpression with CodegenFallback with QueryErrorsBase {
+
+ def this(child: Expression) = this(child, Map.empty[String, String])
+
+ def this(child: Expression, options: Expression) = this(
+ child = child,
+ options = ExprUtils.convertToMapData(options))
+
+ override def dataType: DataType = StringType
+
+ override def nullable: Boolean = false
+
+ @transient
+ private lazy val xmlOptions = new XmlOptions(options, "UTC")
+
+ @transient
+ private lazy val xmlFactory = xmlOptions.buildXmlFactory()
+
+ @transient
+ private lazy val xml = child.eval().asInstanceOf[UTF8String]
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ if (child.foldable && xml != null) {
+ super.checkInputDataTypes()
+ } else if (!child.foldable) {
+ DataTypeMismatch(
+ errorSubClass = "NON_FOLDABLE_INPUT",
+ messageParameters = Map(
+ "inputName" -> "xml",
+ "inputType" -> toSQLType(child.dataType),
+ "inputExpr" -> toSQLExpr(child)))
+ } else {
+ DataTypeMismatch(
+ errorSubClass = "UNEXPECTED_NULL",
+ messageParameters = Map("exprName" -> "xml"))
+ }
+ }
+
+ override def eval(v: InternalRow): Any = {
+ val dataType = XmlInferSchema.infer(xml.toString, xmlOptions).get match {
+ case st: StructType =>
+ XmlInferSchema.canonicalizeType(st).getOrElse(StructType(Nil))
+ case at: ArrayType if at.elementType.isInstanceOf[StructType] =>
+ XmlInferSchema
+ .canonicalizeType(at.elementType)
+ .map(ArrayType(_, containsNull = at.containsNull))
+ .getOrElse(ArrayType(StructType(Nil), containsNull = at.containsNull))
+ case other: DataType =>
+ XmlInferSchema.canonicalizeType(other).getOrElse(StringType)
+ }
+
+ UTF8String.fromString(dataType.sql)
+ }
+
+ override def prettyName: String = "schema_of_xml"
+
+ override protected def withNewChildInternal(newChild: Expression): SchemaOfXml =
+ copy(child = newChild)
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/CreateXmlParser.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/CreateXmlParser.scala
new file mode 100644
index 00000000000..3a7f0b4be28
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/CreateXmlParser.scala
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.xml
+
+import java.io.{ByteArrayInputStream, InputStream, InputStreamReader, StringReader}
+import java.nio.channels.Channels
+import java.nio.charset.{Charset, StandardCharsets}
+import javax.xml.stream.{EventFilter, XMLEventReader, XMLInputFactory, XMLStreamConstants}
+import javax.xml.stream.events.XMLEvent
+
+import org.apache.hadoop.io.Text
+import sun.nio.cs.StreamDecoder
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.unsafe.types.UTF8String
+
+private[sql] object CreateXmlParser extends Serializable {
+ val filter = new EventFilter {
+ override def accept(event: XMLEvent): Boolean =
+ // Ignore comments and processing instructions
+ event.getEventType match {
+ case XMLStreamConstants.COMMENT | XMLStreamConstants.PROCESSING_INSTRUCTION => false
+ case _ => true
+ }
+ }
+
+ def string(xmlInputFactory: XMLInputFactory, record: String): XMLEventReader = {
+ // It does not have to skip for white space, since `XmlInputFormat`
+ // always finds the root tag without a heading space.
+ val eventReader = xmlInputFactory.createXMLEventReader(new StringReader(record))
+ xmlInputFactory.createFilteredReader(eventReader, filter)
+ }
+
+ def utf8String(xmlInputFactory: XMLInputFactory, record: UTF8String): XMLEventReader = {
+ val bb = record.getByteBuffer
+ assert(bb.hasArray)
+
+ val bain = new ByteArrayInputStream(
+ bb.array(), bb.arrayOffset() + bb.position(), bb.remaining())
+
+ val eventReader = xmlInputFactory.createXMLEventReader(
+ new InputStreamReader(bain, StandardCharsets.UTF_8))
+ xmlInputFactory.createFilteredReader(eventReader, filter)
+ }
+
+ def text(xmlInputFactory: XMLInputFactory, record: Text): XMLEventReader = {
+ val bs = new ByteArrayInputStream(record.getBytes, 0, record.getLength)
+
+ val eventReader = xmlInputFactory.createXMLEventReader(bs)
+ xmlInputFactory.createFilteredReader(eventReader, filter)
+ }
+
+ // Jackson parsers can be ranked according to their performance:
+ // 1. Array based with actual encoding UTF-8 in the array. This is the fastest parser
+ // but it doesn't allow to set encoding explicitly. Actual encoding is detected automatically
+ // by checking leading bytes of the array.
+ // 2. InputStream based with actual encoding UTF-8 in the stream. Encoding is detected
+ // automatically by analyzing first bytes of the input stream.
+ // 3. Reader based parser. This is the slowest parser used here but it allows to create
+ // a reader with specific encoding.
+ // The method creates a reader for an array with given encoding and sets size of internal
+ // decoding buffer according to size of input array.
+ private def getStreamDecoder(enc: String, in: Array[Byte], length: Int): StreamDecoder = {
+ val bais = new ByteArrayInputStream(in, 0, length)
+ val byteChannel = Channels.newChannel(bais)
+ val decodingBufferSize = Math.min(length, 8192)
+ val decoder = Charset.forName(enc).newDecoder()
+
+ StreamDecoder.forDecoder(byteChannel, decoder, decodingBufferSize)
+ }
+
+ def text(enc: String, xmlInputFactory: XMLInputFactory, record: Text): XMLEventReader = {
+ val sd = getStreamDecoder(enc, record.getBytes, record.getLength)
+ val eventReader = xmlInputFactory.createXMLEventReader(sd)
+ xmlInputFactory.createFilteredReader(eventReader, filter)
+ }
+
+ def inputStream(xmlInputFactory: XMLInputFactory, is: InputStream): XMLEventReader = {
+ val eventReader = xmlInputFactory.createXMLEventReader(is)
+ xmlInputFactory.createFilteredReader(eventReader, filter)
+ }
+
+ def inputStream(enc: String, xmlInputFactory: XMLInputFactory,
+ is: InputStream): XMLEventReader = {
+ val eventReader = xmlInputFactory.createXMLEventReader(new InputStreamReader(is, enc))
+ xmlInputFactory.createFilteredReader(eventReader, filter)
+ }
+
+ def internalRow(xmlInputFactory: XMLInputFactory, row: InternalRow): XMLEventReader = {
+ val ba = row.getBinary(0)
+
+ val bs = new ByteArrayInputStream(ba, 0, ba.length)
+
+ val eventReader = xmlInputFactory.createXMLEventReader(bs)
+ xmlInputFactory.createFilteredReader(eventReader, filter)
+ }
+
+ def internalRow(enc: String, xmlInputFactory: XMLInputFactory,
+ row: InternalRow): XMLEventReader = {
+ val binary = row.getBinary(0)
+ val sd = getStreamDecoder(enc, binary, binary.length)
+
+ val eventReader = xmlInputFactory.createXMLEventReader(sd)
+ xmlInputFactory.createFilteredReader(eventReader, filter)
+ }
+}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/parsers/StaxXmlGenerator.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlGenerator.scala
similarity index 63%
rename from sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/parsers/StaxXmlGenerator.scala
rename to sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlGenerator.scala
index eb4c94fe280..f1cbc8996b0 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/parsers/StaxXmlGenerator.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlGenerator.scala
@@ -14,42 +14,45 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.spark.sql.execution.datasources.xml.parsers
+package org.apache.spark.sql.catalyst.xml
-import java.sql.{Date, Timestamp}
-import java.time.format.DateTimeFormatter
+import java.sql.Timestamp
import javax.xml.stream.XMLStreamWriter
import scala.collection.Map
-import org.apache.spark.sql.Row
-import org.apache.spark.sql.execution.datasources.xml.XmlOptions
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayData, MapData}
import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
// This class is borrowed from Spark json datasource.
-private[xml] object StaxXmlGenerator {
+private[sql] object StaxXmlGenerator {
/**
* Transforms a single Row to XML
*
- * @param schema the schema object used for conversion
- * @param writer a XML writer object
- * @param options options for XML datasource.
- * @param row The row to convert
+ * @param schema
+ * the schema object used for conversion
+ * @param writer
+ * a XML writer object
+ * @param options
+ * options for XML datasource.
+ * @param row
+ * The row to convert
*/
- def apply(
- schema: StructType,
- writer: XMLStreamWriter,
- options: XmlOptions)(row: Row): Unit = {
+ def apply(schema: StructType, writer: XMLStreamWriter, options: XmlOptions)(
+ row: InternalRow): Unit = {
- require(options.attributePrefix.nonEmpty,
+ require(
+ options.attributePrefix.nonEmpty,
"'attributePrefix' option should not be empty string.")
def writeChildElement(name: String, dt: DataType, v: Any): Unit = (name, dt, v) match {
// If this is meant to be value but in no child, write only a value
case (_, _, null) | (_, NullType, _) if options.nullValue == null =>
- // Because usually elements having `null` do not exist, just do not write
- // elements when given values are `null`.
+ // Because usually elements having `null` do not exist, just do not write
+ // elements when given values are `null`.
case (_, _, _) if name == options.valueTag =>
// If this is meant to be value but in no child, write only a value
writeElement(dt, v, options)
@@ -63,7 +66,7 @@ private[xml] object StaxXmlGenerator {
(dt, v) match {
// If this is meant to be attribute, write an attribute
case (_, null) | (NullType, _)
- if name.startsWith(options.attributePrefix) && name != options.valueTag =>
+ if name.startsWith(options.attributePrefix) && name != options.valueTag =>
Option(options.nullValue).foreach {
writer.writeAttribute(name.substring(options.attributePrefix.length), _)
}
@@ -71,9 +74,9 @@ private[xml] object StaxXmlGenerator {
writer.writeAttribute(name.substring(options.attributePrefix.length), v.toString)
// For ArrayType, we just need to write each as XML element.
- case (ArrayType(ty, _), v: scala.collection.Seq[_]) =>
- v.foreach { e =>
- writeChildElement(name, ty, e)
+ case (ArrayType(ty, _), v: ArrayData) =>
+ (0 until v.numElements()).foreach { i =>
+ writeChildElement(name, ty, v.get(i, ty))
}
// For other datatypes, we just write normal elements.
case _ =>
@@ -83,21 +86,21 @@ private[xml] object StaxXmlGenerator {
def writeElement(dt: DataType, v: Any, options: XmlOptions): Unit = (dt, v) match {
case (_, null) | (NullType, _) => writer.writeCharacters(options.nullValue)
+ case (StringType, v: UTF8String) => writer.writeCharacters(v.toString)
case (StringType, v: String) => writer.writeCharacters(v)
case (TimestampType, v: Timestamp) =>
- val formatter = options.timestampFormat.map(DateTimeFormatter.ofPattern).
- getOrElse(DateTimeFormatter.ISO_INSTANT)
- writer.writeCharacters(formatter.format(v.toInstant()))
- case (DateType, v: Date) =>
- val formatter = options.dateFormat.map(DateTimeFormatter.ofPattern).
- getOrElse(DateTimeFormatter.ISO_DATE)
- writer.writeCharacters(formatter.format(v.toLocalDate()))
+ writer.writeCharacters(options.timestampFormatterInWrite.format(v.toInstant()))
+ case (TimestampType, v: Long) =>
+ writer.writeCharacters(options.timestampFormatterInWrite.format(v))
+ case (DateType, v: Int) =>
+ writer.writeCharacters(options.dateFormatterInWrite.format(v))
case (IntegerType, v: Int) => writer.writeCharacters(v.toString)
case (ShortType, v: Short) => writer.writeCharacters(v.toString)
case (FloatType, v: Float) => writer.writeCharacters(v.toString)
case (DoubleType, v: Double) => writer.writeCharacters(v.toString)
case (LongType, v: Long) => writer.writeCharacters(v.toString)
case (DecimalType(), v: java.math.BigDecimal) => writer.writeCharacters(v.toString)
+ case (DecimalType(), v: Decimal) => writer.writeCharacters(v.toString)
case (ByteType, v: Byte) => writer.writeCharacters(v.toString)
case (BooleanType, v: Boolean) => writer.writeCharacters(v.toString)
@@ -106,9 +109,9 @@ private[xml] object StaxXmlGenerator {
// this case only can happen when we convert a normal [[DataFrame]] to XML file.
// When [[ArrayType]] has [[ArrayType]] as elements, it is confusing what is element name
// for XML file.
- case (ArrayType(ty, _), v: scala.collection.Seq[_]) =>
- v.foreach { e =>
- writeChild(options.arrayElementName, ty, e)
+ case (ArrayType(ty, _), v: ArrayData) =>
+ (0 until v.numElements()).foreach { i =>
+ writeChild(options.arrayElementName, ty, v.get(i, ty))
}
case (MapType(_, vt, _), mv: Map[_, _]) =>
@@ -116,19 +119,19 @@ private[xml] object StaxXmlGenerator {
f.toString.startsWith(options.attributePrefix) && f.toString != options.valueTag
}
// We need to write attributes first before the value.
- (attributes ++ elements).foreach {
- case (k, v) =>
- writeChild(k.toString, vt, v)
+ (attributes ++ elements).foreach { case (k, v) =>
+ writeChild(k.toString, vt, v)
}
- case (StructType(ty), r: Row) =>
- val (attributes, elements) = ty.zip(r.toSeq).partition { case (f, _) =>
+ case (mt: MapType, mv: MapData) => writeMapData(mt, mv)
+
+ case (st: StructType, r: InternalRow) =>
+ val (attributes, elements) = st.zip(r.toSeq(st)).partition { case (f, _) =>
f.name.startsWith(options.attributePrefix) && f.name != options.valueTag
}
// We need to write attributes first before the value.
- (attributes ++ elements).foreach {
- case (field, value) =>
- writeChild(field.name, field.dataType, value)
+ (attributes ++ elements).foreach { case (field, value) =>
+ writeChild(field.name, field.dataType, value)
}
case (_, _) =>
@@ -136,7 +139,23 @@ private[xml] object StaxXmlGenerator {
s"Failed to convert value $v (class of ${v.getClass}) in type $dt to XML.")
}
- val (attributes, elements) = schema.zip(row.toSeq).partition { case (f, _) =>
+ def writeMapData(mapType: MapType, map: MapData): Unit = {
+ val keyArray = map.keyArray()
+ val valueArray = map.valueArray()
+ // write attributes first
+ Seq(true, false).foreach { writeAttribute =>
+ (0 until map.numElements()).foreach { i =>
+ val key = keyArray.get(i, mapType.keyType).toString
+ val isAttribute = key.startsWith(options.attributePrefix) && key != options.valueTag
+ if (writeAttribute == isAttribute) {
+ writeChild(key, mapType.valueType, valueArray.get(i, mapType.valueType))
+ }
+ }
+ }
+ }
+
+ val rowSeq = row.toSeq(schema)
+ val (attributes, elements) = schema.zip(rowSeq).partition { case (f, _) =>
f.name.startsWith(options.attributePrefix) && f.name != options.valueTag
}
// Writing attributes
@@ -152,7 +171,8 @@ private[xml] object StaxXmlGenerator {
// Writing elements
val (names, values) = elements.unzip
val elementSchema = StructType(schema.filter(names.contains))
- val elementRow = Row.fromSeq(row.toSeq.filter(values.contains))
+
+ val elementRow = InternalRow.fromSeq(rowSeq.filter(values.contains))
writeElement(elementSchema, elementRow, options)
writer.writeEndElement()
}
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
new file mode 100644
index 00000000000..e8dbc85f295
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlParser.scala
@@ -0,0 +1,608 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.catalyst.xml
+
+import java.io.{CharConversionException, InputStream, InputStreamReader, StringReader}
+import java.nio.charset.{Charset, MalformedInputException}
+import javax.xml.stream.{XMLEventReader, XMLStreamException}
+import javax.xml.stream.events._
+import javax.xml.transform.stream.StreamSource
+import javax.xml.validation.Schema
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable.ArrayBuffer
+import scala.util.control.NonFatal
+import scala.xml.SAXException
+
+import org.apache.spark.SparkUpgradeException
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, BadRecordException, DropMalformedMode, FailureSafeParser, GenericArrayData, MapData, ParseMode, PartialResultArrayException, PartialResultException, PermissiveMode}
+import org.apache.spark.sql.catalyst.xml.StaxXmlParser.convertStream
+import org.apache.spark.sql.catalyst.xml.TypeCast._
+import org.apache.spark.sql.errors.QueryExecutionErrors
+import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf}
+import org.apache.spark.sql.sources.Filter
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+
+class StaxXmlParser(
+ schema: StructType,
+ val options: XmlOptions,
+ filters: Seq[Filter] = Seq.empty) extends Logging {
+
+ private val factory = options.buildXmlFactory()
+
+ // Flags to signal if we need to fall back to the backward compatible behavior of parsing
+ // dates and timestamps.
+ // For more information, see comments for "enableDateTimeParsingFallback" option in XmlOptions.
+ private val enableParsingFallbackForTimestampType =
+ options.enableDateTimeParsingFallback
+ .orElse(SQLConf.get.jsonEnableDateTimeParsingFallback)
+ .getOrElse {
+ SQLConf.get.legacyTimeParserPolicy == LegacyBehaviorPolicy.LEGACY ||
+ options.timestampFormatInRead.isEmpty
+ }
+ private val enableParsingFallbackForDateType =
+ options.enableDateTimeParsingFallback
+ .orElse(SQLConf.get.jsonEnableDateTimeParsingFallback)
+ .getOrElse {
+ SQLConf.get.legacyTimeParserPolicy == LegacyBehaviorPolicy.LEGACY ||
+ options.dateFormatInRead.isEmpty
+ }
+
+ /**
+ * Parses a single XML string and turns it into either one resulting row or no row (if the
+ * the record is malformed).
+ */
+ val parse: String => Option[InternalRow] = {
+ // This is intentionally a val to create a function once and reuse.
+ if (schema.isEmpty) {
+ (_: String) => Some(InternalRow.empty)
+ } else {
+ val xsdSchema = Option(options.rowValidationXSDPath).map(ValidatorUtil.getSchema)
+ (input: String) => doParseColumn(input, options.parseMode, xsdSchema)
+ }
+ }
+
+ def parseStream(
+ inputStream: InputStream,
+ schema: StructType): Iterator[InternalRow] = {
+ val xsdSchema = Option(options.rowValidationXSDPath).map(ValidatorUtil.getSchema)
+ val safeParser = new FailureSafeParser[String](
+ input => doParseColumn(input, options.parseMode, xsdSchema),
+ options.parseMode,
+ schema,
+ options.columnNameOfCorruptRecord)
+
+ val xmlTokenizer = new XmlTokenizer(inputStream, options)
+ convertStream(xmlTokenizer) { tokens =>
+ safeParser.parse(tokens)
+ }.flatten
+ }
+
+ def parseColumn(xml: String, schema: StructType): InternalRow = {
+ // The user=specified schema from from_xml, etc will typically not include a
+ // "corrupted record" column. In PERMISSIVE mode, which puts bad records in
+ // such a column, this would cause an error. In this mode, if such a column
+ // is not manually specified, then fall back to DROPMALFORMED, which will return
+ // null column values where parsing fails.
+ val parseMode =
+ if (options.parseMode == PermissiveMode &&
+ !schema.fields.exists(_.name == options.columnNameOfCorruptRecord)) {
+ DropMalformedMode
+ } else {
+ options.parseMode
+ }
+ val xsdSchema = Option(options.rowValidationXSDPath).map(ValidatorUtil.getSchema)
+ doParseColumn(xml, parseMode, xsdSchema).orNull
+ }
+
+ def doParseColumn(xml: String,
+ parseMode: ParseMode,
+ xsdSchema: Option[Schema]): Option[InternalRow] = {
+ val xmlRecord = UTF8String.fromString(xml)
+ try {
+ xsdSchema.foreach { schema =>
+ schema.newValidator().validate(new StreamSource(new StringReader(xml)))
+ }
+ val parser = StaxXmlParserUtils.filteredReader(xml)
+ val rootAttributes = StaxXmlParserUtils.gatherRootAttributes(parser)
+ Some(convertObject(parser, schema, options, rootAttributes))
+ } catch {
+ case e: SparkUpgradeException => throw e
+ case e@(_: RuntimeException | _: XMLStreamException | _: MalformedInputException
+ | _: SAXException) =>
+ // XML parser currently doesn't support partial results for corrupted records.
+ // For such records, all fields other than the field configured by
+ // `columnNameOfCorruptRecord` are set to `null`.
+ throw BadRecordException(() => xmlRecord, () => Array.empty, e)
+ case e: CharConversionException if options.charset.isEmpty =>
+ val msg =
+ """JSON parser cannot handle a character in its input.
+ |Specifying encoding as an input option explicitly might help to resolve the issue.
+ |""".stripMargin + e.getMessage
+ val wrappedCharException = new CharConversionException(msg)
+ wrappedCharException.initCause(e)
+ throw BadRecordException(() => xmlRecord, () => Array.empty,
+ wrappedCharException)
+ case PartialResultException(row, cause) =>
+ throw BadRecordException(
+ record = () => xmlRecord,
+ partialResults = () => Array(row),
+ cause)
+ case PartialResultArrayException(rows, cause) =>
+ throw BadRecordException(
+ record = () => xmlRecord,
+ partialResults = () => rows,
+ cause)
+ }
+ }
+
+ /**
+ * Parse the current token (and related children) according to a desired schema
+ */
+ private[xml] def convertField(
+ parser: XMLEventReader,
+ dataType: DataType,
+ options: XmlOptions,
+ attributes: Array[Attribute] = Array.empty): Any = {
+
+ def convertComplicatedType(dt: DataType, attributes: Array[Attribute]): Any = dt match {
+ case st: StructType => convertObject(parser, st, options)
+ case MapType(StringType, vt, _) => convertMap(parser, vt, options, attributes)
+ case ArrayType(st, _) => convertField(parser, st, options)
+ case _: StringType =>
+ convertTo(StaxXmlParserUtils.currentStructureAsString(parser), StringType, options)
+ }
+
+ (parser.peek, dataType) match {
+ case (_: StartElement, dt: DataType) => convertComplicatedType(dt, attributes)
+ case (_: EndElement, _: StringType) =>
+ // Empty. It's null if these are explicitly treated as null, or "" is the null value
+ if (options.treatEmptyValuesAsNulls || options.nullValue == "") {
+ null
+ } else {
+ UTF8String.fromString("")
+ }
+ case (_: EndElement, _: DataType) => null
+ case (c: Characters, ArrayType(st, _)) =>
+ // For `ArrayType`, it needs to return the type of element. The values are merged later.
+ convertTo(c.getData, st, options)
+ case (c: Characters, st: StructType) =>
+ // If a value tag is present, this can be an attribute-only element whose values is in that
+ // value tag field. Or, it can be a mixed-type element with both some character elements
+ // and other complex structure. Character elements are ignored.
+ val attributesOnly = st.fields.forall { f =>
+ f.name == options.valueTag || f.name.startsWith(options.attributePrefix)
+ }
+ if (attributesOnly) {
+ // If everything else is an attribute column, there's no complex structure.
+ // Just return the value of the character element, or null if we don't have a value tag
+ st.find(_.name == options.valueTag).map(
+ valueTag => convertTo(c.getData, valueTag.dataType, options)).orNull
+ } else {
+ // Otherwise, ignore this character element, and continue parsing the following complex
+ // structure
+ parser.next
+ parser.peek match {
+ case _: EndElement => null // no struct here at all; done
+ case _ => convertObject(parser, st, options)
+ }
+ }
+ case (_: Characters, _: StringType) =>
+ convertTo(StaxXmlParserUtils.currentStructureAsString(parser), StringType, options)
+ case (c: Characters, _: DataType) if c.isWhiteSpace =>
+ // When `Characters` is found, we need to look further to decide
+ // if this is really data or space between other elements.
+ val data = c.getData
+ parser.next
+ parser.peek match {
+ case _: StartElement => convertComplicatedType(dataType, attributes)
+ case _: EndElement if data.isEmpty => null
+ case _: EndElement if options.treatEmptyValuesAsNulls => null
+ case _: EndElement => convertTo(data, dataType, options)
+ case _ => convertField(parser, dataType, options, attributes)
+ }
+ case (c: Characters, dt: DataType) =>
+ convertTo(c.getData, dt, options)
+ case (e: XMLEvent, dt: DataType) =>
+ throw new IllegalArgumentException(
+ s"Failed to parse a value for data type $dt with event ${e.toString}")
+ }
+ }
+
+ /**
+ * Parse an object as map.
+ */
+ private def convertMap(
+ parser: XMLEventReader,
+ valueType: DataType,
+ options: XmlOptions,
+ attributes: Array[Attribute]): MapData = {
+ val kvPairs = ArrayBuffer.empty[(UTF8String, Any)]
+ attributes.foreach { attr =>
+ kvPairs += (UTF8String.fromString(options.attributePrefix + attr.getName.getLocalPart)
+ -> convertTo(attr.getValue, valueType, options))
+ }
+ var shouldStop = false
+ while (!shouldStop) {
+ parser.nextEvent match {
+ case e: StartElement =>
+ kvPairs +=
+ (UTF8String.fromString(StaxXmlParserUtils.getName(e.asStartElement.getName, options)) ->
+ convertField(parser, valueType, options))
+ case _: EndElement =>
+ shouldStop = StaxXmlParserUtils.checkEndElement(parser)
+ case _ => // do nothing
+ }
+ }
+ ArrayBasedMapData(kvPairs.toMap)
+ }
+
+ /**
+ * Convert XML attributes to a map with the given schema types.
+ */
+ private def convertAttributes(
+ attributes: Array[Attribute],
+ schema: StructType,
+ options: XmlOptions): Map[String, Any] = {
+ val convertedValuesMap = collection.mutable.Map.empty[String, Any]
+ val valuesMap = StaxXmlParserUtils.convertAttributesToValuesMap(attributes, options)
+ valuesMap.foreach { case (f, v) =>
+ val nameToIndex = schema.map(_.name).zipWithIndex.toMap
+ nameToIndex.get(f).foreach { i =>
+ convertedValuesMap(f) = convertTo(v, schema(i).dataType, options)
+ }
+ }
+ convertedValuesMap.toMap
+ }
+
+ /**
+ * [[convertObject()]] calls this in order to convert the nested object to a row.
+ * [[convertObject()]] contains some logic to find out which events are the start
+ * and end of a nested row and this function converts the events to a row.
+ */
+ private def convertObjectWithAttributes(
+ parser: XMLEventReader,
+ schema: StructType,
+ options: XmlOptions,
+ attributes: Array[Attribute] = Array.empty): InternalRow = {
+ // TODO: This method might have to be removed. Some logics duplicate `convertObject()`
+ val row = new Array[Any](schema.length)
+
+ // Read attributes first.
+ val attributesMap = convertAttributes(attributes, schema, options)
+
+ // Then, we read elements here.
+ val fieldsMap = convertField(parser, schema, options) match {
+ case internalRow: InternalRow =>
+ Map(schema.map(_.name).zip(internalRow.toSeq(schema)): _*)
+ case v if schema.fieldNames.contains(options.valueTag) =>
+ // If this is the element having no children, then it wraps attributes
+ // with a row So, we first need to find the field name that has the real
+ // value and then push the value.
+ val valuesMap = schema.fieldNames.map((_, null)).toMap
+ valuesMap + (options.valueTag -> v)
+ case _ => Map.empty
+ }
+
+ // Here we merge both to a row.
+ val valuesMap = fieldsMap ++ attributesMap
+ valuesMap.foreach { case (f, v) =>
+ val nameToIndex = schema.map(_.name).zipWithIndex.toMap
+ nameToIndex.get(f).foreach { row(_) = v }
+ }
+
+ if (valuesMap.isEmpty) {
+ // Return an empty row with all nested elements by the schema set to null.
+ InternalRow.fromSeq(Seq.fill(schema.fieldNames.length)(null))
+ } else {
+ InternalRow.fromSeq(row.toIndexedSeq)
+ }
+ }
+
+ /**
+ * Parse an object from the event stream into a new InternalRow representing the schema.
+ * Fields in the xml that are not defined in the requested schema will be dropped.
+ */
+ private def convertObject(
+ parser: XMLEventReader,
+ schema: StructType,
+ options: XmlOptions,
+ rootAttributes: Array[Attribute] = Array.empty): InternalRow = {
+ val row = new Array[Any](schema.length)
+ val nameToIndex = schema.map(_.name).zipWithIndex.toMap
+ // If there are attributes, then we process them first.
+ convertAttributes(rootAttributes, schema, options).toSeq.foreach { case (f, v) =>
+ nameToIndex.get(f).foreach { row(_) = v }
+ }
+
+ val wildcardColName = options.wildcardColName
+ val hasWildcard = schema.exists(_.name == wildcardColName)
+
+ var badRecordException: Option[Throwable] = None
+
+ var shouldStop = false
+ while (!shouldStop) {
+ parser.nextEvent match {
+ case e: StartElement => try {
+ val attributes = e.getAttributes.asScala.map(_.asInstanceOf[Attribute]).toArray
+ val field = StaxXmlParserUtils.getName(e.asStartElement.getName, options)
+
+ nameToIndex.get(field) match {
+ case Some(index) => schema(index).dataType match {
+ case st: StructType =>
+ row(index) = convertObjectWithAttributes(parser, st, options, attributes)
+
+ case ArrayType(dt: DataType, _) =>
+ val values = Option(row(index))
+ .map(_.asInstanceOf[ArrayBuffer[Any]])
+ .getOrElse(ArrayBuffer.empty[Any])
+ val newValue = dt match {
+ case st: StructType =>
+ convertObjectWithAttributes(parser, st, options, attributes)
+ case dt: DataType =>
+ convertField(parser, dt, options)
+ }
+ row(index) = values :+ newValue
+
+ case dt: DataType =>
+ row(index) = convertField(parser, dt, options, attributes)
+ }
+
+ case None =>
+ if (hasWildcard) {
+ // Special case: there's an 'any' wildcard element that matches anything else
+ // as a string (or array of strings, to parse multiple ones)
+ val newValue = convertField(parser, StringType, options)
+ val anyIndex = schema.fieldIndex(wildcardColName)
+ schema(wildcardColName).dataType match {
+ case StringType =>
+ row(anyIndex) = newValue
+ case ArrayType(StringType, _) =>
+ val values = Option(row(anyIndex))
+ .map(_.asInstanceOf[ArrayBuffer[String]])
+ .getOrElse(ArrayBuffer.empty[String])
+ row(anyIndex) = values :+ newValue
+ }
+ } else {
+ StaxXmlParserUtils.skipChildren(parser)
+ }
+ }
+ } catch {
+ case e: SparkUpgradeException => throw e
+ case NonFatal(e) =>
+ badRecordException = badRecordException.orElse(Some(e))
+ }
+
+ case _: EndElement =>
+ shouldStop = StaxXmlParserUtils.checkEndElement(parser)
+
+ case _ => // do nothing
+ }
+ }
+
+ // TODO: find a more efficient way to convert ArrayBuffer to GenericArrayData
+ val newRow = new Array[Any](schema.length)
+ var i = 0
+ while (i < schema.length) {
+ if (row(i).isInstanceOf[ArrayBuffer[_]]) {
+ newRow(i) = new GenericArrayData(row(i).asInstanceOf[ArrayBuffer[Any]])
+ } else {
+ newRow(i) = row(i)
+ }
+ i += 1;
+ }
+
+ if (badRecordException.isEmpty) {
+ InternalRow.fromSeq(newRow.toIndexedSeq)
+ } else {
+ throw PartialResultException(InternalRow.fromSeq(newRow.toIndexedSeq),
+ badRecordException.get)
+ }
+ }
+}
+
+/**
+ * XMLRecordReader class to read through a given xml document to output xml blocks as records
+ * as specified by the start tag and end tag.
+ *
+ * This implementation is ultimately loosely based on LineRecordReader in Hadoop.
+ */
+private[xml] class XmlTokenizer(
+ inputStream: InputStream,
+ options: XmlOptions) {
+ private val reader = new InputStreamReader(inputStream, Charset.forName(options.charset))
+ private var currentStartTag: String = _
+ private var buffer = new StringBuilder()
+ private val startTag = s"<${options.rowTag}>"
+ private val endTag = s"</${options.rowTag}>"
+
+ /**
+ * Finds the start of the next record.
+ * It treats data from `startTag` and `endTag` as a record.
+ *
+ * @param key the current key that will be written
+ * @param value the object that will be written
+ * @return whether it reads successfully
+ */
+ def next(): Option[String] = {
+ if (readUntilStartElement()) {
+ try {
+ buffer.append(currentStartTag)
+ // Don't check whether the end element was found. Even if not, return everything
+ // that was read, which will invariably cause a parse error later
+ readUntilEndElement(currentStartTag.endsWith(">"))
+ return Some(buffer.toString())
+ } finally {
+ buffer = new StringBuilder()
+ }
+ }
+ None
+ }
+
+ private def readUntilStartElement(): Boolean = {
+ currentStartTag = startTag
+ var i = 0
+ while (true) {
+ val cOrEOF = reader.read()
+ if (cOrEOF == -1) { // || (i == 0 && getFilePosition() > end)) {
+ // End of file or end of split.
+ return false
+ }
+ val c = cOrEOF.toChar
+ if (c == startTag(i)) {
+ if (i >= startTag.length - 1) {
+ // Found start tag.
+ return true
+ }
+ // else in start tag
+ i += 1
+ } else {
+ // if doesn't match the closing angle bracket, check if followed by attributes
+ if (i == (startTag.length - 1) && Character.isWhitespace(c)) {
+ // Found start tag with attributes. Remember to write with following whitespace
+ // char, not angle bracket
+ currentStartTag = startTag.dropRight(1) + c
+ return true
+ }
+ // else not in start tag
+ i = 0
+ }
+ }
+ // Unreachable.
+ false
+ }
+
+ private def readUntilEndElement(startTagClosed: Boolean): Boolean = {
+ // Index into the start or end tag that has matched so far
+ var si = 0
+ var ei = 0
+ // How many other start tags enclose the one that's started already?
+ var depth = 0
+ // Previously read character
+ var prevC = '\u0000'
+
+ // The current start tag already found may or may not have terminated with
+ // a '>' as it may have attributes we read here. If not, we search for
+ // a self-close tag, but only until a non-self-closing end to the start
+ // tag is found
+ var canSelfClose = !startTagClosed
+
+ while (true) {
+
+ val cOrEOF = reader.read()
+ if (cOrEOF == -1) {
+ // End of file (ignore end of split).
+ return false
+ }
+
+ val c = cOrEOF.toChar
+ buffer.append(c)
+
+ if (c == '>' && prevC != '/') {
+ canSelfClose = false
+ }
+
+ // Still matching a start tag?
+ if (c == startTag(si)) {
+ // Still also matching an end tag?
+ if (c == endTag(ei)) {
+ // In start tag or end tag.
+ si += 1
+ ei += 1
+ } else {
+ if (si >= startTag.length - 1) {
+ // Found start tag.
+ si = 0
+ ei = 0
+ depth += 1
+ } else {
+ // In start tag.
+ si += 1
+ ei = 0
+ }
+ }
+ } else if (c == endTag(ei)) {
+ if (ei >= endTag.length - 1) {
+ if (depth == 0) {
+ // Found closing end tag.
+ return true
+ }
+ // else found nested end tag.
+ si = 0
+ ei = 0
+ depth -= 1
+ } else {
+ // In end tag.
+ si = 0
+ ei += 1
+ }
+ } else if (c == '>' && prevC == '/' && canSelfClose) {
+ if (depth == 0) {
+ // found a self-closing tag (end tag)
+ return true
+ }
+ // else found self-closing nested tag (end tag)
+ si = 0
+ ei = 0
+ depth -= 1
+ } else if (si == (startTag.length - 1) && Character.isWhitespace(c)) {
+ // found a start tag with attributes
+ si = 0
+ ei = 0
+ depth += 1
+ } else {
+ // Not in start tag or end tag.
+ si = 0
+ ei = 0
+ }
+ prevC = c
+ }
+ // Unreachable.
+ false
+ }
+}
+
+private[sql] object StaxXmlParser {
+ /**
+ * Parses a stream that contains CSV strings and turns it into an iterator of tokens.
+ */
+ def tokenizeStream(inputStream: InputStream, options: XmlOptions): Iterator[String] = {
+ val xmlTokenizer = new XmlTokenizer(inputStream, options)
+ convertStream(xmlTokenizer)(tokens => tokens)
+ }
+
+ private def convertStream[T](
+ xmlTokenizer: XmlTokenizer)(
+ convert: String => T) = new Iterator[T] {
+
+ private var nextRecord = xmlTokenizer.next()
+
+ override def hasNext: Boolean = nextRecord.nonEmpty
+
+ override def next(): T = {
+ if (!hasNext) {
+ throw QueryExecutionErrors.endOfStreamError()
+ }
+ val curRecord = convert(nextRecord.get)
+ nextRecord = xmlTokenizer.next()
+ curRecord
+ }
+ }
+}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/parsers/StaxXmlParserUtils.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlParserUtils.scala
similarity index 93%
rename from sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/parsers/StaxXmlParserUtils.scala
rename to sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlParserUtils.scala
index c7f31b33839..2bfc6ef4051 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/parsers/StaxXmlParserUtils.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlParserUtils.scala
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.spark.sql.execution.datasources.xml.parsers
+package org.apache.spark.sql.catalyst.xml
import java.io.StringReader
import javax.xml.namespace.QName
@@ -24,11 +24,9 @@ import javax.xml.stream.events._
import scala.annotation.tailrec
import scala.collection.JavaConverters._
-import org.apache.spark.sql.execution.datasources.xml.XmlOptions
+private[sql] object StaxXmlParserUtils {
-private[xml] object StaxXmlParserUtils {
-
- private[xml] val factory: XMLInputFactory = {
+ private[sql] val factory: XMLInputFactory = {
val factory = XMLInputFactory.newInstance()
factory.setProperty(XMLInputFactory.IS_NAMESPACE_AWARE, false)
factory.setProperty(XMLInputFactory.IS_COALESCING, true)
@@ -112,7 +110,7 @@ private[xml] object StaxXmlParserUtils {
def getName(name: QName, options: XmlOptions): String = {
val localPart = name.getLocalPart
// Ignore namespace prefix up to last : if configured
- if (options.ignoreNamespace) {
+ if (options.ignoreNamespace) {
localPart.split(":").last
} else {
localPart
@@ -131,8 +129,12 @@ private[xml] object StaxXmlParserUtils {
xmlString.append('<').append(e.getName)
e.getAttributes.asScala.foreach { a =>
val att = a.asInstanceOf[Attribute]
- xmlString.append(' ').append(att.getName).append("=\"").
- append(att.getValue).append('"')
+ xmlString
+ .append(' ')
+ .append(att.getName)
+ .append("=\"")
+ .append(att.getValue)
+ .append('"')
}
xmlString.append('>')
indent += 1
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/util/TypeCast.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/TypeCast.scala
similarity index 66%
rename from sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/util/TypeCast.scala
rename to sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/TypeCast.scala
index c91f423a62c..a00f372da7f 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/util/TypeCast.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/TypeCast.scala
@@ -14,25 +14,25 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.spark.sql.execution.datasources.xml.util
+package org.apache.spark.sql.catalyst.xml
import java.math.BigDecimal
-import java.sql.{Date, Timestamp}
import java.text.NumberFormat
-import java.time.{Instant, LocalDate, ZoneId}
-import java.time.format.{DateTimeFormatter, DateTimeFormatterBuilder}
import java.util.Locale
import scala.util.Try
import scala.util.control.Exception._
+import scala.util.control.NonFatal
-import org.apache.spark.sql.execution.datasources.xml.XmlOptions
+import org.apache.spark.sql.catalyst.util.DateTimeUtils
+import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf}
import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
/**
* Utility functions for type casting
*/
-private[xml] object TypeCast {
+private[sql] object TypeCast {
/**
* Casts given string datum to specified type.
@@ -44,7 +44,7 @@ private[xml] object TypeCast {
* @param datum string value
* @param castType SparkSQL type
*/
- private[xml] def castTo(
+ private[sql] def castTo(
datum: String,
castType: DataType,
options: XmlOptions): Any = {
@@ -66,7 +66,7 @@ private[xml] object TypeCast {
Decimal(new BigDecimal(datum.replaceAll(",", "")), dt.precision, dt.scale)
case _: TimestampType => parseXmlTimestamp(datum, options)
case _: DateType => parseXmlDate(datum, options)
- case _: StringType => datum
+ case _: StringType => UTF8String.fromString(datum)
case _ => throw new IllegalArgumentException(s"Unsupported type: ${castType.typeName}")
}
}
@@ -80,76 +80,34 @@ private[xml] object TypeCast {
}
}
- private val supportedXmlDateFormatters = Seq(
- // 2011-12-03
- // 2011-12-03+01:00
- DateTimeFormatter.ISO_DATE
- )
-
- private def parseXmlDate(value: String, options: XmlOptions): Date = {
- val formatters = options.dateFormat.map(DateTimeFormatter.ofPattern).
- map(supportedXmlDateFormatters :+ _).getOrElse(supportedXmlDateFormatters)
- formatters.foreach { format =>
- try {
- return Date.valueOf(LocalDate.parse(value, format))
- } catch {
- case _: Exception => // continue
- }
- }
- throw new IllegalArgumentException(s"cannot convert value $value to Date")
+ private def parseXmlDate(value: String, options: XmlOptions): Int = {
+ options.dateFormatter.parse(value)
}
- private val supportedXmlTimestampFormatters = Seq(
- // 2002-05-30 21:46:54
- new DateTimeFormatterBuilder()
- .parseCaseInsensitive()
- .append(DateTimeFormatter.ISO_LOCAL_DATE)
- .appendLiteral(' ')
- .append(DateTimeFormatter.ISO_LOCAL_TIME)
- .toFormatter()
- .withZone(ZoneId.of("UTC")),
- // 2002-05-30T21:46:54
- DateTimeFormatter.ISO_LOCAL_DATE_TIME.withZone(ZoneId.of("UTC")),
- // 2002-05-30T21:46:54+06:00
- DateTimeFormatter.ISO_OFFSET_DATE_TIME,
- // 2002-05-30T21:46:54.1234Z
- DateTimeFormatter.ISO_INSTANT
- )
-
- private def parseXmlTimestamp(value: String, options: XmlOptions): Timestamp = {
- supportedXmlTimestampFormatters.foreach { format =>
- try {
- return Timestamp.from(Instant.from(format.parse(value)))
- } catch {
- case _: Exception => // continue
- }
- }
- options.timestampFormat.foreach { formatString =>
- // Check if there is offset or timezone and apply Spark timeZone if not
- // Useful to support Java 8 and Java 11+ as they prioritize zone and offset differently
- val hasTemporalInformation = formatString.indexOf("V") +
- formatString.indexOf("z") +
- formatString.indexOf("O") +
- formatString.indexOf("X") +
- formatString.indexOf("x") +
- formatString.indexOf("Z") != (-6)
- val format = if (hasTemporalInformation) {
- DateTimeFormatter.ofPattern(formatString)
- } else {
- DateTimeFormatter.ofPattern(formatString).withZone(options.timezone.map(ZoneId.of).orNull)
- }
- try {
- return Timestamp.from(Instant.from(format.parse(value)))
- } catch {
- case _: Exception => // continue
- }
+ private def parseXmlTimestamp(value: String, options: XmlOptions): Long = {
+ try {
+ options.timestampFormatter.parse(value)
+ } catch {
+ case NonFatal(e) =>
+ // If fails to parse, then tries the way used in 2.0 and 1.x for backwards
+ // compatibility if enabled.
+ val enableParsingFallbackForTimestampType =
+ options.enableDateTimeParsingFallback
+ .orElse(SQLConf.get.jsonEnableDateTimeParsingFallback)
+ .getOrElse {
+ SQLConf.get.legacyTimeParserPolicy == LegacyBehaviorPolicy.LEGACY ||
+ options.timestampFormatInRead.isEmpty
+ }
+ if (!enableParsingFallbackForTimestampType) {
+ throw e
+ }
+ val str = DateTimeUtils.cleanLegacyTimestampStr(UTF8String.fromString(value))
+ DateTimeUtils.stringToTimestamp(str, options.zoneId).getOrElse(throw e)
}
- throw new IllegalArgumentException(s"cannot convert value $value to Timestamp")
}
-
- // TODO: This function unnecessarily does type dispatch. Should merge it with `castTo`.
- private[xml] def convertTo(
+ // TODO: This function unnecessarily does type dispatch. Should merge it with `castTo`.
+ private[sql] def convertTo(
datum: String,
dataType: DataType,
options: XmlOptions): Any = {
@@ -184,14 +142,14 @@ private[xml] object TypeCast {
/**
* Helper method that checks and cast string representation of a numeric types.
*/
- private[xml] def isBoolean(value: String): Boolean = {
+ private[sql] def isBoolean(value: String): Boolean = {
value.toLowerCase(Locale.ROOT) match {
case "true" | "false" => true
case _ => false
}
}
- private[xml] def isDouble(value: String): Boolean = {
+ private[sql] def isDouble(value: String): Boolean = {
val signSafeValue = if (value.startsWith("+") || value.startsWith("-")) {
value.substring(1)
} else {
@@ -207,7 +165,7 @@ private[xml] object TypeCast {
(allCatch opt signSafeValue.toDouble).isDefined
}
- private[xml] def isInteger(value: String): Boolean = {
+ private[sql] def isInteger(value: String): Boolean = {
val signSafeValue = if (value.startsWith("+") || value.startsWith("-")) {
value.substring(1)
} else {
@@ -216,7 +174,7 @@ private[xml] object TypeCast {
(allCatch opt signSafeValue.toInt).isDefined
}
- private[xml] def isLong(value: String): Boolean = {
+ private[sql] def isLong(value: String): Boolean = {
val signSafeValue = if (value.startsWith("+") || value.startsWith("-")) {
value.substring(1)
} else {
@@ -225,25 +183,19 @@ private[xml] object TypeCast {
(allCatch opt signSafeValue.toLong).isDefined
}
- private[xml] def isTimestamp(value: String, options: XmlOptions): Boolean = {
+ private[sql] def isTimestamp(value: String, options: XmlOptions): Boolean = {
try {
- parseXmlTimestamp(value, options)
- true
+ options.timestampFormatter.parseOptional(value).isDefined
} catch {
case _: IllegalArgumentException => false
}
}
- private[xml] def isDate(value: String, options: XmlOptions): Boolean = {
- try {
- parseXmlDate(value, options)
- true
- } catch {
- case _: IllegalArgumentException => false
- }
+ private[sql] def isDate(value: String, options: XmlOptions): Boolean = {
+ (allCatch opt options.dateFormatter.parse(value)).isDefined
}
- private[xml] def signSafeToLong(value: String, options: XmlOptions): Long = {
+ private[sql] def signSafeToLong(value: String, options: XmlOptions): Long = {
if (value.startsWith("+")) {
val data = value.substring(1)
TypeCast.castTo(data, LongType, options).asInstanceOf[Long]
@@ -256,7 +208,7 @@ private[xml] object TypeCast {
}
}
- private[xml] def signSafeToDouble(value: String, options: XmlOptions): Double = {
+ private[sql] def signSafeToDouble(value: String, options: XmlOptions): Double = {
if (value.startsWith("+")) {
val data = value.substring(1)
TypeCast.castTo(data, DoubleType, options).asInstanceOf[Double]
@@ -269,7 +221,7 @@ private[xml] object TypeCast {
}
}
- private[xml] def signSafeToInt(value: String, options: XmlOptions): Int = {
+ private[sql] def signSafeToInt(value: String, options: XmlOptions): Int = {
if (value.startsWith("+")) {
val data = value.substring(1)
TypeCast.castTo(data, IntegerType, options).asInstanceOf[Int]
@@ -282,7 +234,7 @@ private[xml] object TypeCast {
}
}
- private[xml] def signSafeToFloat(value: String, options: XmlOptions): Float = {
+ private[sql] def signSafeToFloat(value: String, options: XmlOptions): Float = {
if (value.startsWith("+")) {
val data = value.substring(1)
TypeCast.castTo(data, FloatType, options).asInstanceOf[Float]
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/util/ValidatorUtil.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/ValidatorUtil.scala
similarity index 95%
rename from sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/util/ValidatorUtil.scala
rename to sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/ValidatorUtil.scala
index f3a5214ee0c..6509842fc6d 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/util/ValidatorUtil.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/ValidatorUtil.scala
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.spark.sql.execution.datasources.xml.util
+package org.apache.spark.sql.catalyst.xml
import java.nio.file.Paths
import javax.xml.XMLConstants
@@ -27,7 +27,7 @@ import org.apache.spark.SparkFiles
/**
* Utilities for working with XSD validation.
*/
-private[xml] object ValidatorUtil {
+private[sql] object ValidatorUtil {
// Parsing XSDs may be slow, so cache them by path:
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/util/XSDToSchema.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/XSDToSchema.scala
similarity index 99%
rename from sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/util/XSDToSchema.scala
rename to sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/XSDToSchema.scala
index dc1c524b7a7..ecd601832e1 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/util/XSDToSchema.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/XSDToSchema.scala
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.spark.sql.execution.datasources.xml.util
+package org.apache.spark.sql.catalyst.xml
import java.io.{File, FileInputStream, InputStreamReader, StringReader}
import java.nio.charset.StandardCharsets
@@ -25,7 +25,6 @@ import scala.collection.JavaConverters._
import org.apache.ws.commons.schema._
import org.apache.ws.commons.schema.constants.Constants
-import org.apache.spark.sql.execution.datasources.xml.XmlOptions
import org.apache.spark.sql.types._
/**
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/util/InferSchema.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/XmlInferSchema.scala
similarity index 91%
rename from sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/util/InferSchema.scala
rename to sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/XmlInferSchema.scala
index 4f540d12752..43d1b8fdb86 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/util/InferSchema.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/XmlInferSchema.scala
@@ -14,12 +14,13 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.spark.sql.execution.datasources.xml.util
+package org.apache.spark.sql.catalyst.xml
import java.io.StringReader
import javax.xml.stream.XMLEventReader
-import javax.xml.stream.events.{Attribute, Characters, EndElement, StartElement, XMLEvent}
+import javax.xml.stream.events._
import javax.xml.transform.stream.StreamSource
+import javax.xml.validation.Schema
import scala.annotation.tailrec
import scala.collection.JavaConverters._
@@ -28,12 +29,10 @@ import scala.util.control.NonFatal
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.util.PermissiveMode
-import org.apache.spark.sql.execution.datasources.xml.XmlOptions
-import org.apache.spark.sql.execution.datasources.xml.parsers.StaxXmlParserUtils
-import org.apache.spark.sql.execution.datasources.xml.util.TypeCast._
+import org.apache.spark.sql.catalyst.xml.TypeCast._
import org.apache.spark.sql.types._
-private[xml] object InferSchema {
+private[sql] object XmlInferSchema {
/**
* Copied from internal Spark api
@@ -78,20 +77,7 @@ private[xml] object InferSchema {
val xsdSchema = Option(options.rowValidationXSDPath).map(ValidatorUtil.getSchema)
iter.flatMap { xml =>
- try {
- xsdSchema.foreach { schema =>
- schema.newValidator().validate(new StreamSource(new StringReader(xml)))
- }
-
- val parser = StaxXmlParserUtils.filteredReader(xml)
- val rootAttributes = StaxXmlParserUtils.gatherRootAttributes(parser)
- Some(inferObject(parser, options, rootAttributes))
- } catch {
- case NonFatal(_) if options.parseMode == PermissiveMode =>
- Some(StructType(Seq(StructField(options.columnNameOfCorruptRecord, StringType))))
- case NonFatal(_) =>
- None
- }
+ infer(xml, options, xsdSchema)
}
}.fold(StructType(Seq()))(compatibleType(options))
@@ -103,6 +89,25 @@ private[xml] object InferSchema {
}
}
+ def infer(xml: String,
+ options: XmlOptions,
+ xsdSchema: Option[Schema] = None): Option[DataType] = {
+ try {
+ val xsd = xsdSchema.orElse(Option(options.rowValidationXSDPath).map(ValidatorUtil.getSchema))
+ xsd.foreach { schema =>
+ schema.newValidator().validate(new StreamSource(new StringReader(xml)))
+ }
+ val parser = StaxXmlParserUtils.filteredReader(xml)
+ val rootAttributes = StaxXmlParserUtils.gatherRootAttributes(parser)
+ Some(inferObject(parser, options, rootAttributes))
+ } catch {
+ case NonFatal(_) if options.parseMode == PermissiveMode =>
+ Some(StructType(Seq(StructField(options.columnNameOfCorruptRecord, StringType))))
+ case NonFatal(_) =>
+ None
+ }
+ }
+
private def inferFrom(datum: String, options: XmlOptions): DataType = {
val value = if (datum != null && options.ignoreSurroundingSpaces) {
datum.trim()
@@ -118,8 +123,8 @@ private[xml] object InferSchema {
case v if isInteger(v) => IntegerType
case v if isDouble(v) => DoubleType
case v if isBoolean(v) => BooleanType
- case v if isTimestamp(v, options) => TimestampType
case v if isDate(v, options) => DateType
+ case v if isTimestamp(v, options) => TimestampType
case _ => StringType
}
} else {
@@ -224,7 +229,7 @@ private[xml] object InferSchema {
// This can be inferred as ArrayType.
nameToDataType.foreach {
case (field, dataTypes) if dataTypes.length > 1 =>
- val elementType = dataTypes.reduceLeft(InferSchema.compatibleType(options))
+ val elementType = dataTypes.reduceLeft(XmlInferSchema.compatibleType(options))
builder += StructField(field, ArrayType(elementType), nullable = true)
case (field, dataTypes) =>
builder += StructField(field, dataTypes.head, nullable = true)
@@ -237,7 +242,7 @@ private[xml] object InferSchema {
/**
* Convert NullType to StringType and remove StructTypes with no fields
*/
- private def canonicalizeType(dt: DataType): Option[DataType] = dt match {
+ def canonicalizeType(dt: DataType): Option[DataType] = dt match {
case at @ ArrayType(elementType, _) =>
for {
canonicalType <- canonicalizeType(elementType)
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
new file mode 100644
index 00000000000..3aef9916a89
--- /dev/null
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/XmlOptions.scala
@@ -0,0 +1,229 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.catalyst.xml
+
+import java.nio.charset.StandardCharsets
+import java.time.ZoneId
+import java.util.Locale
+import javax.xml.stream.XMLInputFactory
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.{DataSourceOptions, FileSourceOptions}
+import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, CompressionCodecs, DateFormatter, DateTimeUtils, ParseMode, PermissiveMode, TimestampFormatter}
+import org.apache.spark.sql.catalyst.util.LegacyDateFormats.FAST_DATE_FORMAT
+import org.apache.spark.sql.errors.QueryExecutionErrors
+import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf}
+
+/**
+ * Options for the XML data source.
+ */
+private[sql] class XmlOptions(
+ @transient val parameters: CaseInsensitiveMap[String],
+ defaultTimeZoneId: String,
+ defaultColumnNameOfCorruptRecord: String)
+ extends FileSourceOptions(parameters) with Logging {
+
+ import XmlOptions._
+
+ def this(
+ parameters: Map[String, String] = Map.empty,
+ defaultTimeZoneId: String = SQLConf.get.sessionLocalTimeZone,
+ defaultColumnNameOfCorruptRecord: String = "") = {
+ this(
+ CaseInsensitiveMap(parameters),
+ defaultTimeZoneId,
+ defaultColumnNameOfCorruptRecord)
+ }
+
+ private def getBool(paramName: String, default: Boolean = false): Boolean = {
+ val param = parameters.getOrElse(paramName, default.toString)
+ if (param == null) {
+ default
+ } else if (param.toLowerCase(Locale.ROOT) == "true") {
+ true
+ } else if (param.toLowerCase(Locale.ROOT) == "false") {
+ false
+ } else {
+ throw QueryExecutionErrors.paramIsNotBooleanValueError(paramName)
+ }
+ }
+
+ val compressionCodec = parameters.get("compression").orElse(parameters.get("codec"))
+ .map(CompressionCodecs.getCodecClassName)
+ val rowTag = parameters.getOrElse("rowTag", XmlOptions.DEFAULT_ROW_TAG)
+ require(rowTag.nonEmpty, "'rowTag' option should not be empty string.")
+ require(!rowTag.startsWith("<") && !rowTag.endsWith(">"),
+ "'rowTag' should not include angle brackets")
+ val rootTag = parameters.getOrElse("rootTag", XmlOptions.DEFAULT_ROOT_TAG)
+ require(!rootTag.startsWith("<") && !rootTag.endsWith(">"),
+ "'rootTag' should not include angle brackets")
+ val declaration = parameters.getOrElse("declaration", XmlOptions.DEFAULT_DECLARATION)
+ require(!declaration.startsWith("<") && !declaration.endsWith(">"),
+ "'declaration' should not include angle brackets")
+ val arrayElementName = parameters.getOrElse("arrayElementName",
+ XmlOptions.DEFAULT_ARRAY_ELEMENT_NAME)
+ val samplingRatio = parameters.get("samplingRatio").map(_.toDouble).getOrElse(1.0)
+ require(samplingRatio > 0, s"samplingRatio ($samplingRatio) should be greater than 0")
+ val excludeAttributeFlag = parameters.get("excludeAttribute").map(_.toBoolean).getOrElse(false)
+ val treatEmptyValuesAsNulls =
+ parameters.get("treatEmptyValuesAsNulls").map(_.toBoolean).getOrElse(false)
+ val attributePrefix =
+ parameters.getOrElse("attributePrefix", XmlOptions.DEFAULT_ATTRIBUTE_PREFIX)
+ val valueTag = parameters.getOrElse("valueTag", XmlOptions.DEFAULT_VALUE_TAG)
+ require(valueTag.nonEmpty, "'valueTag' option should not be empty string.")
+ require(valueTag != attributePrefix,
+ "'valueTag' and 'attributePrefix' options should not be the same.")
+ val nullValue = parameters.getOrElse("nullValue", XmlOptions.DEFAULT_NULL_VALUE)
+ val columnNameOfCorruptRecord =
+ parameters.getOrElse("columnNameOfCorruptRecord", "_corrupt_record")
+ val ignoreSurroundingSpaces =
+ parameters.get("ignoreSurroundingSpaces").map(_.toBoolean).getOrElse(false)
+ val parseMode = ParseMode.fromString(parameters.getOrElse("mode", PermissiveMode.name))
+ val inferSchema = parameters.get("inferSchema").map(_.toBoolean).getOrElse(true)
+ val rowValidationXSDPath = parameters.get("rowValidationXSDPath").orNull
+ val wildcardColName =
+ parameters.getOrElse("wildcardColName", XmlOptions.DEFAULT_WILDCARD_COL_NAME)
+ val ignoreNamespace = parameters.get("ignoreNamespace").map(_.toBoolean).getOrElse(false)
+
+ /**
+ * Infer columns with all valid date entries as date type (otherwise inferred as string or
+ * timestamp type) if schema inference is enabled.
+ *
+ * Enabled by default.
+ *
+ * Not compatible with legacyTimeParserPolicy == LEGACY since legacy date parser will accept
+ * extra trailing characters. Thus, disabled when legacyTimeParserPolicy == LEGACY
+ */
+ val preferDate = {
+ if (SQLConf.get.legacyTimeParserPolicy == LegacyBehaviorPolicy.LEGACY) {
+ false
+ } else {
+ getBool(PREFER_DATE, true)
+ }
+ }
+
+ val dateFormatOption: Option[String] = parameters.get(DATE_FORMAT)
+ // Provide a default value for dateFormatInRead when preferDate. This ensures that the
+ // Iso8601DateFormatter (with strict date parsing) is used for date inference
+ val dateFormatInRead: Option[String] =
+ if (preferDate) {
+ Option(dateFormatOption.getOrElse(DateFormatter.defaultPattern))
+ } else {
+ dateFormatOption
+ }
+ val dateFormatInWrite: String = parameters.getOrElse(DATE_FORMAT, DateFormatter.defaultPattern)
+
+
+ val timestampFormatInRead: Option[String] =
+ if (SQLConf.get.legacyTimeParserPolicy == LegacyBehaviorPolicy.LEGACY) {
+ Some(parameters.getOrElse(TIMESTAMP_FORMAT,
+ s"${DateFormatter.defaultPattern}'T'HH:mm:ss.SSSXXX"))
+ } else {
+ parameters.get(TIMESTAMP_FORMAT)
+ }
+ val timestampFormatInWrite: String = parameters.getOrElse(TIMESTAMP_FORMAT,
+ if (SQLConf.get.legacyTimeParserPolicy == LegacyBehaviorPolicy.LEGACY) {
+ s"${DateFormatter.defaultPattern}'T'HH:mm:ss.SSSXXX"
+ } else {
+ s"${DateFormatter.defaultPattern}'T'HH:mm:ss[.SSS][XXX]"
+ })
+
+ // SPARK-39731: Enables the backward compatible parsing behavior.
+ // Generally, this config should be set to false to avoid producing potentially incorrect results
+ // which is the current default (see JacksonParser).
+ //
+ // If enabled and the date cannot be parsed, we will fall back to `DateTimeUtils.stringToDate`.
+ // If enabled and the timestamp cannot be parsed, `DateTimeUtils.stringToTimestamp` will be used.
+ // Otherwise, depending on the parser policy and a custom pattern, an exception may be thrown and
+ // the value will be parsed as null.
+ val enableDateTimeParsingFallback: Option[Boolean] =
+ parameters.get(ENABLE_DATETIME_PARSING_FALLBACK).map(_.toBoolean)
+
+ val timezone = parameters.get("timezone")
+
+ val zoneId: ZoneId = DateTimeUtils.getZoneId(
+ parameters.getOrElse(DateTimeUtils.TIMEZONE_OPTION,
+ parameters.getOrElse(TIME_ZONE, defaultTimeZoneId)))
+
+ // A language tag in IETF BCP 47 format
+ val locale: Locale = parameters.get(LOCALE).map(Locale.forLanguageTag).getOrElse(Locale.US)
+
+ val multiLine = parameters.get(MULTI_LINE).map(_.toBoolean).getOrElse(true)
+ val charset = parameters.getOrElse(ENCODING,
+ parameters.getOrElse(CHARSET, XmlOptions.DEFAULT_CHARSET))
+
+ def buildXmlFactory(): XMLInputFactory = {
+ XMLInputFactory.newInstance()
+ }
+
+ val timestampFormatter = TimestampFormatter(
+ timestampFormatInRead,
+ zoneId,
+ locale,
+ legacyFormat = FAST_DATE_FORMAT,
+ isParsing = true)
+
+ val timestampFormatterInWrite = TimestampFormatter(
+ timestampFormatInWrite,
+ zoneId,
+ locale,
+ legacyFormat = FAST_DATE_FORMAT,
+ isParsing = false)
+
+ val dateFormatter = DateFormatter(
+ dateFormatInRead,
+ locale,
+ legacyFormat = FAST_DATE_FORMAT,
+ isParsing = true)
+
+ val dateFormatterInWrite = DateFormatter(
+ dateFormatInWrite,
+ locale,
+ legacyFormat = FAST_DATE_FORMAT,
+ isParsing = false)
+}
+
+private[sql] object XmlOptions extends DataSourceOptions {
+ val DEFAULT_ATTRIBUTE_PREFIX = "_"
+ val DEFAULT_VALUE_TAG = "_VALUE"
+ val DEFAULT_ROW_TAG = "ROW"
+ val DEFAULT_ROOT_TAG = "ROWS"
+ val DEFAULT_DECLARATION = "version=\"1.0\" encoding=\"UTF-8\" standalone=\"yes\""
+ val DEFAULT_ARRAY_ELEMENT_NAME = "item"
+ val DEFAULT_CHARSET: String = StandardCharsets.UTF_8.name
+ val DEFAULT_NULL_VALUE: String = null
+ val DEFAULT_WILDCARD_COL_NAME = "xs_any"
+ val PREFER_DATE = newOption("preferDate")
+ val LOCALE = newOption("locale")
+ val COMPRESSION = newOption("compression")
+ val ENABLE_DATETIME_PARSING_FALLBACK = newOption("enableDateTimeParsingFallback")
+ val MULTI_LINE = newOption("multiLine")
+ val DATE_FORMAT = newOption("dateFormat")
+ val TIMESTAMP_FORMAT = newOption("timestampFormat")
+ // Options with alternative
+ val ENCODING = "encoding"
+ val CHARSET = "charset"
+ newOption(ENCODING, CHARSET)
+ val TIME_ZONE = "timezone"
+ newOption(DateTimeUtils.TIMEZONE_OPTION, TIME_ZONE)
+
+ def apply(parameters: Map[String, String]): XmlOptions =
+ new XmlOptions(parameters, SQLConf.get.sessionLocalTimeZone)
+
+ def apply(): XmlOptions =
+ new XmlOptions(Map.empty, SQLConf.get.sessionLocalTimeZone)
+}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
index 52ccba8541d..3d2c59cebd8 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
@@ -3301,6 +3301,12 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
messageParameters = Map("jsonSchema" -> toSQLType(schema)))
}
+ def invalidXmlSchema(schema: DataType): Throwable = {
+ new AnalysisException(
+ errorClass = "INVALID_XML_SCHEMA_MAP_TYPE",
+ messageParameters = Map("xmlSchema" -> toSQLType(schema)))
+ }
+
def tableIndexNotSupportedError(errorMessage: String): Throwable = {
new AnalysisException(
errorClass = "_LEGACY_ERROR_TEMP_1332",
diff --git a/sql/core/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister b/sql/core/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister
index db418b73636..3169e75031f 100644
--- a/sql/core/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister
+++ b/sql/core/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister
@@ -22,7 +22,7 @@ org.apache.spark.sql.execution.datasources.noop.NoopDataSource
org.apache.spark.sql.execution.datasources.orc.OrcFileFormat
org.apache.spark.sql.execution.datasources.v2.parquet.ParquetDataSourceV2
org.apache.spark.sql.execution.datasources.v2.text.TextDataSourceV2
-org.apache.spark.sql.execution.datasources.xml.DefaultSource
+org.apache.spark.sql.execution.datasources.xml.XmlFileFormat
org.apache.spark.sql.execution.streaming.ConsoleSinkProvider
org.apache.spark.sql.execution.streaming.sources.RateStreamProvider
org.apache.spark.sql.execution.streaming.sources.TextSocketSourceProvider
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/DataFrameReader.scala b/sql/core/src/main/scala/org/apache/spark/sql/DataFrameReader.scala
index 443b18f7281..c57c96f9c07 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/DataFrameReader.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/DataFrameReader.scala
@@ -31,6 +31,7 @@ import org.apache.spark.sql.catalyst.csv.{CSVHeaderChecker, CSVOptions, Univocit
import org.apache.spark.sql.catalyst.expressions.ExprUtils
import org.apache.spark.sql.catalyst.json.{CreateJacksonParser, JacksonParser, JSONOptions}
import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, CharVarcharUtils, FailureSafeParser}
+import org.apache.spark.sql.catalyst.xml.{StaxXmlParser, XmlOptions}
import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.sql.execution.command.DDLUtils
import org.apache.spark.sql.execution.datasources.DataSource
@@ -39,6 +40,8 @@ import org.apache.spark.sql.execution.datasources.jdbc._
import org.apache.spark.sql.execution.datasources.json.JsonUtils.checkJsonSchema
import org.apache.spark.sql.execution.datasources.json.TextInputJsonDataSource
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Utils
+import org.apache.spark.sql.execution.datasources.xml.TextInputXmlDataSource
+import org.apache.spark.sql.execution.datasources.xml.XmlUtils.checkXmlSchema
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.util.CaseInsensitiveStringMap
@@ -537,6 +540,74 @@ class DataFrameReader private[sql](sparkSession: SparkSession) extends Logging {
@scala.annotation.varargs
def csv(paths: String*): DataFrame = format("csv").load(paths : _*)
+ /**
+ * Loads a XML file and returns the result as a `DataFrame`. See the documentation on the
+ * other overloaded `xml()` method for more details.
+ *
+ * @since 4.0.0
+ */
+ def xml(path: String): DataFrame = {
+ // This method ensures that calls that explicit need single argument works, see SPARK-16009
+ xml(Seq(path): _*)
+ }
+
+ /**
+ * Loads XML files and returns the result as a `DataFrame`.
+ *
+ * This function will go through the input once to determine the input schema if `inferSchema`
+ * is enabled. To avoid going through the entire data once, disable `inferSchema` option or
+ * specify the schema explicitly using `schema`.
+ *
+ * You can find the XML-specific options for reading XML files in
+ * <a href="https://spark.apache.org/docs/latest/sql-data-sources-xml.html#data-source-option">
+ * Data Source Option</a> in the version you use.
+ *
+ * @since 4.0.0
+ */
+ @scala.annotation.varargs
+ def xml(paths: String*): DataFrame = format("xml").load(paths: _*)
+
+ /**
+ * Loads an `Dataset[String]` storing XML object and returns the result as a `DataFrame`.
+ *
+ * If the schema is not specified using `schema` function and `inferSchema` option is enabled,
+ * this function goes through the input once to determine the input schema.
+ *
+ * @param xmlDataset input Dataset with one XML object per record
+ * @since 4.0.0
+ */
+ def xml(xmlDataset: Dataset[String]): DataFrame = {
+ val parsedOptions: XmlOptions = new XmlOptions(
+ extraOptions.toMap,
+ sparkSession.sessionState.conf.sessionLocalTimeZone,
+ sparkSession.sessionState.conf.columnNameOfCorruptRecord)
+
+ userSpecifiedSchema.foreach(checkXmlSchema)
+
+ val schema = userSpecifiedSchema.map {
+ case s if !SQLConf.get.getConf(
+ SQLConf.LEGACY_RESPECT_NULLABILITY_IN_TEXT_DATASET_CONVERSION) => s.asNullable
+ case other => other
+ }.getOrElse {
+ TextInputXmlDataSource.inferFromDataset(xmlDataset, parsedOptions)
+ }
+
+ ExprUtils.verifyColumnNameOfCorruptRecord(schema, parsedOptions.columnNameOfCorruptRecord)
+ val actualSchema =
+ StructType(schema.filterNot(_.name == parsedOptions.columnNameOfCorruptRecord))
+
+ val parsed = xmlDataset.rdd.mapPartitions { iter =>
+ val rawParser = new StaxXmlParser(actualSchema, parsedOptions)
+ val parser = new FailureSafeParser[String](
+ input => rawParser.parse(input),
+ parsedOptions.parseMode,
+ schema,
+ parsedOptions.columnNameOfCorruptRecord)
+ iter.flatMap(parser.parse)
+ }
+ sparkSession.internalCreateDataFrame(parsed, schema, isStreaming = xmlDataset.isStreaming)
+ }
+
/**
* Loads a Parquet file, returning the result as a `DataFrame`. See the documentation
* on the other overloaded `parquet()` method for more details.
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala b/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala
index 4de6b944bc8..76424faca28 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala
@@ -850,6 +850,38 @@ final class DataFrameWriter[T] private[sql](ds: Dataset[T]) {
format("csv").save(path)
}
+ /**
+ * Saves the content of the `DataFrame` in XML format at the specified path.
+ * This is equivalent to:
+ * {{{
+ * format("xml").save(path)
+ * }}}
+ *
+ * Note that writing a XML file from `DataFrame` having a field `ArrayType` with
+ * its element as `ArrayType` would have an additional nested field for the element.
+ * For example, the `DataFrame` having a field below,
+ *
+ * {@code fieldA [[data1], [data2]]}
+ *
+ * would produce a XML file below.
+ * {@code
+ * <fieldA>
+ * <item>data1</item>
+ * </fieldA>
+ * <fieldA>
+ * <item>data2</item>
+ * </fieldA>}
+ *
+ * Namely, roundtrip in writing and reading can end up in different schema structure.
+ *
+ * You can find the XML-specific options for writing XML files in
+ * <a href="https://spark.apache.org/docs/latest/sql-data-sources-xml.html#data-source-option">
+ * Data Source Option</a> in the version you use.
+ */
+ def xml(path: String): Unit = {
+ format("xml").save(path)
+ }
+
/**
* Wrap a DataFrameWriter action to track the QueryExecution and time cost, then report to the
* user-registered callback functions.
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala
index 94dd3bc0bd6..b1129dad6b1 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala
@@ -44,6 +44,7 @@ import org.apache.spark.sql.execution.datasources.orc.OrcFileFormat
import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
import org.apache.spark.sql.execution.datasources.v2.FileDataSourceV2
import org.apache.spark.sql.execution.datasources.v2.orc.OrcDataSourceV2
+import org.apache.spark.sql.execution.datasources.xml.XmlFileFormat
import org.apache.spark.sql.execution.streaming._
import org.apache.spark.sql.execution.streaming.sources.{RateStreamProvider, TextSocketSourceProvider}
import org.apache.spark.sql.internal.SQLConf
@@ -567,6 +568,7 @@ object DataSource extends Logging {
private val backwardCompatibilityMap: Map[String, String] = {
val jdbc = classOf[JdbcRelationProvider].getCanonicalName
val json = classOf[JsonFileFormat].getCanonicalName
+ val xml = classOf[XmlFileFormat].getCanonicalName
val parquet = classOf[ParquetFileFormat].getCanonicalName
val csv = classOf[CSVFileFormat].getCanonicalName
val libsvm = "org.apache.spark.ml.source.libsvm.LibSVMFileFormat"
@@ -595,6 +597,8 @@ object DataSource extends Logging {
"org.apache.spark.ml.source.libsvm.DefaultSource" -> libsvm,
"org.apache.spark.ml.source.libsvm" -> libsvm,
"com.databricks.spark.csv" -> csv,
+ "com.databricks.spark.xml" -> xml,
+ "org.apache.spark.sql.execution.datasources.xml" -> xml,
"org.apache.spark.sql.execution.streaming.TextSocketSourceProvider" -> socket,
"org.apache.spark.sql.execution.streaming.RateSourceProvider" -> rate
)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/DefaultSource.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/DefaultSource.scala
deleted file mode 100644
index a1da349321b..00000000000
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/DefaultSource.scala
+++ /dev/null
@@ -1,113 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.spark.sql.execution.datasources.xml
-
-import org.apache.hadoop.fs.Path
-
-import org.apache.spark.sql.{DataFrame, SaveMode, SQLContext}
-import org.apache.spark.sql.execution.datasources.xml.util.XmlFile
-import org.apache.spark.sql.sources._
-import org.apache.spark.sql.types.StructType
-
-/**
- * Provides access to XML data from pure SQL statements (i.e. for users of the
- * JDBC server).
- */
-class DefaultSource
- extends RelationProvider
- with SchemaRelationProvider
- with CreatableRelationProvider
- with DataSourceRegister {
-
- /**
- * Short alias for spark-xml data source.
- */
- override def shortName(): String = "xml"
-
- private def checkPath(parameters: Map[String, String]): String = {
- parameters.getOrElse("path",
- throw new IllegalArgumentException("'path' must be specified for XML data."))
- }
-
- /**
- * Creates a new relation for data store in XML given parameters.
- * Parameters have to include 'path'.
- */
- override def createRelation(
- sqlContext: SQLContext,
- parameters: Map[String, String]): BaseRelation = {
- createRelation(sqlContext, parameters, null)
- }
-
- /**
- * Creates a new relation for data store in XML given parameters and user supported schema.
- * Parameters have to include 'path'.
- */
- override def createRelation(
- sqlContext: SQLContext,
- parameters: Map[String, String],
- schema: StructType): XmlRelation = {
- val path = checkPath(parameters)
- // We need the `charset` and `rowTag` before creating the relation.
- val (charset, rowTag) = {
- val options = XmlOptions(parameters)
- (options.charset, options.rowTag)
- }
-
- val paramsWithTZ =
- sqlContext.sparkContext.getConf.getOption("spark.sql.session.timeZone") match {
- case Some(tz) => parameters.updated("timezone", tz)
- case None => parameters
- }
-
- XmlRelation(
- () => XmlFile.withCharset(sqlContext.sparkContext, path, charset, rowTag),
- Some(path),
- paramsWithTZ,
- schema)(sqlContext)
- }
-
- override def createRelation(
- sqlContext: SQLContext,
- mode: SaveMode,
- parameters: Map[String, String],
- data: DataFrame): BaseRelation = {
- val path = checkPath(parameters)
- val filesystemPath = new Path(path)
- val fs = filesystemPath.getFileSystem(sqlContext.sparkContext.hadoopConfiguration)
- val doSave = if (fs.exists(filesystemPath)) {
- mode match {
- case SaveMode.Append =>
- throw new IllegalArgumentException(
- s"Append mode is not supported by ${this.getClass.getCanonicalName}")
- case SaveMode.Overwrite =>
- fs.delete(filesystemPath, true)
- true
- case SaveMode.ErrorIfExists =>
- throw new IllegalArgumentException(s"path $path already exists.")
- case SaveMode.Ignore => false
- }
- } else {
- true
- }
- if (doSave) {
- // Only save data when the save mode is not ignore.
- XmlFile.saveAsXmlFile(data, filesystemPath.toString, parameters)
- }
- createRelation(sqlContext, parameters, data.schema)
- }
-}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlDataSource.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlDataSource.scala
new file mode 100644
index 00000000000..d96d80d6ce5
--- /dev/null
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlDataSource.scala
@@ -0,0 +1,207 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources.xml
+
+import java.nio.charset.{Charset, StandardCharsets}
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.{FileStatus, Path}
+import org.apache.hadoop.mapreduce.Job
+import org.apache.hadoop.mapreduce.lib.input.FileInputFormat
+
+import org.apache.spark.TaskContext
+import org.apache.spark.input.{PortableDataStream, StreamInputFormat}
+import org.apache.spark.internal.Logging
+import org.apache.spark.rdd.{BinaryFileRDD, RDD}
+import org.apache.spark.sql.{Dataset, Encoders, SparkSession}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.util.FailureSafeParser
+import org.apache.spark.sql.catalyst.xml.{StaxXmlParser, XmlInferSchema, XmlOptions}
+import org.apache.spark.sql.execution.SQLExecution
+import org.apache.spark.sql.execution.datasources._
+import org.apache.spark.sql.execution.datasources.text.TextFileFormat
+import org.apache.spark.sql.types.StructType
+
+/**
+ * Common functions for parsing XML files
+ */
+abstract class XmlDataSource extends Serializable {
+ def isSplitable: Boolean
+
+ /**
+ * Parse a [[PartitionedFile]] into [[InternalRow]] instances.
+ */
+ def readFile(
+ conf: Configuration,
+ file: PartitionedFile,
+ parser: StaxXmlParser,
+ schema: StructType): Iterator[InternalRow]
+
+ /**
+ * Infers the schema from `inputPaths` files.
+ */
+ final def inferSchema(
+ sparkSession: SparkSession,
+ inputPaths: Seq[FileStatus],
+ parsedOptions: XmlOptions): Option[StructType] = {
+ if (inputPaths.nonEmpty) {
+ Some(infer(sparkSession, inputPaths, parsedOptions))
+ } else {
+ None
+ }
+ }
+
+ protected def infer(
+ sparkSession: SparkSession,
+ inputPaths: Seq[FileStatus],
+ parsedOptions: XmlOptions): StructType
+}
+
+object XmlDataSource extends Logging {
+ def apply(options: XmlOptions): XmlDataSource = {
+ if (options.multiLine) {
+ MultiLineXmlDataSource
+ } else {
+ TextInputXmlDataSource
+ }
+ }
+}
+
+object TextInputXmlDataSource extends XmlDataSource {
+ override val isSplitable: Boolean = true
+
+ override def readFile(
+ conf: Configuration,
+ file: PartitionedFile,
+ parser: StaxXmlParser,
+ schema: StructType): Iterator[InternalRow] = {
+ val lines = {
+ val linesReader = new HadoopFileLinesReader(file, None, conf)
+ Option(TaskContext.get()).foreach(_.addTaskCompletionListener[Unit](_ => linesReader.close()))
+ linesReader.map { line =>
+ new String(line.getBytes, 0, line.getLength, parser.options.charset)
+ }
+ }
+
+ val safeParser = new FailureSafeParser[String](
+ input => parser.parse(input),
+ parser.options.parseMode,
+ schema,
+ parser.options.columnNameOfCorruptRecord)
+
+ lines.flatMap(safeParser.parse)
+ }
+
+ override def infer(
+ sparkSession: SparkSession,
+ inputPaths: Seq[FileStatus],
+ parsedOptions: XmlOptions): StructType = {
+ val xml = createBaseDataset(sparkSession, inputPaths, parsedOptions)
+ inferFromDataset(xml, parsedOptions)
+ }
+
+ /**
+ * Infers the schema from `Dataset` that stores CSV string records.
+ */
+ def inferFromDataset(
+ xml: Dataset[String],
+ parsedOptions: XmlOptions): StructType = {
+ XmlInferSchema.infer(xml.rdd, parsedOptions)
+ }
+
+ private def createBaseDataset(
+ sparkSession: SparkSession,
+ inputPaths: Seq[FileStatus],
+ options: XmlOptions): Dataset[String] = {
+ val paths = inputPaths.map(_.getPath.toString)
+ val df = sparkSession.baseRelationToDataFrame(
+ DataSource.apply(
+ sparkSession,
+ paths = paths,
+ className = classOf[TextFileFormat].getName,
+ options = options.parameters ++ Map(DataSource.GLOB_PATHS_KEY -> "false")
+ ).resolveRelation(checkFilesExist = false))
+ .select("value").as[String](Encoders.STRING)
+
+ if (Charset.forName(options.charset) == StandardCharsets.UTF_8) {
+ df
+ } else {
+ val charset = options.charset
+ sparkSession.createDataset(df.queryExecution.toRdd.map { row =>
+ val bytes = row.getBinary(0)
+ new String(bytes, 0, bytes.length, charset)
+ })(Encoders.STRING)
+ }
+ }
+}
+
+object MultiLineXmlDataSource extends XmlDataSource {
+ override val isSplitable: Boolean = false
+
+ override def readFile(
+ conf: Configuration,
+ file: PartitionedFile,
+ parser: StaxXmlParser,
+ requiredSchema: StructType): Iterator[InternalRow] = {
+ parser.parseStream(
+ CodecStreams.createInputStreamWithCloseResource(conf, file.toPath),
+ requiredSchema)
+ }
+
+ override def infer(
+ sparkSession: SparkSession,
+ inputPaths: Seq[FileStatus],
+ parsedOptions: XmlOptions): StructType = {
+ val xml = createBaseRdd(sparkSession, inputPaths, parsedOptions)
+
+ val tokenRDD = xml.flatMap { portableDataStream =>
+ StaxXmlParser.tokenizeStream(
+ CodecStreams.createInputStreamWithCloseResource(
+ portableDataStream.getConfiguration,
+ new Path(portableDataStream.getPath())),
+ parsedOptions)
+ }
+ SQLExecution.withSQLConfPropagated(sparkSession) {
+ val schema = XmlInferSchema.infer(tokenRDD, parsedOptions)
+ schema
+ }
+ }
+
+ private def createBaseRdd(
+ sparkSession: SparkSession,
+ inputPaths: Seq[FileStatus],
+ options: XmlOptions): RDD[PortableDataStream] = {
+ val paths = inputPaths.map(_.getPath)
+ val name = paths.mkString(",")
+ val job = Job.getInstance(sparkSession.sessionState.newHadoopConfWithOptions(
+ options.parameters))
+ FileInputFormat.setInputPaths(job, paths: _*)
+ val conf = job.getConfiguration
+
+ val rdd = new BinaryFileRDD(
+ sparkSession.sparkContext,
+ classOf[StreamInputFormat],
+ classOf[String],
+ classOf[PortableDataStream],
+ conf,
+ sparkSession.sparkContext.defaultMinPartitions)
+
+ // Only returns `PortableDataStream`s without paths.
+ rdd.setName(s"XMLFile: $name").values
+ }
+}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlDataToCatalyst.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlDataToCatalyst.scala
deleted file mode 100644
index 95e9743d302..00000000000
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlDataToCatalyst.scala
+++ /dev/null
@@ -1,62 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.spark.sql.execution.datasources.xml
-
-import org.apache.spark.sql.catalyst.CatalystTypeConverters
-import org.apache.spark.sql.catalyst.expressions.{ExpectsInputTypes, Expression, UnaryExpression}
-import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
-import org.apache.spark.sql.catalyst.util.GenericArrayData
-import org.apache.spark.sql.execution.datasources.xml.parsers.StaxXmlParser
-import org.apache.spark.sql.types._
-import org.apache.spark.unsafe.types.UTF8String
-
-case class XmlDataToCatalyst(
- child: Expression,
- schema: DataType,
- options: XmlOptions)
- extends UnaryExpression with CodegenFallback with ExpectsInputTypes {
-
- override lazy val dataType: DataType = schema
-
- @transient
- lazy val rowSchema: StructType = schema match {
- case st: StructType => st
- case ArrayType(st: StructType, _) => st
- }
-
- override def nullSafeEval(xml: Any): Any = xml match {
- case string: UTF8String =>
- CatalystTypeConverters.convertToCatalyst(
- StaxXmlParser.parseColumn(string.toString, rowSchema, options))
- case string: String =>
- StaxXmlParser.parseColumn(string, rowSchema, options)
- case arr: GenericArrayData =>
- CatalystTypeConverters.convertToCatalyst(
- arr.array.map(s => StaxXmlParser.parseColumn(s.toString, rowSchema, options)))
- case arr: Array[_] =>
- arr.map(s => StaxXmlParser.parseColumn(s.toString, rowSchema, options))
- case _ => null
- }
-
- override def inputTypes: Seq[DataType] = schema match {
- case _: StructType => Seq(StringType)
- case ArrayType(_: StructType, _) => Seq(ArrayType(StringType))
- }
-
- // Overrides, in Spark 3.2.0+
- protected def withNewChildInternal(newChild: Expression): XmlDataToCatalyst = copy(newChild)
-}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlFileFormat.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlFileFormat.scala
new file mode 100644
index 00000000000..baacf7f0748
--- /dev/null
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlFileFormat.scala
@@ -0,0 +1,157 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources.xml
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.{FileStatus, Path}
+import org.apache.hadoop.mapreduce._
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.ExprUtils
+import org.apache.spark.sql.catalyst.util.CompressionCodecs
+import org.apache.spark.sql.catalyst.xml.{StaxXmlParser, XmlOptions}
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.execution.datasources._
+import org.apache.spark.sql.sources._
+import org.apache.spark.sql.types._
+import org.apache.spark.util.SerializableConfiguration
+
+/**
+ * Provides access to XML data from pure SQL statements.
+ */
+class XmlFileFormat extends TextBasedFileFormat with DataSourceRegister {
+
+ override def shortName(): String = "xml"
+
+ def getXmlOptions(
+ sparkSession: SparkSession,
+ parameters: Map[String, String]): XmlOptions = {
+ new XmlOptions(parameters,
+ sparkSession.sessionState.conf.sessionLocalTimeZone,
+ sparkSession.sessionState.conf.columnNameOfCorruptRecord)
+ }
+
+ override def isSplitable(
+ sparkSession: SparkSession,
+ options: Map[String, String],
+ path: Path): Boolean = {
+ val xmlOptions = getXmlOptions(sparkSession, options)
+ val xmlDataSource = XmlDataSource(xmlOptions)
+ xmlDataSource.isSplitable && super.isSplitable(sparkSession, options, path)
+ }
+
+ override def inferSchema(
+ sparkSession: SparkSession,
+ options: Map[String, String],
+ files: Seq[FileStatus]): Option[StructType] = {
+ val xmlOptions = getXmlOptions(sparkSession, options)
+
+ XmlDataSource(xmlOptions).inferSchema(
+ sparkSession, files, xmlOptions)
+ }
+
+ override def prepareWrite(
+ sparkSession: SparkSession,
+ job: Job,
+ options: Map[String, String],
+ dataSchema: StructType): OutputWriterFactory = {
+ val conf = job.getConfiguration
+ val xmlOptions = getXmlOptions(sparkSession, options)
+
+ xmlOptions.compressionCodec.foreach { codec =>
+ CompressionCodecs.setCodecConfiguration(conf, codec)
+ }
+
+ new OutputWriterFactory {
+ override def newInstance(
+ path: String,
+ dataSchema: StructType,
+ context: TaskAttemptContext): OutputWriter = {
+ new XmlOutputWriter(path, dataSchema, context, xmlOptions)
+ }
+
+ override def getFileExtension(context: TaskAttemptContext): String = {
+ ".xml" + CodecStreams.getCompressionExtension(context)
+ }
+ }
+ }
+
+ override def buildReader(
+ sparkSession: SparkSession,
+ dataSchema: StructType,
+ partitionSchema: StructType,
+ requiredSchema: StructType,
+ filters: Seq[Filter],
+ options: Map[String, String],
+ hadoopConf: Configuration): (PartitionedFile) => Iterator[InternalRow] = {
+ val broadcastedHadoopConf =
+ sparkSession.sparkContext.broadcast(new SerializableConfiguration(hadoopConf))
+
+ val xmlOptions = getXmlOptions(sparkSession, options)
+
+ val columnNameOfCorruptRecord = xmlOptions.columnNameOfCorruptRecord
+ // Check a field requirement for corrupt records here to throw an exception in a driver side
+ ExprUtils.verifyColumnNameOfCorruptRecord(dataSchema, columnNameOfCorruptRecord)
+ // Don't push any filter which refers to the "virtual" column which cannot present in the input.
+ // Such filters will be applied later on the upper layer.
+ val actualFilters =
+ filters.filterNot(_.references.contains(columnNameOfCorruptRecord))
+ val actualRequiredSchema = StructType(
+ requiredSchema.filterNot(_.name == columnNameOfCorruptRecord))
+ if (requiredSchema.length == 1 &&
+ requiredSchema.head.name == columnNameOfCorruptRecord) {
+ throw QueryCompilationErrors.queryFromRawFilesIncludeCorruptRecordColumnError()
+ }
+
+ (file: PartitionedFile) => {
+ val parser = new StaxXmlParser(
+ actualRequiredSchema,
+ xmlOptions,
+ actualFilters)
+ XmlDataSource(xmlOptions).readFile(
+ broadcastedHadoopConf.value.value,
+ file,
+ parser,
+ requiredSchema)
+ }
+ }
+
+ override def toString: String = "XML"
+
+ override def hashCode(): Int = getClass.hashCode()
+
+ override def equals(other: Any): Boolean = other.isInstanceOf[XmlFileFormat]
+
+ override def supportDataType(dataType: DataType): Boolean = dataType match {
+ case _: AtomicType => true
+
+ case st: StructType => st.forall { f => supportDataType(f.dataType) }
+
+ case ArrayType(elementType, _) => supportDataType(elementType)
+
+ case MapType(keyType, valueType, _) =>
+ supportDataType(keyType) && supportDataType(valueType)
+
+ case udt: UserDefinedType[_] => supportDataType(udt.sqlType)
+
+ case _: NullType => true
+
+ case _ => false
+ }
+}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlInputFormat.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlInputFormat.scala
index 40c32d3aa76..4359ac02f5f 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlInputFormat.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlInputFormat.scala
@@ -27,6 +27,8 @@ import org.apache.hadoop.io.compress._
import org.apache.hadoop.mapreduce.{InputSplit, RecordReader, TaskAttemptContext}
import org.apache.hadoop.mapreduce.lib.input.{FileSplit, TextInputFormat}
+import org.apache.spark.sql.catalyst.xml.XmlOptions
+
/**
* Reads records that are delimited by a specific start/end tag.
*/
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlOptions.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlOptions.scala
deleted file mode 100644
index ad930d74ffb..00000000000
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlOptions.scala
+++ /dev/null
@@ -1,85 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.spark.sql.execution.datasources.xml
-
-import java.nio.charset.StandardCharsets
-
-import org.apache.spark.sql.catalyst.util.{ParseMode, PermissiveMode}
-
-/**
- * Options for the XML data source.
- */
-private[xml] class XmlOptions(
- @transient private val parameters: Map[String, String])
- extends Serializable {
-
- def this() = this(Map.empty)
-
- val charset = parameters.getOrElse("charset", XmlOptions.DEFAULT_CHARSET)
- val codec = parameters.get("compression").orElse(parameters.get("codec")).orNull
- val rowTag = parameters.getOrElse("rowTag", XmlOptions.DEFAULT_ROW_TAG)
- require(rowTag.nonEmpty, "'rowTag' option should not be empty string.")
- require(!rowTag.startsWith("<") && !rowTag.endsWith(">"),
- "'rowTag' should not include angle brackets")
- val rootTag = parameters.getOrElse("rootTag", XmlOptions.DEFAULT_ROOT_TAG)
- require(!rootTag.startsWith("<") && !rootTag.endsWith(">"),
- "'rootTag' should not include angle brackets")
- val declaration = parameters.getOrElse("declaration", XmlOptions.DEFAULT_DECLARATION)
- require(!declaration.startsWith("<") && !declaration.endsWith(">"),
- "'declaration' should not include angle brackets")
- val arrayElementName = parameters.getOrElse("arrayElementName",
- XmlOptions.DEFAULT_ARRAY_ELEMENT_NAME)
- val samplingRatio = parameters.get("samplingRatio").map(_.toDouble).getOrElse(1.0)
- require(samplingRatio > 0, s"samplingRatio ($samplingRatio) should be greater than 0")
- val excludeAttributeFlag = parameters.get("excludeAttribute").map(_.toBoolean).getOrElse(false)
- val treatEmptyValuesAsNulls =
- parameters.get("treatEmptyValuesAsNulls").map(_.toBoolean).getOrElse(false)
- val attributePrefix =
- parameters.getOrElse("attributePrefix", XmlOptions.DEFAULT_ATTRIBUTE_PREFIX)
- val valueTag = parameters.getOrElse("valueTag", XmlOptions.DEFAULT_VALUE_TAG)
- require(valueTag.nonEmpty, "'valueTag' option should not be empty string.")
- require(valueTag != attributePrefix,
- "'valueTag' and 'attributePrefix' options should not be the same.")
- val nullValue = parameters.getOrElse("nullValue", XmlOptions.DEFAULT_NULL_VALUE)
- val columnNameOfCorruptRecord =
- parameters.getOrElse("columnNameOfCorruptRecord", "_corrupt_record")
- val ignoreSurroundingSpaces =
- parameters.get("ignoreSurroundingSpaces").map(_.toBoolean).getOrElse(false)
- val parseMode = ParseMode.fromString(parameters.getOrElse("mode", PermissiveMode.name))
- val inferSchema = parameters.get("inferSchema").map(_.toBoolean).getOrElse(true)
- val rowValidationXSDPath = parameters.get("rowValidationXSDPath").orNull
- val wildcardColName =
- parameters.getOrElse("wildcardColName", XmlOptions.DEFAULT_WILDCARD_COL_NAME)
- val ignoreNamespace = parameters.get("ignoreNamespace").map(_.toBoolean).getOrElse(false)
- val timestampFormat = parameters.get("timestampFormat")
- val timezone = parameters.get("timezone")
- val dateFormat = parameters.get("dateFormat")
-}
-
-private[xml] object XmlOptions {
- val DEFAULT_ATTRIBUTE_PREFIX = "_"
- val DEFAULT_VALUE_TAG = "_VALUE"
- val DEFAULT_ROW_TAG = "ROW"
- val DEFAULT_ROOT_TAG = "ROWS"
- val DEFAULT_DECLARATION = "version=\"1.0\" encoding=\"UTF-8\" standalone=\"yes\""
- val DEFAULT_ARRAY_ELEMENT_NAME = "item"
- val DEFAULT_CHARSET: String = StandardCharsets.UTF_8.name
- val DEFAULT_NULL_VALUE: String = null
- val DEFAULT_WILDCARD_COL_NAME = "xs_any"
-
- def apply(parameters: Map[String, String]): XmlOptions = new XmlOptions(parameters)
-}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlOutputWriter.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlOutputWriter.scala
new file mode 100644
index 00000000000..cde866dcedf
--- /dev/null
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlOutputWriter.scala
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.execution.datasources.xml
+
+import java.nio.charset.Charset
+import javax.xml.stream.XMLOutputFactory
+
+import com.sun.xml.txw2.output.IndentingXMLStreamWriter
+import org.apache.hadoop.fs.Path
+import org.apache.hadoop.mapreduce.TaskAttemptContext
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.{xml, InternalRow}
+import org.apache.spark.sql.catalyst.xml.XmlOptions
+import org.apache.spark.sql.execution.datasources.{CodecStreams, OutputWriter}
+import org.apache.spark.sql.types.StructType
+
+class XmlOutputWriter(
+ val path: String,
+ dataSchema: StructType,
+ context: TaskAttemptContext,
+ options: XmlOptions) extends OutputWriter with Logging {
+
+ private val DEFAULT_INDENT = " "
+ private val charset = Charset.forName(options.charset)
+
+ private val writer = CodecStreams.createOutputStreamWriter(context, new Path(path), charset)
+ private val factory = XMLOutputFactory.newInstance()
+ private val xmlWriter = factory.createXMLStreamWriter(writer)
+ private val indentingXmlWriter = new IndentingXMLStreamWriter(xmlWriter)
+ indentingXmlWriter.setIndentStep(DEFAULT_INDENT)
+
+ // Allow a root tag to be like "rootTag foo='bar'"
+ // This is hacky; won't deal correctly with spaces in attributes, but want
+ // to make this at least work for simple cases without much complication
+ private val rootTagTokens = options.rootTag.split(" ")
+ private val rootElementName = rootTagTokens.head
+ private val rootAttributes: Map[String, String] =
+ if (rootTagTokens.length > 1) {
+ rootTagTokens.tail.map { kv =>
+ val Array(k, v) = kv.split("=")
+ k -> v.replaceAll("['\"]", "")
+ }.toMap
+ } else {
+ Map.empty
+ }
+ private val declaration = options.declaration
+
+
+ // private val gen = new UnivocityGenerator(dataSchema, writer, params)
+
+ private var firstRow: Boolean = true
+
+ override def write(row: InternalRow): Unit = {
+ if (firstRow) {
+ if (declaration != null && declaration.nonEmpty) {
+ indentingXmlWriter.writeProcessingInstruction("xml", declaration)
+ indentingXmlWriter.writeCharacters("\n")
+ }
+ indentingXmlWriter.writeStartElement(rootElementName)
+ rootAttributes.foreach { case (k, v) =>
+ indentingXmlWriter.writeAttribute(k, v)
+ }
+ firstRow = false
+ }
+ xml.StaxXmlGenerator(
+ dataSchema,
+ indentingXmlWriter,
+ options)(row)
+ }
+
+ override def close(): Unit = {
+ if (!firstRow) {
+ indentingXmlWriter.writeEndElement()
+ indentingXmlWriter.close()
+ }
+ writer.close()
+ }
+}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlReader.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlReader.scala
deleted file mode 100644
index 80be4fb0072..00000000000
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlReader.scala
+++ /dev/null
@@ -1,205 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.spark.sql.execution.datasources.xml
-
-import org.apache.spark.rdd.RDD
-import org.apache.spark.sql.{DataFrame, Dataset, SparkSession, SQLContext}
-import org.apache.spark.sql.catalyst.util.FailFastMode
-import org.apache.spark.sql.execution.datasources.xml.util.XmlFile
-import org.apache.spark.sql.types.StructType
-
-/**
- * A collection of static functions for working with XML files in Spark SQL
- */
-class XmlReader(private var schema: StructType,
- private val options: Map[String, Any]) extends Serializable {
-
- private val parameters = collection.mutable.Map.empty[String, String]
- options.foreach { case (k, v) => parameters(k) = v.toString }
-
- // Explicit constructors for Java compatibility
-
- def this() = {
- this(null, Map.empty)
- }
-
- def this(schema: StructType) = {
- this(schema, Map.empty)
- }
-
- def this(options: Map[String, Any]) = {
- this(null, options)
- }
-
- @deprecated("Use XmlReader(Map) with key 'charset' to specify options", "0.13.0")
- def withCharset(charset: String): XmlReader = {
- parameters += ("charset" -> charset)
- this
- }
-
- @deprecated("Use XmlReader(Map) with key 'codec' to specify options", "0.13.0")
- def withCompression(codec: String): XmlReader = {
- parameters += ("codec" -> codec)
- this
- }
-
- @deprecated("Use XmlReader(Map) with key 'rowTag' to specify options", "0.13.0")
- def withRowTag(rowTag: String): XmlReader = {
- parameters += ("rowTag" -> rowTag)
- this
- }
-
- @deprecated("Use XmlReader(Map) with key 'samplingRatio' to specify options", "0.13.0")
- def withSamplingRatio(samplingRatio: Double): XmlReader = {
- parameters += ("samplingRatio" -> samplingRatio.toString)
- this
- }
-
- @deprecated("Use XmlReader(Map) with key 'excludeAttribute' to specify options", "0.13.0")
- def withExcludeAttribute(exclude: Boolean): XmlReader = {
- parameters += ("excludeAttribute" -> exclude.toString)
- this
- }
-
- @deprecated("Use XmlReader(Map) with key 'treatEmptyValuesAsNulls' to specify options", "0.13.0")
- def withTreatEmptyValuesAsNulls(treatAsNull: Boolean): XmlReader = {
- parameters += ("treatEmptyValuesAsNulls" -> treatAsNull.toString)
- this
- }
-
- @deprecated("Use XmlReader(Map) with key 'mode' as 'FAILFAST' to specify options", "0.10.0")
- def withFailFast(failFast: Boolean): XmlReader = {
- if (failFast) {
- parameters += ("mode" -> FailFastMode.name)
- } else {
- parameters -= "mode"
- }
- this
- }
-
- @deprecated("Use XmlReader(Map) with key 'mode' to specify options", "0.13.0")
- def withParseMode(mode: String): XmlReader = {
- parameters += ("mode" -> mode)
- this
- }
-
- @deprecated("Use XmlReader(Map) with key 'attributePrefix' to specify options", "0.13.0")
- def withAttributePrefix(attributePrefix: String): XmlReader = {
- parameters += ("attributePrefix" -> attributePrefix)
- this
- }
-
- @deprecated("Use XmlReader(Map) with key 'valueTag' to specify options", "0.13.0")
- def withValueTag(valueTag: String): XmlReader = {
- parameters += ("valueTag" -> valueTag)
- this
- }
-
- @deprecated("Use XmlReader(Map) with key 'columnNameOfCorruptRecord' to specify options",
- "0.13.0")
- def withColumnNameOfCorruptRecord(name: String): XmlReader = {
- parameters += ("columnNameOfCorruptRecord" -> name)
- this
- }
-
- @deprecated("Use XmlReader(Map) with key 'ignoreSurroundingSpaces' to specify options", "0.13.0")
- def withIgnoreSurroundingSpaces(ignore: Boolean): XmlReader = {
- parameters += ("ignoreSurroundingSpaces" -> ignore.toString)
- this
- }
-
- @deprecated("Use XmlReader(StructType) to specify schema", "0.13.0")
- def withSchema(schema: StructType): XmlReader = {
- this.schema = schema
- this
- }
-
- @deprecated("Use XmlReader(Map) with key 'rowValidationXSDPath' to specify options", "0.13.0")
- def withRowValidationXSDPath(path: String): XmlReader = {
- parameters += ("rowValidationXSDPath" -> path)
- this
- }
-
- /**
- * @param spark current SparkSession
- * @param path path to XML files to parse
- * @return XML parsed as a DataFrame
- */
- def xmlFile(spark: SparkSession, path: String): DataFrame = {
- // We need the `charset` and `rowTag` before creating the relation.
- val (charset, rowTag) = {
- val options = XmlOptions(parameters.toMap)
- (options.charset, options.rowTag)
- }
- val relation = XmlRelation(
- () => XmlFile.withCharset(spark.sparkContext, path, charset, rowTag),
- Some(path),
- parameters.toMap,
- schema)(spark.sqlContext)
- spark.baseRelationToDataFrame(relation)
- }
-
- /**
- * @param spark current SparkSession
- * @param ds XML for individual 'rows' as Strings
- * @return XML parsed as a DataFrame
- */
- def xmlDataset(spark: SparkSession, ds: Dataset[String]): DataFrame = {
- xmlRdd(spark, ds.rdd)
- }
-
- /**
- * @param spark current SparkSession
- * @param xmlRDD XML for individual 'rows' as Strings
- * @return XML parsed as a DataFrame
- */
- def xmlRdd(spark: SparkSession, xmlRDD: RDD[String]): DataFrame = {
- val relation = XmlRelation(
- () => xmlRDD,
- None,
- parameters.toMap,
- schema)(spark.sqlContext)
- spark.baseRelationToDataFrame(relation)
- }
-
- /** Returns a Schema RDD for the given XML path. */
- @deprecated("Use xmlFile(SparkSession, ...)", "0.5.0")
- def xmlFile(sqlContext: SQLContext, path: String): DataFrame = {
- // We need the `charset` and `rowTag` before creating the relation.
- val (charset, rowTag) = {
- val options = XmlOptions(parameters.toMap)
- (options.charset, options.rowTag)
- }
- val relation = XmlRelation(
- () => XmlFile.withCharset(sqlContext.sparkContext, path, charset, rowTag),
- Some(path),
- parameters.toMap,
- schema)(sqlContext)
- sqlContext.baseRelationToDataFrame(relation)
- }
-
- @deprecated("Use xmlRdd(SparkSession, ...)", "0.5.0")
- def xmlRdd(sqlContext: SQLContext, xmlRDD: RDD[String]): DataFrame = {
- val relation = XmlRelation(
- () => xmlRDD,
- None,
- parameters.toMap,
- schema)(sqlContext)
- sqlContext.baseRelationToDataFrame(relation)
- }
-
-}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlRelation.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlRelation.scala
deleted file mode 100644
index 18bdb87fa61..00000000000
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlRelation.scala
+++ /dev/null
@@ -1,86 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.spark.sql.execution.datasources.xml
-
-import java.io.IOException
-
-import org.apache.hadoop.fs.Path
-
-import org.apache.spark.rdd.RDD
-import org.apache.spark.sql._
-import org.apache.spark.sql.execution.datasources.xml.parsers.StaxXmlParser
-import org.apache.spark.sql.execution.datasources.xml.util.{InferSchema, XmlFile}
-import org.apache.spark.sql.sources.{BaseRelation, InsertableRelation, PrunedScan}
-import org.apache.spark.sql.types._
-
-case class XmlRelation protected[spark] (
- baseRDD: () => RDD[String],
- location: Option[String],
- parameters: Map[String, String],
- userSchema: StructType = null)(@transient val sqlContext: SQLContext)
- extends BaseRelation
- with InsertableRelation
- with PrunedScan {
-
- // Hacky: ensure RDD's underlying data actually already exists early on
- baseRDD().partitions
-
- private val options = XmlOptions(parameters)
-
- override val schema: StructType = {
- Option(userSchema).getOrElse {
- InferSchema.infer(
- baseRDD(),
- options)
- }
- }
-
- override def buildScan(requiredColumns: Array[String]): RDD[Row] = {
- val requiredFields = requiredColumns.map(schema(_))
- val requestedSchema = StructType(requiredFields)
- StaxXmlParser.parse(
- baseRDD(),
- requestedSchema,
- options)
- }
-
- // The function below was borrowed from JSONRelation
- override def insert(data: DataFrame, overwrite: Boolean): Unit = {
- val filesystemPath = location match {
- case Some(p) => new Path(p)
- case None =>
- throw new IOException(s"Cannot INSERT into table with no path defined")
- }
-
- val fs = filesystemPath.getFileSystem(sqlContext.sparkContext.hadoopConfiguration)
-
- if (overwrite) {
- try {
- fs.delete(filesystemPath, true)
- } catch {
- case e: IOException =>
- throw new IOException(
- s"Unable to clear output directory ${filesystemPath.toString} prior"
- + s" to INSERT OVERWRITE a XML table:\n${e.toString}")
- }
- // Write the data. We assume that schema isn't changed, and we won't update it.
- XmlFile.saveAsXmlFile(data, filesystemPath.toString, parameters)
- } else {
- throw new IllegalArgumentException("XML tables only support INSERT OVERWRITE for now.")
- }
- }
-}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlUtils.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlUtils.scala
new file mode 100644
index 00000000000..01aa1fad75d
--- /dev/null
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlUtils.scala
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources.xml
+
+import org.apache.spark.SparkException
+import org.apache.spark.input.PortableDataStream
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.Dataset
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{DataTypeMismatch, TypeCheckSuccess}
+import org.apache.spark.sql.catalyst.expressions.ExprUtils
+import org.apache.spark.sql.catalyst.xml.XmlOptions
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.types.DataType
+
+object XmlUtils {
+ /**
+ * Sample XML dataset as configured by `samplingRatio`.
+ */
+ def sample(xml: Dataset[String], options: XmlOptions): Dataset[String] = {
+ require(options.samplingRatio > 0,
+ s"samplingRatio (${options.samplingRatio}) should be greater than 0")
+ if (options.samplingRatio > 0.99) {
+ xml
+ } else {
+ xml.sample(withReplacement = false, options.samplingRatio, 1)
+ }
+ }
+
+ /**
+ * Sample XML RDD as configured by `samplingRatio`.
+ */
+ def sample(xml: RDD[PortableDataStream], options: XmlOptions): RDD[PortableDataStream] = {
+ require(options.samplingRatio > 0,
+ s"samplingRatio (${options.samplingRatio}) should be greater than 0")
+ if (options.samplingRatio > 0.99) {
+ xml
+ } else {
+ xml.sample(withReplacement = false, options.samplingRatio, 1)
+ }
+ }
+
+ def checkXmlSchema(schema: DataType): Unit = {
+ ExprUtils.checkXmlSchema(schema) match {
+ case DataTypeMismatch("INVALID_XML_MAP_KEY_TYPE", _) =>
+ throw QueryCompilationErrors.invalidXmlSchema(schema)
+ case TypeCheckSuccess =>
+ case result =>
+ throw SparkException.internalError(s"Unknown type check result: $result.")
+ }
+ }
+}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/functions.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/functions.scala
deleted file mode 100644
index 1a22652ac45..00000000000
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/functions.scala
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.spark.sql.execution.datasources.xml
-
-import org.apache.spark.sql.Column
-import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
-import org.apache.spark.sql.types.DataType
-
-/**
- * Support functions for working with XML columns directly.
- */
-// scalastyle:off: object.name
-object functions {
-// scalastyle:on: object.name
- /**
- * Parses a column containing a XML string into a `StructType` with the specified schema.
- *
- * @param e a string column containing XML data
- * @param schema the schema to use when parsing the XML string. Must be a StructType if
- * column is string-valued, or ArrayType[StructType] if column is an array of strings
- * @param options key-value pairs that correspond to those supported by [[XmlOptions]]
- */
- def from_xml(e: Column, schema: DataType, options: Map[String, String] = Map.empty): Column = {
- val expr = CatalystSqlParser.parseExpression(e.toString())
- new Column(XmlDataToCatalyst(expr, schema, XmlOptions(options)))
- }
-
-}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/package.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/package.scala
deleted file mode 100644
index 7531f1f6f48..00000000000
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/package.scala
+++ /dev/null
@@ -1,162 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.spark.sql.execution.datasources
-
-import org.apache.hadoop.io.compress.CompressionCodec
-
-import org.apache.spark.sql.{DataFrame, DataFrameReader, DataFrameWriter, Dataset}
-import org.apache.spark.sql.{Encoders, Row, SparkSession, SQLContext}
-import org.apache.spark.sql.execution.datasources.xml.parsers.StaxXmlParser
-import org.apache.spark.sql.execution.datasources.xml.util.{InferSchema, XmlFile}
-import org.apache.spark.sql.types.{ArrayType, StructType}
-
-package object xml {
- /**
- * Adds a method, `xmlFile`, to [[SQLContext]] that allows reading XML data.
- */
- implicit class XmlContext(sqlContext: SQLContext) extends Serializable {
- @deprecated("Use read.format(\"xml\") or read.xml", "0.4.0")
- def xmlFile(
- filePath: String,
- rowTag: String = XmlOptions.DEFAULT_ROW_TAG,
- samplingRatio: Double = 1.0,
- excludeAttribute: Boolean = false,
- treatEmptyValuesAsNulls: Boolean = false,
- failFast: Boolean = false,
- attributePrefix: String = XmlOptions.DEFAULT_ATTRIBUTE_PREFIX,
- valueTag: String = XmlOptions.DEFAULT_VALUE_TAG,
- charset: String = XmlOptions.DEFAULT_CHARSET): DataFrame = {
-
- val parameters = Map(
- "rowTag" -> rowTag,
- "samplingRatio" -> samplingRatio.toString,
- "excludeAttribute" -> excludeAttribute.toString,
- "treatEmptyValuesAsNulls" -> treatEmptyValuesAsNulls.toString,
- "failFast" -> failFast.toString,
- "attributePrefix" -> attributePrefix,
- "valueTag" -> valueTag,
- "charset" -> charset)
- val xmlRelation = XmlRelation(
- () => XmlFile.withCharset(sqlContext.sparkContext, filePath, charset, rowTag),
- location = Some(filePath),
- parameters = parameters)(sqlContext)
- sqlContext.baseRelationToDataFrame(xmlRelation)
- }
- }
-
- /**
- * Adds a method, `saveAsXmlFile`, to [[DataFrame]] that allows writing XML data.
- * If compressionCodec is not null the resulting output will be compressed.
- * Note that a codec entry in the parameters map will be ignored.
- */
- implicit class XmlSchemaRDD(dataFrame: DataFrame) {
- @deprecated("Use write.format(\"xml\") or write.xml", "0.4.0")
- def saveAsXmlFile(
- path: String, parameters: scala.collection.Map[String, String] = Map(),
- compressionCodec: Class[_ <: CompressionCodec] = null): Unit = {
- val mutableParams = collection.mutable.Map(parameters.toSeq: _*)
- val safeCodec = mutableParams.get("codec")
- .orElse(Option(compressionCodec).map(_.getCanonicalName))
- .orNull
- mutableParams.put("codec", safeCodec)
- XmlFile.saveAsXmlFile(dataFrame, path, mutableParams.toMap)
- }
- }
-
- /**
- * Adds a method, `xml`, to DataFrameReader that allows you to read XML files using
- * the DataFileReader
- */
- implicit class XmlDataFrameReader(reader: DataFrameReader) {
- def xml: String => DataFrame = reader
- .format("org.apache.spark.sql.execution.datasources.xml").load
-
- @deprecated("Use XmlReader directly", "0.13.0")
- def xml(xmlDataset: Dataset[String]): DataFrame = {
- val spark = SparkSession.builder().getOrCreate()
- new XmlReader().xmlDataset(spark, xmlDataset)
- }
- }
-
- /**
- * Adds a method, `xml`, to DataFrameWriter that allows you to write XML files using
- * the DataFileWriter
- */
- implicit class XmlDataFrameWriter[T](writer: DataFrameWriter[T]) {
- // Note that writing a XML file from [[DataFrame]] having a field [[ArrayType]] with
- // its element as [[ArrayType]] would have an additional nested field for the element.
- // For example, the [[DataFrame]] having a field below,
- //
- // fieldA [[data1, data2]]
- //
- // would produce a XML file below.
- //
- // <fieldA>
- // <item>data1</item>
- // </fieldA>
- // <fieldA>
- // <item>data2</item>
- // </fieldA>
- //
- // Namely, roundtrip in writing and reading can end up in different schema structure.
- def xml: String => Unit = writer
- .format("org.apache.spark.sql.execution.datasources.xml").save
- }
-
- /**
- * Infers the schema of XML documents as strings.
- *
- * @param ds Dataset of XML strings
- * @param options additional XML parsing options
- * @return inferred schema for XML
- */
- def schema_of_xml(ds: Dataset[String], options: Map[String, String] = Map.empty): StructType =
- InferSchema.infer(ds.rdd, XmlOptions(options))
-
- /**
- * Infers the schema of XML documents as strings.
- *
- * @param df one-column DataFrame of XML strings
- * @param options additional XML parsing options
- * @return inferred schema for XML
- */
- def schema_of_xml_df(df: DataFrame, options: Map[String, String] = Map.empty): StructType =
- schema_of_xml(df.as[String](Encoders.STRING), options)
-
- /**
- * Infers the schema of XML documents when inputs are arrays of strings, each an XML doc.
- *
- * @param ds Dataset of XML strings
- * @param options additional XML parsing options
- * @return inferred schema for XML. Will be an ArrayType[StructType].
- */
- def schema_of_xml_array(ds: Dataset[Array[String]],
- options: Map[String, String] = Map.empty): ArrayType =
- ArrayType(InferSchema.infer(ds.rdd.flatMap(a => a), XmlOptions(options)))
-
- /**
- * @param xml XML document to parse, as string
- * @param schema the schema to use when parsing the XML string
- * @param options key-value pairs that correspond to those supported by [[XmlOptions]]
- * @return [[Row]] representing the parsed XML structure
- */
- def from_xml_string(xml: String, schema: StructType,
- options: Map[String, String] = Map.empty): Row = {
- StaxXmlParser.parseColumn(xml, schema, XmlOptions(options))
- }
-
-}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/parsers/StaxXmlParser.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/parsers/StaxXmlParser.scala
deleted file mode 100644
index 89b140fe28e..00000000000
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/parsers/StaxXmlParser.scala
+++ /dev/null
@@ -1,375 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.spark.sql.execution.datasources.xml.parsers
-
-import java.io.StringReader
-import javax.xml.stream.XMLEventReader
-import javax.xml.stream.events.{Attribute, Characters, EndElement, StartElement, XMLEvent}
-import javax.xml.transform.stream.StreamSource
-import javax.xml.validation.Schema
-
-import scala.collection.JavaConverters._
-import scala.collection.mutable.ArrayBuffer
-import scala.util.Try
-import scala.util.control.NonFatal
-
-import org.apache.spark.internal.Logging
-import org.apache.spark.rdd.RDD
-import org.apache.spark.sql.Row
-import org.apache.spark.sql.catalyst.util.{DropMalformedMode, FailFastMode, ParseMode, PermissiveMode}
-import org.apache.spark.sql.execution.datasources.xml.XmlOptions
-import org.apache.spark.sql.execution.datasources.xml.util._
-import org.apache.spark.sql.execution.datasources.xml.util.TypeCast._
-import org.apache.spark.sql.types._
-
-/**
- * Wraps parser to iteration process.
- */
-private[xml] object StaxXmlParser extends Serializable with Logging {
- def parse(
- xml: RDD[String],
- schema: StructType,
- options: XmlOptions): RDD[Row] = {
- xml.mapPartitions { iter =>
- val xsdSchema = Option(options.rowValidationXSDPath).map(ValidatorUtil.getSchema)
- iter.flatMap { xml =>
- doParseColumn(xml, schema, options, options.parseMode, xsdSchema)
- }
- }
- }
-
- def parseColumn(xml: String, schema: StructType, options: XmlOptions): Row = {
- // The user=specified schema from from_xml, etc will typically not include a
- // "corrupted record" column. In PERMISSIVE mode, which puts bad records in
- // such a column, this would cause an error. In this mode, if such a column
- // is not manually specified, then fall back to DROPMALFORMED, which will return
- // null column values where parsing fails.
- val parseMode =
- if (options.parseMode == PermissiveMode &&
- !schema.fields.exists(_.name == options.columnNameOfCorruptRecord)) {
- DropMalformedMode
- } else {
- options.parseMode
- }
- val xsdSchema = Option(options.rowValidationXSDPath).map(ValidatorUtil.getSchema)
- doParseColumn(xml, schema, options, parseMode, xsdSchema).orNull
- }
-
- private def doParseColumn(xml: String,
- schema: StructType,
- options: XmlOptions,
- parseMode: ParseMode,
- xsdSchema: Option[Schema]): Option[Row] = {
- try {
- xsdSchema.foreach { schema =>
- schema.newValidator().validate(new StreamSource(new StringReader(xml)))
- }
- val parser = StaxXmlParserUtils.filteredReader(xml)
- val rootAttributes = StaxXmlParserUtils.gatherRootAttributes(parser)
- Some(convertObject(parser, schema, options, rootAttributes))
- } catch {
- case e: PartialResultException =>
- failedRecord(xml, options, parseMode, schema,
- e.cause, Some(e.partialResult))
- case NonFatal(e) =>
- failedRecord(xml, options, parseMode, schema, e)
- }
- }
-
- private def failedRecord(record: String,
- options: XmlOptions,
- parseMode: ParseMode,
- schema: StructType,
- cause: Throwable = null,
- partialResult: Option[Row] = None): Option[Row] = {
- // create a row even if no corrupt record column is present
- val abbreviatedRecord =
- (if (record.length() > 1000) record.substring(0, 1000) + "..." else record).
- replaceAll("\n", "")
- parseMode match {
- case FailFastMode =>
- logInfo(s"Malformed line: $abbreviatedRecord")
- logDebug("Caused by:", cause)
- throw new IllegalArgumentException("Malformed line in FAILFAST mode", cause)
- case DropMalformedMode =>
- logInfo(s"Malformed line: $abbreviatedRecord")
- logDebug("Caused by:", cause)
- None
- case PermissiveMode =>
- logDebug(s"Malformed line: $abbreviatedRecord")
- logDebug("Caused by:", cause)
- // The logic below is borrowed from Apache Spark's FailureSafeParser.
- val resultRow = new Array[Any](schema.length)
- schema.filterNot(_.name == options.columnNameOfCorruptRecord).foreach { from =>
- val sourceIndex = schema.fieldIndex(from.name)
- resultRow(sourceIndex) = partialResult.map(_.get(sourceIndex)).orNull
- }
- val corruptFieldIndex = Try(schema.fieldIndex(options.columnNameOfCorruptRecord)).toOption
- corruptFieldIndex.foreach(resultRow(_) = record)
- Some(Row.fromSeq(resultRow.toIndexedSeq))
- }
- }
-
- /**
- * Parse the current token (and related children) according to a desired schema
- */
- private[xml] def convertField(
- parser: XMLEventReader,
- dataType: DataType,
- options: XmlOptions,
- attributes: Array[Attribute] = Array.empty): Any = {
-
- def convertComplicatedType(dt: DataType, attributes: Array[Attribute]): Any = dt match {
- case st: StructType => convertObject(parser, st, options)
- case MapType(StringType, vt, _) => convertMap(parser, vt, options, attributes)
- case ArrayType(st, _) => convertField(parser, st, options)
- case _: StringType =>
- convertTo(StaxXmlParserUtils.currentStructureAsString(parser), StringType, options)
- }
-
- (parser.peek, dataType) match {
- case (_: StartElement, dt: DataType) => convertComplicatedType(dt, attributes)
- case (_: EndElement, _: StringType) =>
- // Empty. It's null if these are explicitly treated as null, or "" is the null value
- if (options.treatEmptyValuesAsNulls || options.nullValue == "") {
- null
- } else {
- ""
- }
- case (_: EndElement, _: DataType) => null
- case (c: Characters, ArrayType(st, _)) =>
- // For `ArrayType`, it needs to return the type of element. The values are merged later.
- convertTo(c.getData, st, options)
- case (c: Characters, st: StructType) =>
- // If a value tag is present, this can be an attribute-only element whose values is in that
- // value tag field. Or, it can be a mixed-type element with both some character elements
- // and other complex structure. Character elements are ignored.
- val attributesOnly = st.fields.forall { f =>
- f.name == options.valueTag || f.name.startsWith(options.attributePrefix)
- }
- if (attributesOnly) {
- // If everything else is an attribute column, there's no complex structure.
- // Just return the value of the character element, or null if we don't have a value tag
- st.find(_.name == options.valueTag).map(
- valueTag => convertTo(c.getData, valueTag.dataType, options)).orNull
- } else {
- // Otherwise, ignore this character element, and continue parsing the following complex
- // structure
- parser.next
- parser.peek match {
- case _: EndElement => null // no struct here at all; done
- case _ => convertObject(parser, st, options)
- }
- }
- case (_: Characters, _: StringType) =>
- convertTo(StaxXmlParserUtils.currentStructureAsString(parser), StringType, options)
- case (c: Characters, _: DataType) if c.isWhiteSpace =>
- // When `Characters` is found, we need to look further to decide
- // if this is really data or space between other elements.
- val data = c.getData
- parser.next
- parser.peek match {
- case _: StartElement => convertComplicatedType(dataType, attributes)
- case _: EndElement if data.isEmpty => null
- case _: EndElement if options.treatEmptyValuesAsNulls => null
- case _: EndElement => convertTo(data, dataType, options)
- case _ => convertField(parser, dataType, options, attributes)
- }
- case (c: Characters, dt: DataType) =>
- convertTo(c.getData, dt, options)
- case (e: XMLEvent, dt: DataType) =>
- throw new IllegalArgumentException(
- s"Failed to parse a value for data type $dt with event ${e.toString}")
- }
- }
-
- /**
- * Parse an object as map.
- */
- private def convertMap(
- parser: XMLEventReader,
- valueType: DataType,
- options: XmlOptions,
- attributes: Array[Attribute]): Map[String, Any] = {
- val kvPairs = ArrayBuffer.empty[(String, Any)]
- attributes.foreach { attr =>
- kvPairs += (options.attributePrefix + attr.getName.getLocalPart -> attr.getValue)
- }
- var shouldStop = false
- while (!shouldStop) {
- parser.nextEvent match {
- case e: StartElement =>
- kvPairs +=
- (StaxXmlParserUtils.getName(e.asStartElement.getName, options) ->
- convertField(parser, valueType, options))
- case _: EndElement =>
- shouldStop = StaxXmlParserUtils.checkEndElement(parser)
- case _ => // do nothing
- }
- }
- kvPairs.toMap
- }
-
- /**
- * Convert XML attributes to a map with the given schema types.
- */
- private def convertAttributes(
- attributes: Array[Attribute],
- schema: StructType,
- options: XmlOptions): Map[String, Any] = {
- val convertedValuesMap = collection.mutable.Map.empty[String, Any]
- val valuesMap = StaxXmlParserUtils.convertAttributesToValuesMap(attributes, options)
- valuesMap.foreach { case (f, v) =>
- val nameToIndex = schema.map(_.name).zipWithIndex.toMap
- nameToIndex.get(f).foreach { i =>
- convertedValuesMap(f) = convertTo(v, schema(i).dataType, options)
- }
- }
- convertedValuesMap.toMap
- }
-
- /**
- * [[convertObject()]] calls this in order to convert the nested object to a row.
- * [[convertObject()]] contains some logic to find out which events are the start
- * and end of a nested row and this function converts the events to a row.
- */
- private def convertObjectWithAttributes(
- parser: XMLEventReader,
- schema: StructType,
- options: XmlOptions,
- attributes: Array[Attribute] = Array.empty): Row = {
- // TODO: This method might have to be removed. Some logics duplicate `convertObject()`
- val row = new Array[Any](schema.length)
-
- // Read attributes first.
- val attributesMap = convertAttributes(attributes, schema, options)
-
- // Then, we read elements here.
- val fieldsMap = convertField(parser, schema, options) match {
- case row: Row =>
- Map(schema.map(_.name).zip(row.toSeq): _*)
- case v if schema.fieldNames.contains(options.valueTag) =>
- // If this is the element having no children, then it wraps attributes
- // with a row So, we first need to find the field name that has the real
- // value and then push the value.
- val valuesMap = schema.fieldNames.map((_, null)).toMap
- valuesMap + (options.valueTag -> v)
- case _ => Map.empty
- }
-
- // Here we merge both to a row.
- val valuesMap = fieldsMap ++ attributesMap
- valuesMap.foreach { case (f, v) =>
- val nameToIndex = schema.map(_.name).zipWithIndex.toMap
- nameToIndex.get(f).foreach { row(_) = v }
- }
-
- if (valuesMap.isEmpty) {
- // Return an empty row with all nested elements by the schema set to null.
- Row.fromSeq(Seq.fill(schema.fieldNames.length)(null))
- } else {
- Row.fromSeq(row.toIndexedSeq)
- }
- }
-
- /**
- * Parse an object from the event stream into a new Row representing the schema.
- * Fields in the xml that are not defined in the requested schema will be dropped.
- */
- private def convertObject(
- parser: XMLEventReader,
- schema: StructType,
- options: XmlOptions,
- rootAttributes: Array[Attribute] = Array.empty): Row = {
- val row = new Array[Any](schema.length)
- val nameToIndex = schema.map(_.name).zipWithIndex.toMap
- // If there are attributes, then we process them first.
- convertAttributes(rootAttributes, schema, options).toSeq.foreach { case (f, v) =>
- nameToIndex.get(f).foreach { row(_) = v }
- }
-
- val wildcardColName = options.wildcardColName
- val hasWildcard = schema.exists(_.name == wildcardColName)
-
- var badRecordException: Option[Throwable] = None
-
- var shouldStop = false
- while (!shouldStop) {
- parser.nextEvent match {
- case e: StartElement => try {
- val attributes = e.getAttributes.asScala.map(_.asInstanceOf[Attribute]).toArray
- val field = StaxXmlParserUtils.getName(e.asStartElement.getName, options)
-
- nameToIndex.get(field) match {
- case Some(index) => schema(index).dataType match {
- case st: StructType =>
- row(index) = convertObjectWithAttributes(parser, st, options, attributes)
-
- case ArrayType(dt: DataType, _) =>
- val values = Option(row(index))
- .map(_.asInstanceOf[ArrayBuffer[Any]])
- .getOrElse(ArrayBuffer.empty[Any])
- val newValue = dt match {
- case st: StructType =>
- convertObjectWithAttributes(parser, st, options, attributes)
- case dt: DataType =>
- convertField(parser, dt, options)
- }
- row(index) = values :+ newValue
-
- case dt: DataType =>
- row(index) = convertField(parser, dt, options, attributes)
- }
-
- case None =>
- if (hasWildcard) {
- // Special case: there's an 'any' wildcard element that matches anything else
- // as a string (or array of strings, to parse multiple ones)
- val newValue = convertField(parser, StringType, options)
- val anyIndex = schema.fieldIndex(wildcardColName)
- schema(wildcardColName).dataType match {
- case StringType =>
- row(anyIndex) = newValue
- case ArrayType(StringType, _) =>
- val values = Option(row(anyIndex))
- .map(_.asInstanceOf[ArrayBuffer[String]])
- .getOrElse(ArrayBuffer.empty[String])
- row(anyIndex) = values :+ newValue
- }
- } else {
- StaxXmlParserUtils.skipChildren(parser)
- }
- }
- } catch {
- case NonFatal(exception) if options.parseMode == PermissiveMode =>
- badRecordException = badRecordException.orElse(Some(exception))
- }
-
- case _: EndElement =>
- shouldStop = StaxXmlParserUtils.checkEndElement(parser)
-
- case _ => // do nothing
- }
- }
-
- if (badRecordException.isEmpty) {
- Row.fromSeq(row.toIndexedSeq)
- } else {
- throw PartialResultException(Row.fromSeq(row.toIndexedSeq), badRecordException.get)
- }
- }
-}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/util/PartialResultException.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/util/PartialResultException.scala
deleted file mode 100644
index 26acd814ba2..00000000000
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/util/PartialResultException.scala
+++ /dev/null
@@ -1,29 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.spark.sql.execution.datasources.xml.util
-
-import org.apache.spark.sql.Row
-
-/**
- * Exception thrown when the underlying parser returns a partial result of parsing.
- * @param partialResult the partial result of parsing a bad record.
- * @param cause the actual exception about why the parser cannot return full result.
- */
-case class PartialResultException(
- partialResult: Row,
- cause: Throwable)
- extends Exception(cause)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/util/XmlFile.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/util/XmlFile.scala
deleted file mode 100644
index ef73652d9de..00000000000
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/util/XmlFile.scala
+++ /dev/null
@@ -1,163 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.spark.sql.execution.datasources.xml.util
-
-import java.io.CharArrayWriter
-import java.nio.charset.Charset
-import javax.xml.stream.XMLOutputFactory
-
-import scala.collection.Map
-
-import com.sun.xml.txw2.output.IndentingXMLStreamWriter
-import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.io.{LongWritable, Text}
-import org.apache.hadoop.io.compress.CompressionCodec
-
-import org.apache.spark.SparkContext
-import org.apache.spark.rdd.RDD
-import org.apache.spark.sql.DataFrame
-import org.apache.spark.sql.catalyst.util.CompressionCodecs
-import org.apache.spark.sql.execution.datasources.xml.{XmlInputFormat, XmlOptions}
-import org.apache.spark.sql.execution.datasources.xml.parsers.StaxXmlGenerator
-
-private[xml] object XmlFile {
- val DEFAULT_INDENT = " "
-
- def withCharset(
- context: SparkContext,
- location: String,
- charset: String,
- rowTag: String): RDD[String] = {
- // This just checks the charset's validity early, to keep behavior
- Charset.forName(charset)
- val config = new Configuration(context.hadoopConfiguration)
- config.set(XmlInputFormat.START_TAG_KEY, s"<$rowTag>")
- config.set(XmlInputFormat.END_TAG_KEY, s"</$rowTag>")
- config.set(XmlInputFormat.ENCODING_KEY, charset)
- context.newAPIHadoopFile(location,
- classOf[XmlInputFormat],
- classOf[LongWritable],
- classOf[Text],
- config).map { case (_, text) => text.toString }
- }
-
- /**
- * Note that writing a XML file from [[DataFrame]] having a field
- * [[org.apache.spark.sql.types.ArrayType]] with its element as nested array would have
- * an additional nested field for the element. For example, the [[DataFrame]] having
- * a field below,
- *
- * fieldA Array(Array(data1, data2))
- *
- * would produce a XML file below.
- *
- * <fieldA>
- * <item>data1</item>
- * </fieldA>
- * <fieldA>
- * <item>data2</item>
- * </fieldA>
- *
- * Namely, roundtrip in writing and reading can end up in different schema structure.
- */
- def saveAsXmlFile(
- dataFrame: DataFrame,
- path: String,
- parameters: Map[String, String] = Map()): Unit = {
- val options = XmlOptions(parameters.toMap)
- val codec = Option(options.codec).map(CompressionCodecs.getCodecClassName)
- // scalastyle:off classforname
- val codecClass: Option[Class[_ <: CompressionCodec]] =
- codec.map(Class.forName(_).asInstanceOf[Class[CompressionCodec]])
- val rowSchema = dataFrame.schema
- val indent = XmlFile.DEFAULT_INDENT
-
- // Allow a root tag to be like "rootTag foo='bar'"
- // This is hacky; won't deal correctly with spaces in attributes, but want
- // to make this at least work for simple cases without much complication
- val rootTagTokens = options.rootTag.split(" ")
- val rootElementName = rootTagTokens.head
- val rootAttributes: Map[String, String] =
- if (rootTagTokens.length > 1) {
- rootTagTokens.tail.map { kv =>
- val Array(k, v) = kv.split("=")
- k -> v.replaceAll("['\"]", "")
- }.toMap
- } else {
- Map.empty
- }
- val declaration = options.declaration
-
- val xmlRDD = dataFrame.rdd.mapPartitions { iter =>
- val factory = XMLOutputFactory.newInstance()
- val writer = new CharArrayWriter()
- val xmlWriter = factory.createXMLStreamWriter(writer)
- val indentingXmlWriter = new IndentingXMLStreamWriter(xmlWriter)
- indentingXmlWriter.setIndentStep(indent)
-
- new Iterator[String] {
- var firstRow: Boolean = true
- var lastRow: Boolean = true
-
- override def hasNext: Boolean = iter.hasNext || firstRow || lastRow
-
- override def next(): String = {
- if (iter.nonEmpty) {
- if (firstRow) {
- if (declaration != null && declaration.nonEmpty) {
- indentingXmlWriter.writeProcessingInstruction("xml", declaration)
- indentingXmlWriter.writeCharacters("\n")
- }
- indentingXmlWriter.writeStartElement(rootElementName)
- rootAttributes.foreach { case (k, v) =>
- indentingXmlWriter.writeAttribute(k, v)
- }
- firstRow = false
- }
- val xml = {
- StaxXmlGenerator(
- rowSchema,
- indentingXmlWriter,
- options)(iter.next())
- indentingXmlWriter.flush()
- writer.toString
- }
- writer.reset()
- xml
- } else {
- if (!firstRow) {
- lastRow = false
- indentingXmlWriter.writeEndElement()
- indentingXmlWriter.close()
- writer.toString
- } else {
- // This means the iterator was initially empty.
- firstRow = false
- lastRow = false
- ""
- }
- }
- }
- }
- }
-
- codecClass match {
- case None => xmlRDD.saveAsTextFile(path)
- case Some(codec) => xmlRDD.saveAsTextFile(path, codec)
- }
- }
-}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/functions.scala b/sql/core/src/main/scala/org/apache/spark/sql/functions.scala
index 679b211ee6b..a9f9a8ffe5b 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/functions.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/functions.scala
@@ -7310,6 +7310,103 @@ object functions {
*/
def to_csv(e: Column): Column = to_csv(e, Map.empty[String, String].asJava)
+ // scalastyle:off line.size.limit
+
+ /**
+ * Parses a column containing a XML string into the data type corresponding to the specified schema.
+ * Returns `null`, in the case of an unparseable string.
+ *
+ * @param e a string column containing XML data.
+ * @param schema the schema to use when parsing the XML string
+ * @param options options to control how the XML is parsed. accepts the same options and the
+ * XML data source.
+ * See
+ * <a href=
+ * "https://spark.apache.org/docs/latest/sql-data-sources-xml.html#data-source-option">
+ * Data Source Option</a> in the version you use.
+ * @group collection_funcs
+ * @since
+ */
+ // scalastyle:on line.size.limit
+ def from_xml(e: Column, schema: StructType, options: Map[String, String]): Column = withExpr {
+ XmlToStructs(CharVarcharUtils.failIfHasCharVarchar(schema), options, e.expr)
+ }
+
+ // scalastyle:off line.size.limit
+
+ /**
+ * (Java-specific) Parses a column containing a XML string into a `StructType`
+ * with the specified schema. Returns `null`, in the case of an unparseable string.
+ *
+ * @param e a string column containing XML data.
+ * @param schema the schema to use when parsing the XML string
+ * @param options options to control how the XML is parsed. accepts the same options and the
+ * XML data source.
+ * See
+ * <a href=
+ * "https://spark.apache.org/docs/latest/sql-data-sources-xml.html#data-source-option">
+ * Data Source Option</a> in the version you use.
+ * @group collection_funcs
+ * @since
+ */
+ // scalastyle:on line.size.limit
+ def from_xml(e: Column, schema: Column, options: java.util.Map[String, String]): Column = {
+ withExpr(new XmlToStructs(e.expr, schema.expr, options.asScala.toMap))
+ }
+
+ /**
+ * Parses a column containing a XML string into the data type
+ * corresponding to the specified schema.
+ * Returns `null`, in the case of an unparseable string.
+ *
+ * @param e a string column containing XML data.
+ * @param schema the schema to use when parsing the XML string
+
+ * @group collection_funcs
+ * @since
+ */
+ def from_xml(e: Column, schema: StructType): Column =
+ from_xml(e, schema, Map.empty[String, String])
+
+ /**
+ * Parses a XML string and infers its schema in DDL format.
+ *
+ * @param xml a XML string.
+ * @group collection_funcs
+ * @since 4.0.0
+ */
+ def schema_of_xml(xml: String): Column = schema_of_xml(lit(xml))
+
+ /**
+ * Parses a XML string and infers its schema in DDL format.
+ *
+ * @param xml a foldable string column containing a XML string.
+ * @group collection_funcs
+ * @since 4.0.0
+ */
+ def schema_of_xml(xml: Column): Column = withExpr(new SchemaOfXml(xml.expr))
+
+ // scalastyle:off line.size.limit
+
+ /**
+ * Parses a XML string and infers its schema in DDL format using options.
+ *
+ * @param xml a foldable string column containing XML data.
+ * @param options options to control how the xml is parsed. accepts the same options and the
+ * XML data source.
+ * See
+ * <a href=
+ * "https://spark.apache.org/docs/latest/sql-data-sources-xml.html#data-source-option">
+ * Data Source Option</a> in the version you use.
+ * @return a column with string literal containing schema in DDL format.
+ * @group collection_funcs
+ * @since 4.0.0
+ */
+ // scalastyle:on line.size.limit
+ def schema_of_xml(xml: Column, options: java.util.Map[String, String]): Column = {
+ withExpr(SchemaOfXml(xml.expr, options.asScala.toMap))
+ }
+
/**
* A transform for timestamps and dates to partition data into years.
*
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/streaming/DataStreamReader.scala b/sql/core/src/main/scala/org/apache/spark/sql/streaming/DataStreamReader.scala
index e3cd7379dad..c7b3473bbb6 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/streaming/DataStreamReader.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/streaming/DataStreamReader.scala
@@ -259,6 +259,27 @@ final class DataStreamReader private[sql](sparkSession: SparkSession) extends Lo
*/
def csv(path: String): DataFrame = format("csv").load(path)
+ /**
+ * Loads a XML file stream and returns the result as a `DataFrame`.
+ *
+ * This function will go through the input once to determine the input schema if `inferSchema`
+ * is enabled. To avoid going through the entire data once, disable `inferSchema` option or
+ * specify the schema explicitly using `schema`.
+ *
+ * You can set the following option(s):
+ * <ul>
+ * <li>`maxFilesPerTrigger` (default: no max limit): sets the maximum number of new files to be
+ * considered in every trigger.</li>
+ * </ul>
+ *
+ * You can find the XML-specific options for reading XML file stream in
+ * <a href="https://spark.apache.org/docs/latest/sql-data-sources-xml.html#data-source-option">
+ * Data Source Option</a> in the version you use.
+ *
+ * @since 4.0.0
+ */
+ def xml(path: String): DataFrame = format("xml").load(path)
+
/**
* Loads a ORC file stream, returning the result as a `DataFrame`.
*
diff --git a/sql/core/src/test/java/test/org/apache/spark/sql/execution/datasources/xml/JavaXmlSuite.java b/sql/core/src/test/java/test/org/apache/spark/sql/execution/datasources/xml/JavaXmlSuite.java
index dae6d93a37b..af3734b4b99 100644
--- a/sql/core/src/test/java/test/org/apache/spark/sql/execution/datasources/xml/JavaXmlSuite.java
+++ b/sql/core/src/test/java/test/org/apache/spark/sql/execution/datasources/xml/JavaXmlSuite.java
@@ -31,8 +31,7 @@ import org.junit.Test;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
-import org.apache.spark.sql.execution.datasources.xml.XmlOptions;
-import org.apache.spark.sql.execution.datasources.xml.XmlReader;
+import org.apache.spark.sql.catalyst.xml.XmlOptions;
public final class JavaXmlSuite {
@@ -81,7 +80,9 @@ public final class JavaXmlSuite {
@Test
public void testXmlParser() {
- Dataset<Row> df = (new XmlReader()).withRowTag(booksFileTag).xmlFile(spark, booksFile);
+ Map<String, String> options = new HashMap<>();
+ options.put("rowTag", booksFileTag);
+ Dataset<Row> df = spark.read().options(options).format("xml").load(booksFile);
String prefix = XmlOptions.DEFAULT_ATTRIBUTE_PREFIX();
long result = df.select(prefix + "id").count();
Assert.assertEquals(result, numBooks);
@@ -98,12 +99,14 @@ public final class JavaXmlSuite {
@Test
public void testSave() throws IOException {
+ Map<String, String> options = new HashMap<>();
+ options.put("rowTag", booksFileTag);
Path booksPath = getEmptyTempDir().resolve("booksFile");
- Dataset<Row> df = (new XmlReader()).withRowTag(booksFileTag).xmlFile(spark, booksFile);
+ Dataset<Row> df = spark.read().options(options).format("xml").load(booksFile);
df.select("price", "description").write().format("xml").save(booksPath.toString());
- Dataset<Row> newDf = (new XmlReader()).xmlFile(spark, booksPath.toString());
+ Dataset<Row> newDf = spark.read().format("xml").load(booksPath.toString());
long result = newDf.select("price").count();
Assert.assertEquals(result, numBooks);
}
diff --git a/sql/core/src/test/resources/test-data/xml-resources/basket_invalid.xml b/sql/core/src/test/resources/test-data/xml-resources/basket_invalid.xml
index cdb1b159698..16ac60c0e4c 100644
--- a/sql/core/src/test/resources/test-data/xml-resources/basket_invalid.xml
+++ b/sql/core/src/test/resources/test-data/xml-resources/basket_invalid.xml
@@ -1,4 +1,14 @@
<baskets>
+ <basket>
+ <entry>
+ <key>1</key>
+ <value>fork</value>
+ </entry>
+ <entry>
+ <key>2</key>
+ <value>cup</value>
+ </entry>
+ </basket>
<basket>
<entry>
<key>9027</key>
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameFunctionsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameFunctionsSuite.scala
index c7dcb575ff0..0d197da63fd 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameFunctionsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameFunctionsSuite.scala
@@ -82,7 +82,10 @@ class DataFrameFunctionsSuite extends QueryTest with SharedSparkSession {
"bucket", "days", "hours", "months", "years", // Datasource v2 partition transformations
"product", // Discussed in https://github.com/apache/spark/pull/30745
"unwrap_udt",
- "collect_top_k"
+ "collect_top_k",
+ // TODO: XML functions will soon be added to SQL Function registry and removed from this list
+ // https://issues.apache.org/jira/browse/SPARK-44787
+ "from_xml", "schema_of_xml"
)
// We only consider functions matching this pattern, this excludes symbolic and other
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/TestUtils.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/TestUtils.scala
index 1cc111517f2..3f595e00d1c 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/TestUtils.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/TestUtils.scala
@@ -25,13 +25,14 @@ private[xml] object TestUtils {
def field(name: String, dataType: DataType = StringType, nullable: Boolean = true): StructField =
StructField(name, dataType, nullable)
- def struct(fields: StructField*): StructType = buildSchema(fields: _*)
+ def structField(fields: StructField*): StructType = buildSchema(fields: _*)
- def struct(name: String, fields: StructField*): StructField = field(name, struct(fields: _*))
+ def structField(name: String, fields: StructField*): StructField =
+ field(name, structField(fields: _*))
def structArray(name: String, fields: StructField*): StructField =
- field(name, ArrayType(struct(fields: _*)))
+ field(name, ArrayType(structField(fields: _*)))
- def array(name: String, dataType: DataType): StructField = field(name, ArrayType(dataType))
+ def arrayField(name: String, dataType: DataType): StructField = field(name, ArrayType(dataType))
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/XmlSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/XmlSuite.scala
index 6fbc8b98dfe..beea14e3cf8 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/XmlSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/XmlSuite.scala
@@ -25,23 +25,24 @@ import scala.collection.JavaConverters._
import scala.collection.mutable
import scala.io.Source
+import org.apache.commons.lang3.exception.ExceptionUtils
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.io.{LongWritable, Text}
import org.apache.hadoop.io.compress.GzipCodec
-import org.apache.hadoop.mapreduce.lib.input.InvalidInputException
import org.apache.spark.SparkException
-import org.apache.spark.sql.{Row, SaveMode}
+import org.apache.spark.sql.{AnalysisException, Encoders, QueryTest, Row, SaveMode}
import org.apache.spark.sql.catalyst.util._
+import org.apache.spark.sql.catalyst.xml.XmlOptions
+import org.apache.spark.sql.catalyst.xml.XmlOptions._
import org.apache.spark.sql.execution.datasources.xml.TestUtils._
-import org.apache.spark.sql.execution.datasources.xml.XmlOptions._
-import org.apache.spark.sql.execution.datasources.xml.functions._
-import org.apache.spark.sql.functions.{column, explode}
+import org.apache.spark.sql.functions._
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types._
-final class XmlSuite extends SharedSparkSession {
+class XmlSuite extends QueryTest with SharedSparkSession {
+ import testImplicits._
private val resDir = "test-data/xml-resources/"
@@ -64,6 +65,7 @@ final class XmlSuite extends SharedSparkSession {
test("DSL test") {
val results = spark.read.format("xml")
+ .option("multiLine", "true")
.load(getTestResourcePath(resDir + "cars.xml"))
.select("year")
.collect()
@@ -74,6 +76,7 @@ final class XmlSuite extends SharedSparkSession {
test("DSL test with xml having unbalanced datatypes") {
val results = spark.read
.option("treatEmptyValuesAsNulls", "true")
+ .option("multiLine", "true")
.xml(getTestResourcePath(resDir + "gps-empty-field.xml"))
assert(results.collect().length === 2)
@@ -100,7 +103,8 @@ final class XmlSuite extends SharedSparkSession {
// This should not throw an exception `java.lang.ArrayIndexOutOfBoundsException`
// as non-existing values are represented as `null`s.
- assert(results.collect()(0).getStruct(0).get(1) === null)
+ val res = results.collect();
+ assert(res(0).getStruct(0).get(1) === null)
}
test("DSL test with mixed elements (struct, string)") {
@@ -124,8 +128,9 @@ final class XmlSuite extends SharedSparkSession {
}
test("DSL test for iso-8859-1 encoded file") {
- val dataFrame = new XmlReader(Map("charset" -> StandardCharsets.ISO_8859_1.name))
- .xmlFile(spark, getTestResourcePath(resDir + "cars-iso-8859-1.xml"))
+ val dataFrame = spark.read
+ .option("charset", StandardCharsets.ISO_8859_1.name)
+ .xml(getTestResourcePath(resDir + "cars-iso-8859-1.xml"))
assert(dataFrame.select("year").collect().length === 3)
val results = dataFrame
@@ -154,13 +159,15 @@ final class XmlSuite extends SharedSparkSession {
}
test("DSL test bad charset name") {
- val exception = intercept[UnsupportedCharsetException] {
+ // val exception = intercept[UnsupportedCharsetException] {
+ val exception = intercept[SparkException] {
spark.read
.option("charset", "1-9588-osi")
.xml(getTestResourcePath(resDir + "cars.xml"))
.select("year")
.collect()
}
+ ExceptionUtils.getRootCause(exception).isInstanceOf[UnsupportedCharsetException]
assert(exception.getMessage.contains("1-9588-osi"))
}
@@ -185,15 +192,17 @@ final class XmlSuite extends SharedSparkSession {
}
test("DSL test for parsing a malformed XML file") {
- val results = new XmlReader(Map("mode" -> DropMalformedMode.name))
- .xmlFile(spark, getTestResourcePath(resDir + "cars-malformed.xml"))
+ val results = spark.read
+ .option("mode", DropMalformedMode.name)
+ .xml(getTestResourcePath(resDir + "cars-malformed.xml"))
assert(results.count() === 1)
}
test("DSL test for dropping malformed rows") {
- val cars = new XmlReader(Map("mode" -> DropMalformedMode.name))
- .xmlFile(spark, getTestResourcePath(resDir + "cars-malformed.xml"))
+ val cars = spark.read
+ .option("mode", DropMalformedMode.name)
+ .xml(getTestResourcePath(resDir + "cars-malformed.xml"))
assert(cars.count() == 1)
assert(cars.head() === Row("Chevy", "Volt", 2015))
@@ -201,34 +210,49 @@ final class XmlSuite extends SharedSparkSession {
test("DSL test for failing fast") {
val exceptionInParse = intercept[SparkException] {
- new XmlReader(Map("mode" -> FailFastMode.name))
- .xmlFile(spark, getTestResourcePath(resDir + "cars-malformed.xml"))
+ spark.read
+ .option("mode", FailFastMode.name)
+ .xml(getTestResourcePath(resDir + "cars-malformed.xml"))
.collect()
}
- assert(exceptionInParse.getMessage.contains("Malformed line in FAILFAST mode"))
+ checkError(
+ // TODO: Exception was nested two level deep as opposed to just one like json/csv
+ exception = exceptionInParse.getCause.getCause.asInstanceOf[SparkException],
+ errorClass = "MALFORMED_RECORD_IN_PARSING.WITHOUT_SUGGESTION",
+ parameters = Map(
+ "badRecord" -> "[null,null,null]",
+ "failFastMode" -> FailFastMode.name)
+ )
}
test("test FAILFAST with unclosed tag") {
val exceptionInParse = intercept[SparkException] {
spark.read
.option("rowTag", "book")
- .option("mode", "FAILFAST")
+ .option("mode", FailFastMode.name)
.xml(getTestResourcePath(resDir + "unclosed_tag.xml"))
.show()
}
- assert(exceptionInParse.getMessage.contains("Malformed line in FAILFAST mode"))
+ checkError(
+ // TODO: Exception was nested two level deep as opposed to just one like json/csv
+ exception = exceptionInParse.getCause.getCause.asInstanceOf[SparkException],
+ errorClass = "MALFORMED_RECORD_IN_PARSING.WITHOUT_SUGGESTION",
+ parameters = Map(
+ "badRecord" -> "[empty row]",
+ "failFastMode" -> FailFastMode.name)
+ )
}
test("DSL test for permissive mode for corrupt records") {
- val carsDf = new XmlReader(Map(
- "mode" -> PermissiveMode.name,
- "columnNameOfCorruptRecord" -> "_malformed_records"))
- .xmlFile(spark, getTestResourcePath(resDir + "cars-malformed.xml"))
+ val carsDf = spark.read
+ .option("mode", PermissiveMode.name)
+ .option("columnNameOfCorruptRecord", "_malformed_records")
+ .xml(getTestResourcePath(resDir + "cars-malformed.xml"))
val cars = carsDf.collect()
assert(cars.length === 3)
- val malformedRowOne = carsDf.select("_malformed_records").first().get(0).toString
- val malformedRowTwo = carsDf.select("_malformed_records").take(2).last.get(0).toString
+ val malformedRowOne = carsDf.cache().select("_malformed_records").first().get(0).toString
+ val malformedRowTwo = carsDf.cache().select("_malformed_records").take(2).last.get(0).toString
val expectedMalformedRowOne = "<ROW><year>2012</year><make>Tesla</make><model>>S" +
"<comment>No comment</comment></ROW>"
val expectedMalformedRowTwo = "<ROW></year><make>Ford</make><model>E350</model>model></model>" +
@@ -243,8 +267,9 @@ final class XmlSuite extends SharedSparkSession {
}
test("DSL test with empty file and known schema") {
- val results = new XmlReader(buildSchema(field("column", StringType, false)))
- .xmlFile(spark, getTestResourcePath(resDir + "empty.xml"))
+ val results = spark.read
+ .schema(buildSchema(field("column", StringType, false)))
+ .xml(getTestResourcePath(resDir + "empty.xml"))
.count()
assert(results === 0)
@@ -257,8 +282,8 @@ final class XmlSuite extends SharedSparkSession {
field("make"),
field("model"),
field("comment"))
- val results = new XmlReader(schema)
- .xmlFile(spark, getTestResourcePath(resDir + "cars-unbalanced-elements.xml"))
+ val results = spark.read.schema(schema)
+ .xml(getTestResourcePath(resDir + "cars-unbalanced-elements.xml"))
.count()
assert(results === 3)
@@ -310,7 +335,8 @@ final class XmlSuite extends SharedSparkSession {
.options(Map("codec" -> classOf[GzipCodec].getName))
.xml(copyFilePath.toString)
// Check that the part file has a .gz extension
- assert(Files.exists(copyFilePath.resolve("part-00000.gz")))
+ assert(Files.list(copyFilePath).iterator().asScala
+ .count(_.getFileName.toString().endsWith(".xml.gz")) === 1)
val carsCopy = spark.read.xml(copyFilePath.toString)
@@ -328,7 +354,8 @@ final class XmlSuite extends SharedSparkSession {
.xml(copyFilePath.toString)
// Check that the part file has a .gz extension
- assert(Files.exists(copyFilePath.resolve("part-00000.gz")))
+ assert(Files.list(copyFilePath).iterator().asScala
+ .count(_.getFileName.toString().endsWith(".xml.gz")) === 1)
val carsCopy = spark.read.xml(copyFilePath.toString)
@@ -364,7 +391,11 @@ final class XmlSuite extends SharedSparkSession {
.options(Map("rootTag" -> "books", "rowTag" -> "book", "declaration" -> ""))
.xml(copyFilePath1.toString)
- assert(getLines(copyFilePath1.resolve("part-00000")).head === "<books>")
+ val xmlFile1 =
+ Files.list(copyFilePath1).iterator.asScala
+ .filter(_.getFileName.toString.startsWith("part-")).next()
+ val firstLine = getLines(xmlFile1).head
+ assert(firstLine === "<books>")
val copyFilePath2 = getEmptyTempDir().resolve("books-copy.xml")
@@ -372,7 +403,10 @@ final class XmlSuite extends SharedSparkSession {
.options(Map("rootTag" -> "books", "rowTag" -> "book"))
.xml(copyFilePath2.toString)
- assert(getLines(copyFilePath2.resolve("part-00000")).head ===
+ val xmlFile2 =
+ Files.list(copyFilePath2).iterator.asScala
+ .filter(_.getFileName.toString.startsWith("part-")).next()
+ assert(getLines(xmlFile2).head ===
"<?xml version=\"1.0\" encoding=\"UTF-8\" standalone=\"yes\"?>")
}
@@ -380,7 +414,11 @@ final class XmlSuite extends SharedSparkSession {
val tempPath = getEmptyTempDir().resolve("items-temp.xml")
val items = spark.createDataFrame(Seq(Tuple1(Array(Array(3, 4))))).toDF("thing").repartition(1)
items.write.option("arrayElementName", "foo").xml(tempPath.toString)
- assert(getLines(tempPath.resolve("part-00000")).count(_.contains("<foo>")) === 2)
+
+ val xmlFile =
+ Files.list(tempPath).iterator.asScala
+ .filter(_.getFileName.toString.startsWith("part-")).next()
+ assert(getLines(xmlFile).count(_.contains("<foo>")) === 2)
}
test("DSL save with nullValue and treatEmptyValuesAsNulls") {
@@ -432,9 +470,9 @@ final class XmlSuite extends SharedSparkSession {
test("DSL save dataframe not read from a XML file") {
val copyFilePath = getEmptyTempDir().resolve("data-copy.xml")
- val schema = buildSchema(array("a", ArrayType(StringType)))
+ val schema = buildSchema(arrayField("a", ArrayType(StringType)))
val data = spark.sparkContext.parallelize(
- List(List(List("aa", "bb"), List("aa", "bb")))).map(Row(_))
+ List(List(List("aa", "bb"), List("aa", "bb"))).map(Row(_)))
val df = spark.createDataFrame(data, schema)
df.write.xml(copyFilePath.toString)
@@ -482,8 +520,8 @@ final class XmlSuite extends SharedSparkSession {
val df = spark.createDataFrame(data, schema)
df.write.xml(copyFilePath.toString)
- val dfCopy = new XmlReader(schema)
- .xmlFile(spark, copyFilePath.toString)
+ val dfCopy = spark.read.schema(schema)
+ .xml(copyFilePath.toString)
assert(dfCopy.collect() === df.collect())
assert(dfCopy.schema === df.schema)
@@ -536,7 +574,7 @@ final class XmlSuite extends SharedSparkSession {
field("description"),
field("genre"),
field("price", DoubleType),
- struct("publish_dates",
+ structField("publish_dates",
field("publish_date", DateType)),
field("title")))
@@ -554,7 +592,7 @@ final class XmlSuite extends SharedSparkSession {
field("description"),
field("genre"),
field("price", DoubleType),
- array("publish_date", DateType),
+ arrayField("publish_date", DateType),
field("title")))
assert(results.collect().length === 12)
@@ -568,13 +606,13 @@ final class XmlSuite extends SharedSparkSession {
assert(results.schema == buildSchema(
field(s"${DEFAULT_ATTRIBUTE_PREFIX}id"),
field("author"),
- struct("genre",
+ structField("genre",
field("genreid", LongType),
field("name")),
field("price", DoubleType),
- struct("publish_dates",
- array("publish_date",
- struct(
+ structField("publish_dates",
+ arrayField("publish_date",
+ structField(
field(s"${DEFAULT_ATTRIBUTE_PREFIX}tag"),
field("day", LongType),
field("month", LongType),
@@ -586,13 +624,13 @@ final class XmlSuite extends SharedSparkSession {
test("DSL test parsing and inferring attribute in elements having no child element") {
// Default value.
- val resultsOne = new XmlReader(Map("rowTag" -> "book"))
- .xmlFile(spark, getTestResourcePath(resDir + "books-attributes-in-no-child.xml"))
+ val resultsOne = spark.read.option("rowTag", "book")
+ .xml(getTestResourcePath(resDir + "books-attributes-in-no-child.xml"))
val schemaOne = buildSchema(
field("_id"),
field("author"),
- struct("price",
+ structField("price",
field("_VALUE"),
field(s"_unit")),
field("publish_date", DateType),
@@ -604,15 +642,16 @@ final class XmlSuite extends SharedSparkSession {
// Explicitly set
val attributePrefix = "@#"
val valueTag = "#@@value"
- val resultsTwo = new XmlReader(Map(
- "rowTag" -> "book", "attributePrefix" -> attributePrefix,
- "valueTag" -> valueTag))
- .xmlFile(spark, getTestResourcePath(resDir + "books-attributes-in-no-child.xml"))
+ val resultsTwo = spark.read
+ .option("rowTag", "book")
+ .option("attributePrefix", attributePrefix)
+ .option("valueTag", valueTag)
+ .xml(getTestResourcePath(resDir + "books-attributes-in-no-child.xml"))
val schemaTwo = buildSchema(
field(s"${attributePrefix}id"),
field("author"),
- struct("price",
+ structField("price",
field(valueTag),
field(s"${attributePrefix}unit")),
field("publish_date", DateType),
@@ -623,8 +662,10 @@ final class XmlSuite extends SharedSparkSession {
}
test("DSL test schema (excluding tags) inferred correctly") {
- val results = new XmlReader(Map("excludeAttribute" -> true, "rowTag" -> "book"))
- .xmlFile(spark, getTestResourcePath(resDir + "books.xml"))
+ val results = spark.read
+ .option("excludeAttribute", true)
+ .option("rowTag", "book")
+ .xml(getTestResourcePath(resDir + "books.xml"))
val schema = buildSchema(
field("author"),
@@ -644,8 +685,8 @@ final class XmlSuite extends SharedSparkSession {
field("comment"),
field("color"),
field("year", IntegerType))
- val results = new XmlReader(schema)
- .xmlFile(spark, getTestResourcePath(resDir + "cars-unbalanced-elements.xml"))
+ val results = spark.read.schema(schema)
+ .xml(getTestResourcePath(resDir + "cars-unbalanced-elements.xml"))
.count()
assert(results === 3)
@@ -665,8 +706,9 @@ final class XmlSuite extends SharedSparkSession {
val schema = buildSchema(
field("name", StringType, false),
field("age"))
- val results = new XmlReader(schema)
- .xmlFile(spark, getTestResourcePath(resDir + "null-numbers.xml"))
+ val results = spark.read.schema(schema)
+ .xml(getTestResourcePath(resDir + "null-numbers.xml"))
+ .select("name", "age")
.collect()
assert(results(0) === Row("alice", "35"))
@@ -678,8 +720,10 @@ final class XmlSuite extends SharedSparkSession {
val schema = buildSchema(
field("name", StringType, false),
field("age", IntegerType))
- val results = new XmlReader(schema, Map("treatEmptyValuesAsNulls" -> true))
- .xmlFile(spark, getTestResourcePath(resDir + "null-numbers.xml"))
+ val results = spark.read.schema(schema)
+ .option("treatEmptyValuesAsNulls", true)
+ .xml(getTestResourcePath(resDir + "null-numbers.xml"))
+ .select("name", "age")
.collect()
assert(results(1) === Row("bob", null))
@@ -718,7 +762,7 @@ final class XmlSuite extends SharedSparkSession {
field("author"),
field("description"),
field("genre"),
- array("xs_any", StringType))
+ arrayField("xs_any", StringType))
val results = spark.read.schema(schema).option("rowTag", "book")
.xml(getTestResourcePath(resDir + "books.xml"))
.collect()
@@ -740,8 +784,8 @@ final class XmlSuite extends SharedSparkSession {
test("Produces correct result for empty vs non-existent rows") {
val schema = buildSchema(
- struct("b",
- struct("es",
+ structField("b",
+ structField("es",
field("e"),
field("f"))))
val result = spark.read
@@ -759,12 +803,12 @@ final class XmlSuite extends SharedSparkSession {
test("Produces correct order of columns for nested rows when user specifies a schema") {
val schema = buildSchema(
- struct("c",
+ structField("c",
field("b", IntegerType),
field("a", IntegerType)))
- val result = new XmlReader(schema)
- .xmlFile(spark, getTestResourcePath(resDir + "simple-nested-objects.xml"))
+ val result = spark.read.schema(schema)
+ .xml(getTestResourcePath(resDir + "simple-nested-objects.xml"))
.select("c.a", "c.b")
.collect()
@@ -803,12 +847,12 @@ final class XmlSuite extends SharedSparkSession {
}
test("Nested element with same name as parent schema inference") {
- val df = new XmlReader(Map("rowTag" -> "parent"))
- .xmlFile(spark, getTestResourcePath(resDir + "nested-element-with-name-of-parent.xml"))
+ val df = spark.read.option("rowTag", "parent")
+ .xml(getTestResourcePath(resDir + "nested-element-with-name-of-parent.xml"))
val schema = buildSchema(
field("child"),
- struct("parent",
+ structField("parent",
field("child")))
assert(df.schema === schema)
}
@@ -907,8 +951,10 @@ final class XmlSuite extends SharedSparkSession {
}
test("ignoreSurroundingSpaces with non-string types") {
- val results = new XmlReader(Map("ignoreSurroundingSpaces" -> true, "rowTag" -> "person"))
- .xmlFile(spark, getTestResourcePath(resDir + "ages-with-spaces.xml"))
+ val results = spark.read
+ .option("ignoreSurroundingSpaces", true)
+ .option("rowTag", "person")
+ .xml(getTestResourcePath(resDir + "ages-with-spaces.xml"))
.collect()
val attrValOne = results(0).getStruct(0)(1)
val attrValTwo = results(1).getStruct(0)(0)
@@ -918,9 +964,11 @@ final class XmlSuite extends SharedSparkSession {
}
test("DSL test with malformed attributes") {
- val results = new XmlReader(Map("mode" -> DropMalformedMode.name, "rowTag" -> "book"))
- .xmlFile(spark, getTestResourcePath(resDir + "books-malformed-attributes.xml"))
- .collect()
+ val results = spark.read
+ .option("mode", DropMalformedMode.name)
+ .option("rowTag", "book")
+ .xml(getTestResourcePath(resDir + "books-malformed-attributes.xml"))
+ .collect()
assert(results.length === 2)
assert(results(0)(0) === "bk111")
@@ -967,8 +1015,8 @@ final class XmlSuite extends SharedSparkSession {
field("non-empty-tag", IntegerType),
field("self-closing-tag", IntegerType))
- val result = new XmlReader(schema)
- .xmlFile(spark, getTestResourcePath(resDir + "self-closing-tag.xml"))
+ val result = spark.read.schema(schema)
+ .xml(getTestResourcePath(resDir + "self-closing-tag.xml"))
.collect()
assert(result(0) === Row(1, null))
@@ -993,16 +1041,16 @@ final class XmlSuite extends SharedSparkSession {
test("DSL test nulls out invalid values when set to permissive and given explicit schema") {
val schema = buildSchema(
- struct("integer_value",
+ structField("integer_value",
field("_VALUE", IntegerType),
field("_int", IntegerType)),
- struct("long_value",
+ structField("long_value",
field("_VALUE", LongType),
field("_int", StringType)),
field("float_value", FloatType),
field("double_value", DoubleType),
field("boolean_value", BooleanType),
- field("string_value"), array("integer_array", IntegerType),
+ field("string_value"), arrayField("integer_array", IntegerType),
field("integer_map", MapType(StringType, IntegerType)),
field("_malformed_records", StringType))
val results = spark.read
@@ -1099,8 +1147,12 @@ final class XmlSuite extends SharedSparkSession {
.replace("file:/", "/"))
.option("mode", "PERMISSIVE")
.option("columnNameOfCorruptRecord", "_malformed_records")
- .xml(getTestResourcePath(resDir + "basket_invalid.xml"))
- assert(basketDF.select("_malformed_records").head().getString(0).startsWith("<basket>"))
+ .xml(getTestResourcePath(resDir + "basket_invalid.xml")).cache()
+ assert(basketDF.filter($"_malformed_records".isNotNull).count() == 1)
+ assert(basketDF.filter($"_malformed_records".isNull).count() == 1)
+ val rec = basketDF.select("_malformed_records").collect()(1).getString(0)
+ assert(rec.startsWith("<basket>") && rec.indexOf("<extra>123</extra>") != -1 &&
+ rec.endsWith("</basket>"))
}
test("test XSD validation with addFile() with validation error") {
@@ -1111,17 +1163,22 @@ final class XmlSuite extends SharedSparkSession {
.option("rowValidationXSDPath", "basket.xsd")
.option("mode", "PERMISSIVE")
.option("columnNameOfCorruptRecord", "_malformed_records")
- .xml(getTestResourcePath(resDir + "basket_invalid.xml"))
- assert(basketDF.select("_malformed_records").head().getString(0).startsWith("<basket>"))
+ .xml(getTestResourcePath(resDir + "basket_invalid.xml")).cache()
+ assert(basketDF.filter($"_malformed_records".isNotNull).count() == 1)
+ assert(basketDF.filter($"_malformed_records".isNull).count() == 1)
+ val rec = basketDF.select("_malformed_records").collect()(1).getString(0)
+ assert(rec.startsWith("<basket>") && rec.indexOf("<extra>123</extra>") != -1 &&
+ rec.endsWith("</basket>"))
}
- test("test xmlRdd") {
+ test("test xmlDataset") {
val data = Seq(
"<ROW><year>2012</year><make>Tesla</make><model>S</model><comment>No comment</comment></ROW>",
"<ROW><year>1997</year><make>Ford</make><model>E350</model><comment>Get one</comment></ROW>",
"<ROW><year>2015</year><make>Chevy</make><model>Volt</model><comment>No</comment></ROW>")
- val rdd = spark.sparkContext.parallelize(data)
- assert(new XmlReader().xmlRdd(spark, rdd).collect().length === 3)
+ val xmlRDD = spark.sparkContext.parallelize(data)
+ val ds = spark.createDataset(xmlRDD)(Encoders.STRING)
+ assert(spark.read.xml(ds).collect().length === 3)
}
import testImplicits._
@@ -1132,30 +1189,39 @@ final class XmlSuite extends SharedSparkSession {
|</parent>
""".stripMargin
val df = Seq((8, xmlData)).toDF("number", "payload")
- val xmlSchema = schema_of_xml_df(df.select("payload"))
- val expectedSchema = df.schema.add("decoded", xmlSchema)
- val result = df.withColumn("decoded", from_xml(df.col("payload"), xmlSchema))
+ val xmlSchema = schema_of_xml(xmlData)
+ val schema = buildSchema(
+ field("_foo", StringType),
+ field("name", StringType),
+ field("pid", StringType))
+ val expectedSchema = df.schema.add("decoded", schema)
+ val result = df.withColumn("decoded",
+ from_xml(df.col("payload"), xmlSchema, Map[String, String]().asJava))
assert(expectedSchema === result.schema)
assert(result.select("decoded.pid").head().getString(0) === "14ft3")
assert(result.select("decoded._foo").head().getString(0) === "bar")
}
-
+ /*
test("from_xml array basic test") {
- val xmlData = Array(
- "<parent><pid>14ft3</pid><name>dave guy</name></parent>",
- "<parent><pid>12345</pid><name>other guy</name></parent>")
+ val xmlData =
+ """<parent><pid>12345</pid><name>dave guy</name></parent>
+ |<parent><pid>67890</pid><name>other guy</name></parent>""".stripMargin
val df = Seq((8, xmlData)).toDF("number", "payload")
- val xmlSchema = schema_of_xml_array(df.select("payload").as[Array[String]])
+ val xmlSchema = ArrayType(
+ StructType(
+ StructField("pid", IntegerType) ::
+ StructField("name", StringType) :: Nil))
val expectedSchema = df.schema.add("decoded", xmlSchema)
- val result = df.withColumn("decoded", from_xml(df.col("payload"), xmlSchema))
+ val result = df.withColumn("decoded",
+ from_xml(df.col("payload"), xmlSchema))
assert(expectedSchema === result.schema)
- // TBD: Following asserts fail when SharedSparkSession was used instead of SQLTestUtils
- // Disabling them for now
- // assert(result.selectExpr("decoded[0].pid").head().getString(0) === "14ft3")
- // assert(result.selectExpr("decoded[1].pid").head().getString(0) === "12345")
+ // TODO: ArrayType and MapType support in from_xml
+ // assert(result.selectExpr("decoded[0].pid").head().getInt(0) === 12345)
+ // assert(result.selectExpr("decoded[1].pid").head().getInt(1) === 67890)
}
+ */
test("from_xml error test") {
// XML contains error
@@ -1165,26 +1231,12 @@ final class XmlSuite extends SharedSparkSession {
|</parent>
""".stripMargin
val df = spark.createDataFrame(Seq((8, xmlData))).toDF("number", "payload")
- val xmlSchema = schema_of_xml_df(df.select("payload"))
- val result = df.withColumn("decoded", from_xml(df.col("payload"), xmlSchema))
+ val xmlSchema = schema_of_xml(xmlData)
+ val result = df.withColumn("decoded",
+ from_xml(df.col("payload"), xmlSchema, Map[String, String]().asJava))
assert(result.select("decoded._corrupt_record").head().getString(0).nonEmpty)
}
- test("from_xml_string basic test") {
- val xmlData =
- """<parent foo="bar"><pid>14ft3</pid>
- | <name>dave guy</name>
- |</parent>
- """.stripMargin
- val df = spark.createDataFrame(Seq((8, xmlData))).toDF("number", "payload")
- val xmlSchema = schema_of_xml_df(df.select("payload"))
- val result = from_xml_string(xmlData, xmlSchema)
-
- assert(result.getString(0) === "bar")
- assert(result.getString(1) === "dave guy")
- assert(result.getString(2) === "14ft3")
- }
-
test("from_xml with PERMISSIVE parse mode with no corrupt col schema") {
// XML contains error
val xmlData =
@@ -1198,10 +1250,11 @@ final class XmlSuite extends SharedSparkSession {
|</parent>
""".stripMargin
val dfNoError = spark.createDataFrame(Seq((8, xmlDataNoError))).toDF("number", "payload")
- val xmlSchema = schema_of_xml_df(dfNoError.select("payload"))
+ val xmlSchema = schema_of_xml(xmlDataNoError)
val df = spark.createDataFrame(Seq((8, xmlData))).toDF("number", "payload")
- val result = df.withColumn("decoded", from_xml(df.col("payload"), xmlSchema))
- assert(result.select("decoded").head().get(0) === null)
+ val result = df.withColumn("decoded",
+ from_xml(df.col("payload"), xmlSchema, Map[String, String]().asJava))
+ assert(result.select("decoded").head().get(0) === Row(null, null))
}
test("decimals with scale greater than precision") {
@@ -1220,7 +1273,8 @@ final class XmlSuite extends SharedSparkSession {
}
test("double field encounters whitespace-only value") {
- val schema = buildSchema(struct("Book", field("Price", DoubleType)), field("_corrupt_record"))
+ val schema = buildSchema(structField("Book", field("Price", DoubleType)),
+ field("_corrupt_record"))
val whitespaceDF = spark.read
.option("rowTag", "Books")
.schema(schema)
@@ -1231,7 +1285,8 @@ final class XmlSuite extends SharedSparkSession {
}
test("struct with only attributes and no value tag does not crash") {
- val schema = buildSchema(struct("book", field("_id", StringType)), field("_corrupt_record"))
+ val schema = buildSchema(structField("book", field("_id", StringType)),
+ field("_corrupt_record"))
val booksDF = spark.read
.option("rowTag", "book")
.schema(schema)
@@ -1291,6 +1346,8 @@ final class XmlSuite extends SharedSparkSession {
test("StructType with missing optional StructType child") {
val df = spark.read.option("rowTag", "Foo")
.xml(getTestResourcePath(resDir + "struct_with_optional_child.xml"))
+ val res = df.collect()
+ assert(res.length == 1)
assert(df.selectExpr("SIZE(Bar)").collect().head.getInt(0) === 2)
}
@@ -1396,8 +1453,9 @@ final class XmlSuite extends SharedSparkSession {
.option("dateFormat", "MM-dd-yyyy")
.xml(getTestResourcePath(resDir + "date.xml"))
val expectedSchema =
- buildSchema(field("author"), field("date", DateType), field("date2", DateType))
+ buildSchema(field("author"), field("date", TimestampType), field("date2", DateType))
assert(df.schema === expectedSchema)
+ assert(df.collect().head.getAs[Timestamp](1).toString === "2021-01-31 16:00:00.0")
assert(df.collect().head.getAs[Date](2).toString === "2021-02-01")
}
@@ -1409,32 +1467,40 @@ final class XmlSuite extends SharedSparkSession {
val expectedSchema =
buildSchema(
field("author"),
- field("time", TimestampType),
+ field("time", StringType),
field("time2", TimestampType),
field("time3", StringType),
field("time4", StringType)
)
assert(df.schema === expectedSchema)
- assert(df.collect().head.getAs[Timestamp](1).getTime === 1322907330000L)
+ assert(df.collect().head.get(1) === "2011-12-03T10:15:30Z")
assert(df.collect().head.getAs[Timestamp](2).getTime === 1322936130000L)
}
test("Test custom timestampFormat without timezone") {
+ val xml = s"""<book>
+ | <author>John Smith</author>
+ | <time>2011-12-03T10:15:30Z</time>
+ | <time2>12-03-2011 10:15:30 PST</time2>
+ | <time3>2011/12/03 06:15:30</time3>
+ |</book>""".stripMargin
+ val input = spark.createDataset(Seq(xml))
val df = spark.read
.option("rowTag", "book")
.option("timestampFormat", "yyyy/MM/dd HH:mm:ss")
- .xml(getTestResourcePath(resDir + "time.xml"))
+ .xml(input)
val expectedSchema =
buildSchema(
field("author"),
- field("time", TimestampType),
+ field("time", StringType),
field("time2", StringType),
- field("time3", TimestampType),
- field("time4", StringType)
+ field("time3", TimestampType)
)
assert(df.schema === expectedSchema)
- assert(df.collect().head.getAs[Timestamp](1).getTime === 1322907330000L)
- assert(df.collect().head.getAs[Timestamp](3).getTime === 1322892930000L)
+ val res = df.collect()
+ assert(res.head.get(1) === "2011-12-03T10:15:30Z")
+ assert(res.head.get(2) === "12-03-2011 10:15:30 PST")
+ assert(res.head.getAs[Timestamp](3).getTime === 1322892930000L)
}
test("Test custom timestampFormat with offset") {
@@ -1445,19 +1511,19 @@ final class XmlSuite extends SharedSparkSession {
val expectedSchema =
buildSchema(
field("author"),
- field("time", TimestampType),
+ field("time", StringType),
field("time2", StringType),
field("time3", StringType),
field("time4", TimestampType)
)
assert(df.schema === expectedSchema)
- assert(df.collect().head.getAs[Timestamp](1).getTime === 1322907330000L)
+ assert(df.collect().head.get(1) === "2011-12-03T10:15:30Z")
assert(df.collect().head.getAs[Timestamp](4).getTime === 1322892930000L)
}
test("Test null number type is null not 0.0") {
val schema = buildSchema(
- struct("Header",
+ structField("Header",
field("_Name"), field("_SequenceNumber", LongType)),
structArray("T",
field("_Number", LongType), field("_VALUE", DoubleType), field("_Volume", DoubleType)))
@@ -1507,12 +1573,22 @@ final class XmlSuite extends SharedSparkSession {
}
test("Issue 588: Ensure fails when data is not present, with or without schema") {
- intercept[InvalidInputException] {
+ val exception1 = intercept[AnalysisException] {
spark.read.xml("/this/file/does/not/exist")
}
- intercept[InvalidInputException] {
+ checkError(
+ exception = exception1,
+ errorClass = "PATH_NOT_FOUND",
+ parameters = Map("path" -> "file:/this/file/does/not/exist")
+ )
+ val exception2 = intercept[AnalysisException] {
spark.read.schema(buildSchema(field("dummy"))).xml("/this/file/does/not/exist")
}
+ checkError(
+ exception = exception2,
+ errorClass = "PATH_NOT_FOUND",
+ parameters = Map("path" -> "file:/this/file/does/not/exist")
+ )
}
test("Issue 614: mixed content element parsed as string in schema") {
@@ -1546,4 +1622,27 @@ final class XmlSuite extends SharedSparkSession {
}
}
+ test("read utf-8 encoded file") {
+ val df = spark.read
+ .option("charset", StandardCharsets.UTF_8.name)
+ .option("rowTag", "book")
+ .xml(getTestResourcePath(resDir + "books.xml"))
+ assert(df.collect().length === 12)
+ }
+
+ test("read file with unicode chars in row tag name") {
+ val df = spark.read
+ .option("charset", StandardCharsets.UTF_8.name)
+ .option("rowTag", "\u66F8") // scalastyle:ignore
+ .xml(getTestResourcePath(resDir + "books-unicode-in-tag-name.xml"))
+ assert(df.collect().length === 3)
+ }
+
+ test("read utf-8 encoded file with empty tag 2") {
+ val df = spark.read
+ .option("charset", StandardCharsets.UTF_8.name)
+ .option("rowTag", "House")
+ .xml(getTestResourcePath(resDir + "fias_house.xml"))
+ assert(df.collect().length === 37)
+ }
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/parsers/StaxXmlParserUtilsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/parsers/StaxXmlParserUtilsSuite.scala
index 0feadab8284..c239cfb92f8 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/parsers/StaxXmlParserUtilsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/parsers/StaxXmlParserUtilsSuite.scala
@@ -25,7 +25,7 @@ import scala.collection.JavaConverters._
import org.scalatest.BeforeAndAfterAll
import org.apache.spark.SparkFunSuite
-import org.apache.spark.sql.execution.datasources.xml.XmlOptions
+import org.apache.spark.sql.catalyst.xml.{StaxXmlParserUtils, XmlOptions}
final class StaxXmlParserUtilsSuite extends SparkFunSuite with BeforeAndAfterAll {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/util/TypeCastSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/util/TypeCastSuite.scala
index 45f6f820cbb..096fb3d83a5 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/util/TypeCastSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/util/TypeCastSuite.scala
@@ -17,15 +17,17 @@
package org.apache.spark.sql.execution.datasources.xml.util
import java.math.BigDecimal
-import java.sql.{Date, Timestamp}
-import java.time.{ZonedDateTime, ZoneId}
import java.util.Locale
-import org.apache.spark.SparkFunSuite
-import org.apache.spark.sql.execution.datasources.xml.XmlOptions
+import org.apache.spark.sql.catalyst.util.{DateFormatter, DateTimeUtils, TimestampFormatter}
+import org.apache.spark.sql.catalyst.util.LegacyDateFormats.FAST_DATE_FORMAT
+import org.apache.spark.sql.catalyst.xml.{TypeCast, XmlOptions}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
-final class TypeCastSuite extends SparkFunSuite {
+final class TypeCastSuite extends SharedSparkSession {
test("Can parse decimal type values") {
val options = new XmlOptions()
@@ -50,7 +52,7 @@ final class TypeCastSuite extends SparkFunSuite {
test("String type should always return the same as the input") {
val options = new XmlOptions()
- assert(TypeCast.castTo("", StringType, options) === "")
+ assert(TypeCast.castTo("", StringType, options) === UTF8String.fromString(""))
}
test("Types are cast correctly") {
@@ -65,73 +67,76 @@ final class TypeCastSuite extends SparkFunSuite {
assert(TypeCast.castTo("1", BooleanType, options) === true)
assert(TypeCast.castTo("false", BooleanType, options) === false)
assert(TypeCast.castTo("0", BooleanType, options) === false)
- assert(
- TypeCast.castTo("2002-05-30 21:46:54", TimestampType, options) ===
- Timestamp.from(
- ZonedDateTime.of(2002, 5, 30, 21, 46, 54, 0, ZoneId.of("UTC"))
- .toInstant()
- )
- )
- assert(
- TypeCast.castTo("2002-05-30T21:46:54", TimestampType, options) ===
- Timestamp.from(
- ZonedDateTime.of(2002, 5, 30, 21, 46, 54, 0, ZoneId.of("UTC"))
- .toInstant()
- )
- )
- assert(
- TypeCast.castTo("2002-05-30T21:46:54.1234", TimestampType, options) ===
- Timestamp.from(
- ZonedDateTime.of(2002, 5, 30, 21, 46, 54, 123400000, ZoneId.of("UTC"))
- .toInstant()
- )
- )
- assert(
- TypeCast.castTo("2002-05-30T21:46:54Z", TimestampType, options) ===
- Timestamp.from(
- ZonedDateTime.of(2002, 5, 30, 21, 46, 54, 0, ZoneId.of("UTC"))
- .toInstant()
- )
- )
- assert(
- TypeCast.castTo("2002-05-30T21:46:54-06:00", TimestampType, options) ===
- Timestamp.from(
- ZonedDateTime.of(2002, 5, 30, 21, 46, 54, 0, ZoneId.of("-06:00"))
- .toInstant()
- )
- )
- assert(
- TypeCast.castTo("2002-05-30T21:46:54+06:00", TimestampType, options) ===
- Timestamp.from(
- ZonedDateTime.of(2002, 5, 30, 21, 46, 54, 0, ZoneId.of("+06:00"))
- .toInstant()
- )
- )
- assert(
- TypeCast.castTo("2002-05-30T21:46:54.1234Z", TimestampType, options) ===
- Timestamp.from(
- ZonedDateTime.of(2002, 5, 30, 21, 46, 54, 123400000, ZoneId.of("UTC"))
- .toInstant()
- )
- )
- assert(
- TypeCast.castTo("2002-05-30T21:46:54.1234-06:00", TimestampType, options) ===
- Timestamp.from(
- ZonedDateTime.of(2002, 5, 30, 21, 46, 54, 123400000, ZoneId.of("-06:00"))
- .toInstant()
- )
- )
- assert(
- TypeCast.castTo("2002-05-30T21:46:54.1234+06:00", TimestampType, options) ===
- Timestamp.from(
- ZonedDateTime.of(2002, 5, 30, 21, 46, 54, 123400000, ZoneId.of("+06:00"))
- .toInstant()
- )
- )
- assert(TypeCast.castTo("2002-09-24", DateType, options) === Date.valueOf("2002-09-24"))
- assert(TypeCast.castTo("2002-09-24Z", DateType, options) === Date.valueOf("2002-09-24"))
- assert(TypeCast.castTo("2002-09-24-06:00", DateType, options) === Date.valueOf("2002-09-24"))
- assert(TypeCast.castTo("2002-09-24+06:00", DateType, options) === Date.valueOf("2002-09-24"))
+
+ {
+ val ts = TypeCast.castTo("2002-05-30 21:46:54", TimestampType, options)
+ assert(ts === 1022820414000000L)
+ assert(ts ===
+ TimestampFormatter(None, DateTimeUtils.getZoneId(SQLConf.get.sessionLocalTimeZone),
+ Locale.US, FAST_DATE_FORMAT, true).parse("2002-05-30 21:46:54"))
+ }
+ {
+ val ts = TypeCast.castTo("2002-05-30T21:46:54", TimestampType, options)
+ assert(ts === 1022820414000000L)
+ assert(ts ===
+ TimestampFormatter(None, DateTimeUtils.getZoneId(SQLConf.get.sessionLocalTimeZone),
+ Locale.US, FAST_DATE_FORMAT, true).parse("2002-05-30T21:46:54"))
+ }
+ {
+ val ts = TypeCast.castTo("2002-05-30T21:46:54.1234", TimestampType, options)
+ assert(ts === 1022820414123400L)
+ assert(ts ===
+ TimestampFormatter(None, DateTimeUtils.getZoneId(SQLConf.get.sessionLocalTimeZone),
+ Locale.US, FAST_DATE_FORMAT, true).parse("2002-05-30T21:46:54.1234"))
+ }
+ {
+ val ts = TypeCast.castTo("2002-05-30T21:46:54Z", TimestampType, options)
+ assert(ts === 1022795214000000L)
+ assert(ts ===
+ TimestampFormatter(None, DateTimeUtils.getZoneId("UTC"),
+ Locale.US, FAST_DATE_FORMAT, true).parse("2002-05-30T21:46:54Z"))
+ }
+ {
+ val ts = TypeCast.castTo("2002-05-30T21:46:54-06:00", TimestampType, options)
+ assert(ts === 1022816814000000L)
+ assert(ts ===
+ TimestampFormatter(None, DateTimeUtils.getZoneId("-06:00"),
+ Locale.US, FAST_DATE_FORMAT, true).parse("2002-05-30T21:46:54-06:00"))
+ }
+ {
+ val ts = TypeCast.castTo("2002-05-30T21:46:54+06:00", TimestampType, options)
+ assert(ts === 1022773614000000L)
+ assert(ts ===
+ TimestampFormatter(None, DateTimeUtils.getZoneId("+06:00"),
+ Locale.US, FAST_DATE_FORMAT, true).parse("2002-05-30T21:46:54+06:00"))
+ }
+ {
+ val ts = TypeCast.castTo("2002-05-30T21:46:54.1234Z", TimestampType, options)
+ assert(ts === 1022795214123400L)
+ assert(ts ===
+ TimestampFormatter(None, DateTimeUtils.getZoneId("UTC"),
+ Locale.US, FAST_DATE_FORMAT, true).parse("2002-05-30T21:46:54.1234Z"))
+ }
+ {
+ val ts = TypeCast.castTo("2002-05-30T21:46:54.1234-06:00", TimestampType, options)
+ assert(ts === 1022816814123400L)
+ assert(ts ===
+ TimestampFormatter(None, DateTimeUtils.getZoneId("-06:00"),
+ Locale.US, FAST_DATE_FORMAT, true).parse("2002-05-30T21:46:54.1234-06:00"))
+ }
+ {
+ val ts = TypeCast.castTo("2002-05-30T21:46:54.1234+06:00", TimestampType, options)
+ assert(ts === 1022773614123400L)
+ assert(ts ===
+ TimestampFormatter(None, DateTimeUtils.getZoneId("+06:00"),
+ Locale.US, FAST_DATE_FORMAT, true).parse("2002-05-30T21:46:54.1234+06:00"))
+ }
+ {
+ val date = TypeCast.castTo("2002-09-24", DateType, options)
+ assert(date === 11954)
+ assert(date === DateFormatter(DateFormatter.defaultPattern,
+ Locale.US, FAST_DATE_FORMAT, true).parse("2002-09-24"))
+ }
}
test("Types with sign are cast correctly") {
@@ -167,10 +172,9 @@ final class TypeCastSuite extends SparkFunSuite {
test("Parsing built-in timestamp formatters") {
val options = XmlOptions(Map())
- val expectedResult = Timestamp.from(
- ZonedDateTime.of(2002, 5, 30, 21, 46, 54, 0, ZoneId.of("UTC"))
- .toInstant
- )
+ val expectedResult =
+ TimestampFormatter(None, DateTimeUtils.getZoneId(SQLConf.get.sessionLocalTimeZone),
+ Locale.US, FAST_DATE_FORMAT, true).parse("2002-05-30 21:46:54")
assert(
TypeCast.castTo("2002-05-30 21:46:54", TimestampType, options) === expectedResult
)
@@ -178,10 +182,14 @@ final class TypeCastSuite extends SparkFunSuite {
TypeCast.castTo("2002-05-30T21:46:54", TimestampType, options) === expectedResult
)
assert(
- TypeCast.castTo("2002-05-30T21:46:54+00:00", TimestampType, options) === expectedResult
+ TypeCast.castTo("2002-05-30T21:46:54+00:00", TimestampType, options) ===
+ TimestampFormatter(None, DateTimeUtils.getZoneId(SQLConf.get.sessionLocalTimeZone),
+ Locale.US, FAST_DATE_FORMAT, true).parse("2002-05-30T21:46:54+00:00")
)
assert(
- TypeCast.castTo("2002-05-30T21:46:54.0000Z", TimestampType, options) === expectedResult
+ TypeCast.castTo("2002-05-30T21:46:54.0000Z", TimestampType, options) ===
+ TimestampFormatter(None, DateTimeUtils.getZoneId(SQLConf.get.sessionLocalTimeZone),
+ Locale.US, FAST_DATE_FORMAT, true).parse("2002-05-30T21:46:54.0000Z")
)
}
@@ -189,48 +197,38 @@ final class TypeCastSuite extends SparkFunSuite {
var options = XmlOptions(Map("timestampFormat" -> "MM-dd-yyyy HH:mm:ss", "timezone" -> "UTC"))
assert(
TypeCast.castTo("12-03-2011 10:15:30", TimestampType, options) ===
- Timestamp.from(
- ZonedDateTime.of(2011, 12, 3, 10, 15, 30, 0, ZoneId.of("UTC"))
- .toInstant
- )
+ TimestampFormatter("MM-dd-yyyy HH:mm:ss", DateTimeUtils.getZoneId("UTC"),
+ Locale.US, FAST_DATE_FORMAT, true).parse("12-03-2011 10:15:30")
)
options = XmlOptions(Map("timestampFormat" -> "yyyy/MM/dd HH:mm:ss", "timezone" -> "UTC"))
assert(
TypeCast.castTo("2011/12/03 10:15:30", TimestampType, options) ===
- Timestamp.from(
- ZonedDateTime.of(2011, 12, 3, 10, 15, 30, 0, ZoneId.of("UTC"))
- .toInstant
- )
+ TimestampFormatter("yyyy/MM/dd HH:mm:ss", DateTimeUtils.getZoneId("UTC"),
+ Locale.US, FAST_DATE_FORMAT, true).parse("2011/12/03 10:15:30")
)
options = XmlOptions(Map("timestampFormat" -> "yyyy/MM/dd HH:mm:ss",
"timezone" -> "Asia/Shanghai"))
assert(
- TypeCast.castTo("2011/12/03 10:15:30", TimestampType, options) !==
- Timestamp.from(
- ZonedDateTime.of(2011, 12, 3, 10, 15, 30, 0, ZoneId.of("UTC"))
- .toInstant
- )
+ TypeCast.castTo("2011/12/03 10:15:30", TimestampType, options) ===
+ TimestampFormatter("yyyy/MM/dd HH:mm:ss", DateTimeUtils.getZoneId("Asia/Shanghai"),
+ Locale.US, FAST_DATE_FORMAT, true).parse("2011/12/03 10:15:30")
)
options = XmlOptions(Map("timestampFormat" -> "yyyy/MM/dd HH:mm:ss",
"timezone" -> "Asia/Shanghai"))
assert(
TypeCast.castTo("2011/12/03 10:15:30", TimestampType, options) ===
- Timestamp.from(
- ZonedDateTime.of(2011, 12, 3, 10, 15, 30, 0, ZoneId.of("Asia/Shanghai"))
- .toInstant
- )
+ TimestampFormatter("yyyy/MM/dd HH:mm:ss", DateTimeUtils.getZoneId("Asia/Shanghai"),
+ Locale.US, FAST_DATE_FORMAT, true).parse("2011/12/03 10:15:30")
)
options = XmlOptions(Map("timestampFormat" -> "yyyy/MM/dd HH:mm:ss"))
- intercept[IllegalArgumentException](
- TypeCast.castTo("2011/12/03 10:15:30", TimestampType, options) ===
- Timestamp.from(
- ZonedDateTime.of(2011, 12, 3, 10, 15, 30, 0, ZoneId.of("UTC"))
- .toInstant
- )
+ assert(TypeCast.castTo("2011/12/03 10:15:30", TimestampType, options) ===
+ TimestampFormatter("yyyy/MM/dd HH:mm:ss",
+ DateTimeUtils.getZoneId(SQLConf.get.sessionLocalTimeZone),
+ Locale.US, FAST_DATE_FORMAT, true).parse("2011/12/03 10:15:30")
)
}
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/util/XSDToSchemaSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/util/XSDToSchemaSuite.scala
index 10b14a3f6bc..e163c05846d 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/util/XSDToSchemaSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/util/XSDToSchemaSuite.scala
@@ -18,6 +18,7 @@ package org.apache.spark.sql.execution.datasources.xml.util
import java.nio.file.Paths
+import org.apache.spark.sql.catalyst.xml.XSDToSchema
import org.apache.spark.sql.execution.datasources.xml.TestUtils._
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types.{ArrayType, DecimalType, FloatType, LongType, StringType}
@@ -31,7 +32,7 @@ class XSDToSchemaSuite extends SharedSparkSession {
.replace("file:/", "/")))
val expectedSchema = buildSchema(
field("basket",
- struct(
+ structField(
structArray("entry",
field("key"),
field("value"))), nullable = false))
@@ -43,7 +44,7 @@ class XSDToSchemaSuite extends SharedSparkSession {
.replace("file:/", "/")))
val expectedSchema = buildSchema(
field("basket",
- struct(
+ structField(
structArray("entry",
field("key"),
field("value"))), nullable = false))
@@ -55,9 +56,9 @@ class XSDToSchemaSuite extends SharedSparkSession {
.replace("file:/", "/")))
val expectedSchema = buildSchema(
field("catalog",
- struct(
+ structField(
field("product",
- struct(
+ structField(
structArray("catalog_item",
field("item_number", nullable = false),
field("price", FloatType, nullable = false),
@@ -78,7 +79,7 @@ class XSDToSchemaSuite extends SharedSparkSession {
val parsedSchema = XSDToSchema.read(Paths.get(testFile(resDir + "choice.xsd")
.replace("file:/", "/")))
val expectedSchema = buildSchema(
- field("el", struct(field("foo"), field("bar"), field("baz")), nullable = false))
+ field("el", structField(field("foo"), field("bar"), field("baz")), nullable = false))
assert(expectedSchema === parsedSchema)
}
@@ -94,21 +95,21 @@ class XSDToSchemaSuite extends SharedSparkSession {
.replace("file:/", "/")))
val expectedSchema = buildSchema(
field("root",
- struct(
+ structField(
field("foo",
- struct(
+ structField(
field("xs_any")),
nullable = false),
field("bar",
- struct(
+ structField(
field("xs_any", nullable = false)),
nullable = false),
field("baz",
- struct(
+ structField(
field("xs_any", ArrayType(StringType), nullable = false)),
nullable = false),
field("bing",
- struct(
+ structField(
field("xs_any")),
nullable = false)),
nullable = false))
@@ -120,7 +121,7 @@ class XSDToSchemaSuite extends SharedSparkSession {
.replace("file:/", "/")))
val expectedSchema = buildSchema(
field("test",
- struct(field("userId", LongType, nullable = false)), nullable = false))
+ structField(field("userId", LongType, nullable = false)), nullable = false))
assert(parsedSchema === expectedSchema)
}
@@ -141,7 +142,7 @@ class XSDToSchemaSuite extends SharedSparkSession {
val expectedSchema = buildSchema(
field(
"book",
- struct(
+ structField(
field("name", StringType, false),
field("author", StringType, false),
field("isbn", StringType, false)
@@ -150,7 +151,7 @@ class XSDToSchemaSuite extends SharedSparkSession {
),
field(
"bookList",
- struct(
+ structField(
structArray(
"book",
field("name", StringType, false),
@@ -171,7 +172,7 @@ class XSDToSchemaSuite extends SharedSparkSession {
val expectedSchema = buildSchema(
field(
"employee",
- struct(
+ structField(
field("firstname", StringType, false),
field("lastname", StringType, false),
field("address", StringType, false),
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/util/XmlFileSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/util/XmlFileSuite.scala
deleted file mode 100644
index 075f38153ad..00000000000
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/util/XmlFileSuite.scala
+++ /dev/null
@@ -1,69 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.spark.sql.execution.datasources.xml.util
-
-import java.nio.charset.{StandardCharsets, UnsupportedCharsetException}
-
-import org.apache.spark.sql.test.SharedSparkSession
-
-final class XmlFileSuite extends SharedSparkSession {
-
- private val resourcePrefix = "test-data/xml-resources/"
- private val booksFile = testFile(resourcePrefix + "books.xml")
- private val booksUnicodeInTagNameFile = testFile(resourcePrefix + "books-unicode-in-tag-name.xml")
- private val booksFileTag = "book"
- private val booksUnicodeFileTag = "\u66F8" // scalastyle:ignore
- private val numBooks = 12
- private val numBooksUnicodeInTagName = 3
- private val fiasHouse = testFile(resourcePrefix + "fias_house.xml")
- private val fiasRowTag = "House"
- private val numHouses = 37
- private val utf8 = StandardCharsets.UTF_8.name
-
- override def beforeAll(): Unit = {
- super.beforeAll()
- }
-
- override def afterAll(): Unit = {
- super.afterAll()
- }
-
- test("read utf-8 encoded file") {
- val baseRDD = XmlFile.withCharset(sparkContext, booksFile, utf8, rowTag = booksFileTag)
- assert(baseRDD.count() === numBooks)
- }
-
- test("read file with unicode chars in row tag name") {
- val baseRDD = XmlFile.withCharset(
- sparkContext, booksUnicodeInTagNameFile, utf8, rowTag = booksUnicodeFileTag)
- assert(baseRDD.count() === numBooksUnicodeInTagName)
- }
-
- test("read utf-8 encoded file with empty tag") {
- val baseRDD = XmlFile.withCharset(sparkContext, fiasHouse, utf8, rowTag = fiasRowTag)
- assert(baseRDD.count() == numHouses)
- baseRDD.collect().foreach(x => assert(x.contains("/>")))
- }
-
- test("unsupported charset") {
- val exception = intercept[UnsupportedCharsetException] {
- XmlFile.withCharset(sparkContext, booksFile, "frylock", rowTag = booksFileTag).count()
- }
- assert(exception.getMessage.contains("frylock"))
- }
-
-}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org