cloud-fan commented on code in PR #56919:
URL: https://github.com/apache/spark/pull/56919#discussion_r3504645804
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/XmlInferSchema.scala:
##########
@@ -312,41 +338,67 @@ 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 {
+ case NullType | LongType => tryParseLong(value)
+ case _: DecimalType => tryParseDecimal(value)
Review Comment:
Good catch, and confirmed empirically — you are right. My equivalence test
missed it only because the two records landed in separate partitions under
default parallelism (each inferred fresh, then merged identically to batch);
forcing a single partition reproduces `Decimal(5,2)` (incremental) vs
`Decimal(22,2)` (legacy), order-dependent as you noted.
Fixed by routing `Long`/`Double`/`Decimal` `typeSoFar` back through
`tryParseLong` (the top of the numeric cascade), so an integer infers as `Long`
exactly as from-scratch inference does and merges through `compatibleType`
identically to the legacy path. Only the "skip the Long check"
micro-optimization is given up on the numeric branch; the non-numeric branches
keep entering mid-cascade (cross-family values merge to String either way).
Pushed in c5711dfbf83.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/XmlInferSchemaSuite.scala:
##########
@@ -700,6 +700,62 @@ 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, and cross-row widening to StringType.
+ val cases = Seq(
+ Seq("""<ROW><a>1</a><b>1.5</b></ROW>""",
"""<ROW><a>2.5</a><b>3</b></ROW>"""),
+ Seq("""<ROW><n><x>1</x><y>t</y></n></ROW>""",
"""<ROW><n><x>2.0</x><y>u</y></n></ROW>"""),
+ Seq("""<ROW><arr>1</arr><arr>2</arr><arr>3.5</arr></ROW>"""),
+ Seq("""<ROW><e k="1">text</e></ROW>""", """<ROW><e
k="2.5">other</e></ROW>"""),
+ Seq("""<ROW><v>2024-01-15</v></ROW>""",
"""<ROW><v>2024-01-15T10:00:00</v></ROW>"""),
+ Seq("""<ROW><v>1</v></ROW>""", """<ROW><v>not-a-number</v></ROW>"""))
+
+ cases.foreach { xml =>
+ val incremental = withSQLConf(
+ SQLConf.XML_SCHEMA_INFERENCE_INCREMENTAL_TYPECASTING.key -> "true") {
+ readData(xml).schema
+ }
+ val batch = withSQLConf(
+ SQLConf.XML_SCHEMA_INFERENCE_INCREMENTAL_TYPECASTING.key -> "false") {
+ readData(xml).schema
+ }
+ assert(incremental === batch, s"incremental and batch schemas differ
for: $xml")
Review Comment:
Done. The equivalence test now reads all records as a single partition (so
the incremental path is actually exercised within the partition rather than
trivially agreeing via the final cross-partition merge) and includes your exact
`prefersDecimal=true` decimal-then-integer case in both row orders. It failed
before the fix and passes after. Also added a direct `inferFrom("5",
Decimal(5,2)) == Decimal(22,2)` unit assertion in
`XmlInferSchemaTypeCastingSuite`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
