fix write path

Author: scwf

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

@@ -1722,7 +1722,8 @@ https://cwiki.apache.org/confluence/display/Hive/Enhanced+Aggregation%2C+Cube%2C
           case "parquet" =>
             createTableDesc
               .setOutputFormat("org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat")
-            createTableDesc.setSerName("org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe")
+            createTableDesc
+              .setSerName("org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe")
 
           case "rcfile" =>
             createTableDesc.setOutputFormat("org.apache.hadoop.hive.ql.io.RCFileOutputFormat")

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

@@ -55,14 +55,14 @@ case class WriteToDirectory(
     val jobConfSer = new SerializableJobConf(jobConf)
     val targetPath = new Path(path)
 
-    val (tmpPath, destPath) = if (isLocal) {
+    val writeToPath = if (isLocal) {
       val localFileSystem = FileSystem.getLocal(jobConf)
       val localPath = localFileSystem.makeQualified(targetPath)
       // remove old dir
       if (localFileSystem.exists(localPath)) {
         localFileSystem.delete(localPath, true)
       }
-      (context.getExternalTmpPath(localPath), localPath)
+      localPath
     } else {
       val qualifiedPath = FileUtils.makeQualified(targetPath, hiveContext.hiveconf)
       val dfs = qualifiedPath.getFileSystem(jobConf)
@@ -71,10 +71,10 @@ case class WriteToDirectory(
       } else {
         dfs.mkdirs(qualifiedPath.getParent)
       }
-      (context.getExternalTmpPath(qualifiedPath), qualifiedPath)
+      qualifiedPath
     }
 
-    val fileSinkConf = new FileSinkDesc(tmpPath.toString, desc, false)
+    val fileSinkConf = new FileSinkDesc(writeToPath.toString, desc, false)
     val isCompressed = hiveContext.hiveconf.getBoolean(
       ConfVars.COMPRESSRESULT.varname, ConfVars.COMPRESSRESULT.defaultBoolVal)
 
@@ -100,15 +100,6 @@ case class WriteToDirectory(
       jobConfSer,
       writerContainer)
 
-    val fs = tmpPath.getFileSystem(jobConf)
-
-    // move tmp file to dest dir
-    if (isLocal) {
-      fs.moveToLocalFile(tmpPath, destPath)
-    } else if (!fs.rename(tmpPath, destPath)) {
-      throw new IOException("Unable to write data to " + destPath)
-    }
-
     Seq.empty[Row]
   }