Update conf key and unit test.

Author: yhuai

sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveContext.scala

@@ -61,8 +61,20 @@ class HiveContext(sc: SparkContext) extends SQLContext(sc) {
   protected[sql] def convertMetastoreParquet: Boolean =
     getConf("spark.sql.hive.convertMetastoreParquet", "true") == "true"
 
-  protected[sql] def convertHiveCTASWithoutStorageSpec: Boolean =
-    getConf("spark.sql.sources.convertHiveCTASWithoutStorageSpec", "false") == "true"
+  /**
+   * When true, a table created by a Hive CTAS statement (no USING clause) will be
+   * converted to a data source table, using the data source set by spark.sql.sources.default.
+   * The table in CTAS statement will be converted when it meets any of the following conditions:
+   *   - The CTAS does not specify any of a SerDe (ROW FORMAT SERDE), a File Format (STORED AS), or
+   *     a Storage Hanlder (STORED BY), and the value of hive.default.fileformat in hive-site.xml
+   *     is either TextFile or SequenceFile.
+   *   - The CTAS statement specifies TextFile (STORED AS TEXTFILE) as the file format and no SerDe
+   *     is specified (no ROW FORMAT SERDE clause).
+   *   - The CTAS statement specifies SequenceFile (STORED AS SEQUENCEFILE) as the file format
+   *     and no SerDe is specified (no ROW FORMAT SERDE clause).
+   */
+  protected[sql] def convertCTAS: Boolean =
+    getConf("spark.sql.hive.convertCTAS", "false") == "true"
 
   override protected[sql] def executePlan(plan: LogicalPlan): this.QueryExecution =
     new this.QueryExecution(plan)

sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveMetastoreCatalog.scala

@@ -506,13 +506,13 @@ private[hive] class HiveMetastoreCatalog(hive: HiveContext) extends Catalog with
           case None => false
         }
 
-        if (hive.convertHiveCTASWithoutStorageSpec && !hasStorageSpec) {
+        if (hive.convertCTAS && !hasStorageSpec) {
           // Do the conversion when convertHiveCTASWithoutStorageSpec is true and the query
           // does not specify any storage format (file format and storage handler).
           if (dbName.isDefined) {
             throw new AnalysisException(
               "Cannot specify database name in a CTAS statement " +
-              "when spark.sql.sources.convertHiveCTASWithoutStorageSpec is set to true.")
+              "when spark.sql.hive.convertCTAS is set to true.")
           }
 
           val mode = if (allowExisting) SaveMode.Ignore else SaveMode.ErrorIfExists
@@ -537,11 +537,11 @@ private[hive] class HiveMetastoreCatalog(hive: HiveContext) extends Catalog with
 
       case p @ CreateTableAsSelect(db, tableName, child, allowExisting, None) =>
         val (dbName, tblName) = processDatabaseAndTableName(db, tableName)
-        if (hive.convertHiveCTASWithoutStorageSpec) {
+        if (hive.convertCTAS) {
           if (dbName.isDefined) {
             throw new AnalysisException(
               "Cannot specify database name in a CTAS statement " +
-              "when spark.sql.sources.convertHiveCTASWithoutStorageSpec is set to true.")
+              "when spark.sql.hive.convertCTAS is set to true.")
           }
 
           val mode = if (allowExisting) SaveMode.Ignore else SaveMode.ErrorIfExists

sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/commands.scala

@@ -165,8 +165,10 @@ case class CreateMetastoreDataSourceAsSelect(
       mode match {
         case SaveMode.ErrorIfExists =>
           sys.error(s"Table $tableName already exists. " +
-            s"If you want to append into it, please set mode to SaveMode.Append. " +
-            s"Or, if you want to overwrite it, please set mode to SaveMode.Overwrite.")
+            s"If you are using saveAsTable, you can set SaveMode to SaveMode.Append to" +
+            s"insert data into the table or set SaveMode to SaveMode.Overwrite to overwrite" +
+            s"the existing data. " +
+            s"Or, if you are using SQL CREATE TABLE, you need to drop $tableName first.")
         case SaveMode.Ignore =>
           // Since the table already exists and the save mode is Ignore, we will just return.
           return Seq.empty[Row]

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

@@ -65,22 +65,27 @@ class SQLQuerySuite extends QueryTest {
       }
     }
 
-    val originalConf = getConf("spark.sql.sources.convertHiveCTASWithoutStorageSpec", "false")
+    val originalConf = getConf("spark.sql.hive.convertCTAS", "false")
 
-    setConf("spark.sql.sources.convertHiveCTASWithoutStorageSpec", "true")
+    setConf("spark.sql.hive.convertCTAS", "true")
 
     sql("CREATE TABLE ctas1 AS SELECT key k, value FROM src ORDER BY k, value")
+    sql("CREATE TABLE IF NOT EXISTS ctas1 AS SELECT key k, value FROM src ORDER BY k, value")
+    var message = intercept[RuntimeException] {
+      sql("CREATE TABLE ctas1 AS SELECT key k, value FROM src ORDER BY k, value")
+    }.getMessage
+    assert(message.contains("Table ctas1 already exists"))
     checkRelation("ctas1", true)
     sql("DROP TABLE ctas1")
 
     // Specifying database name for query can be converted to data source write path
     // is not allowed right now.
-    val message = intercept[AnalysisException] {
+    message = intercept[AnalysisException] {
       sql("CREATE TABLE default.ctas1 AS SELECT key k, value FROM src ORDER BY k, value")
     }.getMessage
     assert(
       message.contains("Cannot specify database name in a CTAS statement"),
-      "When spark.sql.sources.convertHiveCTASWithoutStorageSpec is true, we should not allow " +
+      "When spark.sql.hive.convertCTAS is true, we should not allow " +
       "database name specified.")
 
     sql("CREATE TABLE ctas1 stored as textfile AS SELECT key k, value FROM src ORDER BY k, value")
@@ -105,7 +110,7 @@ class SQLQuerySuite extends QueryTest {
     // checkRelation("ctas1", false)
     // sql("DROP TABLE ctas1")
 
-    setConf("spark.sql.sources.convertHiveCTASWithoutStorageSpec", originalConf)
+    setConf("spark.sql.hive.convertCTAS", originalConf)
   }
 
   test("CTAS with serde") {