@Override
  public int execute() {
    LOG.info("Started DirCopyTask for table {} from source: {} to target: {}", work.getTableName(), work.getFullyQualifiedSourcePath(),
        work.getFullyQualifiedTargetPath());
    HiveConf clonedConf = getConf(conf);
    String distCpDoAsUser = clonedConf.getVar(HiveConf.ConfVars.HIVE_DISTCP_DOAS_USER);
    Retryable retryable = Retryable.builder()
      .withHiveConf(clonedConf)
      .withRetryOnException(IOException.class).withFailOnException(SnapshotException.class).build();
    long startTime = System.currentTimeMillis();
    AtomicInteger retries = new AtomicInteger(-1);
    AtomicBoolean result = new AtomicBoolean(false);
    try {
      return retryable.executeCallable(() -> {
        retries.getAndIncrement();
        UserGroupInformation proxyUser = null;
        Path sourcePath = work.getFullyQualifiedSourcePath();
        Path targetPath = work.getFullyQualifiedTargetPath();
        try {
          if (clonedConf.getBoolVar(HiveConf.ConfVars.REPL_ADD_RAW_RESERVED_NAMESPACE)) {
            sourcePath = reservedRawPath(work.getFullyQualifiedSourcePath().toUri());
            targetPath = reservedRawPath(work.getFullyQualifiedTargetPath().toUri());
          }
          UserGroupInformation ugi = Utils.getUGI();
          String currentUser = ugi.getShortUserName();
          if (distCpDoAsUser != null && !currentUser.equals(distCpDoAsUser)) {
            proxyUser = UserGroupInformation.createProxyUser(
              distCpDoAsUser, UserGroupInformation.getLoginUser());
          }

          setTargetPathOwner(targetPath, sourcePath, proxyUser, clonedConf);
          try {
            if (!checkIfPathExist(sourcePath, proxyUser, clonedConf)) {
              LOG.info("Source path is missing. Ignoring exception.");
              return 0;
            }
          } catch (Exception ex) {
            LOG.warn("Source path missing check failed. ", ex);
            //Should be retried
            throw new IOException(ex);
          }
          if (!getWork().getCopyMode().equals(SnapshotUtils.SnapshotCopyMode.FALLBACK_COPY)) {
            LOG.info("Using Snapshot mode of copy for source: {} and target: {}", sourcePath, targetPath);
            // Use distcp with snapshots for copy.
            result.set(copyUsingDistCpSnapshots(sourcePath, targetPath, proxyUser, clonedConf));
          } else {
            LOG.info("Using Normal copy for source: {} and target: {}", sourcePath, targetPath);
            result.set(runFallbackDistCp(sourcePath, targetPath, proxyUser, clonedConf));
          }
          return 0;
        } finally {
          if (proxyUser != null) {
            FileSystem.closeAllForUGI(proxyUser);
          }
        }
      });
    } catch (Exception e) {
      LOG.error("Replication failed ", e);
      Exception ex = new SecurityException(ErrorMsg.REPL_RETRY_EXHAUSTED.format(e.getMessage()), e);
      setException(ex);
      return ReplUtils.handleException(true, ex, work.getDumpDirectory(), work.getMetricCollector(), getName(), clonedConf);
    } finally {
      String jobId = clonedConf.get(ReplUtils.DISTCP_JOB_ID_CONF, ReplUtils.DISTCP_JOB_ID_CONF_DEFAULT);
      LOG.info("DirCopyTask status for source: {} to  target: {}. Took {}. DistCp JobId {}. Number of retries {}. "
              + "Result: {}", work.getFullyQualifiedSourcePath(), work.getFullyQualifiedTargetPath(),
          ReplUtils.convertToHumanReadableTime(System.currentTimeMillis() - startTime), jobId, retries.get(),
          result.get() ? "SUCCEEDED" : "FAILED");

      // if distcp succeeded then update bytes copied counter
      if (result.get()) {
        FileSystem srcFs = null;
        try {
          srcFs = work.getFullyQualifiedSourcePath().getFileSystem(clonedConf);
        } catch (IOException e) {
          throw new RuntimeException(e);
        }
        ContentSummary summary = null;
        try {
          summary = srcFs.getContentSummary(work.getFullyQualifiedSourcePath());
        } catch (IOException e) {
          throw new RuntimeException(e);
        }
        final long totalBytesCopied = summary.getLength();
        LOG.debug("DirCopyTask copied {} number of bytes by using distcp", totalBytesCopied);
        // increment total bytes replicated counter
        if (work.getMetricCollector() != null) {
          work.getMetricCollector().incrementSizeOfDataReplicated(totalBytesCopied);
        }
      }
    }
  }
DirCopyTask, mainly to be used to copy External table data.
ql/src/java/org/apache/hadoop/hive/ql/exec/repl/DirCopyTask.java