You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ho...@apache.org on 2026/09/24 22:47:17 UTC
(spark) branch branch-4.2 updated: [SPARK-59405][SQL] XSD Validation improvements
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new a2414d99c90c [SPARK-59405][SQL] XSD Validation improvements
a2414d99c90c is described below
commit a2414d99c90c9ec98ccf65743a6f51d1ad305dbc
Author: Holden Karau <ho...@pigscanfly.ca>
AuthorDate: Thu Sep 24 15:39:04 2026 -0700
[SPARK-59405][SQL] XSD Validation improvements
### What changes were proposed in this pull request?
Change validation when using rowValidationXSDPath to match the regular validation flow path and re-use
### Why are the changes needed?
Current XSD validation is too slow when using rowValidationXSDPath.
### Does this PR introduce _any_ user-facing change?
Maybe slight performance improvement
### How was this patch tested?
Existing unit tests + new unit test.
### Was this patch authored or co-authored using generative AI tooling?
Claude and Cursor
Closes #58698 from holdenk/bug10-xsd-validation-improvements.
Lead-authored-by: Holden Karau <ho...@pigscanfly.ca>
Co-authored-by: Holden Karau <ho...@snowflake.com>
Signed-off-by: Holden Karau <ho...@snowflake.com>
(cherry picked from commit f07d18cc7fafcb0044f6b010e88518da46de3335)
Signed-off-by: Holden Karau <ho...@snowflake.com>
---
.../sql/catalyst/xml/StaxXMLRecordReader.scala | 6 ++-
.../spark/sql/catalyst/xml/StaxXmlParser.scala | 5 ++-
.../spark/sql/catalyst/xml/ValidatorUtil.scala | 43 +++++++++++++++++++++-
.../spark/sql/catalyst/xml/XmlInferSchema.scala | 5 ++-
.../sql/execution/datasources/xml/XmlSuite.scala | 25 ++++++++++++-
5 files changed, 77 insertions(+), 7 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXMLRecordReader.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXMLRecordReader.scala
index 8793eebc9720..d087fdd8f799 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXMLRecordReader.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXMLRecordReader.scala
@@ -41,7 +41,7 @@ case class StaxXMLRecordReader(inputStream: () => InputStream, options: XmlOptio
private lazy val primaryEventReader = StaxXmlParserUtils.filteredReader(in1, options)
private val xsdSchemaValidator = Option(options.rowValidationXSDPath)
- .map(path => ValidatorUtil.getSchema(path).newValidator())
+ .map(path => ValidatorUtil.newValidator(ValidatorUtil.getSchema(path)))
// Reader for the XSD validation, if an XSD schema is provided.
private lazy val in2 = xsdSchemaValidator.map(_ => inputStream())
// An XMLStreamReader used by StAXSource for XSD validation.
@@ -103,7 +103,9 @@ case class StaxXMLRecordReader(inputStream: () => InputStream, options: XmlOptio
while (!rowTagStarted && streamReader.hasNext) {
streamReader.next()
}
- xsdSchemaValidator.get.reset()
+ // Reuse the Validator across records: Validator.reset() drops the secure-processing
+ // configuration applied at construction, so re-apply it on every record.
+ ValidatorUtil.reset(xsdSchemaValidator.get)
xsdSchemaValidator.get.validate(new StAXSource(streamReader))
}
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
index 593e6a3a97e1..fc7202097ff9 100644
--- 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
@@ -146,7 +146,7 @@ class StaxXmlParser(
lazy val xmlRecord = UTF8String.fromString(xml)
try {
xsdSchema.foreach { schema =>
- schema.newValidator().validate(new StreamSource(new StringReader(xml)))
+ ValidatorUtil.newValidator(schema).validate(new StreamSource(new StringReader(xml)))
}
options.singleVariantColumn match {
case Some(_) =>
@@ -163,6 +163,9 @@ class StaxXmlParser(
}
} catch {
case e: SparkUpgradeException => throw e
+ // ValidatorUtil.newValidator throws this when the JAXP implementation cannot
+ // disable external access; that is an environment error, not a bad record.
+ case e: UnsupportedOperationException => throw e
case e@(_: RuntimeException | _: XMLStreamException | _: MalformedInputException
| _: SAXException) =>
// XML parser currently doesn't support partial results for corrupted records.
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/ValidatorUtil.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/ValidatorUtil.scala
index dccbb40fdd98..6f0f7948d513 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/ValidatorUtil.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/ValidatorUtil.scala
@@ -19,10 +19,11 @@ package org.apache.spark.sql.catalyst.xml
import java.io.{File, FileInputStream, InputStream}
import javax.xml.XMLConstants
import javax.xml.transform.stream.StreamSource
-import javax.xml.validation.{Schema, SchemaFactory}
+import javax.xml.validation.{Schema, SchemaFactory, Validator}
import com.google.common.cache.{CacheBuilder, CacheLoader}
import org.apache.hadoop.fs.Path
+import org.xml.sax.{SAXNotRecognizedException, SAXNotSupportedException}
import org.apache.spark.SparkFiles
import org.apache.spark.deploy.SparkHadoopUtil
@@ -75,4 +76,44 @@ object ValidatorUtil extends Logging {
* @return Schema for the file at that path
*/
def getSchema(path: String): Schema = cache.get(path)
+
+ /**
+ * Creates a [[Validator]] for the given schema that does not resolve external DTDs,
+ * entities, or schema references found in the document being validated. The record
+ * parser does not process DTDs, so validation applies the same restrictions to keep
+ * the two consistent. All validation of record data must use a Validator returned by
+ * this method, or one reused through [[reset]]. Validators must not be shared between
+ * threads.
+ */
+ def newValidator(schema: Schema): Validator = {
+ val validator = schema.newValidator()
+ configureForRecordValidation(validator)
+ validator
+ }
+
+ /**
+ * Resets a [[Validator]] created by [[newValidator]] and re-applies the
+ * record-validation configuration. [[Validator.reset]] does not retain configuration
+ * applied after construction, so the secure-processing settings must be re-applied on
+ * every reset. Reusing one Validator across records avoids the per-record allocation
+ * cost of [[newValidator]]. Validators must not be shared between threads.
+ */
+ def reset(validator: Validator): Unit = {
+ validator.reset()
+ configureForRecordValidation(validator)
+ }
+
+ private def configureForRecordValidation(validator: Validator): Unit = {
+ try {
+ validator.setProperty(XMLConstants.ACCESS_EXTERNAL_DTD, "")
+ validator.setProperty(XMLConstants.ACCESS_EXTERNAL_SCHEMA, "")
+ validator.setFeature(XMLConstants.FEATURE_SECURE_PROCESSING, true)
+ } catch {
+ case e @ (_: SAXNotRecognizedException | _: SAXNotSupportedException) =>
+ // Throw rather than validate with a Validator that cannot disable external access.
+ throw new UnsupportedOperationException(
+ "The JAXP Validator implementation in use does not support disabling external " +
+ "DTD/schema access; refusing to validate with it.", e)
+ }
+ }
}
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 a0139e98a266..f657baa44e71 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
@@ -151,7 +151,7 @@ class XmlInferSchema(private val options: XmlOptions, private val caseSensitive:
try {
val xsd = xsdSchema.orElse(Option(options.rowValidationXSDPath).map(ValidatorUtil.getSchema))
xsd.foreach { schema =>
- schema.newValidator().validate(new StreamSource(new StringReader(xml)))
+ ValidatorUtil.newValidator(schema).validate(new StreamSource(new StringReader(xml)))
}
parser = StaxXmlParserUtils.filteredReader(xml)
val rootAttributes = StaxXmlParserUtils.gatherRootAttributes(parser)
@@ -178,6 +178,9 @@ class XmlInferSchema(private val options: XmlOptions, private val caseSensitive:
Some(StructType(Nil))
case e: FileNotFoundException if !options.ignoreMissingFiles => throw e
case e @ (_ : AccessControlException | _ : BlockMissingException) => throw e
+ // ValidatorUtil.newValidator throws this when the JAXP implementation cannot
+ // disable external access; that is an environment error, not a bad record.
+ case e: UnsupportedOperationException => throw e
case e @ (_: IOException | _: RuntimeException) if options.ignoreCorruptFiles =>
logWarning("Skipped the rest of the content in the corrupted file", e)
Some(StructType(Nil))
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 a1aae1481f81..e62ceff1a715 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
@@ -16,7 +16,7 @@
*/
package org.apache.spark.sql.execution.datasources.xml
-import java.io.{EOFException, File, FileOutputStream, StringWriter}
+import java.io.{EOFException, File, FileOutputStream, StringReader, StringWriter}
import java.nio.charset.{StandardCharsets, UnsupportedCharsetException}
import java.nio.file.{Files, Path, Paths}
import java.sql.{Date, Timestamp}
@@ -24,6 +24,7 @@ import java.time.{Instant, LocalDateTime, Year}
import java.util.TimeZone
import java.util.concurrent.ConcurrentHashMap
import javax.xml.stream.{XMLOutputFactory, XMLStreamException}
+import javax.xml.transform.stream.StreamSource
import scala.collection.immutable.ArraySeq
import scala.collection.mutable
@@ -34,6 +35,7 @@ import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.FSDataInputStream
import org.apache.hadoop.io.{LongWritable, Text}
import org.apache.hadoop.io.compress.{CompressionCodecFactory, GzipCodec}
+import org.xml.sax.SAXException
import org.apache.spark.{DebugFilesystem, SparkConf, SparkException}
import org.apache.spark.io.ZStdCompressionCodec
@@ -41,7 +43,7 @@ import org.apache.spark.sql.{AnalysisException, DataFrame, Dataset, Encoders, Ro
import org.apache.spark.sql.catalyst.encoders.AgnosticEncoders.UDTEncoder
import org.apache.spark.sql.catalyst.util._
import org.apache.spark.sql.catalyst.util.TypeUtils.ordinalNumber
-import org.apache.spark.sql.catalyst.xml.{IndentingXMLStreamWriter, XmlOptions}
+import org.apache.spark.sql.catalyst.xml.{IndentingXMLStreamWriter, ValidatorUtil, XmlOptions}
import org.apache.spark.sql.catalyst.xml.XmlOptions._
import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.sql.execution.datasources.CommonFileDataSourceSuite
@@ -1274,6 +1276,25 @@ class XmlSuite
}
}
+ test("XSD validation does not resolve external references in records") {
+ val schema = ValidatorUtil.getSchema(
+ getTestResourcePath(resDir + "basket.xsd").replace("file:/", "/"))
+
+ withTempDir { dir =>
+ val dataFile = new File(dir, "data.txt")
+ Files.write(dataFile.toPath, "9027".getBytes(StandardCharsets.UTF_8))
+ // The validator sees the raw record before the record parser does, so it must
+ // reject DOCTYPE references the same way the parser does.
+ val record =
+ s"""<?xml version="1.0"?>
+ |<!DOCTYPE basket [<!ENTITY ext SYSTEM "${dataFile.toURI}">]>
+ |<basket><entry><key>&ext;</key><value>1</value></entry></basket>""".stripMargin
+ intercept[SAXException] {
+ ValidatorUtil.newValidator(schema).validate(new StreamSource(new StringReader(record)))
+ }
+ }
+ }
+
test("test XSD validation with validation error") {
val basketDF = spark.read
.option("rowTag", "basket")
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org