private Partition loadPartitionInternal(Path loadPath, Table tbl, Map<String, String> partSpec,
                        Partition oldPart, LoadFileType loadFileType, boolean inheritTableSpecs,
                        boolean inheritLocation, boolean isSkewedStoreAsSubdir,
                        boolean isSrcLocal, boolean isAcidIUDoperation, boolean resetStatistics,
                        Long writeId, int stmtId, boolean isInsertOverwrite,
                        boolean isTxnTable, List<FileStatus> newFiles, boolean isDirectInsert) throws HiveException {
    Path tblDataLocationPath =  tbl.getDataLocation();
    boolean isMmTableWrite = AcidUtils.isInsertOnlyTable(tbl.getParameters());
    assert tbl.getPath() != null : "null==getPath() for " + tbl.getTableName();
    boolean isFullAcidTable = AcidUtils.isFullAcidTable(tbl);
    List<FileStatus> newFileStatuses = null;
    try {
      PerfLogger perfLogger = SessionState.getPerfLogger();

      /**
       * Move files before creating the partition since down stream processes
       * check for existence of partition in metadata before accessing the data.
       * If partition is created before data is moved, downstream waiting
       * processes might move forward with partial data
       */

      Path oldPartPath = (oldPart != null) ? oldPart.getDataLocation() : null;
      Path newPartPath = null;

      if (inheritLocation) {
        newPartPath = genPartPathFromTable(tbl, partSpec, tblDataLocationPath);

        if(oldPart != null) {
          /*
           * If we are moving the partition across filesystem boundaries
           * inherit from the table properties. Otherwise (same filesystem) use the
           * original partition location.
           *
           * See: HIVE-1707 and HIVE-2117 for background
           */
          FileSystem oldPartPathFS = oldPartPath.getFileSystem(getConf());
          FileSystem tblPathFS = tblDataLocationPath.getFileSystem(getConf());
          if (FileUtils.isEqualFileSystemAndSameOzoneBucket(oldPartPathFS, tblPathFS, oldPartPath, tblDataLocationPath)) {
            newPartPath = oldPartPath;
          }
        }
      } else {
        newPartPath = oldPartPath == null
          ? genPartPathFromTable(tbl, partSpec, tblDataLocationPath) : oldPartPath;
      }

      perfLogger.perfLogBegin("MoveTask", PerfLogger.FILE_MOVES);

      // Note: the stats for ACID tables do not have any coordination with either Hive ACID logic
      //       like txn commits, time outs, etc.; nor the lower level sync in metastore pertaining
      //       to ACID updates. So the are not themselves ACID.

      // Note: this assumes both paths are qualified; which they are, currently.
      if (((isMmTableWrite || isDirectInsert || isFullAcidTable) && loadPath.equals(newPartPath)) ||
              (loadFileType == LoadFileType.IGNORE)) {
        // MM insert query or direct insert; move itself is a no-op.
        if (Utilities.FILE_OP_LOGGER.isTraceEnabled()) {
          Utilities.FILE_OP_LOGGER.trace("not moving " + loadPath + " to " + newPartPath + " (MM = " + isMmTableWrite
              + ", Direct insert = " + isDirectInsert + ")");
        }
        if (newFiles != null) {
          if (!newFiles.isEmpty()) {
            newFileStatuses = new ArrayList<>();
            newFileStatuses.addAll(newFiles);
          } else {
            newFileStatuses = listFilesCreatedByQuery(loadPath, writeId, stmtId);
            newFiles.addAll(newFileStatuses);
          }
        }
        if (Utilities.FILE_OP_LOGGER.isTraceEnabled()) {
          Utilities.FILE_OP_LOGGER.trace("maybe deleting stuff from " + oldPartPath
              + " (new " + newPartPath + ") for replace");
        }
      } else {
        // Either a non-MM query, or a load into MM table from an external source.
        Path destPath = newPartPath;
        if (isMmTableWrite) {
          // We will load into MM directory, and hide previous directories if needed.
          destPath = new Path(destPath, isInsertOverwrite
              ? AcidUtils.baseDir(writeId) : AcidUtils.deltaSubdir(writeId, writeId, stmtId));
        }
        if (!isAcidIUDoperation && isFullAcidTable) {
          destPath = fixFullAcidPathForLoadData(loadFileType, destPath, writeId, stmtId, tbl);
        }
        if (Utilities.FILE_OP_LOGGER.isTraceEnabled()) {
          Utilities.FILE_OP_LOGGER.trace("moving " + loadPath + " to " + destPath);
        }

        boolean isManaged = tbl.getTableType() == TableType.MANAGED_TABLE;
        // TODO: why is "&& !isAcidIUDoperation" needed here?
        if (!isTxnTable && ((loadFileType == LoadFileType.REPLACE_ALL) || (oldPart == null && !isAcidIUDoperation))) {
          //for fullAcid tables we don't delete files for commands with OVERWRITE - we create a new
          // base_x.  (there is Insert Overwrite and Load Data Overwrite)
          boolean isSkipTrash = MetaStoreUtils.isSkipTrash(tbl.getParameters());
          boolean needRecycle = !tbl.isTemporary()
              && ReplChangeManager.shouldEnableCm(getDatabase(tbl.getDbName()), tbl.getTTable());
          replaceFiles(tbl.getPath(), loadPath, destPath, oldPartPath, getConf(), isSrcLocal,
              isSkipTrash, newFiles, FileUtils.HIDDEN_FILES_PATH_FILTER, needRecycle, isManaged, isInsertOverwrite);
        } else {
          FileSystem fs = destPath.getFileSystem(conf);
          copyFiles(conf, loadPath, destPath, fs, isSrcLocal, isAcidIUDoperation,
              (loadFileType == LoadFileType.OVERWRITE_EXISTING), newFiles,
              tbl.getNumBuckets() > 0, isFullAcidTable, isManaged, false);
        }
      }
      perfLogger.perfLogEnd("MoveTask", PerfLogger.FILE_MOVES);
      Partition newTPart = oldPart != null ? oldPart : new Partition(tbl, partSpec, newPartPath);
      alterPartitionSpecInMemory(tbl, partSpec, newTPart.getTPartition(), inheritTableSpecs, newPartPath.toString());
      validatePartition(newTPart);

      // If config is set, table is not temporary and partition being inserted exists, capture
      // the list of files added. For not yet existing partitions (insert overwrite to new partition
      // or dynamic partition inserts), the add partition event will capture the list of files added.
      // Generate an insert event only if inserting into an existing partition
      // When inserting into a new partition, the add partition event takes care of insert event
      if ((null != oldPart) && (null != newFiles)) {
        if (isTxnTable) {
          addWriteNotificationLog(tbl, partSpec, newFiles, writeId, null);
        } else {
          fireInsertEvent(tbl, partSpec, (loadFileType == LoadFileType.REPLACE_ALL), newFiles);
        }
      } else {
        LOG.debug("No new files were created, and is not a replace, or we're inserting into a "
                + "partition that does not exist yet. Skipping generating INSERT event.");
      }

      // recreate the partition if it existed before
      if (isSkewedStoreAsSubdir) {
        org.apache.hadoop.hive.metastore.api.Partition newCreatedTpart = newTPart.getTPartition();
        SkewedInfo skewedInfo = newCreatedTpart.getSd().getSkewedInfo();
        /* Construct list bucketing location mappings from sub-directory name. */
        Map<List<String>, String> skewedColValueLocationMaps = constructListBucketingLocationMap(
            newPartPath, skewedInfo);
        /* Add list bucketing location mappings. */
        skewedInfo.setSkewedColValueLocationMaps(skewedColValueLocationMaps);
        newCreatedTpart.getSd().setSkewedInfo(skewedInfo);
      }

      // If there is no column stats gather stage present in the plan. So we don't know the accuracy of the stats or
      // auto gather stats is turn off explicitly. We need to reset the stats in both cases.
      if (resetStatistics || !this.getConf().getBoolVar(HiveConf.ConfVars.HIVE_STATS_AUTOGATHER)) {
        LOG.debug(
            "Clear partition column statistics by setting basic stats to false for " + newTPart.getCompleteName());
        StatsSetupConst.setBasicStatsState(newTPart.getParameters(), StatsSetupConst.FALSE);
      }

      if (oldPart == null) {
        newTPart.getTPartition().setParameters(new HashMap<String,String>());
        if (this.getConf().getBoolVar(HiveConf.ConfVars.HIVE_STATS_AUTOGATHER)) {
          StatsSetupConst.setStatsStateForCreateTable(newTPart.getParameters(),
              MetaStoreUtils.getColumnNames(tbl.getCols()), StatsSetupConst.TRUE);
        }
        // Note: we are creating a brand new the partition, so this is going to be valid for ACID.
        List<FileStatus> filesForStats = null;
        if (newFileStatuses != null && !newFileStatuses.isEmpty()) {
          filesForStats = newFileStatuses;
        } else {
          if (isTxnTable) {
            filesForStats = AcidUtils.getAcidFilesForStats(newTPart.getTable(), newPartPath, conf, null);
          } else {
            filesForStats = HiveStatsUtils.getFileStatusRecurse(newPartPath, -1, newPartPath.getFileSystem(conf));
          }
        }
        if (filesForStats != null) {
          MetaStoreServerUtils.populateQuickStats(filesForStats, newTPart.getParameters());
        } else {
          // The ACID state is probably absent. Warning is logged in the get method.
          MetaStoreServerUtils.clearQuickStats(newTPart.getParameters());
        }
      }
      return newTPart;
    } catch (IOException | MetaException | InvalidOperationException e) {
      LOG.error("Error in loadPartitionInternal", e);
      throw new HiveException(e);
    }
  }
Move all the files from loadPath into Hive. If the partition
does not exist - one is created - files in loadPath are moved into Hive. But the
directory itself is not removed.

@param loadPath
         Directory containing files to load into Table
@param tbl
         name of table to be loaded.
@param partSpec
         defines which partition needs to be loaded
@param oldPart
         already existing partition object, can be null
@param loadFileType
         if REPLACE_ALL - replace files in the table,
         otherwise add files to table (KEEP_EXISTING, OVERWRITE_EXISTING)
@param inheritTableSpecs if true, on [re]creating the partition, take the
         location/inputformat/outputformat/serde details from table spec
@param inheritLocation
         if true, partition path is generated from table
@param isSkewedStoreAsSubdir
         if true, skewed is stored as sub-directory
@param isSrcLocal
         If the source directory is LOCAL
@param isAcidIUDoperation
         true if this is an ACID operation Insert/Update/Delete operation
@param resetStatistics
         if true, reset the statistics. Do not reset statistics if false.
@param writeId
         write ID allocated for the current load operation
@param stmtId
         statement ID of the current load statement
@param isInsertOverwrite
@param isTxnTable

@return Partition object being loaded with data
@throws HiveException
ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java