private void replaceFiles(Path tablePath, Path srcf, Path destf, Path oldPath, HiveConf conf,
          boolean isSrcLocal, boolean purge, List<FileStatus> newFiles, PathFilter deletePathFilter,
      boolean isNeedRecycle, boolean isManaged, boolean isInsertOverwrite) throws HiveException {
    try {

      FileSystem destFs = destf.getFileSystem(conf);
      // check if srcf contains nested sub-directories
      FileStatus[] srcs;
      FileSystem srcFs;
      try {
        srcFs = srcf.getFileSystem(conf);
        srcs = srcFs.globStatus(srcf);
      } catch (IOException e) {
        throw new HiveException("Getting globStatus " + srcf.toString(), e);
      }

      // For insert/load overwrite cases, where external.table.purge is disabled for the table, there may be stale
      // partitions present in the table location after Alter table drop partition operation. In such cases, oldPath will be
      // null, since those partitions will not be present in metastore. Added below check to clean up those stale partitions.
      if (oldPath == null && isInsertOverwrite) {
        deleteOldPathForReplace(destf, destf, conf, purge, deletePathFilter, isNeedRecycle);
      }

      // the extra check is required to make ALTER TABLE ... CONCATENATE work
      if (oldPath != null && (srcs != null || isInsertOverwrite)) {
        deleteOldPathForReplace(destf, oldPath, conf, purge, deletePathFilter, isNeedRecycle);
      }

      if (srcs == null) {
        LOG.info("No sources specified to move: " + srcf);
        return;
      }

      // first call FileUtils.mkdir to make sure that destf directory exists, if not, it creates
      // destf
      boolean destfExist = FileUtils.mkdir(destFs, destf, conf);
      if(!destfExist) {
        throw new IOException("Directory " + destf.toString()
            + " does not exist and could not be created.");
      }

      // Two cases:
      // 1. srcs has only a src directory, if rename src directory to destf, we also need to
      // Copy/move each file under the source directory to avoid to delete the destination
      // directory if it is the root of an HDFS encryption zone.
      // 2. srcs must be a list of files -- ensured by LoadSemanticAnalyzer
      // in both cases, we move the file under destf
      if (srcs.length == 1 && srcs[0].isDirectory()) {
        if (!moveFile(conf, srcs[0].getPath(), destf, true, isSrcLocal, isManaged)) {
          throw new IOException("Error moving: " + srcf + " into: " + destf);
        }

        // Add file paths of the files that will be moved to the destination if the caller needs it
        if (newFiles != null) {
          newFiles.addAll(HdfsUtils.listLocatedFileStatus(destFs, destf, null, true));
        }
      } else {
        final Map<Future<Boolean>, Path> moveFutures = Maps.newLinkedHashMapWithExpectedSize(srcs.length);
        final int moveFilesThreadCount = HiveConf.getIntVar(conf, ConfVars.HIVE_MOVE_FILES_THREAD_COUNT);
        final ExecutorService pool = moveFilesThreadCount > 0
            ? Executors.newFixedThreadPool(
                moveFilesThreadCount,
                new ThreadFactoryBuilder().setDaemon(true).setNameFormat("Replace-Thread-%d").build())
            : MoreExecutors.newDirectExecutorService();
        final SessionState parentSession = SessionState.get();
        // its either a file or glob
        for (FileStatus src : srcs) {
          Path destFile = new Path(destf, src.getPath().getName());
          moveFutures.put(
              pool.submit(
                  new Callable<Boolean>() {
                    @Override
                    public Boolean call() throws Exception {
                      SessionState.setCurrentSessionState(parentSession);
                      return moveFile(
                          conf, src.getPath(), destFile, true, isSrcLocal, isManaged);
                    }
                  }),
              destFile);
        }

        pool.shutdown();
        for (Map.Entry<Future<Boolean>, Path> moveFuture : moveFutures.entrySet()) {
          boolean moveFailed;
          try {
            moveFailed = !moveFuture.getKey().get();
          } catch (InterruptedException | ExecutionException e) {
            pool.shutdownNow();
            if (e.getCause() instanceof IOException) {
              throw (IOException) e.getCause();
            }
            if (e.getCause() instanceof HiveException) {
              throw (HiveException) e.getCause();
            }
            throw handlePoolException(pool, e);
          }
          if (moveFailed) {
            throw new IOException("Error moving: " + srcf + " into: " + destf);
          }

          // Add file paths of the files that will be moved to the destination if the caller needs it
          if (null != newFiles) {
            newFiles.add(destFs.getFileStatus(moveFuture.getValue()));
          }
        }
      }
    } catch (IOException e) {
      throw new HiveException(e.getMessage(), e);
    }
  }
Replaces files in the partition with new data set specified by srcf. Works
by renaming directory of srcf to the destination file.
srcf, destf, and tmppath should resident in the same DFS, but the oldPath can be in a
different DFS.

@param tablePath path of the table.  Used to identify permission inheritance.
@param srcf
         Source directory to be renamed to tmppath. It should be a
         leaf directory where the final data files reside. However it
         could potentially contain subdirectories as well.
@param destf
         The directory where the final data needs to go
@param oldPath
         The directory where the old data location, need to be cleaned up.  Most of time, will be the same
         as destf, unless its across FileSystem boundaries.
@param purge
         When set to true files which needs to be deleted are not moved to Trash
@param isSrcLocal
         If the source directory is LOCAL
@param newFiles
         Output the list of new files replaced in the destination path
@param isManaged
         If the table is managed.
ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java