[SPARK-19667][SQL]create table with hiveenabled in default database u…

…se 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 HIVE_CREATETABLE_DEFAULTDB_USEWAREHOUSE_PATH =
+  buildConf("spark.hive.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/hive/src/main/scala/org/apache/spark/sql/hive/HiveExternalCatalog.scala

@@ -42,7 +42,7 @@ import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
 import org.apache.spark.sql.execution.command.DDLUtils
 import org.apache.spark.sql.execution.datasources.PartitioningUtils
 import org.apache.spark.sql.hive.client.HiveClient
-import org.apache.spark.sql.internal.HiveSerDe
+import org.apache.spark.sql.internal.{HiveSerDe, SQLConf}
 import org.apache.spark.sql.internal.StaticSQLConf._
 import org.apache.spark.sql.types.{DataType, StructType}
 
@@ -407,8 +407,14 @@ private[spark] class HiveExternalCatalog(conf: SparkConf, hadoopConf: Configurat
     properties
   }
 
+  // if the database is default, the value of WAREHOUSE_PATH in conf returned
   private def defaultTablePath(tableIdent: TableIdentifier): String = {
-    val dbLocation = getDatabase(tableIdent.database.get).locationUri
+    val dbLocation = if (tableIdent.database.get == SessionCatalog.DEFAULT_DATABASE
+      || conf.get(SQLConf.HIVE_CREATETABLE_DEFAULTDB_USEWAREHOUSE_PATH)) {
+      conf.get(WAREHOUSE_PATH)
+    } else {
+      getDatabase(tableIdent.database.get).locationUri
+    }
     new Path(new Path(dbLocation), tableIdent.table).toString
   }
 

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

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