cloud-fan commented on code in PR #57414:
URL: https://github.com/apache/spark/pull/57414#discussion_r3637784172
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala:
##########
@@ -357,6 +270,93 @@ 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. A corrupt/missing input is skipped as a unit when the
ignore flags are set --
+ * across the whole input, since `readArchiveEntries` advances to later
entries lazily (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, skipping the whole input when the
ignore flags are set and
+ * it is missing/corrupt. `readArchiveEntries` opens only the first entry
eagerly and advances to
+ * later entries lazily, so a corrupt later entry throws while the returned
iterator is consumed,
+ * not while it is built -- guarding only construction would let that
escape. The returned iterator
+ * therefore catches on both construction and advancement (`hasNext`),
ending the input on a
+ * skipped error rather than propagating it.
+ */
+ 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:
Please add a regression test where inference consumes a valid first archive
entry and then fails while advancing to a later entry, with
`ignoreCorruptFiles=true`, for both multi-line JSON and XML. The existing
corrupt-archive inference test fails while opening a separate bad archive, so
it never exercises this new `hasNext` recovery path.
--
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]
