From 01ce31def35c886a757fa3afc1b45ca5fadd8a62 Mon Sep 17 00:00:00 2001 From: Holden Karau Date: Tue, 1 Sep 2026 02:38:25 +0000 Subject: [PATCH 1/2] [SQL] Disable external reference resolution in XSD row validation Validators used for rowValidationXSDPath are now created through ValidatorUtil.newValidator, which disables external DTD and schema access and enables secure processing on the validator. A JDK-default validator resolves external DTDs and entities found in record data during validate(), before the record parser, which rejects DTDs, sees the record. Records containing such references now fail validation and are handled per parse mode. StaxXMLRecordReader creates a fresh validator per record instead of reusing one through Validator.reset(), which does not retain configuration applied after creation. A validator whose JAXP implementation cannot disable external access now fails the query uniformly instead of marking records corrupt. Co-authored-by: Cursor Co-Authored-By: Holden Karau --- .../catalyst/xml/StaxXMLRecordReader.scala | 10 ++++---- .../sql/catalyst/xml/StaxXmlParser.scala | 5 +++- .../sql/catalyst/xml/ValidatorUtil.scala | 25 ++++++++++++++++++- .../sql/catalyst/xml/XmlInferSchema.scala | 5 +++- .../execution/datasources/xml/XmlSuite.scala | 25 +++++++++++++++++-- 5 files changed, 60 insertions(+), 10 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 8793eebc97202..a42a340539b58 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 @@ -40,10 +40,9 @@ case class StaxXMLRecordReader(inputStream: () => InputStream, options: XmlOptio private lazy val in1 = inputStream() private lazy val primaryEventReader = StaxXmlParserUtils.filteredReader(in1, options) - private val xsdSchemaValidator = Option(options.rowValidationXSDPath) - .map(path => ValidatorUtil.getSchema(path).newValidator()) + private val xsdSchema = Option(options.rowValidationXSDPath).map(ValidatorUtil.getSchema) // Reader for the XSD validation, if an XSD schema is provided. - private lazy val in2 = xsdSchemaValidator.map(_ => inputStream()) + private lazy val in2 = xsdSchema.map(_ => inputStream()) // An XMLStreamReader used by StAXSource for XSD validation. private lazy val xsdValidationStreamReader = in2.map(in => StaxXmlParserUtils.filteredStreamReader(in, options)) @@ -103,8 +102,9 @@ case class StaxXMLRecordReader(inputStream: () => InputStream, options: XmlOptio while (!rowTagStarted && streamReader.hasNext) { streamReader.next() } - xsdSchemaValidator.get.reset() - xsdSchemaValidator.get.validate(new StAXSource(streamReader)) + // Create a fresh Validator per record: Validator.reset() does not retain the + // configuration that ValidatorUtil.newValidator applies. + ValidatorUtil.newValidator(xsdSchema.get).validate(new StAXSource(streamReader)) } override def close(): Unit = { 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 340dc61fb5112..d964564fbde1f 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))) } if (options.singleVariantColumn.isDefined || options.rootVariantType) { // If the singleVariantColumn is specified or the requested output is a root Variant, @@ -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 dccbb40fdd985..f93eaf99ffa15 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,26 @@ 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 go through this method. + */ + def newValidator(schema: Schema): Validator = { + val validator = schema.newValidator() + 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) + } + validator + } } 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 d2a59498549a3..3c1334c4c215b 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 @@ -178,7 +178,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) @@ -205,6 +205,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 8a8ab4de3a8f8..3c2000214b983 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""" + |]> + |&ext;1""".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") From 0d2b9b53264ec39b56b10b2188ba3ebbabb0edff Mon Sep 17 00:00:00 2001 From: Holden Karau Date: Wed, 2 Sep 2026 07:02:16 +0000 Subject: [PATCH 2/2] [SQL] Speed up XSD row validation by reusing the Validator StaxXMLRecordReader now keeps one Validator for the lifetime of the reader and re-applies the record-validation configuration after each Validator.reset(), instead of allocating a fresh Validator per record. Validator.reset() does not retain configuration applied after construction, so the secure-processing settings must be re-applied on every reset; ValidatorUtil.reset factors that out so construction and per-record reset share one code path. Reusing the Validator avoids the per-record allocation cost of Schema.newValidator(). Co-authored-by: Cursor Co-Authored-By: Holden Karau --- .../catalyst/xml/StaxXMLRecordReader.scala | 12 ++++++----- .../sql/catalyst/xml/ValidatorUtil.scala | 21 +++++++++++++++++-- 2 files changed, 26 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 a42a340539b58..d087fdd8f7992 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 @@ -40,9 +40,10 @@ case class StaxXMLRecordReader(inputStream: () => InputStream, options: XmlOptio private lazy val in1 = inputStream() private lazy val primaryEventReader = StaxXmlParserUtils.filteredReader(in1, options) - private val xsdSchema = Option(options.rowValidationXSDPath).map(ValidatorUtil.getSchema) + private val xsdSchemaValidator = Option(options.rowValidationXSDPath) + .map(path => ValidatorUtil.newValidator(ValidatorUtil.getSchema(path))) // Reader for the XSD validation, if an XSD schema is provided. - private lazy val in2 = xsdSchema.map(_ => inputStream()) + private lazy val in2 = xsdSchemaValidator.map(_ => inputStream()) // An XMLStreamReader used by StAXSource for XSD validation. private lazy val xsdValidationStreamReader = in2.map(in => StaxXmlParserUtils.filteredStreamReader(in, options)) @@ -102,9 +103,10 @@ case class StaxXMLRecordReader(inputStream: () => InputStream, options: XmlOptio while (!rowTagStarted && streamReader.hasNext) { streamReader.next() } - // Create a fresh Validator per record: Validator.reset() does not retain the - // configuration that ValidatorUtil.newValidator applies. - ValidatorUtil.newValidator(xsdSchema.get).validate(new StAXSource(streamReader)) + // 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)) } override def close(): Unit = { 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 f93eaf99ffa15..436c81a49df4c 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 @@ -81,10 +81,28 @@ object ValidatorUtil extends Logging { * 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 go through this method. + * the two consistent. All validation of record data must use a Validator returned by + * this method, or one reused through [[reset]]. */ 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]]. + */ + 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, "") @@ -96,6 +114,5 @@ object ValidatorUtil extends Logging { "The JAXP Validator implementation in use does not support disabling external " + "DTD/schema access; refusing to validate with it.", e) } - validator } }