cloud-fan commented on code in PR #57414:
URL: https://github.com/apache/spark/pull/57414#discussion_r3630096461
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/csv/CSVTable.scala:
##########
@@ -46,11 +46,7 @@ case class CSVTable(
columnPruning = sparkSession.sessionState.conf.csvColumnPruning,
sparkSession.sessionState.conf.sessionLocalTimeZone)
- // The DSv2 reader does not route archives to `readArchive` (it calls
`readFile` directly), so
- // archive scans aren't supported here; pass supportsArchiveScan = false
so an archive input
- // keeps failing with UNABLE_TO_INFER_SCHEMA rather than having its raw
bytes parsed as CSV.
- CSVDataSource(parsedOptions)
- .inferSchema(sparkSession, files, parsedOptions, supportsArchiveScan =
false)
+ CSVDataSource(parsedOptions).inferSchema(sparkSession, files,
parsedOptions)
Review Comment:
Please keep the DSv2 refusal guard for both CSV and JSON. This call now
infers a schema from decompressed archive entries, but the DSv2 scan still
parses the raw archive bytes. Forcing the source onto V2 therefore changes a
clear unsupported-input error into silent wrong results.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala:
##########
@@ -357,6 +263,63 @@ 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.
+ */
+ 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
+
+ // Each archive entry and each loose file is one JSON document: stream it
straight through (an
+ // archive entry by entry -- streamed, never unpacked -- a loose file
directly), skipping a
+ // whole input when it is corrupt/missing and the ignore flags are set. An
archive entry's
+ // stream is a `CloseShieldInputStream` view over the one shared archive
cursor, valid only
+ // until `readArchiveEntries` advances; `JsonInferSchema.infer` fully
consumes each document
+ // before pulling the next, so the cursor is never read after it advances.
+ val docs: RDD[InputStream] = baseRdd.flatMap { stream =>
+ val path = new Path(stream.getPath())
+ try {
+ if (SupportsArchiveFormat.isArchivePath(path)) {
+ SupportsArchiveFormat.readArchiveEntries(path,
stream.getConfiguration) { (_, in) =>
+ Iterator.single(in)
+ }
+ } else {
+ Iterator.single(
+
CodecStreams.createInputStreamWithCloseResource(stream.getConfiguration, path))
+ }
+ } catch {
+ case e: FileNotFoundException if ignoreMissingFiles =>
+ logWarning(log"Skipped missing input: ${MDC(PATH,
stream.getPath())}", e)
+ Iterator.empty
+ case e: FileNotFoundException => throw e
+ case e @ (_: RuntimeException | _: IOException) if ignoreCorruptFiles
=>
Review Comment:
Please make this handler cover exceptions raised while the returned iterator
is consumed, and apply the same fix to the XML helper. `readArchiveEntries`
only probes the first entry before returning. A failure while advancing to a
later entry happens after this `try` has returned, so `ignoreCorruptFiles` does
not reliably skip the corrupt input as the new Scaladoc claims.
--
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]
