private SnapshotUtils.SnapshotCopyMode createSnapshotsAtSource(Path sourcePath, String snapshotPrefix,
      boolean isSnapshotEnabled, HiveConf conf, SnapshotUtils.ReplSnapshotCount replSnapshotCount, FileList snapPathFileList,
      ArrayList<String> prevSnaps, boolean isBootstrap) throws IOException {
    if (!isSnapshotEnabled) {
      LOG.info("Snapshot copy not enabled for path {} Will use normal distCp for copying data.", sourcePath);
      return FALLBACK_COPY;
    }
    DistributedFileSystem sourceDfs = SnapshotUtils.getDFS(sourcePath, conf);
    try {
      if(isBootstrap) {
        // Delete any pre existing snapshots.
        SnapshotUtils.deleteSnapshotIfExists(sourceDfs, sourcePath, firstSnapshot(snapshotPrefix), conf);
        SnapshotUtils.deleteSnapshotIfExists(sourceDfs, sourcePath, secondSnapshot(snapshotPrefix), conf);
        allowAndCreateInitialSnapshot(sourcePath, snapshotPrefix, conf, replSnapshotCount, snapPathFileList, sourceDfs);
        return INITIAL_COPY;
      }

      if (prevSnaps.contains(sourcePath.toString())) {
        // We already created a snapshot for this, just refresh the latest snapshot and leave.
        sourceDfs.deleteSnapshot(sourcePath, secondSnapshot(snapshotPrefix));
        replSnapshotCount.incrementNumDeleted();
        SnapshotUtils.createSnapshot(sourceDfs, sourcePath, secondSnapshot(snapshotPrefix), conf);
        replSnapshotCount.incrementNumCreated();
        snapPathFileList.add(sourcePath.toString());
        return SnapshotUtils
            .isSnapshotAvailable(sourceDfs, sourcePath, snapshotPrefix, OLD_SNAPSHOT, conf) ? DIFF_COPY : INITIAL_COPY;
      }
      // check if second snapshot exists.
      boolean isSecondSnapAvlb = SnapshotUtils.isSnapshotAvailable(sourceDfs, sourcePath, snapshotPrefix,
          OLD_SNAPSHOT, conf);
      if (isSecondSnapAvlb) {
        sourceDfs.deleteSnapshot(sourcePath, firstSnapshot(snapshotPrefix));
        replSnapshotCount.incrementNumDeleted();
        sourceDfs.renameSnapshot(sourcePath, secondSnapshot(snapshotPrefix), firstSnapshot(snapshotPrefix));
        SnapshotUtils.createSnapshot(sourceDfs, sourcePath, secondSnapshot(snapshotPrefix), conf);
        replSnapshotCount.incrementNumCreated();
        snapPathFileList.add(sourcePath.toString());
        return DIFF_COPY;
      } else {
        // Check if first snapshot is available
        boolean isFirstSnapshotAvailable =
            SnapshotUtils.isSnapshotAvailable(sourceDfs, sourcePath, snapshotPrefix, NEW_SNAPSHOT, conf);
        if (isFirstSnapshotAvailable) {
          sourceDfs.renameSnapshot(sourcePath, secondSnapshot(snapshotPrefix), firstSnapshot(snapshotPrefix));
          SnapshotUtils.createSnapshot(sourceDfs, sourcePath, secondSnapshot(snapshotPrefix), conf);
          replSnapshotCount.incrementNumCreated();
          snapPathFileList.add(sourcePath.toString());
          return DIFF_COPY;
        } else {
          allowAndCreateInitialSnapshot(sourcePath, snapshotPrefix, conf, replSnapshotCount, snapPathFileList,
              sourceDfs);
          return INITIAL_COPY;
        }
      }
    } catch (FileNotFoundException fnf) {
      // Source deleted is already handled and is not an abnormal scenario, log and return.
      LOG.debug("Can not enable snapshot for path: {}", sourcePath, fnf);
      return FALLBACK_COPY;
    }
  }
Creates copy task for the paths configured, irrespective of which table/partition belongs to it.
@param singlePathLocations paths to be copied. if the value is true then only the task is created.
@param fileList the tracking file which maintains the list of tasks.
@param conf Hive Configuration.
@throws Exception in case of any error.
ql/src/java/org/apache/hadoop/hive/ql/exec/repl/ReplExternalTables.java