cloud-fan commented on code in PR #57414:
URL: https://github.com/apache/spark/pull/57414#discussion_r3685100433
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala:
##########
@@ -357,6 +270,91 @@ object MultiLineJsonDataSource extends JsonDataSource {
}
}
+ /**
+ * Infers a multi-line JSON schema when at least one input is an archive.
Each archive entry
+ * (streamed via `SupportsArchiveFormat`, never unpacked to disk) and each
loose file is one whole
+ * JSON document, and all feed a single [[JsonInferSchema]] pass -- exactly
as a directory of the
+ * same files would infer. Single-line archive inference does not come here:
the Text data source
+ * reads archives directly, so it flows through
[[TextInputJsonDataSource.infer]] like any
+ * directory read. Corrupt/missing inputs are skipped when the ignore flags
are set (see
+ * [[skipInputOnError]]).
+ */
+ private def inferWithArchives(
+ sparkSession: SparkSession,
+ inputPaths: Seq[FileStatus],
+ parsedOptions: JSONOptions): StructType = {
+ val baseRdd = JsonDataSource.createBaseRdd(sparkSession, inputPaths,
parsedOptions)
+ val encoding = parsedOptions.encoding
+ val ignoreCorruptFiles = parsedOptions.ignoreCorruptFiles
+ val ignoreMissingFiles = parsedOptions.ignoreMissingFiles
+
+ // An archive entry's stream is only valid until the shared cursor
advances, so each document
+ // must be consumed before the next is pulled; `JsonInferSchema.infer`
does.
+ val docs: RDD[InputStream] = baseRdd.flatMap { stream =>
+ val path = new Path(stream.getPath())
+ skipInputOnError(stream.getPath(), ignoreMissingFiles,
ignoreCorruptFiles) {
+ if (SupportsArchiveFormat.isArchivePath(path)) {
+ SupportsArchiveFormat.readArchiveEntries(path,
stream.getConfiguration) { (_, in) =>
+ Iterator.single(in)
+ }
+ } else {
+ Iterator.single(
+
CodecStreams.createInputStreamWithCloseResource(stream.getConfiguration, path))
+ }
+ }
+ }
+
+ SQLExecution.withSQLConfPropagated(sparkSession) {
+ val docParser: (JsonFactory, InputStream) => JsonParser = encoding
+ .map(enc => CreateJacksonParser.inputStream(enc, _: JsonFactory, _:
InputStream))
+ .getOrElse(CreateJacksonParser.inputStream(_: JsonFactory, _:
InputStream))
+ new JsonInferSchema(parsedOptions).infer[InputStream](
+ JsonUtils.sample(docs, parsedOptions), docParser, isReadFile = true)
+ }
+ }
+
+ /**
+ * Builds one input's document iterator, catching a missing/corrupt error
when the ignore flags
+ * are set. `readArchiveEntries` advances to later entries lazily, so a
corrupt later entry throws
+ * on `hasNext`, not at construction; the returned iterator catches both. A
construction failure
+ * skips the whole input; a mid-advance failure keeps the entries already
yielded and skips only
+ * the remainder of the archive.
+ */
+ private def skipInputOnError(
+ inputPath: String,
+ ignoreMissingFiles: Boolean,
+ ignoreCorruptFiles: Boolean)(
+ build: => Iterator[InputStream]): Iterator[InputStream] = {
+ def handle(e: Throwable): Iterator[InputStream] = e match {
+ case e: FileNotFoundException if ignoreMissingFiles =>
+ logWarning(log"Skipped missing input: ${MDC(PATH, inputPath)}", e)
+ Iterator.empty
+ case e: FileNotFoundException => throw e
+ case e @ (_: RuntimeException | _: IOException) if ignoreCorruptFiles =>
+ logWarning(log"Skipped the corrupted input: ${MDC(PATH, inputPath)}",
e)
+ Iterator.empty
+ case NonFatal(e) =>
+ throw QueryExecutionErrors.cannotReadFilesError(
+ e, SparkPath.fromPathString(inputPath).urlEncoded)
+ }
+
+ val underlying =
+ try build
+ catch { case NonFatal(e) => return handle(e) }
+
+ new Iterator[InputStream] {
+ private var delegate = underlying
+ override def hasNext: Boolean =
+ try delegate.hasNext
+ catch {
+ case NonFatal(e) =>
+ delegate = handle(e)
Review Comment:
Close the failed `Closeable` iterator before replacing it.
`readArchiveEntries` releases its archive stream only on normal exhaustion or
explicit close, so this catch retains one archive/decoder stream per
late-corrupt input until task completion; enough ignored corrupt archives in
one partition can exhaust file descriptors. Please apply the same fix to the
XML helper.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
