[SPARK-20594][SQL]The staging directory should be a child directory starts with "." to avoid being deleted if we set hive.exec.stagingdir under the table directory.

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

@@ -17,7 +17,7 @@
 
 package org.apache.spark.sql.hive.execution
 
-import java.io.IOException
+import java.io.{File, IOException}
 import java.net.URI
 import java.text.SimpleDateFormat
 import java.util.{Date, Locale, Random}
@@ -97,12 +97,23 @@ case class InsertIntoHiveTable(
     val inputPathUri: URI = inputPath.toUri
     val inputPathName: String = inputPathUri.getPath
     val fs: FileSystem = inputPath.getFileSystem(hadoopConf)
-    val stagingPathName: String =
+    var stagingPathName: String =
       if (inputPathName.indexOf(stagingDir) == -1) {
         new Path(inputPathName, stagingDir).toString
       } else {
         inputPathName.substring(0, inputPathName.indexOf(stagingDir) + stagingDir.length)
       }
+
+    // SPARK-20594: The staging directory should be a child directory starts with "." to avoid
+    // being deleted if we set hive.exec.stagingdir under the table directory.
+    if (FileUtils.isSubDir(new Path(stagingPathName), inputPath, fs)
+      && !stagingPathName.stripPrefix(inputPathName).stripPrefix(File.separator).startsWith(".")) {
+      logDebug(s"The staging dir '$stagingPathName' should be a child directory starts " +
+        s"with '.' to avoid being deleted if we set hive.exec.stagingdir under the table " +
+        s"directory.")
+      stagingPathName = new Path(inputPathName, ".hive-staging").toString
+    }
+
     val dir: Path =
       fs.makeQualified(
         new Path(stagingPathName + "_" + executionId + "-" + TaskRunner.getTaskRunnerID))

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

@@ -494,4 +494,45 @@ class InsertIntoHiveTableSuite extends QueryTest with TestHiveSingleton with Bef
         spark.table("t").write.insertInto(tableName)
       }
   }
+
+  private def dropTables(tableNames: String*): Unit = {
+    tableNames.foreach { name =>
+      sql(s"DROP TABLE IF EXISTS $name")
+    }
+  }
+
+  test(
+    """SPARK-20594: The staging directory should be appended with ".hive-staging"
+      |to avoid being deleted if we set hive.exec.stagingdir under the table directory
+      |without start with "."""".stripMargin) {
+
+    dropTables("test_table", "test_table1")
+
+    sql("CREATE TABLE test_table (key int, value string)")
+
+    // Add some data.
+    testData.write.mode(SaveMode.Append).insertInto("test_table")
+
+    // Make sure the table has also been updated.
+    checkAnswer(
+      sql("SELECT * FROM test_table"),
+      testData.collect().toSeq
+    )
+
+    sql("CREATE TABLE test_table1 (key int, value string)")
+
+    // Set hive.exec.stagingdir under the table directory without start with ".".
+    sql("set hive.exec.stagingdir=./test")
+
+    // Now overwrite.
+    sql("INSERT OVERWRITE TABLE test_table1 SELECT * FROM test_table")
+
+    // Make sure the table has also been updated.
+    checkAnswer(
+      sql("SELECT * FROM test_table1"),
+      testData.collect().toSeq
+    )
+
+    dropTables("test_table", "test_table1")
+  }
 }