HDFS-17581. Add FastCopy tool and support dfs -fastcp command

Author: LiuGuH

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileSystem.java

@@ -2544,6 +2544,10 @@ public void moveFromLocalFile(Path src, Path dst)
     copyFromLocalFile(true, src, dst);
   }
 
+  public void fastCopy(Path src, Path dst, boolean overwrite) throws IOException {
+    throw new UnsupportedOperationException("FileSystem does not support fastcopy method.");
+  }
+
   /**
    * The src file is on the local disk.  Add it to the filesystem at
    * the given dst name.

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/shell/CommandWithDestination.java

@@ -365,6 +365,14 @@ protected void copyFileToTarget(PathData src, PathData target)
       IOUtils.closeStream(in);
     }
   }
+
+  protected void copyFileToTargetWithFastCp(PathData src,PathData target) throws IOException {
+    final boolean preserveRawXattrs =
+        checkPathsForReservedRaw(src.path, target.path);
+    src.fs.setVerifyChecksum(verifyChecksum);
+    fastCopyToTarget(src, target);
+    preserveAttributes(src, target, preserveRawXattrs);
+  }
 
   /**
    * Check the source and target paths to ensure that they are either both in
@@ -433,6 +441,24 @@ protected void copyStreamToTarget(InputStream in, PathData target)
     }
   }
 
+  protected void fastCopyToTarget(PathData src, PathData target)
+      throws IOException {
+    if (target.exists && (target.stat.isDirectory() || !overwrite)) {
+      throw new PathExistsException(target.toString());
+    }
+    TargetFileSystem targetFs = new TargetFileSystem(target.fs);
+    try {
+      PathData tempTarget = direct ? target : target.suffix("._COPYING_");
+      targetFs.setWriteChecksum(writeChecksum);
+      src.fs.fastCopy(src.path, tempTarget.path, overwrite);
+      if (!direct) {
+        targetFs.rename(tempTarget, target);
+      }
+    } finally {
+      targetFs.close(); // last ditch effort to ensure temp file is removed
+    }
+  }
+
   /**
    * Preserve the attributes of the source to the target.
    * The method calls {@link #shouldPreserve(FileAttribute)} to check what

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/shell/CopyCommands.java

@@ -34,6 +34,7 @@
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.PathIsDirectoryException;
+import org.apache.hadoop.fs.PathOperationException;
 import org.apache.hadoop.io.IOUtils;
 
 /** Various commands for copy files */
@@ -44,6 +45,7 @@ class CopyCommands {
   public static void registerCommands(CommandFactory factory) {
     factory.addClass(Merge.class, "-getmerge");
     factory.addClass(Cp.class, "-cp");
+    factory.addClass(FastCp.class, "-fastcp");
     factory.addClass(CopyFromLocal.class, "-copyFromLocal");
     factory.addClass(CopyToLocal.class, "-copyToLocal");
     factory.addClass(Get.class, "-get");
@@ -209,7 +211,81 @@ private void popPreserveOption(List<String> args) {
       }
     }
   }
-  
+
+  static class FastCp extends CopyCommandWithMultiThread {
+    public static final String NAME = "fastcp";
+    public static final String USAGE =
+        "[-f] [-p | -p[topax]] [-d] [-t <thread count>]"
+            + " [-q <thread pool queue size>] <src> ... <dst>";
+    public static final String DESCRIPTION =
+        "FastCopy files that match the file pattern <src> to a destination."
+            + " When copying multiple files, the destination must be a "
+            + "directory.\nFlags :\n"
+            + "  -p[topax] : Preserve file attributes [topx] (timestamps, "
+            + "ownership, permission, ACL, XAttr). If -p is specified with "
+            + "no arg, then preserves timestamps, ownership, permission. "
+            + "If -pa is specified, then preserves permission also because "
+            + "ACL is a super-set of permission. Determination of whether raw "
+            + "namespace extended attributes are preserved is independent of "
+            + "the -p flag.\n"
+            + "  -f : Overwrite the destination if it already exists.\n"
+            + "  -d : Skip creation of temporary file(<dst>._COPYING_).\n"
+            + "  -t <thread count> : Number of threads to be used, "
+            + "default is 1.\n"
+            + "  -q <thread pool queue size> : Thread pool queue size to be "
+            + "used, default is 1024.\n";
+
+    @Override
+    protected void processOptions(LinkedList<String> args) throws IOException {
+      popPreserveOption(args);
+      CommandFormat cf = new CommandFormat(2, Integer.MAX_VALUE, "f", "d");
+      cf.addOptionWithValue("t");
+      cf.addOptionWithValue("q");
+      cf.parse(args);
+      setDirectWrite(cf.getOpt("d"));
+      setOverwrite(cf.getOpt("f"));
+      setThreadCount(cf.getOptValue("t"));
+      setThreadPoolQueueSize(cf.getOptValue("q"));
+      // should have a -r option
+      setRecursive(true);
+      getRemoteDestination(args);
+    }
+
+    private void popPreserveOption(List<String> args) {
+      for (Iterator<String> iter = args.iterator(); iter.hasNext(); ) {
+        String cur = iter.next();
+        if (cur.equals("--")) {
+          // stop parsing arguments when you see --
+          break;
+        } else if (cur.startsWith("-p")) {
+          iter.remove();
+          if (cur.length() == 2) {
+            setPreserve(true);
+          } else {
+            String attributes = cur.substring(2);
+            for (int index = 0; index < attributes.length(); index++) {
+              preserve(FileAttribute.getAttribute(attributes.charAt(index)));
+            }
+          }
+          return;
+        }
+      }
+    }
+
+    @Override
+    protected void processPath(PathData src, PathData dst) throws IOException {
+      if (src.stat.isSymlink()) {
+        // TODO: remove when FileContext is supported, this needs to either
+        // copy the symlink or deref the symlink
+        throw new PathOperationException(src.toString());
+      } else if (src.stat.isFile()) {
+        copyFileToTargetWithFastCp(src, dst);
+      } else if (src.stat.isDirectory() && !isRecursive()) {
+        throw new PathIsDirectoryException(src.toString());
+      }
+    }
+  }
+
   /** 
    * Copy local files to a remote filesystem
    */

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/TestFilterFileSystem.java

@@ -137,6 +137,7 @@ public Token<?>[] addDelegationTokens(String renewer, Credentials creds)
     void setQuota(Path f, long namespaceQuota, long storagespaceQuota);
     void setQuotaByStorageType(Path f, StorageType type, long quota);
     StorageStatistics getStorageStatistics();
+    public void fastCopy(Path src, Path dst, boolean overwrite) throws IOException;
 
     /*
     Not passed through as the inner implementation will miss features

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/TestHarFileSystem.java

@@ -234,6 +234,8 @@ public Collection<? extends BlockStoragePolicySpi> getAllStoragePolicies()
     public Collection<FileStatus> getTrashRoots(boolean allUsers) throws IOException;
     StorageStatistics getStorageStatistics();
 
+    public void fastCopy(Path src, Path dst, boolean overwrite) throws IOException;
+
     FutureDataInputStreamBuilder openFile(Path path)
         throws IOException, UnsupportedOperationException;
 

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java

@@ -693,6 +693,12 @@ public HdfsDataOutputStream next(final FileSystem fs, final Path p)
     }.resolve(this, absF);
   }
 
+  @Override
+  public void fastCopy(Path src, Path dst, boolean overwrite) throws IOException {
+    FastCopy fastCopy = new FastCopy(getConf(), src, dst, overwrite);
+    fastCopy.copyFile();
+  }
+
   /**
    * Same as create(), except fails if parent directory doesn't already exist.
    */