brkyvz commented on code in PR #59435:
URL: https://github.com/apache/spark/pull/59435#discussion_r4439402531
##########
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:
Is it possible to push the `parsedOptions.decompressionCodecSplittable` into
super.isSplittable? It is copied 6 times across 3 files. The options are
available in `TextBasedFileScan` and since `decompressionCodecSplittable` is a
shared option, maybe we can do something there?
##########
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:
this feels overkill in terms of how much it may be likely saving us. I
imagine the code below may already be pretty inefficient comparatively
--
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]
