brkyvz commented on code in PR #59435:
URL: https://github.com/apache/spark/pull/59435#discussion_r4380367365
##########
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:
this means it was set to an empty string?
##########
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:
instead of passing in a default hadoop conf, can we just provide an option
of HadoopConf?
##########
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:
so I can't provide `decompression("zstd")`? Do we not expect users to
provide the extension or short name of the codec?
##########
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:
should we accept `zst` as well for zstandard? Since that's a more common
suffix for it?
##########
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:
is this common? I thought you could technically provide session/default
confs. I doubt setting this as a global conf will be common, but if it is set -
I don't know why we would remove it.
--
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]
