diff options
42 files changed, 368 insertions, 671 deletions
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala b/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala index c0811f6a4f..6a600c1379 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/DataFrameWriter.scala @@ -297,7 +297,7 @@ final class DataFrameWriter private[sql](df: DataFrame) { // If offsets have already been created, we trying to resume a query. val checkpointPath = new Path(checkpointLocation, "offsets") - val fs = checkpointPath.getFileSystem(df.sparkSession.sessionState.hadoopConf) + val fs = checkpointPath.getFileSystem(df.sparkSession.sessionState.newHadoopConf()) if (fs.exists(checkpointPath)) { throw new AnalysisException( s"Unable to resume query written to memory sink. Delete $checkpointPath to start over.") diff --git a/sql/core/src/main/scala/org/apache/spark/sql/RuntimeConfig.scala b/sql/core/src/main/scala/org/apache/spark/sql/RuntimeConfig.scala index e90a042431..bf97d728b8 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/RuntimeConfig.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/RuntimeConfig.scala @@ -17,33 +17,44 @@ package org.apache.spark.sql +import org.apache.spark.sql.internal.SQLConf + /** - * Runtime configuration interface for Spark. To access this, use `SparkSession.conf`. + * Runtime configuration interface for Spark. To access this, use [[SparkSession.conf]]. + * + * Options set here are automatically propagated to the Hadoop configuration during I/O. * * @since 2.0.0 */ -abstract class RuntimeConfig { +class RuntimeConfig private[sql](sqlConf: SQLConf = new SQLConf) { /** * Sets the given Spark runtime configuration property. * * @since 2.0.0 */ - def set(key: String, value: String): RuntimeConfig + def set(key: String, value: String): RuntimeConfig = { + sqlConf.setConfString(key, value) + this + } /** * Sets the given Spark runtime configuration property. * * @since 2.0.0 */ - def set(key: String, value: Boolean): RuntimeConfig + def set(key: String, value: Boolean): RuntimeConfig = { + set(key, value.toString) + } /** * Sets the given Spark runtime configuration property. * * @since 2.0.0 */ - def set(key: String, value: Long): RuntimeConfig + def set(key: String, value: Long): RuntimeConfig = { + set(key, value.toString) + } /** * Returns the value of Spark runtime configuration property for the given key. @@ -52,49 +63,27 @@ abstract class RuntimeConfig { * @since 2.0.0 */ @throws[NoSuchElementException]("if the key is not set") - def get(key: String): String + def get(key: String): String = { + sqlConf.getConfString(key) + } /** * Returns the value of Spark runtime configuration property for the given key. * * @since 2.0.0 */ - def getOption(key: String): Option[String] + def getOption(key: String): Option[String] = { + try Option(get(key)) catch { + case _: NoSuchElementException => None + } + } /** * Resets the configuration property for the given key. * * @since 2.0.0 */ - def unset(key: String): Unit - - /** - * Sets the given Hadoop configuration property. This is passed directly to Hadoop during I/O. - * - * @since 2.0.0 - */ - def setHadoop(key: String, value: String): RuntimeConfig - - /** - * Returns the value of the Hadoop configuration property. - * - * @throws NoSuchElementException if the key is not set - * @since 2.0.0 - */ - @throws[NoSuchElementException]("if the key is not set") - def getHadoop(key: String): String - - /** - * Returns the value of the Hadoop configuration property. - * - * @since 2.0.0 - */ - def getHadoopOption(key: String): Option[String] - - /** - * Resets the Hadoop configuration property for the given key. - * - * @since 2.0.0 - */ - def unsetHadoop(key: String): Unit + def unset(key: String): Unit = { + sqlConf.unsetConf(key) + } } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala b/sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala index 6477f42680..f05546a567 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala @@ -40,7 +40,7 @@ import org.apache.spark.sql.catalyst.plans.logical.{LocalRelation, LogicalPlan, import org.apache.spark.sql.execution._ import org.apache.spark.sql.execution.datasources.LogicalRelation import org.apache.spark.sql.execution.ui.SQLListener -import org.apache.spark.sql.internal.{CatalogImpl, RuntimeConfigImpl, SessionState, SharedState} +import org.apache.spark.sql.internal.{CatalogImpl, SessionState, SharedState} import org.apache.spark.sql.sources.BaseRelation import org.apache.spark.sql.types.{DataType, LongType, StructType} import org.apache.spark.sql.util.ExecutionListenerManager @@ -201,9 +201,7 @@ class SparkSession private( * @group config * @since 2.0.0 */ - @transient lazy val conf: RuntimeConfig = { - new RuntimeConfigImpl(sessionState.conf, sessionState.hadoopConf) - } + @transient lazy val conf: RuntimeConfig = new RuntimeConfig(sessionState.conf) /** * Set Spark SQL configuration properties. diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/AnalyzeTable.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/AnalyzeTable.scala index b6f7808398..54ff5ae7d9 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/AnalyzeTable.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/AnalyzeTable.scala @@ -77,7 +77,7 @@ case class AnalyzeTable(tableName: String) extends RunnableCommand { catalogTable.storage.locationUri.map { p => val path = new Path(p) try { - val fs = path.getFileSystem(sparkSession.sessionState.hadoopConf) + val fs = path.getFileSystem(sparkSession.sessionState.newHadoopConf()) calculateTableSize(fs, path) } catch { case NonFatal(e) => diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/SetCommand.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/SetCommand.scala index 952a0d676f..bbb2a2235f 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/SetCommand.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/SetCommand.scala @@ -62,70 +62,6 @@ case class SetCommand(kv: Option[(String, Option[String])]) extends RunnableComm } (keyValueOutput, runFunc) - case Some((SQLConf.Deprecated.EXTERNAL_SORT, Some(value))) => - val runFunc = (sparkSession: SparkSession) => { - logWarning( - s"Property ${SQLConf.Deprecated.EXTERNAL_SORT} is deprecated and will be ignored. " + - s"External sort will continue to be used.") - Seq(Row(SQLConf.Deprecated.EXTERNAL_SORT, "true")) - } - (keyValueOutput, runFunc) - - case Some((SQLConf.Deprecated.USE_SQL_AGGREGATE2, Some(value))) => - val runFunc = (sparkSession: SparkSession) => { - logWarning( - s"Property ${SQLConf.Deprecated.USE_SQL_AGGREGATE2} is deprecated and " + - s"will be ignored. ${SQLConf.Deprecated.USE_SQL_AGGREGATE2} will " + - s"continue to be true.") - Seq(Row(SQLConf.Deprecated.USE_SQL_AGGREGATE2, "true")) - } - (keyValueOutput, runFunc) - - case Some((SQLConf.Deprecated.TUNGSTEN_ENABLED, Some(value))) => - val runFunc = (sparkSession: SparkSession) => { - logWarning( - s"Property ${SQLConf.Deprecated.TUNGSTEN_ENABLED} is deprecated and " + - s"will be ignored. Tungsten will continue to be used.") - Seq(Row(SQLConf.Deprecated.TUNGSTEN_ENABLED, "true")) - } - (keyValueOutput, runFunc) - - case Some((SQLConf.Deprecated.CODEGEN_ENABLED, Some(value))) => - val runFunc = (sparkSession: SparkSession) => { - logWarning( - s"Property ${SQLConf.Deprecated.CODEGEN_ENABLED} is deprecated and " + - s"will be ignored. Codegen will continue to be used.") - Seq(Row(SQLConf.Deprecated.CODEGEN_ENABLED, "true")) - } - (keyValueOutput, runFunc) - - case Some((SQLConf.Deprecated.UNSAFE_ENABLED, Some(value))) => - val runFunc = (sparkSession: SparkSession) => { - logWarning( - s"Property ${SQLConf.Deprecated.UNSAFE_ENABLED} is deprecated and " + - s"will be ignored. Unsafe mode will continue to be used.") - Seq(Row(SQLConf.Deprecated.UNSAFE_ENABLED, "true")) - } - (keyValueOutput, runFunc) - - case Some((SQLConf.Deprecated.SORTMERGE_JOIN, Some(value))) => - val runFunc = (sparkSession: SparkSession) => { - logWarning( - s"Property ${SQLConf.Deprecated.SORTMERGE_JOIN} is deprecated and " + - s"will be ignored. Sort merge join will continue to be used.") - Seq(Row(SQLConf.Deprecated.SORTMERGE_JOIN, "true")) - } - (keyValueOutput, runFunc) - - case Some((SQLConf.Deprecated.PARQUET_UNSAFE_ROW_RECORD_READER_ENABLED, Some(value))) => - val runFunc = (sparkSession: SparkSession) => { - logWarning( - s"Property ${SQLConf.Deprecated.PARQUET_UNSAFE_ROW_RECORD_READER_ENABLED} is " + - s"deprecated and will be ignored. Vectorized parquet reader will be used instead.") - Seq(Row(SQLConf.PARQUET_VECTORIZED_READER_ENABLED, "true")) - } - (keyValueOutput, runFunc) - // Configures a single property. case Some((key, Some(value))) => val runFunc = (sparkSession: SparkSession) => { diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala index 8d9feec9b8..f38e260bc9 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala @@ -210,7 +210,7 @@ case class LoadData( // Follow Hive's behavior: // If no schema or authority is provided with non-local inpath, // we will use hadoop configuration "fs.default.name". - val defaultFSConf = sparkSession.sessionState.hadoopConf.get("fs.default.name") + val defaultFSConf = sparkSession.sessionState.newHadoopConf().get("fs.default.name") val defaultFS = if (defaultFSConf == null) { new URI("") } else { diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala index 8e72e06b1f..2f3826f72b 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala @@ -131,7 +131,7 @@ case class DataSource( val allPaths = caseInsensitiveOptions.get("path") val globbedPaths = allPaths.toSeq.flatMap { path => val hdfsPath = new Path(path) - val fs = hdfsPath.getFileSystem(sparkSession.sessionState.hadoopConf) + val fs = hdfsPath.getFileSystem(sparkSession.sessionState.newHadoopConf()) val qualified = hdfsPath.makeQualified(fs.getUri, fs.getWorkingDirectory) SparkHadoopUtil.get.globPathIfNecessary(qualified) }.toArray @@ -226,7 +226,7 @@ case class DataSource( case Seq(singlePath) => try { val hdfsPath = new Path(singlePath) - val fs = hdfsPath.getFileSystem(sparkSession.sessionState.hadoopConf) + val fs = hdfsPath.getFileSystem(sparkSession.sessionState.newHadoopConf()) val metadataPath = new Path(hdfsPath, FileStreamSink.metadataDir) val res = fs.exists(metadataPath) res @@ -284,7 +284,7 @@ case class DataSource( val allPaths = caseInsensitiveOptions.get("path") ++ paths val globbedPaths = allPaths.flatMap { path => val hdfsPath = new Path(path) - val fs = hdfsPath.getFileSystem(sparkSession.sessionState.hadoopConf) + val fs = hdfsPath.getFileSystem(sparkSession.sessionState.newHadoopConf()) val qualified = hdfsPath.makeQualified(fs.getUri, fs.getWorkingDirectory) val globPath = SparkHadoopUtil.get.globPathIfNecessary(qualified) @@ -374,7 +374,7 @@ case class DataSource( val path = new Path(caseInsensitiveOptions.getOrElse("path", { throw new IllegalArgumentException("'path' is not specified") })) - val fs = path.getFileSystem(sparkSession.sessionState.hadoopConf) + val fs = path.getFileSystem(sparkSession.sessionState.newHadoopConf()) path.makeQualified(fs.getUri, fs.getWorkingDirectory) } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala index c26cae84d7..615906a52e 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategy.scala @@ -19,7 +19,6 @@ package org.apache.spark.sql.execution.datasources import scala.collection.mutable.ArrayBuffer -import org.apache.hadoop.conf.Configuration import org.apache.hadoop.fs.{BlockLocation, FileStatus, LocatedFileStatus, Path} import org.apache.spark.internal.Logging @@ -107,9 +106,6 @@ private[sql] object FileSourceStrategy extends Strategy with Logging { val pushedDownFilters = dataFilters.flatMap(DataSourceStrategy.translateFilter) logInfo(s"Pushed Filters: ${pushedDownFilters.mkString(",")}") - val hadoopConf = new Configuration(files.sparkSession.sessionState.hadoopConf) - files.options.foreach { case (k, v) => if (v ne null) hadoopConf.set(k, v) } - val readFile = files.fileFormat.buildReader( sparkSession = files.sparkSession, dataSchema = files.dataSchema, @@ -117,7 +113,7 @@ private[sql] object FileSourceStrategy extends Strategy with Logging { requiredSchema = prunedDataSchema, filters = pushedDownFilters, options = files.options, - hadoopConf = hadoopConf) + hadoopConf = files.sparkSession.sessionState.newHadoopConfWithOptions(files.options)) val plannedPartitions = files.bucketSpec match { case Some(bucketing) if files.sparkSession.sessionState.conf.bucketingEnabled => diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/InsertIntoHadoopFsRelation.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/InsertIntoHadoopFsRelation.scala index fa954975b8..4df7d0ce4c 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/InsertIntoHadoopFsRelation.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/InsertIntoHadoopFsRelation.scala @@ -78,7 +78,7 @@ private[sql] case class InsertIntoHadoopFsRelation( s"cannot save to file.") } - val hadoopConf = new Configuration(sparkSession.sessionState.hadoopConf) + val hadoopConf = sparkSession.sessionState.newHadoopConfWithOptions(options) val fs = outputPath.getFileSystem(hadoopConf) val qualifiedOutputPath = outputPath.makeQualified(fs.getUri, fs.getWorkingDirectory) @@ -106,10 +106,6 @@ private[sql] case class InsertIntoHadoopFsRelation( val job = Job.getInstance(hadoopConf) job.setOutputKeyClass(classOf[Void]) job.setOutputValueClass(classOf[InternalRow]) - - // Also set the options in Hadoop Configuration - options.foreach { case (k, v) => if (v ne null) job.getConfiguration.set(k, v) } - FileOutputFormat.setOutputPath(job, qualifiedOutputPath) val partitionSet = AttributeSet(partitionColumns) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/fileSourceInterfaces.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/fileSourceInterfaces.scala index 3058e79201..25f88d9c39 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/fileSourceInterfaces.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/fileSourceInterfaces.scala @@ -274,7 +274,7 @@ class HDFSFileCatalog( partitionSchema: Option[StructType]) extends FileCatalog with Logging { - private val hadoopConf = new Configuration(sparkSession.sessionState.hadoopConf) + private val hadoopConf = sparkSession.sessionState.newHadoopConfWithOptions(parameters) var leafFiles = mutable.LinkedHashMap.empty[Path, FileStatus] var leafDirToChildrenFiles = mutable.Map.empty[Path, Array[FileStatus]] diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JSONRelation.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JSONRelation.scala index b6b3907e3e..62446583a5 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JSONRelation.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JSONRelation.scala @@ -128,7 +128,7 @@ class DefaultSource extends FileFormat with DataSourceRegister { private def createBaseRdd( sparkSession: SparkSession, inputPaths: Seq[FileStatus]): RDD[String] = { - val job = Job.getInstance(sparkSession.sessionState.hadoopConf) + val job = Job.getInstance(sparkSession.sessionState.newHadoopConf()) val conf = job.getConfiguration val paths = inputPaths.map(_.getPath) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRelation.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRelation.scala index 5be8770790..c689ad08ca 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRelation.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRelation.scala @@ -652,7 +652,7 @@ private[sql] object ParquetRelation extends Logging { val assumeInt96IsTimestamp = sparkSession.sessionState.conf.isParquetINT96AsTimestamp val writeLegacyParquetFormat = sparkSession.sessionState.conf.writeLegacyParquetFormat val serializedConf = - new SerializableConfiguration(sparkSession.sessionState.hadoopConf) + new SerializableConfiguration(sparkSession.sessionState.newHadoopConf()) // !! HACK ALERT !! // diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSink.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSink.scala index 61c9c88cb3..70aea7fa49 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSink.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSink.scala @@ -45,7 +45,7 @@ class FileStreamSink( private val basePath = new Path(path) private val logPath = new Path(basePath, FileStreamSink.metadataDir) private val fileLog = new FileStreamSinkLog(sparkSession, logPath.toUri.toString) - private val fs = basePath.getFileSystem(sparkSession.sessionState.hadoopConf) + private val fs = basePath.getFileSystem(sparkSession.sessionState.newHadoopConf()) override def addBatch(batchId: Long, data: DataFrame): Unit = { if (batchId <= fileLog.getLatest().map(_._1).getOrElse(-1L)) { diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala index e22a05bd3b..681addea02 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala @@ -38,7 +38,7 @@ class FileStreamSource( override val schema: StructType, dataFrameBuilder: Array[String] => DataFrame) extends Source with Logging { - private val fs = new Path(path).getFileSystem(sparkSession.sessionState.hadoopConf) + private val fs = new Path(path).getFileSystem(sparkSession.sessionState.newHadoopConf()) private val metadataLog = new HDFSMetadataLog[Seq[String]](sparkSession, metadataPath) private var maxBatchId = metadataLog.getLatest().map(_._1).getOrElse(-1L) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/HDFSMetadataLog.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/HDFSMetadataLog.scala index 99bf20c746..9fe06a6c36 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/HDFSMetadataLog.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/HDFSMetadataLog.scala @@ -211,7 +211,7 @@ class HDFSMetadataLog[T: ClassTag](sparkSession: SparkSession, path: String) } private def createFileManager(): FileManager = { - val hadoopConf = new Configuration(sparkSession.sessionState.hadoopConf) + val hadoopConf = sparkSession.sessionState.newHadoopConf() try { new FileContextManager(metadataPath, hadoopConf) } catch { diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamFileCatalog.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamFileCatalog.scala index b2bc31634c..4f699719c2 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamFileCatalog.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamFileCatalog.scala @@ -30,7 +30,7 @@ class StreamFileCatalog(sparkSession: SparkSession, path: Path) extends FileCata val metadataDirectory = new Path(path, FileStreamSink.metadataDir) logInfo(s"Reading streaming file log from $metadataDirectory") val metadataLog = new FileStreamSinkLog(sparkSession, metadataDirectory.toUri.toString) - val fs = path.getFileSystem(sparkSession.sessionState.hadoopConf) + val fs = path.getFileSystem(sparkSession.sessionState.newHadoopConf()) override def paths: Seq[Path] = path :: Nil diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStoreRDD.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStoreRDD.scala index 635bb86607..e16dda8a5b 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStoreRDD.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStoreRDD.scala @@ -46,7 +46,7 @@ class StateStoreRDD[T: ClassTag, U: ClassTag]( // A Hadoop Configuration can be about 10 KB, which is pretty big, so broadcast it private val confBroadcast = dataRDD.context.broadcast( - new SerializableConfiguration(sessionState.hadoopConf)) + new SerializableConfiguration(sessionState.newHadoopConf())) override protected def getPartitions: Array[Partition] = dataRDD.partitions diff --git a/sql/core/src/main/scala/org/apache/spark/sql/internal/RuntimeConfigImpl.scala b/sql/core/src/main/scala/org/apache/spark/sql/internal/RuntimeConfigImpl.scala deleted file mode 100644 index 137323583b..0000000000 --- a/sql/core/src/main/scala/org/apache/spark/sql/internal/RuntimeConfigImpl.scala +++ /dev/null @@ -1,73 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.spark.sql.internal - -import org.apache.hadoop.conf.Configuration - -import org.apache.spark.sql.RuntimeConfig - - -/** - * Implementation for [[RuntimeConfig]]. - */ -class RuntimeConfigImpl( - sqlConf: SQLConf = new SQLConf, - hadoopConf: Configuration = new Configuration) - extends RuntimeConfig { - - override def set(key: String, value: String): RuntimeConfig = { - sqlConf.setConfString(key, value) - this - } - - override def set(key: String, value: Boolean): RuntimeConfig = set(key, value.toString) - - override def set(key: String, value: Long): RuntimeConfig = set(key, value.toString) - - @throws[NoSuchElementException]("if the key is not set") - override def get(key: String): String = sqlConf.getConfString(key) - - override def getOption(key: String): Option[String] = { - try Option(get(key)) catch { - case _: NoSuchElementException => None - } - } - - override def unset(key: String): Unit = sqlConf.unsetConf(key) - - override def setHadoop(key: String, value: String): RuntimeConfig = hadoopConf.synchronized { - hadoopConf.set(key, value) - this - } - - @throws[NoSuchElementException]("if the key is not set") - override def getHadoop(key: String): String = hadoopConf.synchronized { - Option(hadoopConf.get(key)).getOrElse { - throw new NoSuchElementException(key) - } - } - - override def getHadoopOption(key: String): Option[String] = hadoopConf.synchronized { - Option(hadoopConf.get(key)) - } - - override def unsetHadoop(key: String): Unit = hadoopConf.synchronized { - hadoopConf.unset(key) - } - -} diff --git a/sql/core/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/core/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala index b268a4fef7..6fbf32676f 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala @@ -516,13 +516,6 @@ object SQLConf { object Deprecated { val MAPRED_REDUCE_TASKS = "mapred.reduce.tasks" - val EXTERNAL_SORT = "spark.sql.planner.externalSort" - val USE_SQL_AGGREGATE2 = "spark.sql.useAggregate2" - val TUNGSTEN_ENABLED = "spark.sql.tungsten.enabled" - val CODEGEN_ENABLED = "spark.sql.codegen" - val UNSAFE_ENABLED = "spark.sql.unsafe.enabled" - val SORTMERGE_JOIN = "spark.sql.planner.sortMergeJoin" - val PARQUET_UNSAFE_ROW_RECORD_READER_ENABLED = "spark.sql.parquet.enableUnsafeRowRecordReader" } } @@ -764,7 +757,7 @@ private[sql] class SQLConf extends Serializable with CatalystConf with Logging { settings.remove(key) } - private[spark] def unsetConf(entry: ConfigEntry[_]): Unit = { + def unsetConf(entry: ConfigEntry[_]): Unit = { settings.remove(entry.key) } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/internal/SessionState.scala b/sql/core/src/main/scala/org/apache/spark/sql/internal/SessionState.scala index 1bda572e63..63e0dc7e7e 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/internal/SessionState.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/internal/SessionState.scala @@ -48,8 +48,21 @@ private[sql] class SessionState(sparkSession: SparkSession) { * SQL-specific key-value configurations. */ lazy val conf: SQLConf = new SQLConf - lazy val hadoopConf: Configuration = { - new Configuration(sparkSession.sparkContext.hadoopConfiguration) + + def newHadoopConf(): Configuration = { + val hadoopConf = new Configuration(sparkSession.sparkContext.hadoopConfiguration) + conf.getAllConfs.foreach { case (k, v) => if (v ne null) hadoopConf.set(k, v) } + hadoopConf + } + + def newHadoopConfWithOptions(options: Map[String, String]): Configuration = { + val hadoopConf = newHadoopConf() + options.foreach { case (k, v) => + if ((v ne null) && k != "path" && k != "paths") { + hadoopConf.set(k, v) + } + } + hadoopConf } // Automatically extract `spark.sql.*` entries and put it in our SQLConf diff --git a/sql/core/src/test/scala/org/apache/spark/sql/SQLContextSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/SQLContextSuite.scala index d49cc103b5..2f62ad4850 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/SQLContextSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/SQLContextSuite.scala @@ -85,27 +85,4 @@ class SQLContextSuite extends SparkFunSuite with SharedSparkContext { assert(sqlContext.getConf("spark.sql.with.or.without.you") == "my love") } - test("Hadoop conf interaction between SQLContext and SparkContext") { - val mySpecialKey = "mai.special.key" - val mySpecialValue = "msv" - try { - sc.hadoopConfiguration.set(mySpecialKey, mySpecialValue) - val sqlContext = SQLContext.getOrCreate(sc) - val sessionState = sqlContext.sessionState - assert(sessionState.hadoopConf.get(mySpecialKey) === mySpecialValue) - assert(sqlContext.runtimeConf.getHadoop(mySpecialKey) === mySpecialValue) - // mutating hadoop conf in SQL doesn't mutate the underlying one - sessionState.hadoopConf.set(mySpecialKey, "no no no") - assert(sessionState.hadoopConf.get(mySpecialKey) === "no no no") - assert(sqlContext.runtimeConf.getHadoop(mySpecialKey) === "no no no") - assert(sc.hadoopConfiguration.get(mySpecialKey) === mySpecialValue) - sqlContext.runtimeConf.setHadoop(mySpecialKey, "yes yes yes") - assert(sessionState.hadoopConf.get(mySpecialKey) === "yes yes yes") - assert(sqlContext.runtimeConf.getHadoop(mySpecialKey) === "yes yes yes") - assert(sc.hadoopConfiguration.get(mySpecialKey) === mySpecialValue) - } finally { - sc.hadoopConfiguration.unset(mySpecialKey) - } - } - } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategySuite.scala index f73d485acf..ac2af77a6e 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategySuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategySuite.scala @@ -292,56 +292,50 @@ class FileSourceStrategySuite extends QueryTest with SharedSQLContext with Predi } test("Locality support for FileScanRDD - one file per partition") { - withHadoopConf( - "fs.file.impl" -> classOf[LocalityTestFileSystem].getName, - "fs.file.impl.disable.cache" -> "true" - ) { - withSQLConf(SQLConf.FILES_MAX_PARTITION_BYTES.key -> "10") { - val table = - createTable(files = Seq( - "file1" -> 10, - "file2" -> 10 - )) - - checkScan(table) { partitions => - val Seq(p1, p2) = partitions - assert(p1.files.length == 1) - assert(p1.files.flatMap(_.locations).length == 1) - assert(p2.files.length == 1) - assert(p2.files.flatMap(_.locations).length == 1) - - val fileScanRDD = getFileScanRDD(table) - assert(partitions.flatMap(fileScanRDD.preferredLocations).length == 2) - } + withSQLConf( + SQLConf.FILES_MAX_PARTITION_BYTES.key -> "10", + "fs.file.impl" -> classOf[LocalityTestFileSystem].getName, + "fs.file.impl.disable.cache" -> "true") { + val table = + createTable(files = Seq( + "file1" -> 10, + "file2" -> 10 + )) + + checkScan(table) { partitions => + val Seq(p1, p2) = partitions + assert(p1.files.length == 1) + assert(p1.files.flatMap(_.locations).length == 1) + assert(p2.files.length == 1) + assert(p2.files.flatMap(_.locations).length == 1) + + val fileScanRDD = getFileScanRDD(table) + assert(partitions.flatMap(fileScanRDD.preferredLocations).length == 2) } } } test("Locality support for FileScanRDD - large file") { - withHadoopConf( - "fs.file.impl" -> classOf[LocalityTestFileSystem].getName, - "fs.file.impl.disable.cache" -> "true" - ) { - withSQLConf( + withSQLConf( SQLConf.FILES_MAX_PARTITION_BYTES.key -> "10", - SQLConf.FILES_OPEN_COST_IN_BYTES.key -> "0" - ) { - val table = - createTable(files = Seq( - "file1" -> 15, - "file2" -> 5 - )) - - checkScan(table) { partitions => - val Seq(p1, p2) = partitions - assert(p1.files.length == 1) - assert(p1.files.flatMap(_.locations).length == 1) - assert(p2.files.length == 2) - assert(p2.files.flatMap(_.locations).length == 2) - - val fileScanRDD = getFileScanRDD(table) - assert(partitions.flatMap(fileScanRDD.preferredLocations).length == 3) - } + SQLConf.FILES_OPEN_COST_IN_BYTES.key -> "0", + "fs.file.impl" -> classOf[LocalityTestFileSystem].getName, + "fs.file.impl.disable.cache" -> "true") { + val table = + createTable(files = Seq( + "file1" -> 15, + "file2" -> 5 + )) + + checkScan(table) { partitions => + val Seq(p1, p2) = partitions + assert(p1.files.length == 1) + assert(p1.files.flatMap(_.locations).length == 1) + assert(p2.files.length == 2) + assert(p2.files.flatMap(_.locations).length == 2) + + val fileScanRDD = getFileScanRDD(table) + assert(partitions.flatMap(fileScanRDD.preferredLocations).length == 3) } } } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala index 9baae80f15..ceda920ddc 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala @@ -430,42 +430,37 @@ class CSVSuite extends QueryTest with SharedSQLContext with SQLTestUtils { } test("SPARK-13543 Write the output as uncompressed via option()") { - val clonedConf = new Configuration(hadoopConfiguration) - hadoopConfiguration.set("mapreduce.output.fileoutputformat.compress", "true") - hadoopConfiguration - .set("mapreduce.output.fileoutputformat.compress.type", CompressionType.BLOCK.toString) - hadoopConfiguration - .set("mapreduce.output.fileoutputformat.compress.codec", classOf[GzipCodec].getName) - hadoopConfiguration.set("mapreduce.map.output.compress", "true") - hadoopConfiguration.set("mapreduce.map.output.compress.codec", classOf[GzipCodec].getName) + val extraOptions = Map( + "mapreduce.output.fileoutputformat.compress" -> "true", + "mapreduce.output.fileoutputformat.compress.type" -> CompressionType.BLOCK.toString, + "mapreduce.map.output.compress" -> "true", + "mapreduce.map.output.compress.codec" -> classOf[GzipCodec].getName + ) withTempDir { dir => - try { - val csvDir = new File(dir, "csv").getCanonicalPath - val cars = sqlContext.read - .format("csv") - .option("header", "true") - .load(testFile(carsFile)) - - cars.coalesce(1).write - .format("csv") - .option("header", "true") - .option("compression", "none") - .save(csvDir) - - val compressedFiles = new File(csvDir).listFiles() - assert(compressedFiles.exists(!_.getName.endsWith(".csv.gz"))) - - val carsCopy = sqlContext.read - .format("csv") - .option("header", "true") - .load(csvDir) - - verifyCars(carsCopy, withHeader = true) - } finally { - // Hadoop 1 doesn't have `Configuration.unset` - hadoopConfiguration.clear() - clonedConf.asScala.foreach(entry => hadoopConfiguration.set(entry.getKey, entry.getValue)) - } + val csvDir = new File(dir, "csv").getCanonicalPath + val cars = sqlContext.read + .format("csv") + .option("header", "true") + .options(extraOptions) + .load(testFile(carsFile)) + + cars.coalesce(1).write + .format("csv") + .option("header", "true") + .option("compression", "none") + .options(extraOptions) + .save(csvDir) + + val compressedFiles = new File(csvDir).listFiles() + assert(compressedFiles.exists(!_.getName.endsWith(".csv.gz"))) + + val carsCopy = sqlContext.read + .format("csv") + .option("header", "true") + .options(extraOptions) + .load(csvDir) + + verifyCars(carsCopy, withHeader = true) } } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala index e5588bec4b..b1279abd63 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/JsonSuite.scala @@ -1507,23 +1507,11 @@ class JsonSuite extends QueryTest with SharedSQLContext with TestJsonData { df.write.json(path + "/p=2") assert(sqlContext.read.json(path).count() === 4) - val clonedConf = new Configuration(hadoopConfiguration) - try { - // Setting it twice as the name of the propery has changed between hadoop versions. - hadoopConfiguration.setClass( - "mapred.input.pathFilter.class", - classOf[TestFileFilter], - classOf[PathFilter]) - hadoopConfiguration.setClass( - "mapreduce.input.pathFilter.class", - classOf[TestFileFilter], - classOf[PathFilter]) - assert(sqlContext.read.json(path).count() === 2) - } finally { - // Hadoop 1 doesn't have `Configuration.unset` - hadoopConfiguration.clear() - clonedConf.asScala.foreach(entry => hadoopConfiguration.set(entry.getKey, entry.getValue)) - } + val extraOptions = Map( + "mapred.input.pathFilter.class" -> classOf[TestFileFilter].getName, + "mapreduce.input.pathFilter.class" -> classOf[TestFileFilter].getName + ) + assert(sqlContext.read.options(extraOptions).json(path).count() === 2) } } @@ -1609,45 +1597,40 @@ class JsonSuite extends QueryTest with SharedSQLContext with TestJsonData { } test("SPARK-13543 Write the output as uncompressed via option()") { - val clonedConf = new Configuration(hadoopConfiguration) - hadoopConfiguration.set("mapreduce.output.fileoutputformat.compress", "true") - hadoopConfiguration - .set("mapreduce.output.fileoutputformat.compress.type", CompressionType.BLOCK.toString) - hadoopConfiguration - .set("mapreduce.output.fileoutputformat.compress.codec", classOf[GzipCodec].getName) - hadoopConfiguration.set("mapreduce.map.output.compress", "true") - hadoopConfiguration.set("mapreduce.map.output.compress.codec", classOf[GzipCodec].getName) + val extraOptions = Map[String, String]( + "mapreduce.output.fileoutputformat.compress" -> "true", + "mapreduce.output.fileoutputformat.compress.type" -> CompressionType.BLOCK.toString, + "mapreduce.output.fileoutputformat.compress.codec" -> classOf[GzipCodec].getName, + "mapreduce.map.output.compress" -> "true", + "mapreduce.map.output.compress.codec" -> classOf[GzipCodec].getName + ) withTempDir { dir => - try { - val dir = Utils.createTempDir() - dir.delete() - - val path = dir.getCanonicalPath - primitiveFieldAndType.map(record => record.replaceAll("\n", " ")).saveAsTextFile(path) - - val jsonDF = sqlContext.read.json(path) - val jsonDir = new File(dir, "json").getCanonicalPath - jsonDF.coalesce(1).write - .format("json") - .option("compression", "none") - .save(jsonDir) - - val compressedFiles = new File(jsonDir).listFiles() - assert(compressedFiles.exists(!_.getName.endsWith(".json.gz"))) - - val jsonCopy = sqlContext.read - .format("json") - .load(jsonDir) - - assert(jsonCopy.count == jsonDF.count) - val jsonCopySome = jsonCopy.selectExpr("string", "long", "boolean") - val jsonDFSome = jsonDF.selectExpr("string", "long", "boolean") - checkAnswer(jsonCopySome, jsonDFSome) - } finally { - // Hadoop 1 doesn't have `Configuration.unset` - hadoopConfiguration.clear() - clonedConf.asScala.foreach(entry => hadoopConfiguration.set(entry.getKey, entry.getValue)) - } + val dir = Utils.createTempDir() + dir.delete() + + val path = dir.getCanonicalPath + primitiveFieldAndType.map(record => record.replaceAll("\n", " ")).saveAsTextFile(path) + + val jsonDF = sqlContext.read.json(path) + val jsonDir = new File(dir, "json").getCanonicalPath + jsonDF.coalesce(1).write + .format("json") + .option("compression", "none") + .options(extraOptions) + .save(jsonDir) + + val compressedFiles = new File(jsonDir).listFiles() + assert(compressedFiles.exists(!_.getName.endsWith(".json.gz"))) + + val jsonCopy = sqlContext.read + .format("json") + .options(extraOptions) + .load(jsonDir) + + assert(jsonCopy.count == jsonDF.count) + val jsonCopySome = jsonCopy.selectExpr("string", "long", "boolean") + val jsonDFSome = jsonDF.selectExpr("string", "long", "boolean") + checkAnswer(jsonCopySome, jsonDFSome) } } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetCompatibilityTest.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetCompatibilityTest.scala index 4217c81ff3..45cc6810d4 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetCompatibilityTest.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetCompatibilityTest.scala @@ -38,14 +38,15 @@ private[sql] abstract class ParquetCompatibilityTest extends QueryTest with Parq } protected def readParquetSchema(path: String, pathFilter: Path => Boolean): MessageType = { + val hadoopConf = sqlContext.sessionState.newHadoopConf() val fsPath = new Path(path) - val fs = fsPath.getFileSystem(hadoopConfiguration) + val fs = fsPath.getFileSystem(hadoopConf) val parquetFiles = fs.listStatus(fsPath, new PathFilter { override def accept(path: Path): Boolean = pathFilter(path) }).toSeq.asJava val footers = - ParquetFileReader.readAllFootersInParallel(hadoopConfiguration, parquetFiles, true) + ParquetFileReader.readAllFootersInParallel(hadoopConf, parquetFiles, true) footers.asScala.head.getParquetMetadata.getFileMetaData.getSchema } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala index 0aab36ae38..32fe5ba127 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala @@ -113,7 +113,7 @@ class ParquetIOSuite extends QueryTest with ParquetTest with SharedSQLContext { withTempPath { location => val path = new Path(location.getCanonicalPath) - val conf = new Configuration(sqlContext.sessionState.hadoopConf) + val conf = sqlContext.sessionState.newHadoopConf() writeMetadata(parquetSchema, path, conf) readParquetFile(path.toString)(df => { val sparkTypes = df.schema.map(_.dataType) @@ -250,7 +250,7 @@ class ParquetIOSuite extends QueryTest with ParquetTest with SharedSQLContext { withTempPath { location => val path = new Path(location.getCanonicalPath) - val conf = new Configuration(sqlContext.sessionState.hadoopConf) + val conf = sqlContext.sessionState.newHadoopConf() writeMetadata(parquetSchema, path, conf) val errorMessage = intercept[Throwable] { sqlContext.read.parquet(path.toString).printSchema() @@ -271,7 +271,7 @@ class ParquetIOSuite extends QueryTest with ParquetTest with SharedSQLContext { withTempPath { location => val path = new Path(location.getCanonicalPath) - val conf = new Configuration(sqlContext.sessionState.hadoopConf) + val conf = sqlContext.sessionState.newHadoopConf() writeMetadata(parquetSchema, path, conf) val sparkTypes = sqlContext.read.parquet(path.toString).schema.map(_.dataType) assert(sparkTypes === expectedSparkTypes) @@ -279,9 +279,10 @@ class ParquetIOSuite extends QueryTest with ParquetTest with SharedSQLContext { } test("compression codec") { + val hadoopConf = sqlContext.sessionState.newHadoopConf() def compressionCodecFor(path: String, codecName: String): String = { val codecs = for { - footer <- readAllFootersWithoutSummaryFiles(new Path(path), hadoopConfiguration) + footer <- readAllFootersWithoutSummaryFiles(new Path(path), hadoopConf) block <- footer.getParquetMetadata.getBlocks.asScala column <- block.getColumns.asScala } yield column.getCodec.name() @@ -350,17 +351,18 @@ class ParquetIOSuite extends QueryTest with ParquetTest with SharedSQLContext { } test("write metadata") { + val hadoopConf = sqlContext.sessionState.newHadoopConf() withTempPath { file => val path = new Path(file.toURI.toString) - val fs = FileSystem.getLocal(hadoopConfiguration) + val fs = FileSystem.getLocal(hadoopConf) val schema = StructType.fromAttributes(ScalaReflection.attributesFor[(Int, String)]) - writeMetadata(schema, path, hadoopConfiguration) + writeMetadata(schema, path, hadoopConf) assert(fs.exists(new Path(path, ParquetFileWriter.PARQUET_COMMON_METADATA_FILE))) assert(fs.exists(new Path(path, ParquetFileWriter.PARQUET_METADATA_FILE))) val expectedSchema = new CatalystSchemaConverter().convert(schema) - val actualSchema = readFooter(path, hadoopConfiguration).getFileMetaData.getSchema + val actualSchema = readFooter(path, hadoopConf).getFileMetaData.getSchema actualSchema.checkContains(expectedSchema) expectedSchema.checkContains(actualSchema) @@ -431,7 +433,7 @@ class ParquetIOSuite extends QueryTest with ParquetTest with SharedSQLContext { withTempPath { location => val extraMetadata = Map(CatalystReadSupport.SPARK_METADATA_KEY -> sparkSchema.toString) val path = new Path(location.getCanonicalPath) - val conf = new Configuration(sqlContext.sessionState.hadoopConf) + val conf = sqlContext.sessionState.newHadoopConf() writeMetadata(parquetSchema, path, conf, extraMetadata) readParquetFile(path.toString) { df => @@ -446,26 +448,16 @@ class ParquetIOSuite extends QueryTest with ParquetTest with SharedSQLContext { } test("SPARK-8121: spark.sql.parquet.output.committer.class shouldn't be overridden") { + val extraOptions = Map( + SQLConf.OUTPUT_COMMITTER_CLASS.key -> classOf[ParquetOutputCommitter].getCanonicalName, + "spark.sql.parquet.output.committer.class" -> + classOf[JobCommitFailureParquetOutputCommitter].getCanonicalName + ) withTempPath { dir => - val clonedConf = new Configuration(hadoopConfiguration) - - hadoopConfiguration.set( - SQLConf.OUTPUT_COMMITTER_CLASS.key, classOf[ParquetOutputCommitter].getCanonicalName) - - hadoopConfiguration.set( - "spark.sql.parquet.output.committer.class", - classOf[JobCommitFailureParquetOutputCommitter].getCanonicalName) - - try { - val message = intercept[SparkException] { - sqlContext.range(0, 1).write.parquet(dir.getCanonicalPath) - }.getCause.getMessage - assert(message === "Intentional exception for testing purposes") - } finally { - // Hadoop 1 doesn't have `Configuration.unset` - hadoopConfiguration.clear() - clonedConf.asScala.foreach(entry => hadoopConfiguration.set(entry.getKey, entry.getValue)) - } + val message = intercept[SparkException] { + sqlContext.range(0, 1).write.options(extraOptions).parquet(dir.getCanonicalPath) + }.getCause.getMessage + assert(message === "Intentional exception for testing purposes") } } @@ -482,36 +474,29 @@ class ParquetIOSuite extends QueryTest with ParquetTest with SharedSQLContext { } test("SPARK-7837 Do not close output writer twice when commitTask() fails") { - val clonedConf = new Configuration(hadoopConfiguration) - // Using a output committer that always fail when committing a task, so that both // `commitTask()` and `abortTask()` are invoked. - hadoopConfiguration.set( - "spark.sql.parquet.output.committer.class", - classOf[TaskCommitFailureParquetOutputCommitter].getCanonicalName) + val extraOptions = Map[String, String]( + "spark.sql.parquet.output.committer.class" -> + classOf[TaskCommitFailureParquetOutputCommitter].getCanonicalName + ) - try { - // Before fixing SPARK-7837, the following code results in an NPE because both - // `commitTask()` and `abortTask()` try to close output writers. + // Before fixing SPARK-7837, the following code results in an NPE because both + // `commitTask()` and `abortTask()` try to close output writers. - withTempPath { dir => - val m1 = intercept[SparkException] { - sqlContext.range(1).coalesce(1).write.parquet(dir.getCanonicalPath) - }.getCause.getMessage - assert(m1.contains("Intentional exception for testing purposes")) - } + withTempPath { dir => + val m1 = intercept[SparkException] { + sqlContext.range(1).coalesce(1).write.options(extraOptions).parquet(dir.getCanonicalPath) + }.getCause.getMessage + assert(m1.contains("Intentional exception for testing purposes")) + } - withTempPath { dir => - val m2 = intercept[SparkException] { - val df = sqlContext.range(1).select('id as 'a, 'id as 'b).coalesce(1) - df.write.partitionBy("a").parquet(dir.getCanonicalPath) - }.getCause.getMessage - assert(m2.contains("Intentional exception for testing purposes")) - } - } finally { - // Hadoop 1 doesn't have `Configuration.unset` - hadoopConfiguration.clear() - clonedConf.asScala.foreach(entry => hadoopConfiguration.set(entry.getKey, entry.getValue)) + withTempPath { dir => + val m2 = intercept[SparkException] { + val df = sqlContext.range(1).select('id as 'a, 'id as 'b).coalesce(1) + df.write.partitionBy("a").options(extraOptions).parquet(dir.getCanonicalPath) + }.getCause.getMessage + assert(m2.contains("Intentional exception for testing purposes")) } } @@ -519,31 +504,27 @@ class ParquetIOSuite extends QueryTest with ParquetTest with SharedSQLContext { // For dictionary encoding, Parquet changes the encoding types according to its writer // version. So, this test checks one of the encoding types in order to ensure that // the file is written with writer version2. + val extraOptions = Map[String, String]( + // Write a Parquet file with writer version2. + ParquetOutputFormat.WRITER_VERSION -> ParquetProperties.WriterVersion.PARQUET_2_0.toString, + // By default, dictionary encoding is enabled from Parquet 1.2.0 but + // it is enabled just in case. + ParquetOutputFormat.ENABLE_DICTIONARY -> "true" + ) + + val hadoopConf = sqlContext.sessionState.newHadoopConfWithOptions(extraOptions) + withTempPath { dir => - val clonedConf = new Configuration(hadoopConfiguration) - try { - // Write a Parquet file with writer version2. - hadoopConfiguration.set(ParquetOutputFormat.WRITER_VERSION, - ParquetProperties.WriterVersion.PARQUET_2_0.toString) - - // By default, dictionary encoding is enabled from Parquet 1.2.0 but - // it is enabled just in case. - hadoopConfiguration.setBoolean(ParquetOutputFormat.ENABLE_DICTIONARY, true) - val path = s"${dir.getCanonicalPath}/part-r-0.parquet" - sqlContext.range(1 << 16).selectExpr("(id % 4) AS i") - .coalesce(1).write.mode("overwrite").parquet(path) - - val blockMetadata = readFooter(new Path(path), hadoopConfiguration).getBlocks.asScala.head - val columnChunkMetadata = blockMetadata.getColumns.asScala.head - - // If the file is written with version2, this should include - // Encoding.RLE_DICTIONARY type. For version1, it is Encoding.PLAIN_DICTIONARY - assert(columnChunkMetadata.getEncodings.contains(Encoding.RLE_DICTIONARY)) - } finally { - // Manually clear the hadoop configuration for other tests. - hadoopConfiguration.clear() - clonedConf.asScala.foreach(entry => hadoopConfiguration.set(entry.getKey, entry.getValue)) - } + val path = s"${dir.getCanonicalPath}/part-r-0.parquet" + sqlContext.range(1 << 16).selectExpr("(id % 4) AS i") + .coalesce(1).write.options(extraOptions).mode("overwrite").parquet(path) + + val blockMetadata = readFooter(new Path(path), hadoopConf).getBlocks.asScala.head + val columnChunkMetadata = blockMetadata.getColumns.asScala.head + + // If the file is written with version2, this should include + // Encoding.RLE_DICTIONARY type. For version1, it is Encoding.PLAIN_DICTIONARY + assert(columnChunkMetadata.getEncodings.contains(Encoding.RLE_DICTIONARY)) } } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/text/TextSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/text/TextSuite.scala index 47330f1db3..923c0b350e 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/text/TextSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/text/TextSuite.scala @@ -87,28 +87,22 @@ class TextSuite extends QueryTest with SharedSQLContext { } test("SPARK-13543 Write the output as uncompressed via option()") { - val clonedConf = new Configuration(hadoopConfiguration) - hadoopConfiguration.set("mapreduce.output.fileoutputformat.compress", "true") - hadoopConfiguration - .set("mapreduce.output.fileoutputformat.compress.type", CompressionType.BLOCK.toString) - hadoopConfiguration - .set("mapreduce.output.fileoutputformat.compress.codec", classOf[GzipCodec].getName) - hadoopConfiguration.set("mapreduce.map.output.compress", "true") - hadoopConfiguration.set("mapreduce.map.output.compress.codec", classOf[GzipCodec].getName) + val extraOptions = Map[String, String]( + "mapreduce.output.fileoutputformat.compress" -> "true", + "mapreduce.output.fileoutputformat.compress.type" -> CompressionType.BLOCK.toString, + "mapreduce.map.output.compress" -> "true", + "mapreduce.output.fileoutputformat.compress.codec" -> classOf[GzipCodec].getName, + "mapreduce.map.output.compress.codec" -> classOf[GzipCodec].getName + ) withTempDir { dir => - try { - val testDf = sqlContext.read.text(testFile) - val tempDir = Utils.createTempDir() - val tempDirPath = tempDir.getAbsolutePath - testDf.write.option("compression", "none").mode(SaveMode.Overwrite).text(tempDirPath) - val compressedFiles = new File(tempDirPath).listFiles() - assert(compressedFiles.exists(!_.getName.endsWith(".txt.gz"))) - verifyFrame(sqlContext.read.text(tempDirPath).toDF()) - } finally { - // Hadoop 1 doesn't have `Configuration.unset` - hadoopConfiguration.clear() - clonedConf.asScala.foreach(entry => hadoopConfiguration.set(entry.getKey, entry.getValue)) - } + val testDf = sqlContext.read.text(testFile) + val tempDir = Utils.createTempDir() + val tempDirPath = tempDir.getAbsolutePath + testDf.write.option("compression", "none") + .options(extraOptions).mode(SaveMode.Overwrite).text(tempDirPath) + val compressedFiles = new File(tempDirPath).listFiles() + assert(compressedFiles.exists(!_.getName.endsWith(".txt.gz"))) + verifyFrame(sqlContext.read.options(extraOptions).text(tempDirPath).toDF()) } } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/FileStreamSinkLogSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/FileStreamSinkLogSuite.scala index df127d958e..7b413dda1e 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/FileStreamSinkLogSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/FileStreamSinkLogSuite.scala @@ -217,7 +217,7 @@ class FileStreamSinkLogSuite extends SparkFunSuite with SharedSQLContext { SQLConf.FILE_SINK_LOG_COMPACT_INTERVAL.key -> "3", SQLConf.FILE_SINK_LOG_CLEANUP_DELAY.key -> "0") { withFileStreamSinkLog { sinkLog => - val fs = sinkLog.metadataPath.getFileSystem(sqlContext.sessionState.hadoopConf) + val fs = sinkLog.metadataPath.getFileSystem(sqlContext.sessionState.newHadoopConf()) def listBatchFiles(): Set[String] = { fs.listStatus(sinkLog.metadataPath).map(_.getPath.getName).filter { fileName => diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/HDFSMetadataLogSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/HDFSMetadataLogSuite.scala index 129b5a8c36..5f92c5bb9b 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/HDFSMetadataLogSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/HDFSMetadataLogSuite.scala @@ -82,7 +82,7 @@ class HDFSMetadataLogSuite extends SparkFunSuite with SharedSQLContext { } testQuietly("HDFSMetadataLog: fallback from FileContext to FileSystem") { - sqlContext.sessionState.hadoopConf.set( + sqlContext.conf.setConfString( s"fs.$scheme.impl", classOf[FakeFileSystem].getName) withTempDir { temp => diff --git a/sql/core/src/test/scala/org/apache/spark/sql/internal/RuntimeConfigSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/internal/RuntimeConfigSuite.scala index f809e01169..a629b73ac0 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/internal/RuntimeConfigSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/internal/RuntimeConfigSuite.scala @@ -22,7 +22,7 @@ import org.apache.spark.sql.RuntimeConfig class RuntimeConfigSuite extends SparkFunSuite { - private def newConf(): RuntimeConfig = new RuntimeConfigImpl + private def newConf(): RuntimeConfig = new RuntimeConfig test("set and get") { val conf = newConf() @@ -54,33 +54,4 @@ class RuntimeConfigSuite extends SparkFunSuite { conf.get("k1") } } - - test("set and get hadoop configuration") { - val conf = newConf() - conf - .setHadoop("k1", "v1") - .setHadoop("k2", "v2") - - assert(conf.getHadoop("k1") == "v1") - assert(conf.getHadoop("k2") == "v2") - - intercept[NoSuchElementException] { - conf.get("notset") - } - } - - test("getHadoopOption") { - val conf = newConf().setHadoop("k1", "v1") - assert(conf.getHadoopOption("k1") == Some("v1")) - assert(conf.getHadoopOption("notset") == None) - } - - test("unsetHadoop") { - val conf = newConf().setHadoop("k1", "v1") - assert(conf.getHadoop("k1") == "v1") - conf.unsetHadoop("k1") - intercept[NoSuchElementException] { - conf.getHadoop("k1") - } - } } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/test/SQLTestUtils.scala b/sql/core/src/test/scala/org/apache/spark/sql/test/SQLTestUtils.scala index 5577c9f3ee..ffb206af0e 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/test/SQLTestUtils.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/test/SQLTestUtils.scala @@ -85,34 +85,6 @@ private[sql] trait SQLTestUtils } /** - * The Hadoop configuration used by the active [[SQLContext]]. - */ - protected def hadoopConfiguration: Configuration = { - sqlContext.sessionState.hadoopConf - } - - /** - * Sets all Hadoop configurations specified in `pairs`, calls `f`, and then restore all Hadoop - * configurations. - */ - protected def withHadoopConf(pairs: (String, String)*)(f: => Unit): Unit = { - val (keys, _) = pairs.unzip - val originalValues = keys.map(key => Option(hadoopConfiguration.get(key))) - - try { - pairs.foreach { case (key, value) => - hadoopConfiguration.set(key, value) - } - f - } finally { - keys.zip(originalValues).foreach { - case (key, Some(value)) => hadoopConfiguration.set(key, value) - case (key, None) => hadoopConfiguration.unset(key) - } - } - } - - /** * Sets all SQL configurations specified in `pairs`, calls `f`, and then restore all SQL * configurations. * @@ -164,7 +136,7 @@ private[sql] trait SQLTestUtils } finally { // If the test failed part way, we don't want to mask the failure by failing to remove // temp tables that never got created. - try functions.foreach { case (functionName, isTemporary) => + functions.foreach { case (functionName, isTemporary) => val withTemporary = if (isTemporary) "TEMPORARY" else "" sqlContext.sql(s"DROP $withTemporary FUNCTION IF EXISTS $functionName") assert( diff --git a/sql/core/src/test/scala/org/apache/spark/sql/test/TestSQLContext.scala b/sql/core/src/test/scala/org/apache/spark/sql/test/TestSQLContext.scala index 9799c6d42b..5ef80b9aa3 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/test/TestSQLContext.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/test/TestSQLContext.scala @@ -18,8 +18,8 @@ package org.apache.spark.sql.test import org.apache.spark.{SparkConf, SparkContext} -import org.apache.spark.sql.{RuntimeConfig, SparkSession, SQLContext} -import org.apache.spark.sql.internal.{RuntimeConfigImpl, SessionState, SQLConf} +import org.apache.spark.sql.{SparkSession, SQLContext} +import org.apache.spark.sql.internal.{SessionState, SQLConf} /** * A special [[SQLContext]] prepared for testing. diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala index c4db4f307c..58c10b7b1d 100644 --- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala +++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala @@ -549,7 +549,7 @@ private[hive] class MetaStoreFileCatalog( Some(partitionSpecFromHive.partitionColumns)) { override def getStatus(path: Path): Array[FileStatus] = { - val fs = path.getFileSystem(sparkSession.sessionState.hadoopConf) + val fs = path.getFileSystem(sparkSession.sessionState.newHadoopConf()) fs.listStatus(path) } diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcRelation.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcRelation.scala index 4f81967a5b..d6a847f3ba 100644 --- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcRelation.scala +++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcRelation.scala @@ -57,7 +57,7 @@ private[sql] class DefaultSource files: Seq[FileStatus]): Option[StructType] = { OrcFileOperator.readSchema( files.map(_.getPath.toUri.toString), - Some(new Configuration(sparkSession.sessionState.hadoopConf)) + Some(sparkSession.sessionState.newHadoopConf()) ) } @@ -278,7 +278,7 @@ private[orc] case class OrcTableScan( with HiveInspectors { def execute(): RDD[InternalRow] = { - val job = Job.getInstance(new Configuration(sparkSession.sessionState.hadoopConf)) + val job = Job.getInstance(sparkSession.sessionState.newHadoopConf()) val conf = job.getConfiguration // Tries to push down filters if ORC filter push-down is enabled diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/test/TestHive.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/test/TestHive.scala index f74e5cd6f5..1d8f24cb27 100644 --- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/test/TestHive.scala +++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/test/TestHive.scala @@ -34,7 +34,7 @@ import org.apache.hadoop.hive.serde2.`lazy`.LazySimpleSerDe import org.apache.spark.{SparkConf, SparkContext} import org.apache.spark.internal.Logging import org.apache.spark.internal.config.CATALOG_IMPLEMENTATION -import org.apache.spark.sql.{RuntimeConfig, SparkSession, SQLContext} +import org.apache.spark.sql.{SparkSession, SQLContext} import org.apache.spark.sql.catalyst.analysis._ import org.apache.spark.sql.catalyst.analysis.FunctionRegistry.FunctionBuilder import org.apache.spark.sql.catalyst.expressions.ExpressionInfo @@ -43,7 +43,7 @@ import org.apache.spark.sql.execution.QueryExecution import org.apache.spark.sql.execution.command.{CacheTableCommand, HiveNativeCommand} import org.apache.spark.sql.hive._ import org.apache.spark.sql.hive.client.HiveClient -import org.apache.spark.sql.internal.{RuntimeConfigImpl, SQLConf} +import org.apache.spark.sql.internal.SQLConf import org.apache.spark.util.{ShutdownHookManager, Utils} // SPARK-3729: Test key required to check for initialization errors with config. diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/MetastoreDataSourcesSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/MetastoreDataSourcesSuite.scala index 31ba735708..b21ca4f26e 100644 --- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/MetastoreDataSourcesSuite.scala +++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/MetastoreDataSourcesSuite.scala @@ -374,7 +374,7 @@ class MetastoreDataSourcesSuite extends QueryTest with SQLTestUtils with TestHiv val expectedPath = sessionState.catalog.hiveDefaultTableFilePath(TableIdentifier("ctasJsonTable")) val filesystemPath = new Path(expectedPath) - val fs = filesystemPath.getFileSystem(sqlContext.sessionState.hadoopConf) + val fs = filesystemPath.getFileSystem(sqlContext.sessionState.newHadoopConf()) if (fs.exists(filesystemPath)) fs.delete(filesystemPath, true) // It is a managed table when we do not specify the location. diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveCommandSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveCommandSuite.scala index a4c6d3c185..8b3f2d1a0c 100644 --- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveCommandSuite.scala +++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveCommandSuite.scala @@ -266,14 +266,6 @@ class HiveCommandSuite extends QueryTest with SQLTestUtils with TestHiveSingleto intercept[AnalysisException] { sql(s"""LOAD DATA LOCAL INPATH "$incorrectUri" INTO TABLE non_part_table""") } - - // Unset default URI Scheme and Authority: throw exception - val originalFsName = hiveContext.sessionState.hadoopConf.get("fs.default.name") - hiveContext.sessionState.hadoopConf.unset("fs.default.name") - intercept[AnalysisException] { - sql(s"""LOAD DATA INPATH "$testData" INTO TABLE non_part_table""") - } - hiveContext.sessionState.hadoopConf.set("fs.default.name", originalFsName) } } diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala index e23272de85..687a4a7e51 100644 --- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala +++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala @@ -36,7 +36,7 @@ class HiveDDLSuite extends QueryTest with SQLTestUtils with TestHiveSingleton { val expectedTablePath = hiveContext.sessionState.catalog.hiveDefaultTableFilePath(tableIdentifier) val filesystemPath = new Path(expectedTablePath) - val fs = filesystemPath.getFileSystem(hiveContext.sessionState.hadoopConf) + val fs = filesystemPath.getFileSystem(hiveContext.sessionState.newHadoopConf()) fs.exists(filesystemPath) } diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/orc/OrcHadoopFsRelationSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/orc/OrcHadoopFsRelationSuite.scala index f11c055fb9..b97da1ffdc 100644 --- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/orc/OrcHadoopFsRelationSuite.scala +++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/orc/OrcHadoopFsRelationSuite.scala @@ -94,7 +94,7 @@ class OrcHadoopFsRelationSuite extends HadoopFsRelationTest { .orc(path) // Check if this is compressed as ZLIB. - val conf = sqlContext.sessionState.hadoopConf + val conf = sqlContext.sessionState.newHadoopConf() val fs = FileSystem.getLocal(conf) val maybeOrcFile = new File(path).listFiles().find(_.getName.endsWith(".zlib.orc")) assert(maybeOrcFile.isDefined) diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/orc/OrcQuerySuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/orc/OrcQuerySuite.scala index 0a0cdf60e8..62d9f5a339 100644 --- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/orc/OrcQuerySuite.scala +++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/orc/OrcQuerySuite.scala @@ -181,7 +181,7 @@ class OrcQuerySuite extends QueryTest with BeforeAndAfterAll with OrcTest { // Following codec is supported in hive-0.13.1, ignore it now ignore("Other compression options for writing to an ORC file - 0.13.1 and above") { val data = (1 to 100).map(i => (i, s"val_$i")) - val conf = sqlContext.sessionState.hadoopConf + val conf = sqlContext.sessionState.newHadoopConf() conf.set(ConfVars.HIVE_ORC_DEFAULT_COMPRESS.varname, "SNAPPY") withOrcFile(data) { file => diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/sources/hadoopFsRelationSuites.scala b/sql/hive/src/test/scala/org/apache/spark/sql/sources/HadoopFsRelationTest.scala index 3b16468e76..67b403a9bd 100644 --- a/sql/hive/src/test/scala/org/apache/spark/sql/sources/hadoopFsRelationSuites.scala +++ b/sql/hive/src/test/scala/org/apache/spark/sql/sources/HadoopFsRelationTest.scala @@ -17,10 +17,8 @@ package org.apache.spark.sql.sources -import scala.collection.JavaConverters._ import scala.util.Random -import org.apache.hadoop.conf.Configuration import org.apache.hadoop.fs.Path import org.apache.hadoop.mapreduce.{JobContext, TaskAttemptContext} import org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter @@ -117,56 +115,55 @@ abstract class HadoopFsRelationTest extends QueryTest with SQLTestUtils with Tes new UDT.MyDenseVectorUDT() ).filter(supportsDataType) - try { - for (dataType <- supportedDataTypes) { - for (parquetDictionaryEncodingEnabled <- Seq(true, false)) { - test(s"test all data types - $dataType with parquet.enable.dictionary = " + - s"$parquetDictionaryEncodingEnabled") { - - hadoopConfiguration.setBoolean("parquet.enable.dictionary", - parquetDictionaryEncodingEnabled) - - withTempPath { file => - val path = file.getCanonicalPath - - val dataGenerator = RandomDataGenerator.forType( - dataType = dataType, - nullable = true, - new Random(System.nanoTime()) - ).getOrElse { - fail(s"Failed to create data generator for schema $dataType") - } - - // Create a DF for the schema with random data. The index field is used to sort the - // DataFrame. This is a workaround for SPARK-10591. - val schema = new StructType() - .add("index", IntegerType, nullable = false) - .add("col", dataType, nullable = true) - val rdd = - sqlContext.sparkContext.parallelize((1 to 10).map(i => Row(i, dataGenerator()))) - val df = sqlContext.createDataFrame(rdd, schema).orderBy("index").coalesce(1) - - df.write - .mode("overwrite") - .format(dataSourceName) - .option("dataSchema", df.schema.json) - .save(path) - - val loadedDF = sqlContext - .read - .format(dataSourceName) - .option("dataSchema", df.schema.json) - .schema(df.schema) - .load(path) - .orderBy("index") - - checkAnswer(loadedDF, df) + for (dataType <- supportedDataTypes) { + for (parquetDictionaryEncodingEnabled <- Seq(true, false)) { + test(s"test all data types - $dataType with parquet.enable.dictionary = " + + s"$parquetDictionaryEncodingEnabled") { + + val extraOptions = Map[String, String]( + "parquet.enable.dictionary" -> parquetDictionaryEncodingEnabled.toString + ) + + withTempPath { file => + val path = file.getCanonicalPath + + val dataGenerator = RandomDataGenerator.forType( + dataType = dataType, + nullable = true, + new Random(System.nanoTime()) + ).getOrElse { + fail(s"Failed to create data generator for schema $dataType") } + + // Create a DF for the schema with random data. The index field is used to sort the + // DataFrame. This is a workaround for SPARK-10591. + val schema = new StructType() + .add("index", IntegerType, nullable = false) + .add("col", dataType, nullable = true) + val rdd = + sqlContext.sparkContext.parallelize((1 to 10).map(i => Row(i, dataGenerator()))) + val df = sqlContext.createDataFrame(rdd, schema).orderBy("index").coalesce(1) + + df.write + .mode("overwrite") + .format(dataSourceName) + .option("dataSchema", df.schema.json) + .options(extraOptions) + .save(path) + + val loadedDF = sqlContext + .read + .format(dataSourceName) + .option("dataSchema", df.schema.json) + .schema(df.schema) + .options(extraOptions) + .load(path) + .orderBy("index") + + checkAnswer(loadedDF, df) } } } - } finally { - hadoopConfiguration.unset("parquet.enable.dictionary") } test("save()/load() - non-partitioned table - Overwrite") { @@ -209,7 +206,7 @@ abstract class HadoopFsRelationTest extends QueryTest with SQLTestUtils with Tes testDF.write.mode(SaveMode.Ignore).format(dataSourceName).save(file.getCanonicalPath) val path = new Path(file.getCanonicalPath) - val fs = path.getFileSystem(sqlContext.sessionState.hadoopConf) + val fs = path.getFileSystem(sqlContext.sessionState.newHadoopConf()) assert(fs.listStatus(path).isEmpty) } } @@ -510,7 +507,7 @@ abstract class HadoopFsRelationTest extends QueryTest with SQLTestUtils with Tes s"${file.getCanonicalFile}/p1=2/p2=bar" ).map { p => val path = new Path(p) - val fs = path.getFileSystem(sqlContext.sessionState.hadoopConf) + val fs = path.getFileSystem(sqlContext.sessionState.newHadoopConf()) path.makeQualified(fs.getUri, fs.getWorkingDirectory).toString } @@ -605,53 +602,45 @@ abstract class HadoopFsRelationTest extends QueryTest with SQLTestUtils with Tes } test("SPARK-8578 specified custom output committer will not be used to append data") { - val clonedConf = new Configuration(hadoopConfiguration) - try { - val df = sqlContext.range(1, 10).toDF("i") - withTempPath { dir => - df.write.mode("append").format(dataSourceName).save(dir.getCanonicalPath) - hadoopConfiguration.set( - SQLConf.OUTPUT_COMMITTER_CLASS.key, - classOf[AlwaysFailOutputCommitter].getName) - // Since Parquet has its own output committer setting, also set it - // to AlwaysFailParquetOutputCommitter at here. - hadoopConfiguration.set("spark.sql.parquet.output.committer.class", - classOf[AlwaysFailParquetOutputCommitter].getName) - // Because there data already exists, - // this append should succeed because we will use the output committer associated - // with file format and AlwaysFailOutputCommitter will not be used. - df.write.mode("append").format(dataSourceName).save(dir.getCanonicalPath) - checkAnswer( - sqlContext.read - .format(dataSourceName) - .option("dataSchema", df.schema.json) - .load(dir.getCanonicalPath), - df.union(df)) - - // This will fail because AlwaysFailOutputCommitter is used when we do append. - intercept[Exception] { - df.write.mode("overwrite").format(dataSourceName).save(dir.getCanonicalPath) - } + val extraOptions = Map[String, String]( + SQLConf.OUTPUT_COMMITTER_CLASS.key -> classOf[AlwaysFailOutputCommitter].getName, + // Since Parquet has its own output committer setting, also set it + // to AlwaysFailParquetOutputCommitter at here. + "spark.sql.parquet.output.committer.class" -> + classOf[AlwaysFailParquetOutputCommitter].getName + ) + + val df = sqlContext.range(1, 10).toDF("i") + withTempPath { dir => + df.write.mode("append").format(dataSourceName).save(dir.getCanonicalPath) + // Because there data already exists, + // this append should succeed because we will use the output committer associated + // with file format and AlwaysFailOutputCommitter will not be used. + df.write.mode("append").format(dataSourceName).save(dir.getCanonicalPath) + checkAnswer( + sqlContext.read + .format(dataSourceName) + .option("dataSchema", df.schema.json) + .options(extraOptions) + .load(dir.getCanonicalPath), + df.union(df)) + + // This will fail because AlwaysFailOutputCommitter is used when we do append. + intercept[Exception] { + df.write.mode("overwrite") + .options(extraOptions).format(dataSourceName).save(dir.getCanonicalPath) } - withTempPath { dir => - hadoopConfiguration.set( - SQLConf.OUTPUT_COMMITTER_CLASS.key, - classOf[AlwaysFailOutputCommitter].getName) - // Since Parquet has its own output committer setting, also set it - // to AlwaysFailParquetOutputCommitter at here. - hadoopConfiguration.set("spark.sql.parquet.output.committer.class", - classOf[AlwaysFailParquetOutputCommitter].getName) - // Because there is no existing data, - // this append will fail because AlwaysFailOutputCommitter is used when we do append - // and there is no existing data. - intercept[Exception] { - df.write.mode("append").format(dataSourceName).save(dir.getCanonicalPath) - } + } + withTempPath { dir => + // Because there is no existing data, + // this append will fail because AlwaysFailOutputCommitter is used when we do append + // and there is no existing data. + intercept[Exception] { + df.write.mode("append") + .options(extraOptions) + .format(dataSourceName) + .save(dir.getCanonicalPath) } - } finally { - // Hadoop 1 doesn't have `Configuration.unset` - hadoopConfiguration.clear() - clonedConf.asScala.foreach(entry => hadoopConfiguration.set(entry.getKey, entry.getValue)) } } @@ -671,38 +660,38 @@ abstract class HadoopFsRelationTest extends QueryTest with SQLTestUtils with Tes } test("Locality support for FileScanRDD") { - withHadoopConf( + val options = Map[String, String]( "fs.file.impl" -> classOf[LocalityTestFileSystem].getName, "fs.file.impl.disable.cache" -> "true" - ) { - withTempPath { dir => - val path = "file://" + dir.getCanonicalPath - val df1 = sqlContext.range(4) - df1.coalesce(1).write.mode("overwrite").format(dataSourceName).save(path) - df1.coalesce(1).write.mode("append").format(dataSourceName).save(path) - - def checkLocality(): Unit = { - val df2 = sqlContext.read - .format(dataSourceName) - .option("dataSchema", df1.schema.json) - .load(path) - - val Some(fileScanRDD) = df2.queryExecution.executedPlan.collectFirst { - case scan: DataSourceScanExec if scan.rdd.isInstanceOf[FileScanRDD] => - scan.rdd.asInstanceOf[FileScanRDD] - } + ) + withTempPath { dir => + val path = "file://" + dir.getCanonicalPath + val df1 = sqlContext.range(4) + df1.coalesce(1).write.mode("overwrite").options(options).format(dataSourceName).save(path) + df1.coalesce(1).write.mode("append").options(options).format(dataSourceName).save(path) - val partitions = fileScanRDD.partitions - val preferredLocations = partitions.flatMap(fileScanRDD.preferredLocations) + def checkLocality(): Unit = { + val df2 = sqlContext.read + .format(dataSourceName) + .option("dataSchema", df1.schema.json) + .options(options) + .load(path) - assert(preferredLocations.distinct.length == 2) + val Some(fileScanRDD) = df2.queryExecution.executedPlan.collectFirst { + case scan: DataSourceScanExec if scan.rdd.isInstanceOf[FileScanRDD] => + scan.rdd.asInstanceOf[FileScanRDD] } - checkLocality() + val partitions = fileScanRDD.partitions + val preferredLocations = partitions.flatMap(fileScanRDD.preferredLocations) - withSQLConf(SQLConf.PARALLEL_PARTITION_DISCOVERY_THRESHOLD.key -> "0") { - checkLocality() - } + assert(preferredLocations.distinct.length == 2) + } + + checkLocality() + + withSQLConf(SQLConf.PARALLEL_PARTITION_DISCOVERY_THRESHOLD.key -> "0") { + checkLocality() } } } diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/sources/ParquetHadoopFsRelationSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/sources/ParquetHadoopFsRelationSuite.scala index 19749a9713..1d104889fe 100644 --- a/sql/hive/src/test/scala/org/apache/spark/sql/sources/ParquetHadoopFsRelationSuite.scala +++ b/sql/hive/src/test/scala/org/apache/spark/sql/sources/ParquetHadoopFsRelationSuite.scala @@ -132,7 +132,7 @@ class ParquetHadoopFsRelationSuite extends HadoopFsRelationTest { val summaryPath = new Path(path, "_metadata") val commonSummaryPath = new Path(path, "_common_metadata") - val fs = summaryPath.getFileSystem(hadoopConfiguration) + val fs = summaryPath.getFileSystem(sqlContext.sessionState.newHadoopConf()) fs.delete(summaryPath, true) fs.delete(commonSummaryPath, true) |