This is an automated email from the ASF dual-hosted git repository.
MaxGekk pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 7a009e23c0b6 [SPARK-57802][SQL] Make XML schema inference incremental,
aligning with CSV
7a009e23c0b6 is described below
commit 7a009e23c0b6bdf85afb21ab545f31ea783934dc
Author: Wenchen Fan <[email protected]>
AuthorDate: Thu Jul 2 12:48:24 2026 +0200
[SPARK-57802][SQL] Make XML schema inference incremental, aligning with CSV
### What changes were proposed in this pull request?
This makes XML schema inference **incremental**: instead of inferring each
primitive value from scratch and merging afterwards, each value is refined
starting from the type already inferred for that field so far. This aligns XML
with the CSV datasource (`CSVInferSchema.inferField`), which already works this
way.
Concretely:
- `XmlInferSchema.inferFrom(datum, typeSoFar)` dispatches on `typeSoFar`
and enters a `tryParse*` cascade at the parser matching it, so a field already
known to be (say) `Double` is not re-checked against `Long`. The result is
reconciled with `typeSoFar` via `compatibleType`, keeping the type lattice
monotonic (a field's type can only widen, never narrow).
- The RDD driver (`infer(xml: RDD[String])`) threads the schema merged so
far in each partition back into the next record's inference as a hint. The hint
is propagated down through `inferObject`/`inferField` per field (unwrapping
array/value-tag wrappers), mirroring how the field's type is accumulated.
- A new config,
`spark.sql.xml.schemaInference.incrementalTypeCasting.enabled` (default
**true**), gates the behavior. Setting it to false restores the legacy
per-record inference. The old `is*`-predicate helpers are replaced by the
`tryParse*` cascade.
This is **behavior-preserving**: the inferred schema is unchanged. The
change avoids redundant parse attempts once a field has widened, and reduces
divergence between the CSV, JSON, and XML inference implementations. The config
is provided as a kill switch should any corner-case difference surface in the
field.
### Why are the changes needed?
XML was the only text-based datasource still inferring each value from
scratch on every row. CSV (and, structurally, JSON) already refine from the
type-so-far. Aligning XML:
- avoids redundant work (once a field is `Double`, later values no longer
re-run the `Long`/`Decimal` checks);
- makes inference monotonic and easier to reason about;
- reduces gratuitous divergence between the three inference code paths.
### Does this PR introduce _any_ user-facing change?
No change to inferred schemas. A new internal SQL config
`spark.sql.xml.schemaInference.incrementalTypeCasting.enabled` (default true)
is added as a kill switch for the new inference path.
### How was this patch tested?
- New unit suite `XmlInferSchemaTypeCastingSuite` (catalyst) tests
`inferFrom(value, typeSoFar)` directly across the type lattice: widening,
incompatible-value fallback to `StringType`,
timestamp/date/boolean-from-other-types, and empty/null handling. Mirrors
`CSVInferSchemaSuite`.
- New end-to-end test in `XmlInferSchemaSuite` asserts the incremental path
(flag on) and the legacy batch path (flag off) produce **identical** schemas
across mixed numerics, nested structs, repeated elements (arrays), attributes,
value tags, and cross-row widening.
- Existing `XmlInferSchemaSuite`, `XmlSuite`, and `XmlVariantSuite` pass
unchanged with the flag defaulted on.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Anthropic Claude Opus)
Closes #56919 from cloud-fan/xml-incremental-inference.
Authored-by: Wenchen Fan <[email protected]>
Signed-off-by: Max Gekk <[email protected]>
(cherry picked from commit d76b07926ea31988114769c618279d504e3ad805)
Signed-off-by: Max Gekk <[email protected]>
---
.../spark/sql/catalyst/xml/XmlInferSchema.scala | 357 ++++++++++++++++-----
.../org/apache/spark/sql/internal/SQLConf.scala | 18 ++
.../xml/XmlInferSchemaTypeCastingSuite.scala | 165 ++++++++++
.../datasources/xml/XmlInferSchemaSuite.scala | 84 +++++
4 files changed, 539 insertions(+), 85 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/XmlInferSchema.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/XmlInferSchema.scala
index 2471cfc735bd..fb534062ec28 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/XmlInferSchema.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/XmlInferSchema.scala
@@ -33,7 +33,7 @@ import scala.xml.SAXException
import org.apache.hadoop.hdfs.BlockMissingException
import org.apache.hadoop.security.AccessControlException
-import org.apache.spark.SparkIllegalArgumentException
+import org.apache.spark.{SparkException, SparkIllegalArgumentException}
import org.apache.spark.internal.Logging
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.analysis.TypeCoercion
@@ -77,6 +77,8 @@ class XmlInferSchema(private val options: XmlOptions, private
val caseSensitive:
private val isTimeTypeEnabled = SQLConf.get.isTimeTypeEnabled
+ private val incrementalTypeCasting =
SQLConf.get.xmlSchemaInferenceIncrementalTypeCasting
+
override def equals(obj: Any): Boolean = obj match {
case other: XmlInferSchema =>
options == other.options &&
@@ -123,9 +125,27 @@ class XmlInferSchema(private val options: XmlOptions,
private val caseSensitive:
val mergedTypesFromPartitions = schemaData.mapPartitions { iter =>
val xsdSchema =
Option(options.rowValidationXSDPath).map(ValidatorUtil.getSchema)
- iter.flatMap { xml =>
- infer(xml, xsdSchema)
- }.reduceOption(compatibleType(caseSensitive, options.valueTag)).iterator
+ if (incrementalTypeCasting) {
+ // Incremental inference: carry the schema merged so far in this
partition and feed it
+ // back as a hint into the next record's inference, so each field's
type is refined
+ // rather than re-derived from scratch. The final partition type is
the last (fully
+ // merged) schema, matching the batch path's fold up to
`compatibleType`'s associativity.
+ var schemaSoFar: DataType = StructType(Nil)
+ iter.flatMap { xml =>
+ val schemaHints = schemaSoFar match {
+ case st: StructType => st
+ case _ => StructType(Nil)
+ }
+ infer(xml, xsdSchema, schemaHints).map { dataType =>
+ schemaSoFar = compatibleType(caseSensitive,
options.valueTag)(schemaSoFar, dataType)
+ schemaSoFar
+ }
+ }.reduceOption((_, latest) => latest).iterator
+ } else {
+ iter.flatMap { xml =>
+ infer(xml, xsdSchema)
+ }.reduceOption(compatibleType(caseSensitive,
options.valueTag)).iterator
+ }
}
// Here we manually submit a fold-like Spark job, so that we can set the
SQLConf when running
@@ -150,7 +170,10 @@ class XmlInferSchema(private val options: XmlOptions,
private val caseSensitive:
}
}
- def infer(xml: String, xsdSchema: Option[Schema] = None): Option[DataType] =
{
+ def infer(
+ xml: String,
+ xsdSchema: Option[Schema],
+ schemaHints: StructType): Option[DataType] = {
var parser: XMLEventReader = null
try {
val xsd =
xsdSchema.orElse(Option(options.rowValidationXSDPath).map(ValidatorUtil.getSchema))
@@ -159,7 +182,7 @@ class XmlInferSchema(private val options: XmlOptions,
private val caseSensitive:
}
parser = StaxXmlParserUtils.filteredReader(xml)
val rootAttributes = StaxXmlParserUtils.gatherRootAttributes(parser)
- val schema = Some(inferObject(parser, rootAttributes))
+ val schema = Some(inferObject(parser, rootAttributes, Some(schemaHints)))
parser.close()
schema
} catch {
@@ -194,6 +217,9 @@ class XmlInferSchema(private val options: XmlOptions,
private val caseSensitive:
}
}
+ def infer(xml: String, xsdSchema: Option[Schema] = None): Option[DataType] =
+ infer(xml, xsdSchema, StructType(Nil))
+
def inferFromReaders(recordReader: RDD[StaxXMLRecordReader]): StructType = {
val sampledRecordReader = if (options.samplingRatio < 1.0) {
recordReader.sample(withReplacement = false, options.samplingRatio, 1)
@@ -312,41 +338,80 @@ class XmlInferSchema(private val options: XmlOptions,
private val caseSensitive:
}
private def inferFrom(datum: String): DataType = {
+ // Start type inference from scratch, without any prior knowledge of the
field's type.
+ inferFrom(datum, NullType)
+ }
+
+ /**
+ * Infer the type of a primitive value, refining a type already inferred for
the same field
+ * from previous values.
+ *
+ * Rather than probing every candidate type from scratch on every value (the
historical
+ * approach), this dispatches on `typeSoFar` -- the type accumulated for
this field so far --
+ * and only attempts to widen it. Given a value already known to be
`Double`, for example,
+ * there is no point checking whether it is a `Long`, as the merged type can
only be `Double`
+ * or wider. This mirrors the schema inference used by the CSV datasource
+ * (`CSVInferSchema.inferField`) and keeps the type lattice monotonic: each
value can only
+ * hold the type where it is or widen it, never narrow it. `compatibleType`
reconciles the
+ * newly inferred type with `typeSoFar`, falling back to `StringType` (the
top type) when they
+ * are incompatible.
+ */
+ private[xml] def inferFrom(datum: String, typeSoFar: DataType): DataType = {
val value = if (datum != null && options.ignoreSurroundingSpaces) {
datum.trim()
} else {
datum
}
- if (options.inferSchema) {
- lazy val decimalTry = tryParseDecimal(value)
- lazy val timestampNTZTry = tryParseTimestampNTZ(value)
- value match {
- case null => NullType
- case v if v.isEmpty => NullType
- case v if isLong(v) => LongType
- case v if options.prefersDecimal && decimalTry.isDefined =>
decimalTry.get
- case v if isDouble(v) => DoubleType
- case v if isBoolean(v) => BooleanType
- case v if isTimeTypeEnabled && isTime(v) =>
TimeType(TimeType.DEFAULT_PRECISION)
- case v if isDate(v) => DateType
- case v if timestampNTZTry.isDefined => timestampNTZTry.get
- case v if isTimestamp(v) => TimestampType
- case _ => StringType
- }
- } else {
- StringType
- }
+ if (!options.inferSchema) {
+ return StringType
+ }
+ if (value == null || value.isEmpty) {
+ return typeSoFar
+ }
+
+ val inferredType = typeSoFar match {
+ // For a numeric type inferred so far, re-enter the cascade at its top
(`tryParseLong`)
+ // rather than at the narrower numeric parser. Entering at
`tryParseDecimal` would infer an
+ // integer value directly as a narrow `Decimal` (e.g. "5" ->
Decimal(1,0)) instead of the
+ // `Long` that from-scratch inference produces; `compatibleType` then
merges those to
+ // different Decimal precisions (Decimal(5,2) vs Decimal(22,2)), which
would make the
+ // inferred type differ from the legacy path and depend on row order.
Re-entering at the top
+ // yields the same representative as from-scratch inference, so the
merged type matches.
+ case NullType | LongType | DoubleType | _: DecimalType =>
tryParseLong(value)
+ case BooleanType => tryParseBoolean(value)
+ // For a temporal type inferred so far, re-enter the cascade at the top
of the temporal
+ // sub-cascade (`tryParseTime`, which flows Time -> Date -> TimestampNTZ
-> Timestamp)
+ // rather than at the narrower temporal parser. Entering at
`tryParseTimestamp` for a
+ // `Timestamp` type-so-far, for example, would fail to parse a date-only
value (when
+ // `timestampFormat` requires a time part) and fall through to `String`;
`compatibleType`
+ // then merges `Timestamp` with `String` to `String`, whereas
from-scratch inference yields
+ // `Date` which widens with `Timestamp` to `Timestamp` via
`findWiderDateTimeType`. That
+ // would make the inferred type differ from the legacy path and depend
on row order.
+ // Re-entering at the top yields the same representative as from-scratch
inference (which
+ // reaches the temporal parsers through `tryParseDouble`), so the merged
type matches.
+ case _: TimeType | DateType | TimestampNTZType | _:
TimestampNTZNanosType | TimestampType =>
+ tryParseTime(value)
+ case StringType => StringType
+ case other: DataType =>
+ throw SparkException.internalError(s"Unexpected data type $other")
+ }
+ compatibleType(caseSensitive, options.valueTag)(typeSoFar, inferredType)
}
- private def inferField(parser: XMLEventReader): DataType = {
+ private def inferField(
+ parser: XMLEventReader,
+ dataTypeHintOpt: Option[DataType] = None): DataType = {
parser.peek match {
case _: EndElement =>
parser.nextEvent()
NullType
- case _: StartElement => inferObject(parser)
+ case _: StartElement =>
+ inferObject(parser, schemaHintsOpt = structTypeHint(dataTypeHintOpt))
case _: Characters =>
- val structType = inferObject(parser).asInstanceOf[StructType]
+ val structType =
+ inferObject(parser, schemaHintsOpt =
structTypeHintWithValueTag(dataTypeHintOpt))
+ .asInstanceOf[StructType]
structType match {
case _ if structType.fields.isEmpty =>
NullType
@@ -369,7 +434,8 @@ class XmlInferSchema(private val options: XmlOptions,
private val caseSensitive:
*/
private def inferObject(
parser: XMLEventReader,
- rootAttributes: Array[Attribute] = Array.empty): DataType = {
+ rootAttributes: Array[Attribute] = Array.empty,
+ schemaHintsOpt: Option[StructType] = None): DataType = {
/**
* Retrieves the field name with respect to the case sensitivity setting.
* We pick the first name we encountered.
@@ -393,12 +459,22 @@ class XmlInferSchema(private val options: XmlOptions,
private val caseSensitive:
val nameToDataType =
collection.mutable.TreeMap.empty[String,
DataType](caseSensitivityOrdering)
+ // When incremental inference is on, `schemaHintsOpt` carries the schema
inferred for this
+ // object from previous records. `fieldToHint` maps each field name to its
already-inferred
+ // type so it can be threaded into `inferFrom`/`inferField` as the
starting point, refining
+ // it rather than re-deriving from scratch. Empty for the batch path
(`schemaHintsOpt` None).
+ val fieldToHint: Map[String, DataType] = schemaHintsOpt match {
+ case Some(schema) if caseSensitive => schema.nameToDataType
+ case Some(schema) => schema.nameToDataTypeCaseInsensitive
+ case None => Map.empty
+ }
+
// If there are attributes, then we should process them first.
val rootValuesMap =
StaxXmlParserUtils.convertAttributesToValuesMap(rootAttributes, options)
rootValuesMap.foreach {
case (f, v) =>
- addOrUpdateType(nameToDataType, f, inferFrom(v))
+ addOrUpdateType(nameToDataType, f, inferFrom(v,
matchingPrimitiveHint(fieldToHint.get(f))))
}
var shouldStop = false
while (!shouldStop) {
@@ -407,14 +483,16 @@ class XmlInferSchema(private val options: XmlOptions,
private val caseSensitive:
val attributes = e.getAttributes.asScala.toArray
val valuesMap =
StaxXmlParserUtils.convertAttributesToValuesMap(attributes, options)
val field = StaxXmlParserUtils.getName(e.asStartElement.getName,
options)
- val inferredType = inferField(parser) match {
+ val fieldHintOpt = fieldToHint.get(field)
+ val inferredType = inferField(parser, fieldHintOpt) match {
case st: StructType if valuesMap.nonEmpty =>
// Merge attributes to the field
val nestedBuilder = ArrayBuffer[StructField]()
nestedBuilder ++= st.fields
valuesMap.foreach {
case (f, v) =>
- nestedBuilder += StructField(f, inferFrom(v), nullable =
true)
+ val attrHint = matchingAttributeHint(fieldHintOpt, f)
+ nestedBuilder += StructField(f, inferFrom(v, attrHint),
nullable = true)
}
StructType(nestedBuilder.sortBy(_.name).toArray)
@@ -426,7 +504,8 @@ class XmlInferSchema(private val options: XmlOptions,
private val caseSensitive:
}
valuesMap.foreach {
case (f, v) =>
- nestedBuilder += StructField(f, inferFrom(v), nullable =
true)
+ val attrHint = matchingAttributeHint(fieldHintOpt, f)
+ nestedBuilder += StructField(f, inferFrom(v, attrHint),
nullable = true)
}
StructType(nestedBuilder.sortBy(_.name).toArray)
@@ -437,7 +516,8 @@ class XmlInferSchema(private val options: XmlOptions,
private val caseSensitive:
case c: Characters if !c.isWhiteSpace =>
// This is a value tag
- val valueTagType = inferFrom(c.getData)
+ val valueTagType =
+ inferFrom(c.getData,
matchingPrimitiveHint(fieldToHint.get(options.valueTag)))
addOrUpdateType(nameToDataType, options.valueTag, valueTagType)
case _: EndElement | _: EndDocument =>
@@ -453,87 +533,177 @@ class XmlInferSchema(private val options: XmlOptions,
private val caseSensitive:
}.toList.sortBy(_.name))
}
+ // --- Helpers for incremental inference: extracting the type hint for a
nested field ---
+ //
+ // When incremental inference threads a previously-inferred schema back in,
a field's hint may
+ // be wrapped (a leaf value gets represented as a single-`valueTag` struct,
and repeated elements
+ // as an array). These helpers unwrap the hint to the shape the next
inference step expects.
+
/**
- * Helper method that checks and cast string representation of a numeric
types.
+ * Reduce a field's hint to the primitive type to feed into `inferFrom` for
a leaf value.
+ * A leaf value may have been inferred previously either as a bare
primitive, or -- when the
+ * element also carried attributes -- as a struct with a `valueTag` field,
possibly wrapped in
+ * an array for repeated elements. Anything else (a struct without a value
tag) offers no usable
+ * primitive hint, so we fall back to `NullType` (i.e. infer from scratch).
*/
- private def isBoolean(value: String): Boolean = {
- value.toLowerCase(Locale.ROOT) match {
- case "true" | "false" => true
- case _ => false
+ private def matchingPrimitiveHint(dataTypeHintOpt: Option[DataType]):
DataType = {
+ def unwrap(dt: DataType): DataType = dt match {
+ case ArrayType(elementType, _) => unwrap(elementType)
+ case st: StructType if st.fieldNames.contains(options.valueTag) =>
+ st(options.valueTag).dataType match {
+ case ArrayType(elementType, _) => elementType
+ case other => other
+ }
+ case _: StructType => NullType
+ case primitive => primitive
+ }
+ dataTypeHintOpt.map(unwrap).map {
+ // A value tag can only hold a primitive; a nested complex type here
means the hint is not
+ // usable as a primitive starting point, so infer from scratch.
+ case _: StructType | _: MapType => NullType
+ case primitive => primitive
+ }.getOrElse(NullType)
+ }
+
+ /**
+ * Extract the hint for a named attribute from a field's hint. Attributes
live as fields of the
+ * struct (or the struct element of an array) the field was inferred as.
+ */
+ private def matchingAttributeHint(
+ dataTypeHintOpt: Option[DataType],
+ attributeName: String): DataType = {
+ val structOpt = dataTypeHintOpt match {
+ case Some(st: StructType) => Some(st)
+ case Some(ArrayType(st: StructType, _)) => Some(st)
+ case _ => None
+ }
+ matchingPrimitiveHint(structOpt.flatMap(_.fields.find(_.name ==
attributeName).map(_.dataType)))
+ }
+
+ /** Reduce a field's hint to a struct hint for a nested object (an array
uses its element). */
+ private def structTypeHint(dataTypeHintOpt: Option[DataType]):
Option[StructType] =
+ dataTypeHintOpt match {
+ case Some(st: StructType) => Some(st)
+ case Some(ArrayType(st: StructType, _)) => Some(st)
+ case _ => None
+ }
+
+ /**
+ * Like `structTypeHint`, but for a leaf-or-mixed element whose previous
inference may have been a
+ * bare primitive: wrap it as a single-`valueTag` struct so the value tag's
type is still hinted.
+ */
+ private def structTypeHintWithValueTag(dataTypeHintOpt: Option[DataType]):
Option[StructType] =
+ dataTypeHintOpt match {
+ case Some(st: StructType) => Some(st)
+ case Some(ArrayType(st: StructType, _)) => Some(st)
+ case Some(primitive) => Some(new StructType().add(options.valueTag,
primitive))
+ case None => None
+ }
+
+ /**
+ * The `tryParse*` methods below form a cascade: each attempts to parse the
value as its own
+ * type and, on failure, delegates to the next-wider type in the inference
lattice
+ * (Long -> Decimal -> Double -> Time/Date -> TimestampNTZ -> Timestamp ->
Boolean -> String).
+ * Entering the cascade at the parser matching `typeSoFar` (see `inferFrom`)
therefore only ever
+ * widens the type, never narrows it. This mirrors `CSVInferSchema`'s
`tryParse*` chain.
+ */
+ private def tryParseLong(field: String): DataType = {
+ val signSafeValue = if (field.startsWith("+") || field.startsWith("-")) {
+ field.substring(1)
+ } else {
+ field
+ }
+ // A little shortcut to avoid trying many formatters in the common case
that
+ // the input isn't a number. All built-in formats will start with a digit.
+ if (signSafeValue.isEmpty || !Character.isDigit(signSafeValue.head)) {
+ return tryParseDecimal(field)
+ }
+ if ((allCatch opt signSafeValue.toLong).isDefined) {
+ LongType
+ } else {
+ tryParseDecimal(field)
}
}
- private def tryParseDecimal(value: String): Option[DataType] = {
- val signSafeValue = if (value.startsWith("+") || value.startsWith("-")) {
- value.substring(1)
+ private def tryParseDecimal(field: String): DataType = {
+ val signSafeValue = if (field.startsWith("+") || field.startsWith("-")) {
+ field.substring(1)
} else {
- value
+ field
}
// A little shortcut to avoid trying many formatters in the common case
that
// the input isn't a decimal. All built-in formats will start with a digit
or period.
if (signSafeValue.isEmpty ||
!(Character.isDigit(signSafeValue.head) || signSafeValue.head == '.')) {
- return None
+ return tryParseDouble(field)
}
// Rule out strings ending in D or F, as they will parse as double but
should be disallowed
if (signSafeValue.last match {
case 'd' | 'D' | 'f' | 'F' => true
case _ => false
}) {
- return None
+ return tryParseDouble(field)
}
-
- allCatch opt {
- val bigDecimal = decimalParser(value)
- DecimalType(Math.max(bigDecimal.precision, bigDecimal.scale),
bigDecimal.scale)
+ val decimalTry = if (options.prefersDecimal) {
+ allCatch opt {
+ val bigDecimal = decimalParser(field)
+ DecimalType(Math.max(bigDecimal.precision, bigDecimal.scale),
bigDecimal.scale)
+ }
+ } else {
+ None
}
+ decimalTry.getOrElse(tryParseDouble(field))
}
- private def isDouble(value: String): Boolean = {
- val signSafeValue = if (value.startsWith("+") || value.startsWith("-")) {
- value.substring(1)
+ private def tryParseDouble(field: String): DataType = {
+ val signSafeValue = if (field.startsWith("+") || field.startsWith("-")) {
+ field.substring(1)
} else {
- value
+ field
}
// A little shortcut to avoid trying many formatters in the common case
that
// the input isn't a double. All built-in formats will start with a digit
or period.
- if (signSafeValue.isEmpty ||
- !(Character.isDigit(signSafeValue.head) || signSafeValue.head == '.')) {
- return false
- }
- // Rule out strings ending in D or F, as they will parse as double but
should be disallowed
- if (signSafeValue.last match {
- case 'd' | 'D' | 'f' | 'F' => true
- case _ => false
- }) {
- return false
+ val isDouble =
+ if (signSafeValue.isEmpty ||
+ !(Character.isDigit(signSafeValue.head) || signSafeValue.head == '.'))
{
+ false
+ } else if (signSafeValue.last match {
+ // Rule out strings ending in D or F,
+ // as they will parse as double but should be disallowed
+ case 'd' | 'D' | 'f' | 'F' => true
+ case _ => false
+ }) {
+ false
+ } else {
+ (allCatch opt signSafeValue.toDouble).isDefined
+ }
+ if (isDouble) {
+ DoubleType
+ } else if (isTimeTypeEnabled && isTime(field)) {
+ // TIME is tried ahead of date/timestamp, matching the ordering of the
previous cascade.
+ TimeType(TimeType.DEFAULT_PRECISION)
+ } else {
+ tryParseDate(field)
}
- (allCatch opt signSafeValue.toDouble).isDefined
}
- private def isLong(value: String): Boolean = {
- val signSafeValue = if (value.startsWith("+") || value.startsWith("-")) {
- value.substring(1)
+ private def tryParseTime(field: String): DataType = {
+ if (isTimeTypeEnabled && isTime(field)) {
+ TimeType(TimeType.DEFAULT_PRECISION)
} else {
- value
+ tryParseDate(field)
}
- // A little shortcut to avoid trying many formatters in the common case
that
- // the input isn't a number. All built-in formats will start with a digit.
- if (signSafeValue.isEmpty || !Character.isDigit(signSafeValue.head)) {
- return false
- }
- (allCatch opt signSafeValue.toLong).isDefined
}
- private def isTimestamp(value: String): Boolean = {
- try {
- timestampFormatter.parseOptional(value).isDefined
- } catch {
- case _: IllegalArgumentException => false
+ private def tryParseDate(field: String): DataType = {
+ if ((allCatch opt dateFormatter.parse(field)).isDefined) {
+ DateType
+ } else {
+ tryParseTimestampNTZ(field)
}
}
- private def tryParseTimestampNTZ(field: String): Option[DataType] = {
+ private def tryParseTimestampNTZ(field: String): DataType = {
// We can only parse the value as TimestampNTZType if it does not have
zone-offset or
// time-zone component and can be parsed with the timestamp formatter.
// Otherwise, it is likely to be a timestamp with timezone.
@@ -548,18 +718,35 @@ class XmlInferSchema(private val options: XmlOptions,
private val caseSensitive:
timestampNTZFormatter.parseWithoutTimeZoneNanosOptional(field, 9,
false)
.exists(_.nanosWithinMicro != 0)
if (hasSubMicro) {
- return Some(TimestampNTZNanosType(9))
+ return TimestampNTZNanosType(9)
}
- return Some(timestampType)
+ return timestampType
}
} catch {
case _: Exception =>
}
- None
+ tryParseTimestamp(field)
+ }
+
+ private def tryParseTimestamp(field: String): DataType = {
+ val isTimestamp =
+ try {
+ timestampFormatter.parseOptional(field).isDefined
+ } catch {
+ case _: IllegalArgumentException => false
+ }
+ if (isTimestamp) {
+ TimestampType
+ } else {
+ tryParseBoolean(field)
+ }
}
- private def isDate(value: String): Boolean = {
- (allCatch opt dateFormatter.parse(value)).isDefined
+ private def tryParseBoolean(field: String): DataType = {
+ field.toLowerCase(Locale.ROOT) match {
+ case "true" | "false" => BooleanType
+ case _ => StringType
+ }
}
private def isTime(value: String): Boolean = {
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index 8be23cdaef00..1ca873d7da1a 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -7562,6 +7562,21 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val XML_SCHEMA_INFERENCE_INCREMENTAL_TYPECASTING =
+ buildConf("spark.sql.xml.schemaInference.incrementalTypeCasting.enabled")
+ .internal()
+ .doc(
+ "When true (default), XML schema inference refines the type inferred
for each field " +
+ "incrementally: each value is parsed starting from the type inferred
for that field so " +
+ "far, rather than probing every candidate type from scratch. This
avoids redundant parse " +
+ "attempts once a field has widened to a broader type. This matches the
incremental " +
+ "inference already used by the CSV datasource. Set to false to restore
the legacy " +
+ "behavior of inferring each value independently and merging
afterwards.")
+ .version("4.3.0")
+ .withBindingPolicy(ConfigBindingPolicy.SESSION)
+ .booleanConf
+ .createWithDefault(true)
+
val ASSUME_ANSI_FALSE_IF_NOT_PERSISTED =
buildConf("spark.sql.assumeAnsiFalseIfNotPersisted.enabled")
.internal()
@@ -8949,6 +8964,9 @@ class SQLConf extends Serializable with Logging with
SqlApiConf {
def xmlVariantRespectInferSchema: Boolean =
getConf(SQLConf.XML_VARIANT_RESPECT_INFER_SCHEMA)
+ def xmlSchemaInferenceIncrementalTypeCasting: Boolean =
+ getConf(SQLConf.XML_SCHEMA_INFERENCE_INCREMENTAL_TYPECASTING)
+
def coerceMergeNestedTypes: Boolean =
getConf(SQLConf.MERGE_INTO_NESTED_TYPE_COERCION_ENABLED)
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/xml/XmlInferSchemaTypeCastingSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/xml/XmlInferSchemaTypeCastingSuite.scala
new file mode 100644
index 000000000000..c78171d61565
--- /dev/null
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/xml/XmlInferSchemaTypeCastingSuite.scala
@@ -0,0 +1,165 @@
+/*
+ * 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 org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.catalyst.plans.SQLHelper
+import org.apache.spark.sql.types._
+
+/**
+ * Unit tests for [[XmlInferSchema]]'s incremental primitive-type inference.
`inferFrom(value,
+ * typeSoFar)` refines the type inferred for a field so far, entering the
`tryParse*` cascade at
+ * the parser matching `typeSoFar` and reconciling the result with
`compatibleType`. These tests
+ * check that refining from a non-null `typeSoFar` yields the same widened
type that inferring the
+ * value from scratch and then merging would produce -- i.e. the incremental
path is consistent
+ * with the batch path. The structure mirrors `CSVInferSchemaSuite`.
+ */
+class XmlInferSchemaTypeCastingSuite extends SparkFunSuite with SQLHelper {
+
+ private def newInferSchema(options: Map[String, String]): XmlInferSchema =
+ new XmlInferSchema(new XmlOptions(options, "UTC", "_corrupt_record"),
caseSensitive = false)
+
+ test("String field types are inferred correctly from null types") {
+ val inferSchema = newInferSchema(Map("timestampFormat" -> "yyyy-MM-dd
HH:mm:ss"))
+
+ assert(inferSchema.inferFrom("", NullType) == NullType)
+ assert(inferSchema.inferFrom(null, NullType) == NullType)
+ assert(inferSchema.inferFrom("100000000000", NullType) == LongType)
+ // XML infers integral values as LongType (there is no IntegerType
narrowing).
+ assert(inferSchema.inferFrom("60", NullType) == LongType)
+ assert(inferSchema.inferFrom("3.5", NullType) == DoubleType)
+ assert(inferSchema.inferFrom("test", NullType) == StringType)
+ assert(inferSchema.inferFrom("2015-08-20 15:57:00", NullType) ==
TimestampType)
+ assert(inferSchema.inferFrom("True", NullType) == BooleanType)
+ assert(inferSchema.inferFrom("FAlSE", NullType) == BooleanType)
+
+ // A value beyond Long range is not preferred as decimal by default ->
Double.
+ val textValueOne = Long.MaxValue.toString + "0"
+ assert(inferSchema.inferFrom(textValueOne, NullType) == DoubleType)
+ }
+
+ test("String field types are inferred correctly from other types") {
+ val inferSchema = newInferSchema(Map("timestampFormat" -> "yyyy-MM-dd
HH:mm:ss"))
+
+ assert(inferSchema.inferFrom("1.0", LongType) == DoubleType)
+ assert(inferSchema.inferFrom("test", LongType) == StringType)
+ assert(inferSchema.inferFrom(null, DoubleType) == DoubleType)
+ assert(inferSchema.inferFrom("test", DoubleType) == StringType)
+ // A Long-so-far field seeing a timestamp string widens to String
(incompatible).
+ assert(inferSchema.inferFrom("2015-08-20 14:57:00", LongType) ==
StringType)
+ assert(inferSchema.inferFrom("2015-08-20 15:57:00", DoubleType) ==
StringType)
+ assert(inferSchema.inferFrom("True", LongType) == StringType)
+ assert(inferSchema.inferFrom("FALSE", TimestampType) == StringType)
+
+ val textValueOne = Long.MaxValue.toString + "0"
+ assert(inferSchema.inferFrom(textValueOne, LongType) == DoubleType)
+ }
+
+ test("Timestamp field types are inferred correctly via custom data format") {
+ var inferSchema = newInferSchema(Map("timestampFormat" -> "yyyy-mm"))
+ assert(inferSchema.inferFrom("2015-08", TimestampType) == TimestampType)
+
+ inferSchema = newInferSchema(Map("timestampFormat" -> "yyyy"))
+ assert(inferSchema.inferFrom("2015", TimestampType) == TimestampType)
+
+ val combined = Map(
+ "timestampFormat" -> "yyyy-MM-dd'T'HH:mm:ss",
+ "timestampNTZFormat" -> "yyyy-MM-dd HH:mm:ss")
+ inferSchema = newInferSchema(combined)
+ // Merging TimestampNTZ (so far) with a value that parses as Timestamp
yields Timestamp.
+ assert(inferSchema.inferFrom("2016-03-11T20:00:00", TimestampNTZType) ==
TimestampType)
+ }
+
+ test("Timestamp field types are inferred correctly from other types") {
+ val inferSchema = newInferSchema(Map.empty[String, String])
+
+ assert(inferSchema.inferFrom("2015-08-20 14", LongType) == StringType)
+ assert(inferSchema.inferFrom("2015-08-20 14:10", DoubleType) == StringType)
+ assert(inferSchema.inferFrom("2015-08 14:49:00", LongType) == StringType)
+ }
+
+ test("Boolean field types are inferred correctly from other types") {
+ val inferSchema = newInferSchema(Map.empty[String, String])
+
+ assert(inferSchema.inferFrom("Fale", LongType) == StringType)
+ assert(inferSchema.inferFrom("TRUEe", DoubleType) == StringType)
+ }
+
+ test("Empty and null values keep the type inferred so far") {
+ val inferSchema = newInferSchema(Map.empty[String, String])
+
+ // A genuinely empty/null value carries no type information, so
`typeSoFar` is preserved.
+ // (Unlike CSV, XML inference does not treat the `nullValue` option as
null here; a value
+ // equal to `nullValue` is inferred by its content, matching the
pre-existing behavior.)
+ assert(inferSchema.inferFrom("", NullType) == NullType)
+ assert(inferSchema.inferFrom("", LongType) == LongType)
+ assert(inferSchema.inferFrom(null, DoubleType) == DoubleType)
+ assert(inferSchema.inferFrom(null, TimestampType) == TimestampType)
+ }
+
+ test("Refining from a wider type never narrows and is consistent with fresh
inference") {
+ val inferSchema = newInferSchema(Map.empty[String, String])
+ // Once a field is Double, an integral value keeps it Double (does not
narrow to Long).
+ assert(inferSchema.inferFrom("2", DoubleType) == DoubleType)
+ // Refining Double with a Double value stays Double.
+ assert(inferSchema.inferFrom("1.5", DoubleType) == DoubleType)
+ // An incompatible value widens all the way to String.
+ assert(inferSchema.inferFrom("abc", DoubleType) == StringType)
+ }
+
+ test("Refining a numeric type-so-far matches fresh inference (SPARK-57802)")
{
+ // Refining a numeric `typeSoFar` re-enters the cascade at `tryParseLong`,
so an integer value
+ // infers as `Long` (not a narrow `Decimal`) exactly as from-scratch
inference would, and the
+ // merge with `typeSoFar` yields the same type as the legacy path.
Otherwise, under
+ // prefersDecimal, a `Decimal`-so-far field seeing "5" would infer
`Decimal(1,0)` and merge to
+ // a different (order-dependent) precision than the legacy
`Long`-then-merge result.
+ val prefersDecimal = newInferSchema(Map("prefersDecimal" -> "true"))
+ // "5" fresh -> Long; compatibleType(Decimal(5,2), Long) -> Decimal(22,2).
+ assert(prefersDecimal.inferFrom("5", DecimalType(5, 2)) == DecimalType(22,
2))
+ assert(prefersDecimal.inferFrom("5", DoubleType) == DoubleType)
+ // A fractional value still refines through the decimal/double parsers as
usual: "3.5" infers
+ // as Decimal(2,1) and merges with Decimal(5,2) to Decimal(5,2).
+ assert(prefersDecimal.inferFrom("3.5", DecimalType(5, 2)) ==
DecimalType(5, 2))
+ }
+
+ test("Refining a temporal type-so-far matches fresh inference
(SPARK-57802)") {
+ // Refining a temporal `typeSoFar` re-enters the cascade at the top of the
temporal sub-cascade
+ // (`tryParseTime`, flowing Time -> Date -> TimestampNTZ -> Timestamp), so
a date-only value
+ // infers as `Date` exactly as from-scratch inference would, and the merge
with `typeSoFar`
+ // widens through `findWiderDateTimeType` to the same type as the legacy
path. Otherwise, with a
+ // time-requiring `timestampFormat`, a `Timestamp`-so-far field seeing a
date-only value would
+ // fail `tryParseTimestamp`, fall through to `String`, and merge to
`String` -- an
+ // order-dependent divergence from the legacy `Date`-then-merge result.
+ val inferSchema = newInferSchema(Map("timestampFormat" ->
"yyyy-MM-dd'T'HH:mm:ss"))
+ // "2024-01-15" fresh -> Date; compatibleType(Timestamp, Date) ->
Timestamp (DATE adopts the
+ // LTZ family of the other side).
+ assert(inferSchema.inferFrom("2024-01-15", TimestampType) == TimestampType)
+ // Same for a TimestampNTZ-so-far field: DATE adopts the NTZ family, so
the merge stays NTZ.
+ assert(inferSchema.inferFrom("2024-01-15", TimestampNTZType) ==
TimestampNTZType)
+ // A Date-so-far field seeing a timestamp value widens to Timestamp.
+ assert(inferSchema.inferFrom("2024-01-15T10:00:00", DateType) ==
TimestampType)
+ }
+
+ test("date is inferred regardless of preferDate") {
+ Seq("true", "false").foreach { preferDate =>
+ val inferSchema = newInferSchema(Map("preferDate" -> preferDate))
+ assert(inferSchema.inferFrom("2024-01-15", NullType) == DateType,
+ s"expected DateType with preferDate=$preferDate")
+ }
+ }
+}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/XmlInferSchemaSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/XmlInferSchemaSuite.scala
index bd675bd62510..78755d531cd8 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/XmlInferSchemaSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/XmlInferSchemaSuite.scala
@@ -700,6 +700,90 @@ class XmlInferSchemaSuite
val dfTimeDate = readData(xmlTimeDate)
assert(dfTimeDate.schema.fields.head.dataType === StringType)
}
+
+ test("incremental inference widens the type across rows monotonically") {
+ // Long -> Double: once a value forces Double, a later Long value must not
narrow it back.
+ val longThenDouble = Seq(
+ """<ROW><v>1</v></ROW>""",
+ """<ROW><v>1.5</v></ROW>""",
+ """<ROW><v>2</v></ROW>""")
+ assert(readData(longThenDouble).schema.fields.head.dataType === DoubleType)
+
+ // Long -> String: an incompatible later value widens all the way to the
top type.
+ val longThenString = Seq(
+ """<ROW><v>1</v></ROW>""",
+ """<ROW><v>abc</v></ROW>""")
+ assert(readData(longThenString).schema.fields.head.dataType === StringType)
+
+ // The result is independent of row order (the merge is commutative).
+ val doubleThenLong = Seq(
+ """<ROW><v>1.5</v></ROW>""",
+ """<ROW><v>2</v></ROW>""")
+ assert(readData(doubleThenLong).schema.fields.head.dataType === DoubleType)
+ }
+
+ test("date is inferred regardless of preferDate") {
+ val xmlDate = Seq("""<ROW><d>2024-01-15</d></ROW>""")
+ // preferDate governs which date formatter is used, not whether date
inference is attempted.
+ Seq("true", "false").foreach { preferDate =>
+ val df = readData(xmlDate, Map("preferDate" -> preferDate))
+ assert(df.schema.fields.head.dataType === DateType,
+ s"expected DateType with preferDate=$preferDate")
+ }
+ }
+
+ test("incremental type casting yields the same schema as the legacy batch
path") {
+ // Incremental inference (the default) must produce the same inferred
schema as the legacy
+ // per-record path across a range of shapes: mixed numerics, nested
structs, repeated elements
+ // (arrays), attributes, value tags, cross-row widening to StringType, and
-- the case that
+ // makes incremental inference actually differ if the type-so-far is
threaded naively --
+ // prefersDecimal fields mixing a decimal and an integer, in both row
orders.
+ //
+ // All records are read as a SINGLE partition so that the incremental path
actually threads
+ // one record's type into the next within the partition; with the default
parallelism each
+ // record lands in its own partition and both paths trivially agree via
the final merge.
+ def readOnePartition(xml: Seq[String], options: Map[String, String]):
StructType = {
+ val ds = spark.createDataset(spark.sparkContext.parallelize(xml,
1))(Encoders.STRING)
+ spark.read.options(Map("rowTag" -> "ROW") ++ options).xml(ds).schema
+ }
+
+ val cases: Seq[(Seq[String], Map[String, String])] = Seq(
+ Seq("""<ROW><a>1</a><b>1.5</b></ROW>""",
"""<ROW><a>2.5</a><b>3</b></ROW>""") -> Map.empty,
+ Seq("""<ROW><n><x>1</x><y>t</y></n></ROW>""",
+ """<ROW><n><x>2.0</x><y>u</y></n></ROW>""") -> Map.empty,
+ Seq("""<ROW><arr>1</arr><arr>2</arr><arr>3.5</arr></ROW>""") ->
Map.empty,
+ Seq("""<ROW><e k="1">text</e></ROW>""", """<ROW><e
k="2.5">other</e></ROW>""") -> Map.empty,
+ Seq("""<ROW><v>2024-01-15</v></ROW>""",
+ """<ROW><v>2024-01-15T10:00:00</v></ROW>""") -> Map.empty,
+ Seq("""<ROW><v>1</v></ROW>""", """<ROW><v>not-a-number</v></ROW>""") ->
Map.empty,
+ // prefersDecimal: a decimal then an integer, and the reverse order.
+ Seq("""<ROW><a>123.45</a></ROW>""", """<ROW><a>5</a></ROW>""") ->
+ Map("prefersDecimal" -> "true"),
+ Seq("""<ROW><a>5</a></ROW>""", """<ROW><a>123.45</a></ROW>""") ->
+ Map("prefersDecimal" -> "true"),
+ // Temporal family with a time-requiring timestampFormat, in both row
orders. A date-only
+ // value must widen with a timestamp identically whether it is seen
first (so the field is
+ // Date when the timestamp arrives) or last (so the field is Timestamp
when the date arrives):
+ // entering the temporal cascade at its top keeps the date-only value
inferring as Date rather
+ // than falling through to String, so `findWiderDateTimeType` widens
Date + Timestamp to
+ // Timestamp on both the incremental and the legacy path.
+ Seq("""<ROW><v>2024-01-15T10:00:00</v></ROW>""",
"""<ROW><v>2024-01-15</v></ROW>""") ->
+ Map("timestampFormat" -> "yyyy-MM-dd'T'HH:mm:ss"),
+ Seq("""<ROW><v>2024-01-15</v></ROW>""",
"""<ROW><v>2024-01-15T10:00:00</v></ROW>""") ->
+ Map("timestampFormat" -> "yyyy-MM-dd'T'HH:mm:ss"))
+
+ cases.foreach { case (xml, options) =>
+ val incremental = withSQLConf(
+ SQLConf.XML_SCHEMA_INFERENCE_INCREMENTAL_TYPECASTING.key -> "true") {
+ readOnePartition(xml, options)
+ }
+ val batch = withSQLConf(
+ SQLConf.XML_SCHEMA_INFERENCE_INCREMENTAL_TYPECASTING.key -> "false") {
+ readOnePartition(xml, options)
+ }
+ assert(incremental === batch, s"incremental and batch schemas differ
for: $xml")
+ }
+ }
}
trait XmlSchemaInferenceCaseSensitivityTests extends QueryTest {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
