[SPARK-19667][SQL]create table with hiveenabled in default database use warehouse path instead of the location of default database

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

@@ -672,6 +672,14 @@ object SQLConf {
       .stringConf
       .createWithDefault(TimeZone.getDefault().getID())
 
+  // for test
+  val TEST_HIVE_CREATETABLE_DEFAULTDB_USEWAREHOUSE_PATH =
+  buildConf("spark.hive.test.createTable.defaultDB.location.useWarehousePath")
+    .doc("Enables test case to use warehouse path instead of db location when " +
+      "create table in default database.")
+    .booleanConf
+    .createWithDefault(false)
+
   object Deprecated {
     val MAPRED_REDUCE_TASKS = "mapred.reduce.tasks"
   }

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

@@ -90,8 +90,10 @@ private[sql] class SharedState(val sparkContext: SparkContext) extends Logging {
 
   // Create the default database if it doesn't exist.
   {
+    // default database set to empty string,
+    // when reload from metastore using warehouse path to replace it
     val defaultDbDefinition = CatalogDatabase(
-      SessionCatalog.DEFAULT_DATABASE, "default database", warehousePath, Map())
+      SessionCatalog.DEFAULT_DATABASE, "default database", "", Map())
     // Initialize default database if it doesn't exist
     if (!externalCatalog.databaseExists(SessionCatalog.DEFAULT_DATABASE)) {
       // There may be another Spark application creating default database at the same time, here we

sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala

@@ -48,6 +48,8 @@ import org.apache.spark.sql.catalyst.parser.{CatalystSqlParser, ParseException}
 import org.apache.spark.sql.execution.QueryExecutionException
 import org.apache.spark.sql.execution.command.DDLUtils
 import org.apache.spark.sql.hive.client.HiveClientImpl._
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.internal.StaticSQLConf._
 import org.apache.spark.sql.types._
 import org.apache.spark.util.{CircularBuffer, Utils}
 
@@ -311,11 +313,12 @@ private[hive] class HiveClientImpl(
   override def createDatabase(
       database: CatalogDatabase,
       ignoreIfExists: Boolean): Unit = withHiveState {
+    // default database's location always use the warehouse path, here set to emtpy string
     client.createDatabase(
       new HiveDatabase(
         database.name,
         database.description,
-        database.locationUri,
+        if (database.name == SessionCatalog.DEFAULT_DATABASE) "" else database.locationUri,
         Option(database.properties).map(_.asJava).orNull),
         ignoreIfExists)
   }
@@ -339,10 +342,17 @@ private[hive] class HiveClientImpl(
 
   override def getDatabase(dbName: String): CatalogDatabase = withHiveState {
     Option(client.getDatabase(dbName)).map { d =>
+      // default database's location always use the warehouse path
+      // TEST_HIVE_CREATETABLE_DEFAULTDB_USEWAREHOUSE_PATH is a flag fro test
+      val dbLocation = if (dbName == SessionCatalog.DEFAULT_DATABASE
+        || sparkConf.get(SQLConf.TEST_HIVE_CREATETABLE_DEFAULTDB_USEWAREHOUSE_PATH)) {
+        sparkConf.get(WAREHOUSE_PATH)
+      } else d.getLocationUri
+
       CatalogDatabase(
         name = d.getName,
         description = d.getDescription,
-        locationUri = d.getLocationUri,
+        locationUri = dbLocation,
         properties = Option(d.getParameters).map(_.asScala.toMap).orNull)
     }.getOrElse(throw new NoSuchDatabaseException(dbName))
   }

sql/hive/src/test/scala/org/apache/spark/sql/hive/ShowCreateTableSuite.scala

@@ -344,6 +344,8 @@ class ShowCreateTableSuite extends QueryTest with SQLTestUtils with TestHiveSing
       )
 
       table.copy(
+        storage = table.storage.copy(
+          locationUri = table.storage.locationUri.map(_.stripPrefix("file:"))),
         createTime = 0L,
         lastAccessTime = 0L,
         properties = table.properties.filterKeys(!nondeterministicProps.contains(_))

sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala

@@ -1587,4 +1587,51 @@ class HiveDDLSuite
       }
     }
   }
+
+  test("create table with default database use warehouse path instead of database location") {
+    withTable("t") {
+      // default database use warehouse path as its location
+      withTempDir { dir =>
+        spark.sparkContext.conf
+          .set(SQLConf.TEST_HIVE_CREATETABLE_DEFAULTDB_USEWAREHOUSE_PATH.key, "true")
+
+        val sparkWarehousePath = spark.sharedState.warehousePath.stripSuffix("/")
+        spark.sql(s"CREATE DATABASE default_test LOCATION '$dir'" )
+        val db = spark.sessionState.catalog.getDatabaseMetadata("default_test")
+        assert(db.locationUri.stripSuffix("/") == sparkWarehousePath)
+        spark.sql("USE default_test")
+
+        spark.sql("CREATE TABLE t(a string)")
+        val table = spark.sessionState.catalog.getTableMetadata(TableIdentifier("t"))
+        assert(table.location.stripSuffix("/").stripPrefix("file:") ==
+          new File(sparkWarehousePath, "t").getAbsolutePath.stripSuffix("/"))
+
+        // clear
+        spark.sparkContext.conf
+          .remove(SQLConf.TEST_HIVE_CREATETABLE_DEFAULTDB_USEWAREHOUSE_PATH.key)
+
+        spark.sql("DROP TABLE t")
+        spark.sql("DROP DATABASE default_test")
+        spark.sql("USE DEFAULT")
+      }
+
+      // not default database use its's location from the create command
+      withTempDir { dir =>
+        val dirPath = s"file:${dir.getAbsolutePath.stripSuffix("/")}"
+        spark.sql(s"CREATE DATABASE test_not_default LOCATION '$dir'" )
+        val db = spark.sessionState.catalog.getDatabaseMetadata("test_not_default")
+        assert(db.locationUri.stripSuffix("/") == dirPath)
+        spark.sql("USE test_not_default")
+
+        spark.sql("CREATE TABLE t(a string)")
+        val table = spark.sessionState.catalog.getTableMetadata(TableIdentifier("t"))
+        assert(table.location.stripSuffix("/") == s"$dirPath/t" )
+
+        // clear
+        spark.sql("DROP TABLE t")
+        spark.sql("DROP DATABASE test_not_default")
+        spark.sql("USE DEFAULT")
+      }
+    }
+  }
 }