[SPARK-56486][SQL] Add XSDToSchema.extendDecimalPrecision to prevent silent decimal truncation

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XSDToSchema.scala

@@ -66,6 +66,39 @@ object XSDToSchema extends Logging{
     getStructType(xmlSchema)
   }
 
+  /**
+   * Widens every [[DecimalType]] in the given schema so that decimals inferred from an XSD's
+   * `totalDigits`/`fractionDigits` facets can hold their full integer range without silent
+   * truncation. `read` maps `totalDigits=t, fractionDigits=f` to `DecimalType(t, f)`, which is
+   * correct per the W3C XML Schema spec but leaves only `t - f` integer digits. Because Spark's
+   * decimal scale is fixed while the XSD `fractionDigits` facet is only a maximum, values that
+   * use fewer fractional digits and more integer digits overflow the precision and are silently
+   * read as null. This opt-in utility raises each `DecimalType(p, s)` to
+   * `DecimalType(min(maxPrecision, p + s), s)`, recursing into nested struct fields and array
+   * elements. It does not change `read`'s default inference, so existing callers are unaffected.
+   *
+   * @param schema the schema returned by `read`
+   * @param maxPrecision the maximum precision a widened decimal may reach; must not exceed
+   *                     [[DecimalType.MAX_PRECISION]]
+   * @return a copy of the schema with decimal precisions widened
+   */
+  def extendDecimalPrecision(
+      schema: StructType,
+      maxPrecision: Int = DecimalType.MAX_PRECISION): StructType = {
+    StructType(schema.map { field =>
+      field.copy(dataType = extendDecimalType(field.dataType, maxPrecision))
+    })
+  }
+
+  private def extendDecimalType(dataType: DataType, maxPrecision: Int): DataType = dataType match {
+    case DecimalType.Fixed(precision, scale) =>
+      DecimalType(math.min(maxPrecision, precision + scale), scale)
+    case struct: StructType =>
+      extendDecimalPrecision(struct, maxPrecision)
+    case ArrayType(elementType, containsNull) =>
+      ArrayType(extendDecimalType(elementType, maxPrecision), containsNull)
+    case other => other
+  }
 
   private def getStructField(xmlSchema: XmlSchema, schemaType: XmlSchemaType): StructField = {
     schemaType match {

sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/xml/util/XSDToSchemaSuite.scala

@@ -130,6 +130,86 @@ class XSDToSchemaSuite extends SharedSparkSession {
     assert(parsedSchema === expectedSchema)
   }
 
+  test("SPARK-56486: extendDecimalPrecision widens decimals and caps at max precision") {
+    val parsedSchema = XSDToSchema.read(new Path(testFile(resDir + "decimal-with-restriction.xsd")))
+    val widenedSchema = XSDToSchema.extendDecimalPrecision(parsedSchema)
+    // decimal_type_3 (12, 6) -> (18, 6); the other two are already at precision 38 (the cap)
+    // and so are left unchanged.
+    val expectedSchema = buildSchema(
+      field("decimal_type_3", DecimalType(18, 6), nullable = false),
+      field("decimal_type_1", DecimalType(38, 18), nullable = false),
+      field("decimal_type_2", DecimalType(38, 2), nullable = false)
+    )
+    assert(widenedSchema === expectedSchema)
+  }
+
+  test("SPARK-56486: extendDecimalPrecision recurses into nested struct and array decimals") {
+    val xsdString =
+      """<?xml version="1.0" encoding="UTF-8" ?>
+        |<xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema">
+        |  <xs:element name="root">
+        |    <xs:complexType>
+        |      <xs:sequence>
+        |        <xs:element name="nested">
+        |          <xs:complexType>
+        |            <xs:sequence>
+        |              <xs:element name="amount">
+        |                <xs:simpleType>
+        |                  <xs:restriction base="xs:decimal">
+        |                    <xs:totalDigits value="20"/>
+        |                    <xs:fractionDigits value="5"/>
+        |                  </xs:restriction>
+        |                </xs:simpleType>
+        |              </xs:element>
+        |            </xs:sequence>
+        |          </xs:complexType>
+        |        </xs:element>
+        |        <xs:element name="prices" maxOccurs="unbounded">
+        |          <xs:simpleType>
+        |            <xs:restriction base="xs:decimal">
+        |              <xs:totalDigits value="10"/>
+        |              <xs:fractionDigits value="2"/>
+        |            </xs:restriction>
+        |          </xs:simpleType>
+        |        </xs:element>
+        |      </xs:sequence>
+        |    </xs:complexType>
+        |  </xs:element>
+        |</xs:schema>
+        |""".stripMargin
+    val widenedSchema = XSDToSchema.extendDecimalPrecision(XSDToSchema.read(xsdString))
+    val expectedSchema = StructType(StructField("root",
+      StructType(
+        StructField("nested",
+          StructType(StructField("amount", DecimalType(25, 5), false) :: Nil), false) ::
+        StructField("prices", ArrayType(DecimalType(12, 2)), false) :: Nil),
+      false) :: Nil)
+    assert(widenedSchema === expectedSchema)
+  }
+
+  test("SPARK-56486: extendDecimalPrecision respects an explicit maxPrecision") {
+    val xsdString =
+      """<?xml version="1.0" encoding="UTF-8" ?>
+        |<xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema">
+        |  <xs:element name="amount">
+        |    <xs:simpleType>
+        |      <xs:restriction base="xs:decimal">
+        |        <xs:totalDigits value="20"/>
+        |        <xs:fractionDigits value="5"/>
+        |      </xs:restriction>
+        |    </xs:simpleType>
+        |  </xs:element>
+        |</xs:schema>
+        |""".stripMargin
+    val parsedSchema = XSDToSchema.read(xsdString)
+    // read() leaves only 20 - 5 = 15 integer digits, which is where the silent truncation occurs.
+    assert(parsedSchema === buildSchema(field("amount", DecimalType(20, 5), nullable = false)))
+    val widenedSchema = XSDToSchema.extendDecimalPrecision(parsedSchema, maxPrecision = 22)
+    // p + s = 25, but the explicit cap of 22 wins.
+    val expectedSchema = buildSchema(field("amount", DecimalType(22, 5), nullable = false))
+    assert(widenedSchema === expectedSchema)
+  }
+
   test("Test ref attribute / Issue 617") {
     val parsedSchema = XSDToSchema.read(new Path(testFile(resDir + "ref-attribute.xsd")))
     val expectedSchema = buildSchema(