u

Author: xiaonanyang-db

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

@@ -1007,7 +1007,7 @@ class XmlTokenizer(
   }
 }
 
-object StaxXmlParser {
+object StaxXmlParser extends Logging {
   /**
    * Parses a stream that contains CSV strings and turns it into an iterator of tokens.
    */
@@ -1050,14 +1050,16 @@ object StaxXmlParser {
         }
       }
     } catch {
-      case e: Throwable =>
+      case e: FileNotFoundException if options.ignoreMissingFiles =>
+        logWarning("Skipped missing file", e)
+        Iterator.empty
+      case NonFatal(e) =>
         SparkErrorUtils.getRootCause(e) match {
-          case _: IOException | _: RuntimeException | _: InternalError
-              if options.ignoreCorruptFiles =>
-            Iterator.empty
-          case _: FileNotFoundException if options.ignoreMissingFiles =>
+          case _: RuntimeException | _: IOException | _: InternalError
+            if options.ignoreCorruptFiles =>
+            logWarning("Skipped the rest of the content in the corrupted file", e)
             Iterator.empty
-          case _ => throw e
+          case o => throw o
         }
     }
 

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

@@ -235,26 +235,12 @@ object MultiLineXmlDataSource extends XmlDataSource {
 
     val xmlParserRdd: RDD[StaxXMLRecordReader] =
       xml.flatMap { portableDataStream =>
-        try {
-          val inputStream = () =>
-            CodecStreams.createInputStreamWithCloseResource(
-              portableDataStream.getConfiguration,
-              new Path(portableDataStream.getPath())
-            )
-          StaxXmlParser.convertStream(inputStream, parsedOptions)(identity)
-        } catch {
-          case e: FileNotFoundException if parsedOptions.ignoreMissingFiles =>
-            logWarning("Skipped missing file", e)
-            None
-          case NonFatal(e) =>
-            Utils.getRootCause(e) match {
-              case _: RuntimeException | _: IOException | _: InternalError
-                  if parsedOptions.ignoreCorruptFiles =>
-                logWarning("Skipped the rest of the content in the corrupted file", e)
-                None
-              case o => throw o
-            }
-        }
+        val inputStream = () =>
+          CodecStreams.createInputStreamWithCloseResource(
+            portableDataStream.getConfiguration,
+            new Path(portableDataStream.getPath())
+          )
+        StaxXmlParser.convertStream(inputStream, parsedOptions)(identity)
       }
 
     SQLExecution.withSQLConfPropagated(sparkSession) {