[SPARK-18817] [SPARKR] [SQL] Set default warehouse dir to tempdir

R/pkg/R/sparkR.R

@@ -363,6 +363,7 @@ sparkR.session <- function(
   ...) {
 
   sparkConfigMap <- convertNamedListToEnv(sparkConfig)
+
   namedParams <- list(...)
   if (length(namedParams) > 0) {
     paramMap <- convertNamedListToEnv(namedParams)
@@ -376,6 +377,12 @@ sparkR.session <- function(
     overrideEnvs(sparkConfigMap, paramMap)
   }
 
+  # NOTE(shivaram): Set default warehouse dir to tmpdir to meet CRAN requirements
+  # See SPARK-18817 for more details
+  if (!exists("spark.sql.default.warehouse.dir", envir = sparkConfigMap)) {
+    assign("spark.sql.default.warehouse.dir", tempdir(), envir = sparkConfigMap)
+  }
+
   deployMode <- ""
   if (exists("spark.submit.deployMode", envir = sparkConfigMap)) {
     deployMode <- sparkConfigMap[["spark.submit.deployMode"]]

R/pkg/inst/tests/testthat/test_context.R

@@ -72,6 +72,20 @@ test_that("repeatedly starting and stopping SparkSession", {
   }
 })
 
+test_that("Default warehouse dir should be set to tempdir", {
+  sparkR.session.stop()
+  sparkR.session(enableHiveSupport = FALSE)
+
+  # Create a temporary table
+  sql("CREATE TABLE people_warehouse_test")
+  # spark-warehouse should be written only tempdir() and not current working directory
+  res <- list.files(path = ".", pattern = ".*spark-warehouse.*",
+                    recursive = TRUE, include.dirs = TRUE)
+  expect_equal(length(res), 0)
+  result <- sql("DROP TABLE people_warehouse_test")
+  sparkR.session.stop()
+})
+
 test_that("rdd GC across sparkR.stop", {
   sc <- sparkR.sparkContext() # sc should get id 0
   rdd1 <- parallelize(sc, 1:20, 2L) # rdd1 should get id 1

sql/core/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -819,7 +819,13 @@ private[sql] class SQLConf extends Serializable with CatalystConf with Logging {
 
   def variableSubstituteDepth: Int = getConf(VARIABLE_SUBSTITUTE_DEPTH)
 
-  def warehousePath: String = new Path(getConf(StaticSQLConf.WAREHOUSE_PATH)).toString
+  def warehousePath: String = {
+    if (contains(StaticSQLConf.WAREHOUSE_PATH.key)) {
+      new Path(getConf(StaticSQLConf.WAREHOUSE_PATH).get).toString
+    } else {
+      new Path(getConf(StaticSQLConf.DEFAULT_WAREHOUSE_PATH)).toString
+    }
+  }
 
   def ignoreCorruptFiles: Boolean = getConf(IGNORE_CORRUPT_FILES)
 
@@ -964,11 +970,17 @@ object StaticSQLConf {
     }
   }
 
-  val WAREHOUSE_PATH = buildConf("spark.sql.warehouse.dir")
-    .doc("The default location for managed databases and tables.")
+  val DEFAULT_WAREHOUSE_PATH = buildConf("spark.sql.default.warehouse.dir")
+    .doc("The default location for managed databases and tables. " +
+         "Used if spark.sql.warehouse.dir is not set")
     .stringConf
     .createWithDefault(Utils.resolveURI("spark-warehouse").toString)
 
+  val WAREHOUSE_PATH = buildConf("spark.sql.warehouse.dir")
+    .doc("The location for managed databases and tables.")
+    .stringConf
+    .createOptional
+
   val CATALOG_IMPLEMENTATION = buildConf("spark.sql.catalogImplementation")
     .internal()
     .stringConf

sql/core/src/main/scala/org/apache/spark/sql/internal/SharedState.scala

@@ -55,14 +55,19 @@ private[sql] class SharedState(val sparkContext: SparkContext) extends Logging {
         s"is set. Setting ${WAREHOUSE_PATH.key} to the value of " +
         s"hive.metastore.warehouse.dir ('$hiveWarehouseDir').")
       hiveWarehouseDir
-    } else {
+    } else if (sparkContext.conf.contains(WAREHOUSE_PATH.key) &&
+               sparkContext.conf.get(WAREHOUSE_PATH).isDefined) {
       // If spark.sql.warehouse.dir is set, we will override hive.metastore.warehouse.dir using
       // the value of spark.sql.warehouse.dir.
-      // When neither spark.sql.warehouse.dir nor hive.metastore.warehouse.dir is set,
-      // we will set hive.metastore.warehouse.dir to the default value of spark.sql.warehouse.dir.
-      val sparkWarehouseDir = sparkContext.conf.get(WAREHOUSE_PATH)
+      val sparkWarehouseDir = sparkContext.conf.get(WAREHOUSE_PATH).get
       sparkContext.conf.set("hive.metastore.warehouse.dir", sparkWarehouseDir)
       sparkWarehouseDir
+    } else {
+      // When neither spark.sql.warehouse.dir nor hive.metastore.warehouse.dir is set,
+      // we will set hive.metastore.warehouse.dir to the value of spark.sql.default.warehouse.dir.
+      val sparkDefaultWarehouseDir = sparkContext.conf.get(DEFAULT_WAREHOUSE_PATH)
+      sparkContext.conf.set("hive.metastore.warehouse.dir", sparkDefaultWarehouseDir)
+      sparkDefaultWarehouseDir
     }
 
   }

sql/core/src/test/scala/org/apache/spark/sql/internal/SQLConfSuite.scala

@@ -221,6 +221,19 @@ class SQLConfSuite extends QueryTest with SharedSQLContext {
       .sessionState.conf.warehousePath.stripSuffix("/"))
   }
 
+  test("changing default value of warehouse path") {
+    try {
+      val newWarehouseDefault = "spark-warehouse2"
+      val newWarehouseDefaultPath = new Path(Utils.resolveURI(newWarehouseDefault)).toString
+      sparkContext.conf.set("spark.sql.default.warehouse.dir", newWarehouseDefaultPath)
+      val spark = new SparkSession(sparkContext)
+      assert(newWarehouseDefaultPath.stripSuffix("/") === spark
+        .sessionState.conf.warehousePath.stripSuffix("/"))
+    } finally {
+      sparkContext.conf.remove("spark.sql.default.warehouse.dir")
+    }
+  }
+
   test("MAX_CASES_BRANCHES") {
     withTable("tab1") {
       spark.range(10).write.saveAsTable("tab1")