remove SparkConf

Author: sandip-db

core/src/main/scala/org/apache/spark/internal/config/package.scala

@@ -2846,16 +2846,4 @@ package object config {
       .checkValues(Set("connect", "classic"))
       .createWithDefault(
         if (sys.env.get("SPARK_CONNECT_MODE").contains("1")) "connect" else "classic")
-
-  private[spark] val FILE_DATA_SOURCE_ZSTANDARD_ENABLED =
-    ConfigBuilder("spark.io.zStandard.enabled")
-      .internal()
-      .doc("Hadoop library used in Spark has not been compiled with ZSTD support. " +
-        "This conf enables the use of ZStandard codec available within SparkCompressionCodec " +
-        "for file data sources. This conf is not a SQLConf because SQLConf is not accessible " +
-        "from WholeTextFileRecordReader. This is a temporary workaround until Spark's Hadoop " +
-        "library has support for ZSTD.")
-      .version("4.1.0")
-      .booleanConf
-      .createWithDefault(true)
 }

core/src/main/scala/org/apache/spark/io/HadoopCodecStreams.scala

@@ -27,7 +27,6 @@ import org.apache.hadoop.fs.Path
 import org.apache.hadoop.io.compress._
 
 import org.apache.spark.{SparkConf, SparkEnv}
-import org.apache.spark.internal.config
 import org.apache.spark.io.{CompressionCodec => SparkCompressionCodec}
 
 /**
@@ -37,9 +36,6 @@ import org.apache.spark.io.{CompressionCodec => SparkCompressionCodec}
  * non-standard file extensions like `.zstd` and `.gzip` for Zstandard and Gzip codecs.
  */
 object HadoopCodecStreams {
-  private lazy val sparkConf = Option(SparkEnv.get).map(_.conf).getOrElse(new SparkConf)
-  private lazy val isSparkZstdCodecEnabled =
-    sparkConf.get(config.FILE_DATA_SOURCE_ZSTANDARD_ENABLED)
   private val ZSTD_EXTENSIONS = Seq(".zstd", ".zst")
 
   // get codec based on file name extension
@@ -62,16 +58,10 @@ object HadoopCodecStreams {
   def createZstdInputStream(
     file: Path,
     inputStream: InputStream): Option[InputStream] = {
+    val sparkConf = Option(SparkEnv.get).map(_.conf).getOrElse(new SparkConf)
     val fileName = file.getName.toLowerCase(Locale.ROOT)
 
-    // FOR DEBUGGING GITHUB ACTION FAILURE ONLY -- START
-    assert(ZSTD_EXTENSIONS.exists(fileName.endsWith),
-      s"File $fileName does not have a recognized Zstandard extension:" +
-        s"${ZSTD_EXTENSIONS.mkString(", ")}")
-    assert(isSparkZstdCodecEnabled, "Spark Zstandard codec is not enabled.")
-    // FOR DEBUGGING GITHUB ACTION FAILURE ONLY -- END
-
-    val isOpt = if (ZSTD_EXTENSIONS.exists(fileName.endsWith) && isSparkZstdCodecEnabled) {
+    val isOpt = if (ZSTD_EXTENSIONS.exists(fileName.endsWith)) {
       Some(
         SparkCompressionCodec
           .createCodec(sparkConf, SparkCompressionCodec.ZSTD)
@@ -80,8 +70,6 @@ object HadoopCodecStreams {
     } else {
       None
     }
-    assert(isOpt.isDefined,
-      s"Failed to create Zstandard input stream for file: $fileName")
     isOpt
   }
 
@@ -98,7 +86,7 @@ object HadoopCodecStreams {
         } catch {
           case e: RuntimeException =>
             // createInputStream may fail for ZSTD if hadoop is not already compiled with ZSTD
-            // support. In that case, we try to use Spark's Zstandard codec if enabled.
+            // support. In that case, we try to use Spark's Zstandard codec.
             createZstdInputStream(file, inputStream).getOrElse(throw e)
         }
       }.getOrElse(inputStream)

core/src/test/scala/org/apache/spark/input/WholeTextFileRecordReaderSuite.scala

@@ -25,8 +25,7 @@ import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.io.Text
 import org.apache.hadoop.io.compress.{CompressionCodecFactory, GzipCodec}
 
-import org.apache.spark.{SparkConf, SparkContext, SparkException, SparkFunSuite}
-import org.apache.spark.internal.config
+import org.apache.spark.{SparkConf, SparkContext, SparkFunSuite}
 import org.apache.spark.io.ZStdCompressionCodec
 
 /**
@@ -112,43 +111,26 @@ class WholeTextFileRecordReaderSuite extends SparkFunSuite {
           createNativeFile(dir, filename, contents, compressionType)
         }
 
-        if (!sc.conf.get(config.FILE_DATA_SOURCE_ZSTANDARD_ENABLED) &&
-          (compressionType == CompressionType.ZSTD || compressionType == CompressionType.ZST)) {
-          val e = intercept[SparkException] {
-            sc.wholeTextFiles(dir.toString, 3).collect()
-          }
-          assert(e.getCause.isInstanceOf[RuntimeException])
-          assert(e.getCause.getMessage === "native zStandard library not available: " +
-            "this version of libhadoop was built without zstd support.")
-        } else {
-          val res = sc.wholeTextFiles(dir.toString, 3).collect()
-
-          assert(res.length === WholeTextFileRecordReaderSuite.fileNames.length,
-            "Number of files read out does not fit with the actual value.")
-
-          for ((filename, contents) <- res) {
-            val shortName = compressionType match {
-              case CompressionType.NONE => filename.split('/').last
-              case _ => filename.split('/').last.split('.').head
-            }
-            assert(WholeTextFileRecordReaderSuite.fileNames.contains(shortName),
-              s"Missing file name $filename.")
-            assert(contents === new Text(WholeTextFileRecordReaderSuite.files(shortName)).toString,
-              s"file $filename contents can not match.")
+        val res = sc.wholeTextFiles(dir.toString, 3).collect()
+
+        assert(res.length === WholeTextFileRecordReaderSuite.fileNames.length,
+          "Number of files read out does not fit with the actual value.")
+
+        for ((filename, contents) <- res) {
+          val shortName = compressionType match {
+            case CompressionType.NONE => filename.split('/').last
+            case _ => filename.split('/').last.split('.').head
           }
+          assert(WholeTextFileRecordReaderSuite.fileNames.contains(shortName),
+            s"Missing file name $filename.")
+          assert(contents === new Text(WholeTextFileRecordReaderSuite.files(shortName)).toString,
+            s"file $filename contents can not match.")
         }
       }
     }
   }
 }
 
-class WholeTextFileRecordReaderZStandardDisabledSuite extends WholeTextFileRecordReaderSuite {
-
-  override def getSparkConf(): SparkConf = {
-    super.getSparkConf().set(config.FILE_DATA_SOURCE_ZSTANDARD_ENABLED, false)
-  }
-}
-
 /**
  * Files to be tested are defined here.
  */