modify the test case

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

@@ -672,14 +672,6 @@ 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/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala

@@ -48,7 +48,6 @@ 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}
@@ -342,9 +341,7 @@ 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)) {
+      val dbLocation = if (dbName == SessionCatalog.DEFAULT_DATABASE) {
         sparkConf.get(WAREHOUSE_PATH)
       } else d.getLocationUri
 

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

@@ -334,6 +334,35 @@ class HiveSparkSubmitSuite
     runSparkSubmit(argsForShowTables)
   }
 
+  test("SPARK-19667: create table in default database with HiveEnabled use warehouse path " +
+    "instead of the location of default database") {
+    val unusedJar = TestUtils.createJarWithClasses(Seq.empty)
+    val warehousePath1 = Utils.createTempDir("wh1")
+    val argsForCreateTable = Seq(
+      "--class", SPARK_19667_CREATE_TABLE.getClass.getName.stripSuffix("$"),
+      "--name", "SPARK-19667",
+      "--master", "local-cluster[2,1,1024]",
+      "--conf", "spark.ui.enabled=false",
+      "--conf", "spark.master.rest.enabled=false",
+      "--conf", s"spark.sql.warehouse.dir=$warehousePath1",
+      unusedJar.toString)
+    runSparkSubmit(argsForCreateTable)
+
+    val warehousePath2 = Utils.createTempDir("wh2")
+    val argsForShowTables = Seq(
+      "--class", SPARK_19667_VERIFY_TABLE_PATH.getClass.getName.stripSuffix("$"),
+      "--name", "SPARK-19667",
+      "--master", "local-cluster[2,1,1024]",
+      "--conf", "spark.ui.enabled=false",
+      "--conf", "spark.master.rest.enabled=false",
+      "--conf", s"spark.sql.warehouse.dir=$warehousePath2",
+      unusedJar.toString)
+    runSparkSubmit(argsForShowTables)
+
+    Utils.deleteRecursively(warehousePath1)
+    Utils.deleteRecursively(warehousePath2)
+  }
+
   // NOTE: This is an expensive operation in terms of time (10 seconds+). Use sparingly.
   // This is copied from org.apache.spark.deploy.SparkSubmitSuite
   private def runSparkSubmit(args: Seq[String]): Unit = {
@@ -905,3 +934,91 @@ object SPARK_18989_DESC_TABLE {
     }
   }
 }
+
+object SPARK_19667_CREATE_TABLE {
+  def main(args: Array[String]): Unit = {
+    val spark = SparkSession.builder().enableHiveSupport().getOrCreate()
+    try {
+      val warehousePath = spark.sharedState.warehousePath.stripSuffix("/")
+      val defaultDB = spark.sessionState.catalog.getDatabaseMetadata("default")
+      // default database use warehouse path as its location
+      assert(defaultDB.locationUri.stripSuffix("/") == warehousePath)
+      spark.sql("CREATE TABLE t(a string)")
+
+      val table = spark.sessionState.catalog.getTableMetadata(TableIdentifier("t"))
+      // table in default database use the location of default database which is also warehouse path
+      assert(table.location.stripSuffix("/") == s"file:$warehousePath/t")
+      spark.sql("INSERT INTO TABLE t SELECT 1")
+      assert(spark.sql("SELECT * FROM t").count == 1)
+
+      spark.sql("CREATE DATABASE not_default")
+      spark.sql("USE not_default")
+      spark.sql("CREATE TABLE t1(b string)")
+      val table1 = spark.sessionState.catalog.getTableMetadata(TableIdentifier("t1"))
+      // table in not default database use the location of its own database
+      assert(table1.location.stripSuffix("/") == s"file:$warehousePath/not_default.db/t1")
+    } finally {
+      spark.sql("USE default")
+    }
+  }
+}
+
+object SPARK_19667_VERIFY_TABLE_PATH {
+  def main(args: Array[String]): Unit = {
+    val spark = SparkSession.builder().enableHiveSupport().getOrCreate()
+    try {
+      val warehousePath = spark.sharedState.warehousePath.stripSuffix("/")
+      val defaultDB = spark.sessionState.catalog.getDatabaseMetadata("default")
+      // default database use warehouse path as its location
+      assert(defaultDB.locationUri.stripSuffix("/") == warehousePath)
+
+      val table = spark.sessionState.catalog.getTableMetadata(TableIdentifier("t"))
+      // the table in default database created in job(SPARK_19667_CREATE_TABLE) above,
+      // which has different warehouse path from this job, its location still equals to
+      // the location when it's created.
+      assert(table.location.stripSuffix("/") != s"file:$warehousePath/t")
+      assert(spark.sql("SELECT * FROM t").count == 1)
+
+      spark.sql("CREATE TABLE t3(d string)")
+      val table3 = spark.sessionState.catalog.getTableMetadata(TableIdentifier("t3"))
+      // the table in default database created here in this job, it will use the warehouse path
+      // of this job as its location
+      assert(table3.location.stripSuffix("/") == s"file:$warehousePath/t3")
+
+      spark.sql("USE not_default")
+      val table1 = spark.sessionState.catalog.getTableMetadata(TableIdentifier("t1"))
+      // the table in not default database create in job(SPARK_19667_CREATE_TABLE) above,
+      // which has different warehouse path from this job, its location still equals to
+      // the location when it's created.
+      assert(table1.location.stripSuffix("/") != s"$warehousePath/not_default.db/t1")
+      assert(!new File(s"$warehousePath/not_default.db/t1").exists())
+
+      spark.sql("CREATE TABLE t2(c string)")
+      val table2 = spark.sessionState.catalog.getTableMetadata(TableIdentifier("t2"))
+      // the table in not default database created here in this job, it will use the location
+      // of the database as its location, not the warehouse path in this job
+      assert(table2.location.stripSuffix("/") != s"file:$warehousePath/not_default.db/t2")
+
+      spark.sql("CREATE DATABASE not_default_1")
+      spark.sql("USE not_default_1")
+      spark.sql("CREATE TABLE t4(e string)")
+      val table4 = spark.sessionState.catalog.getTableMetadata(TableIdentifier("t4"))
+      // the table created in the database which created in this job, it will use the location
+      // of the database.
+      assert(table4.location.stripSuffix("/") == s"file:$warehousePath/not_default_1.db/t4")
+
+    } finally {
+      spark.sql("DROP TABLE IF EXISTS t4")
+      spark.sql("DROP DATABASE not_default_1")
+
+      spark.sql("USE not_default")
+      spark.sql("DROP TABLE IF EXISTS t1")
+      spark.sql("DROP TABLE IF EXISTS t2")
+      spark.sql("DROP DATABASE not_default")
+
+      spark.sql("USE default")
+      spark.sql("DROP TABLE IF EXISTS t")
+      spark.sql("DROP TABLE IF EXISTS t3")
+    }
+  }
+}

sql/hive/src/test/scala/org/apache/spark/sql/hive/client/VersionsSuite.scala

@@ -655,7 +655,7 @@ class VersionsSuite extends QueryTest with SQLTestUtils with TestHiveSingleton w
 
         val tPath = new Path(spark.sessionState.conf.warehousePath, "t")
         Seq("1").toDF("a").write.saveAsTable("t")
-        val expectedPath = s"file:${tPath.toUri.getPath.stripSuffix("/")}"
+        val expectedPath = tPath.toUri.getPath.stripSuffix("/")
         val table = spark.sessionState.catalog.getTableMetadata(TableIdentifier("t"))
 
         assert(table.location.stripSuffix("/") == expectedPath)
@@ -665,7 +665,7 @@ class VersionsSuite extends QueryTest with SQLTestUtils with TestHiveSingleton w
         val t1Path = new Path(spark.sessionState.conf.warehousePath, "t1")
         spark.sql("create table t1 using parquet as select 2 as a")
         val table1 = spark.sessionState.catalog.getTableMetadata(TableIdentifier("t1"))
-        val expectedPath1 = s"file:${t1Path.toUri.getPath.stripSuffix("/")}"
+        val expectedPath1 = t1Path.toUri.getPath.stripSuffix("/")
 
         assert(table1.location.stripSuffix("/") == expectedPath1)
         assert(t1Path.getFileSystem(spark.sessionState.newHadoopConf()).exists(t1Path))

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

@@ -1587,51 +1587,4 @@ 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")
-      }
-    }
-  }
 }