sandip-db commented on code in PR #59435:
URL: https://github.com/apache/spark/pull/59435#discussion_r4449572247
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileFormat.scala:
##########
@@ -454,14 +455,58 @@ object FileFormat {
abstract class TextBasedFileFormat extends FileFormat with SessionStateHelper {
private var codecFactory: CompressionCodecFactory = _
+ // The `decompression` splittability resolved for the options of the latest
[[isSplitable]]
+ // call. The scan passes the same options map for every file, so this avoids
parsing the codec
+ // name and loading its class again per file.
+ @transient @volatile
+ private var decompressionCodecSplittableCache: (Map[String, String],
Option[Boolean]) = _
+
+ /**
+ * Forces the `decompression` 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(
+ hadoopConf, new FileSourceOptions(options).decompressionCodec)
+ super.buildReaderWithPartitionValues(
+ sparkSession, dataSchema, partitionSchema, requiredSchema, filters,
options, hadoopConf)
+ }
+
+ /**
+ * Whether the file at `path` can be split by its compression codec. When
the `decompression`
+ * read option is set, the splittability of that codec wins over the file
name extension, so
+ * the split decision always agrees with how
[[buildReaderWithPartitionValues]] decompresses.
+ */
override def isSplitable(
sparkSession: SparkSession,
options: Map[String, String],
path: Path): Boolean = {
- if (codecFactory == null) {
- codecFactory = new CompressionCodecFactory(getHadoopConf(sparkSession,
options))
+ decompressionCodecSplittable(options).getOrElse {
Review Comment:
Agreed, I removed the cache. `isSplitable` now just builds
`FileSourceOptions` and checks `decompressionCodecSplittable`, falling back to
the extension-based check.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/csv/CSVScan.scala:
##########
@@ -54,12 +54,16 @@ case class CSVScan(
conf.columnNameOfCorruptRecord)
override def isSplitable(path: Path): Boolean = {
- CSVDataSource(parsedOptions).isSplitable && super.isSplitable(path)
+ // When `decompression` is set, its splittability wins over the file
extension.
+ CSVDataSource(parsedOptions).isSplitable &&
+
parsedOptions.decompressionCodecSplittable.getOrElse(super.isSplitable(path))
Review Comment:
Good idea, done. `TextBasedFileScan.isSplitable` now applies the forced
codec's splittability, parsing the option once per scan, just like
`TextBasedFileFormat.isSplitable` does for V1. `CSVScan`, `JsonScan` and
`TextScan` are back to calling `super.isSplitable(path)` in both `isSplitable`
and `getFileUnSplittableReason`, so their split code is the same as on master.
--
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]
