draft

Author: xiaonanyang-db

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXMLRecordReader.scala

@@ -38,14 +38,14 @@ case class StaxXMLRecordReader(inputStream: () => InputStream, options: XmlOptio
     with Logging {
   // Reader for the XML record parsing.
   private val in1 = inputStream()
-  private val primaryEventReader = StaxXmlParserUtils.filteredReader(in1, options)
+  private lazy val primaryEventReader = StaxXmlParserUtils.filteredReader(in1, options)
 
   private val xsdSchemaValidator = Option(options.rowValidationXSDPath)
     .map(path => ValidatorUtil.getSchema(path).newValidator())
   // Reader for the XSD validation, if an XSD schema is provided.
   private val in2 = xsdSchemaValidator.map(_ => inputStream())
   // An XMLStreamReader used by StAXSource for XSD validation.
-  private val xsdValidationStreamReader =
+  private lazy val xsdValidationStreamReader =
     in2.map(in => StaxXmlParserUtils.filteredStreamReader(in, options))
 
   final var hasMoreRecord: Boolean = true
@@ -108,11 +108,16 @@ case class StaxXMLRecordReader(inputStream: () => InputStream, options: XmlOptio
   }
 
   override def close(): Unit = {
-    primaryEventReader.close()
-    xsdValidationStreamReader.foreach(_.close())
     in1.close()
     in2.foreach(_.close())
     hasMoreRecord = false
+    try {
+      primaryEventReader.close()
+      xsdValidationStreamReader.foreach(_.close())
+    } catch {
+      case NonFatal(e) =>
+        logWarning("Error closing XML stream", e)
+    }
   }
 
   override def nextEvent(): XMLEvent = primaryEventReader.nextEvent()

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/XmlInferSchema.scala

@@ -278,6 +278,10 @@ class XmlInferSchema(options: XmlOptions, caseSensitive: Boolean)
             logWarning("Skipped the rest of the content in the corrupted file", e)
             parser.close()
             Some(StructType(Nil))
+          case _: IOException | _: RuntimeException | _: InternalError
+              if !options.ignoreCorruptFiles =>
+            parser.close()
+            throw e
           case _ =>
             logWarning("Failed to infer schema from XML record", e)
             handleXmlErrorsByParseMode(

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

@@ -3021,6 +3021,13 @@ class XmlSuite
            .xml(inputFile.toURI.toString)
            .collect()
         assert(result.isEmpty)
+
+        val result2 = spark.read
+          .option("rowTag", "ROW")
+          .option("multiLine", true)
+          .xml(inputFile.toURI.toString)
+          .collect()
+        assert(result2.isEmpty)
       }
     })
     withTempPath { dir =>