sandip-db commented on code in PR #59435:
URL: https://github.com/apache/spark/pull/59435#discussion_r4433374281
##########
core/src/main/scala/org/apache/spark/io/HadoopCodecStreams.scala:
##########
@@ -38,8 +40,49 @@ import org.apache.spark.io.{CompressionCodec =>
SparkCompressionCodec}
object HadoopCodecStreams {
private val ZSTD_EXTENSIONS = Seq(".zstd", ".zst")
- // get codec based on file name extension
+ /**
+ * Hadoop `Configuration` key carrying the read-side compression codec
explicitly requested via
+ * the `decompression` data source option. The value is a full codec class
name, an
+ * empty string to force "no compression" (read the file as-is), or absent
to infer the codec
+ * from the file name extension. Set by the file format / scan read paths
and consumed by
+ * [[getDecompressionCodec]] so it uniformly overrides extension-based
detection for line-based,
+ * multi-line, whole-text, and schema-inference reads.
+ *
+ * This is an internal key, deliberately outside the `spark.*` namespace so
it is not mistaken for
+ * a user-facing configuration. The read paths always overwrite or unset it,
so a value inherited
+ * from the session or Hadoop configuration is never honored.
+ */
+ val DECOMPRESSION_CODEC_CONF =
"org.apache.spark.sql.execution.datasources.decompressionCodec"
+
+ /**
+ * Returns the codec to decompress `file` with. When a read compression
option has been set (see
+ * [[DECOMPRESSION_CODEC_CONF]]) it takes precedence over the file name
extension: a codec class
+ * name selects that codec, and an empty value means the file is read as-is.
Otherwise the codec
+ * is inferred from the file name extension.
+ */
def getDecompressionCodec(
+ config: Configuration,
+ file: Path): Option[CompressionCodec] = {
+ Option(config.get(DECOMPRESSION_CODEC_CONF)) match {
+ case Some(codecClassName) if codecClassName.nonEmpty =>
+ // The compression codec was explicitly specified via the read option,
so use it instead of
+ // inferring from the file name extension. Resolve it the same way the
write path does
+ // (see CodecStreams.getCompressionCodec): load the class and
instantiate it, so any
+ // loadable CompressionCodec is supported, not just those registered
with the factory.
+ val codecClass = config.getClassByName(codecClassName)
+ Some(ReflectionUtils.newInstance(codecClass,
config).asInstanceOf[CompressionCodec])
+ case Some(_) =>
+ // The read option was explicitly set to `none`/`uncompressed`: read
the file as-is,
+ // ignoring any compression-related file name extension.
Review Comment:
Yes. `none`/`uncompressed` is stored as an empty string. I made that
explicit: it's now matched as `case Some("")`, before the class-name case.
##########
core/src/main/scala/org/apache/spark/io/HadoopCodecStreams.scala:
##########
@@ -56,12 +99,26 @@ object HadoopCodecStreams {
}
def createZstdInputStream(
+ file: Path,
+ inputStream: InputStream): Option[InputStream] = {
+ createZstdInputStream(new Configuration(false), file, inputStream)
+ }
+
+ def createZstdInputStream(
+ config: Configuration,
Review Comment:
Done: I removed the 2-argument overload instead. It had no callers left.
Both `createInputStream` and `HadoopLineRecordReader` pass the read's Hadoop
conf, so no default conf is built anymore.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/CompressionCodecs.scala:
##########
@@ -21,21 +21,46 @@ import java.util.Locale
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.io.SequenceFile.CompressionType
+import org.apache.hadoop.io.compress.{SplittableCompressionCodec,
ZStandardCodec}
+import org.apache.spark.io.HadoopCodecStreams
import org.apache.spark.sql.errors.QueryExecutionErrors
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.util.Utils
object CompressionCodecs {
private val shortCompressionCodecNames = HadoopCompressionCodec.values().map
{ codec =>
codec.lowerCaseName() ->
Option(codec.getCompressionCodec).map(_.getClass.getName).orNull
}.toMap
+ // Read-side short names extend the write-side set with Zstandard, which
file-format writers do
+ // not support yet (see the TODO in `HadoopCompressionCodec`) but readers
do, via Spark's
+ // built-in Zstandard codec when Hadoop lacks native support (see
`HadoopCodecStreams`).
+ private val readShortCompressionCodecNames = shortCompressionCodecNames ++
Map(
Review Comment:
Good idea, done: `zst` is now a read short name for `ZStandardCodec`
##########
core/src/main/scala/org/apache/spark/io/HadoopCodecStreams.scala:
##########
@@ -56,12 +99,26 @@ object HadoopCodecStreams {
}
def createZstdInputStream(
+ file: Path,
+ inputStream: InputStream): Option[InputStream] = {
+ createZstdInputStream(new Configuration(false), file, inputStream)
+ }
+
+ def createZstdInputStream(
+ config: Configuration,
file: Path,
inputStream: InputStream): Option[InputStream] = {
val sparkConf = Option(SparkEnv.get).map(_.conf).getOrElse(new SparkConf)
val fileName = file.getName.toLowerCase(Locale.ROOT)
- val isOpt = if (ZSTD_EXTENSIONS.exists(fileName.endsWith)) {
+ // Use Spark's Zstandard codec when the read option explicitly requested a
Zstandard codec (in
+ // which case the extension may not match), or, when no codec was forced,
when the file name has
+ // a Zstandard extension. A forced non-Zstandard codec is never replaced
based on the extension.
+ val useZstd = Option(config.get(DECOMPRESSION_CODEC_CONF)) match {
+ case Some(codecClassName) => codecClassName ==
classOf[ZStandardCodec].getName
Review Comment:
You can: `decompression` accepts `zstd` and `zstandard` (and now `zst`) in
any letter case, as well as codec class names. The short name is resolved to
the codec class name when the option is parsed
(`CompressionCodecs.getReadCodecClassName`), and this internal conf only ever
holds that class name, so comparing class names here covers all of them. The
decode tests read with each of these names. I added a comment here to make that
clear.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/CompressionCodecs.scala:
##########
@@ -66,4 +91,42 @@ object CompressionCodecs {
conf.set("mapreduce.map.output.compress", "false")
}
}
+
+ /**
+ * Records the codec requested via the `decompression` data source option
into the Hadoop
+ * `Configuration`, so the file readers decompress with it instead of
inferring the codec from
+ * the file name extension (see
[[HadoopCodecStreams.getDecompressionCodec]]). Called on the
+ * driver while planning the read, so an unsupported setup fails before any
task runs.
+ *
+ * @param decompressionCodec the parsed `decompression` option: `None` when
absent (unset
+ * the conf so the codec is inferred from the
extension),
+ * `Some(null)` for `none`/`uncompressed` (read
the file as-is), or
+ * `Some(className)` to force that codec.
+ */
+ def setReadCodecConfiguration(
+ conf: Configuration,
+ decompressionCodec: Option[String]): Unit = decompressionCodec match {
+ case Some(codec) =>
+ // Hadoop's `LineRecordReader` infers the codec from the file name
extension and cannot be
+ // told otherwise, so fail rather than silently ignore the option.
+ if (!SQLConf.get.hadoopLineRecordReaderEnabled) {
+ throw
QueryExecutionErrors.decompressionCodecWithLegacyLineReaderError()
+ }
+ // An empty string encodes "no compression" (read as-is); a class name
selects that codec.
+ conf.set(HadoopCodecStreams.DECOMPRESSION_CODEC_CONF,
Option(codec).getOrElse(""))
+ case None =>
+ // Unset rather than leave untouched, so a value inherited from the
session or Hadoop
+ // configuration cannot override the codec of a read that did not ask
for it.
+ conf.unset(HadoopCodecStreams.DECOMPRESSION_CODEC_CONF)
Review Comment:
I'd like to keep the unset. The key is internal (it is deliberately outside
the `spark.*` namespace) and is only meant to carry the parsed read option to
the readers, and honoring a value set globally would cause real problems:
- **Splitting would disagree with decompression.** Splits are decided from
the parsed `decompression` option, not from this conf. A global key that forces
gzip would decompress every file as gzip while the file name extensions still
decide splitting, so a large `.csv` file would be split and then fail with
`Cannot seek in GzipCodec compressed stream`, or silently lose rows with
`ignoreCorruptFiles`. This is the same split/decode mismatch as the earlier
libsvm comment.
- **It would skip the option's checks:** short-name validation, the legacy
line reader error and the archive rejection only run when the option is set.
- **It would apply to every text-based read in the session,** with no
documented contract.
If a session-wide default codec turns out to be useful, I think it should be
a proper `spark.sql.*` conf that feeds the option, so it gets the same checks
and split handling.
--
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]
