HIVE-28473:INSERT OVERWRITE LOCAL DIRECTORY writes staging files to wrong hdfs directory (apache#5407)(yu liang, reviewed by Butao Zhang)

Author: liangyu-1

ql/src/java/org/apache/hadoop/hive/ql/exec/MoveTask.java

@@ -438,9 +438,11 @@ public int execute() {
             }
           }
           else {
-            FileSystem targetFs = targetPath.getFileSystem(conf);
-            if (!targetFs.exists(targetPath.getParent())){
-              targetFs.mkdirs(targetPath.getParent());
+            if (lfd.getIsDfsDir()) {
+              FileSystem targetFs = targetPath.getFileSystem(conf);
+              if (!targetFs.exists(targetPath.getParent())) {
+                targetFs.mkdirs(targetPath.getParent());
+              }
             }
             moveFile(sourcePath, targetPath, lfd.getIsDfsDir());
           }

ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java

@@ -8468,6 +8468,7 @@ private FileSinkDesc createFileSinkDesc(String dest, TableDesc table_desc,
                                           RowSchema fsRS, boolean canBeMerged, Table dest_tab, boolean isMmCtas,
                                           Integer dest_type, QB qb, boolean isDirectInsert, AcidUtils.Operation acidOperation, String moveTaskId) throws SemanticException {
     boolean isInsertOverwrite = false;
+    boolean isLocal = false;
     Context.Operation writeOperation = getWriteOperation(dest);
     switch (dest_type) {
     case QBMetaData.DEST_PARTITION:
@@ -8492,6 +8493,7 @@ private FileSinkDesc createFileSinkDesc(String dest, TableDesc table_desc,
 
       break;
     case QBMetaData.DEST_LOCAL_FILE:
+      isLocal = true;
     case QBMetaData.DEST_DFS_FILE:
       //CTAS path or insert into file/directory
       break;
@@ -8545,7 +8547,12 @@ private FileSinkDesc createFileSinkDesc(String dest, TableDesc table_desc,
     fileSinkDesc.setStatsAggPrefix(fileSinkDesc.getDirName().toString());
     if (!destTableIsMaterialization &&
         HiveConf.getVar(conf, HIVE_STATS_DBCLASS).equalsIgnoreCase(StatDB.fs.name())) {
-      String statsTmpLoc = ctx.getTempDirForInterimJobPath(dest_path).toString();
+      String statsTmpLoc;
+      if (isLocal){
+        statsTmpLoc = ctx.getMRTmpPath().toString();
+      } else {
+        statsTmpLoc = ctx.getTempDirForInterimJobPath(dest_path).toString();
+      }
       fileSinkDesc.setStatsTmpDir(statsTmpLoc);
       LOG.debug("Set stats collection dir : " + statsTmpLoc);
     }