MaxGekk commented on code in PR #56919:
URL: https://github.com/apache/spark/pull/56919#discussion_r3504452246
##########
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:
This equivalence check is the main guard for the "flag-on == flag-off"
claim, but none of the `cases` set `prefersDecimal` (and `readData` passes no
options), so `prefersDecimal=false` throughout — the one case where the two
paths diverge (see the comment on the `Decimal` dispatch arm) isn't exercised,
and this test passes despite it. Worth adding a `prefersDecimal=true` field
mixing a decimal and an integer, in both row orders, e.g.
`Seq("<ROW><a>123.45</a></ROW>", "<ROW><a>5</a></ROW>")` and the reverse. It
would fail today and would guard whichever fix you take.
##########
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:
This dispatch makes the incremental path diverge from the legacy path under
`prefersDecimal`, which I think breaks the "flag-on == flag-off" invariant. For
a field with values `123.45` then `5` (prefersDecimal on):
- **Incremental**: `5` arrives with `typeSoFar = Decimal(5,2)` →
`tryParseDecimal("5")` → `Decimal(1,0)` → `compatibleType(Decimal(5,2),
Decimal(1,0))` → `Decimal(5,2)`.
- **Legacy** (flag off): `5` is inferred from scratch as `Long`, then
`compatibleType(Decimal(5,2), Long)` hits the `(DecimalType, IntegralType)`
fallback → `DecimalType.forType(Long) = Decimal(20,0)` → `Decimal(22,2)`.
So `Decimal(5,2)` vs `Decimal(22,2)`. It's also order-dependent: `5` seen
first → `typeSoFar = Long` → a later decimal widens to `Decimal(22,2)`, so the
incremental schema depends on row order / partition layout, whereas legacy is
order-independent (`compatibleType` is associative/commutative). That
contradicts "no change to inferred schemas."
Root cause: entering `tryParseDecimal` for a `Decimal` type-so-far infers an
integer as a narrow `Decimal` instead of the from-scratch `Long` that legacy
widens via `DecimalType.forType`. Simplest fix that preserves behavior: for a
numeric (`Long`/`Decimal`/`Double`) `typeSoFar`, still enter the cascade at
`tryParseLong`, so an integer infers as `Long` and merges through
`compatibleType` exactly as the legacy path does (you give up only the "skip
the Long check" micro-optimization on the numeric branch). Alternatively, if
converging on CSV's tighter behavior is intended, it'd be worth dropping the
behavior-preserving framing and calling out that the default-on flag changes
inferred Decimal schemas.
--
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]
