HyukjinKwon commented on code in PR #59435:
URL: https://github.com/apache/spark/pull/59435#discussion_r4273571562
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala:
##########
@@ -4303,6 +4305,165 @@ abstract class CSVSuite
}
}
}
+
+ // Writes `data` to `file` compressed with `codec` (a short name; "" means
no compression), using
+ // Spark's built-in Zstandard codec for zstd (which Hadoop may not be
compiled with) and Hadoop's
+ // CompressionCodecFactory otherwise. The file name intentionally does NOT
carry the codec's
+ // extension, so extension-based detection cannot pick the right codec and
the
+ // `decompressionCodec` option is what makes the read succeed.
+ private def writeCompressed(file: File, data: String, codec: String): Unit =
{
+ val codecFactory = new
CompressionCodecFactory(spark.sessionState.newHadoopConf())
+ val os = new FileOutputStream(file)
+ val out = codec match {
+ case "" => os
+ case "zstd" | "zstandard" => new
ZStdCompressionCodec(sparkConf).compressedOutputStream(os)
+ case name => codecFactory.getCodecByName(name).createOutputStream(os)
+ }
+ out.write(data.getBytes(StandardCharsets.UTF_8))
+ out.close()
+ }
+
+ test("SPARK-60217 decompressionCodec option overrides file name extension") {
+ val numRecord = 100
+ val records = (1 to numRecord).map(i => (i, s"value_$i"))
+ // Every read-supported codec, including zstd/zstandard which the writers
do not support.
+ val codecs = HadoopCompressionCodec.values().map(_.lowerCaseName())
+ .filterNot(c => c == HadoopCompressionCodec.NONE.lowerCaseName() ||
+ c == HadoopCompressionCodec.UNCOMPRESSED.lowerCaseName()) ++
Seq("zstd", "zstandard")
+ for (multiLine <- Seq(true, false); codec <- codecs) {
+ withTempDir { dir =>
+ val data = ("id,value" +: records.map { case (id, v) => s"$id,$v"
}).mkString("\n")
+ // Misleading `.csv` name (no codec suffix): extension-based detection
would read it as
+ // plain text and fail. The `decompressionCodec` option forces the
correct codec.
+ val file = new File(dir, "data.csv")
+ writeCompressed(file, data, codec)
+
+ val df = spark.read
+ .option("decompressionCodec", codec)
+ .option("header", "true")
+ .option("inferSchema", "true")
+ .option("multiLine", multiLine.toString)
+ .csv(file.getCanonicalPath)
+ checkAnswer(df, records.map { case (id, v) => Row(id, v) })
+ }
+ }
+ }
+
+ test("SPARK-60217 decompressionCodec option decides whether a file is
split") {
Review Comment:
**Nit (P3):** This is the only V1 CSV split test, and it cannot fail if the
new term in `CSVFileFormat.isSplitable` is reverted: bzip2 on a `.csv` name is
splittable by extension anyway, and the multiLine case short-circuits before
the term.
The other formats have the same gap. `JsonSuite` (V1 and V2) has no
partition-count or unsplittable-reason assertion. Every `XmlSuite` test uses
the default `multiLine=true`, so `XmlFileFormat`'s new term never runs.
Cases like the `TextSuite` split table for these formats would close the gap.
**Verification:**
- **Regression:** For CSV V1, JSON V1/V2 and line-mode XML, a forced
non-splittable codec yields exactly one partition with the complete answer, and
forced `none` on a .gz-named plain file yields multiple partitions with the
complete answer.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormat.scala:
##########
@@ -482,6 +483,26 @@ object FileFormat {
abstract class TextBasedFileFormat extends FileFormat with SessionStateHelper {
private var codecFactory: CompressionCodecFactory = _
+ /**
+ * Forces the `decompressionCodec` read option, when set, onto `hadoopConf`
before the format's
+ * [[buildReader]] broadcasts it, so the readers decompress every file with
that codec instead of
+ * inferring it from the file name extension
+ * (see [[CompressionCodecs.setReadCodecConfiguration]]).
+ */
+ override def buildReaderWithPartitionValues(
+ sparkSession: SparkSession,
+ dataSchema: StructType,
+ partitionSchema: StructType,
+ requiredSchema: StructType,
+ filters: Seq[Filter],
+ options: Map[String, String],
+ hadoopConf: Configuration): PartitionedFile => Iterator[InternalRow] = {
+ CompressionCodecs.setReadCodecConfiguration(
Review Comment:
**Nit (P3):** This override forces `decompressionCodec` for every
`TextBasedFileFormat` subclass, not only CSV, JSON, text and XML. mllib's
`LibSVMFileFormat` inherits it but keeps the extension-based `isSplitable`.
For example, `spark.read.format("libsvm").option("decompressionCodec",
"gzip").load(path)` on a gzip file named `data.libsvm` that spans several
splits fails in `HadoopLineRecordReader` with `Cannot seek in GzipCodec
compressed stream`. With `ignoreCorruptFiles=true`, every split after the first
is skipped, so rows go missing without an error.
libsvm's numFeatures inference also honors the option today:
`MLUtils.parseLibSVMFile` reads through `TextFileFormat` with the same options.
The `FileSourceOptions` scaladoc says only the four formats honor the option.
CSV, JSON, text and XML reads are not affected.
**Recommended change:** Make forcing `decompressionCodec` an explicit opt-in
of the CSV, JSON, text and XML file formats, for example an overridable
TextBasedFileFormat hook that defaults to off. Other TextBasedFileFormat
subclasses, including LibSVMFileFormat, then get the internal key unset and
keep merge-target reading and splitting. Also drop `decompressionCodec` from
the options LibSVMFileFormat.inferSchema passes to MLUtils.parseLibSVMFile, so
numFeatures inference ignores it as well. Add libsvm data source regression
tests showing the option does not affect libsvm reads, with and without
numFeatures.
**Why this works:** The base buildReaderWithPartitionValues passes the
parsed codec to setReadCodecConfiguration only when the concrete format opts
in. Otherwise it passes None, which unsets the key, so a format that forces the
codec is always one that also applies decompressionCodecSplittable in its
isSplitable. LibSVMFileFormat.inferSchema removes the option before delegating
to TextFileFormat, so no libsvm read route forces the codec.
**Scope:** Restrict codec forcing in the V1 text-based base class to the
four formats that already own the matching split precedence. Keep libsvm's
TextFileFormat-based schema inference from re-applying the option, and pin
libsvm behavior with regression tests.
**Compatibility:** Forced decompression and split precedence for the CSV,
JSON, text and XML sources on DSv1 and DSv2.
**Risks:** A supporting format that forgets to opt in silently ignores the
option. The four formats must opt in, and the existing per-format decode tests
catch a missed opt-in. A future non-supporting format that delegates inference
to a supporting format with raw user options would re-open the second route;
the libsvm change documents that pattern.
**Constraints:** Keep the DSv2 TextBasedFileScan path unchanged. All its
subclasses already support the option.
**Success:** spark.read.format("libsvm") with `decompressionCodec` set
reads, infers numFeatures and splits files exactly as without the option,
whether or not numFeatures is given. CSV, JSON, text and XML reads keep
honoring the forced codec and its split precedence.
**Verification:**
- **Regression:** With `decompressionCodec` set to a non-splittable codec, a
libsvm read of a multi-split plain file returns all rows with the same
partitioning as without the option, both when numFeatures is given and when it
is inferred, and fails with neither a cannot-seek nor a decompression error.
- **Compatibility:** The existing forced-codec decode and split tests for
CSV, JSON, text and XML still pass, showing the four formats still opt in.
##########
docs/sql-data-sources-csv.md:
##########
@@ -297,6 +297,12 @@ Data source options of CSV can be set via:
<td>Compression codec to use when saving to file. This can be one of the
known case-insensitive shorten names (<code>none</code>, <code>bzip2</code>,
<code>gzip</code>, <code>lz4</code>, <code>snappy</code> and
<code>deflate</code>). CSV built-in functions ignore this option.</td>
<td>write</td>
</tr>
+ <tr>
+ <td><code>decompressionCodec</code></td>
+ <td>(none)</td>
+ <td>Compression codec used to decompress the input files, instead of
inferring it from each file name extension. This can be one of the known
case-insensitive short names (<code>none</code>, <code>uncompressed</code>,
<code>bzip2</code>, <code>deflate</code>, <code>gzip</code>, <code>lz4</code>,
<code>snappy</code>, <code>zstd</code> and <code>zstandard</code>) or a fully
qualified Hadoop <code>CompressionCodec</code> class name. <code>none</code>
and <code>uncompressed</code> read the files as-is, even when the file name
carries a compression extension. The codec applies to every file read; setting
it when the input includes an archive file (<code>.tar</code>,
<code>.tar.gz</code>, <code>.tgz</code>, <code>.zip</code> or <code>.7z</code>)
is not supported and fails the query. Unlike the write option
<code>compression</code>, which is ignored on read, it does not affect writes.
CSV built-in functions ignore this option.</td>
Review Comment:
**Nit (P3):** This archive sentence is unconditional, and so are the ones in
the JSON, text and XML pages. But `checkNoDecompressionCodec` only runs inside
the `archiveFormatEnabled && isArchivePath` branch, and
`spark.sql.files.archive.reader.enabled` defaults to false.
With the default config, `spark.read.option("decompressionCodec",
"gzip").csv("in.tar.gz")` does not fail: the tar stream is gunzipped and parsed
as CSV rows. Could the four rows say that the rejection applies when archive
reading is enabled?
**Verification:**
- **Inspection:** Each updated doc row names the archive-reading condition
and no longer promises an unconditional failure.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormat.scala:
##########
@@ -216,6 +229,8 @@ object SupportsArchiveFormat {
* `(entry, stream)` pairs.
*/
private def openArchiveStream(path: Path, conf: Configuration):
ArchiveEntries = {
+ // Backstop for paths that open archives without planning a scan (e.g.
schema inference).
+ checkNoDecompressionCodec(path,
Option(conf.get(HadoopCodecStreams.DECOMPRESSION_CODEC_CONF)))
Review Comment:
**Nit (P3):** On the archive-aware schema inference paths (CSV
`inferSchema`'s archive branch and multiLine JSON and XML `inferWithArchives`),
this backstop is the only check, and it runs inside an inference task.
`inferWithArchives`/`skipInputOnError` then catch it:
- with `ignoreCorruptFiles=true`, the archive is logged as corrupt and
dropped from inference;
- otherwise, the job fails with `FAILED_READ_FILE` wrapping the cause.
Both differ from the top-level driver-side
`UNSUPPORTED_FEATURE.DECOMPRESSION_CODEC_FOR_ARCHIVE` described in the PR and
in `checkNoDecompressionCodec`'s scaladoc. For example, with archive reading
enabled, `spark.read.option("multiLine", true).option("decompressionCodec",
"gzip").json(dirWithTarGz)` without a schema takes this path. The shared
archive tests always read with `.schema(readSchema)`, so it is never exercised.
The inference entry points already know on the driver whether an input is an
archive, so checking there would keep the error consistent. Only the opt-in
archive reader is affected.
**Verification:**
- **Regression:** Reading an archive with `decompressionCodec` and no user
schema raises the top-level UNSUPPORTED_FEATURE.DECOMPRESSION_CODEC_FOR_ARCHIVE
error, with ignoreCorruptFiles both true and false, for every archive-capable
format that accepts the option.
--
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]
