HDFS-7312. Update DistCp v1 to optionally not use tmp location. (Joseph Prosser via yzhangal)

CHANGES.txt

@@ -265,6 +265,9 @@ Release 1.3.0 - unreleased
     MAPREDUCE-6198. NPE from JobTracker#resolveAndAddToTopology in MR1 cause
     initJob and heartbeat failure. (zxu via rkanter)
 
+    HDFS-7312. Update DistCp v1 to optionally not use tmp location.
+    (Joseph Prosser via yzhangal)
+
 Release 1.2.2 - unreleased
 
   INCOMPATIBLE CHANGES

src/test/org/apache/hadoop/fs/TestCopyFiles.java

@@ -308,8 +308,26 @@ public class TestCopyFiles extends TestCase {
     deldir(localfs, TEST_ROOT_DIR+"/srcdat");
   }
 
-  /** copy files from dfs file system to dfs file system */
-  public void testCopyFromDfsToDfs() throws Exception {
+  private static void addToArgList(List<String> argList, final String... args) {
+	for (String arg : args) {
+	  argList.add(arg);
+	}
+  }
+  
+  private void addSrcDstToArgList(List<String> argList, final boolean skipTmp,
+      final String dst, final String... srcs) {
+    if (skipTmp) {
+      argList.add("-skiptmp");
+    }
+    addToArgList(argList, srcs);
+    argList.add(dst);
+  }
+
+  /**
+   * copy files from dfs file system to dfs file system Pass option to use
+   * -skiptmp flag
+   */
+  private void testCopyFromDfsToDfs(boolean skipTmp) throws Exception {
     String namenode = null;
     MiniDFSCluster cluster = null;
     try {
@@ -319,24 +337,37 @@ public class TestCopyFiles extends TestCase {
       namenode = FileSystem.getDefaultUri(conf).toString();
       if (namenode.startsWith("hdfs://")) {
         MyFile[] files = createFiles(URI.create(namenode), "/srcdat");
-        ToolRunner.run(new DistCp(conf), new String[] {
-                                         "-log",
-                                         namenode+"/logs",
-                                         namenode+"/srcdat",
-                                         namenode+"/destdat"});
+        List<String> argList = new ArrayList<String>();
+        addToArgList(argList, "-log", namenode + "/logs");
+        addSrcDstToArgList(argList, skipTmp, namenode + "/destdat", namenode
+            + "/srcdat");
+        ToolRunner.run(new DistCp(conf),
+            argList.toArray(new String[argList.size()]));
         assertTrue("Source and destination directories do not match.",
-                   checkFiles(hdfs, "/destdat", files));
-        FileSystem fs = FileSystem.get(URI.create(namenode+"/logs"), conf);
+            checkFiles(hdfs, "/destdat", files));
+        FileSystem fs = FileSystem.get(URI.create(namenode + "/logs"), conf);
         assertTrue("Log directory does not exist.",
-                   fs.exists(new Path(namenode+"/logs")));
+            fs.exists(new Path(namenode + "/logs")));
         deldir(hdfs, "/destdat");
         deldir(hdfs, "/srcdat");
         deldir(hdfs, "/logs");
       }
     } finally {
-      if (cluster != null) { cluster.shutdown(); }
+      if (cluster != null) {
+        cluster.shutdown();
+      }
     }
   }
+
+  /** copy files from dfs file system to dfs file system */
+  public void testCopyFromDfsToDfs() throws Exception {
+    testCopyFromDfsToDfs(false);
+  }
+
+  /** copy files from dfs file system to dfs file system with skiptmp */
+  public void testCopyFromDfsToDfsWithSkiptmp() throws Exception {
+    testCopyFromDfsToDfs(true);
+  }
   
   /** copy files from local file system to dfs file system */
   public void testCopyFromLocalToDfs() throws Exception {
@@ -362,7 +393,9 @@ public class TestCopyFiles extends TestCase {
         deldir(FileSystem.get(LOCAL_FS_URI, conf), TEST_ROOT_DIR+"/srcdat");
       }
     } finally {
-      if (cluster != null) { cluster.shutdown(); }
+      if (cluster != null) {
+        cluster.shutdown();
+      }
     }
   }
 
@@ -395,7 +428,8 @@ public class TestCopyFiles extends TestCase {
     }
   }
 
-  public void testCopyDfsToDfsUpdateOverwrite() throws Exception {
+  private void testCopyDfsToDfsUpdateOverwrite(boolean skipTmp)
+      throws Exception {
     MiniDFSCluster cluster = null;
     try {
       Configuration conf = new Configuration();
@@ -404,75 +438,86 @@ public class TestCopyFiles extends TestCase {
       final String namenode = hdfs.getUri().toString();
       if (namenode.startsWith("hdfs://")) {
         MyFile[] files = createFiles(URI.create(namenode), "/srcdat");
-        ToolRunner.run(new DistCp(conf), new String[] {
-                                         "-p",
-                                         "-log",
-                                         namenode+"/logs",
-                                         namenode+"/srcdat",
-                                         namenode+"/destdat"});
+        // run without update
+        List<String> argList = new ArrayList<String>();
+        addToArgList(argList, "-p", "-log", namenode + "/logs");
+        addSrcDstToArgList(argList, skipTmp, namenode + "/destdat", namenode
+            + "/srcdat");
+        ToolRunner.run(new DistCp(conf),
+            argList.toArray(new String[argList.size()]));
         assertTrue("Source and destination directories do not match.",
-                   checkFiles(hdfs, "/destdat", files));
-        FileSystem fs = FileSystem.get(URI.create(namenode+"/logs"), conf);
+            checkFiles(hdfs, "/destdat", files));
+        FileSystem fs = FileSystem.get(URI.create(namenode + "/logs"), conf);
         assertTrue("Log directory does not exist.",
-                    fs.exists(new Path(namenode+"/logs")));
+            fs.exists(new Path(namenode + "/logs")));
 
         FileStatus[] dchkpoint = getFileStatus(hdfs, "/destdat", files);
-        final int nupdate = NFILES>>2;
+        final int nupdate = NFILES >> 2;
         updateFiles(cluster.getFileSystem(), "/srcdat", files, nupdate);
         deldir(hdfs, "/logs");
-
-        ToolRunner.run(new DistCp(conf), new String[] {
-                                         "-p",
-                                         "-update",
-                                         "-log",
-                                         namenode+"/logs",
-                                         namenode+"/srcdat",
-                                         namenode+"/destdat"});
+        argList.clear();
+        // run with update
+        addToArgList(argList, "-p", "-update", "-log", namenode + "/logs");
+        addSrcDstToArgList(argList, skipTmp, namenode + "/destdat", namenode
+            + "/srcdat");
+        ToolRunner.run(new DistCp(conf),
+            argList.toArray(new String[argList.size()]));
         assertTrue("Source and destination directories do not match.",
-                   checkFiles(hdfs, "/destdat", files));
+            checkFiles(hdfs, "/destdat", files));
         assertTrue("Update failed to replicate all changes in src",
-                 checkUpdate(hdfs, dchkpoint, "/destdat", files, nupdate));
+            checkUpdate(hdfs, dchkpoint, "/destdat", files, nupdate));
 
         deldir(hdfs, "/logs");
-        ToolRunner.run(new DistCp(conf), new String[] {
-                                         "-p",
-                                         "-overwrite",
-                                         "-log",
-                                         namenode+"/logs",
-                                         namenode+"/srcdat",
-                                         namenode+"/destdat"});
+        argList.clear();
+        // run with overwrite
+        addToArgList(argList, "-p", "-overwrite", "-log", namenode + "/logs");
+        addSrcDstToArgList(argList, skipTmp, namenode + "/destdat", namenode
+            + "/srcdat");
+        ToolRunner.run(new DistCp(conf),
+            argList.toArray(new String[argList.size()]));
         assertTrue("Source and destination directories do not match.",
-                   checkFiles(hdfs, "/destdat", files));
+            checkFiles(hdfs, "/destdat", files));
         assertTrue("-overwrite didn't.",
-                 checkUpdate(hdfs, dchkpoint, "/destdat", files, NFILES));
+            checkUpdate(hdfs, dchkpoint, "/destdat", files, NFILES));
 
         deldir(hdfs, "/destdat");
         deldir(hdfs, "/srcdat");
         deldir(hdfs, "/logs");
       }
     } finally {
-      if (cluster != null) { cluster.shutdown(); }
+      if (cluster != null) {
+        cluster.shutdown();
+      }
     }
   }
 
-  public void testCopyDfsToDfsUpdateWithSkipCRC() throws Exception {
+  public void testCopyDfsToDfsUpdateOverwrite() throws Exception {
+    testCopyDfsToDfsUpdateOverwrite(false);
+  }
+
+  public void testCopyDfsToDfsUpdateOverwriteSkiptmp() throws Exception {
+    testCopyDfsToDfsUpdateOverwrite(true);
+  }
+
+  private void testCopyDfsToDfsUpdateWithSkipCRC(boolean skipTmp)
+      throws Exception {
     MiniDFSCluster cluster = null;
     try {
       Configuration conf = new Configuration();
       cluster = new MiniDFSCluster(conf, 2, true, null);
       final FileSystem hdfs = cluster.getFileSystem();
       final String namenode = hdfs.getUri().toString();
-      
+
       FileSystem fs = FileSystem.get(URI.create(namenode), new Configuration());
       // Create two files of the same name, same length but different
       // contents
       final String testfilename = "test";
       final String srcData = "act act act";
       final String destData = "cat cat cat";
-      
+
       if (namenode.startsWith("hdfs://")) {
-        deldir(hdfs,"/logs");
-        
+        deldir(hdfs, "/logs");
+
         Path srcPath = new Path("/srcdat", testfilename);
         Path destPath = new Path("/destdat", testfilename);
         FSDataOutputStream out = fs.create(srcPath, true);
@@ -482,17 +527,13 @@ public class TestCopyFiles extends TestCase {
         out = fs.create(destPath, true);
         out.writeUTF(destData);
         out.close();
-        
         // Run with -skipcrccheck option
-        ToolRunner.run(new DistCp(conf), new String[] {
-          "-p",
-          "-update",
-          "-skipcrccheck",
-          "-log",
-          namenode+"/logs",
-          namenode+"/srcdat",
-          namenode+"/destdat"});
-        
+        List<String> argList = new ArrayList<String>();
+        addToArgList(argList, "-p", "-update", "-skipcrccheck", "-log", namenode + "/logs");
+        addSrcDstToArgList(argList, skipTmp, namenode + "/destdat", namenode
+            + "/srcdat");
+        ToolRunner.run(new DistCp(conf),
+            argList.toArray(new String[argList.size()]));
         // File should not be overwritten
         FSDataInputStream in = hdfs.open(destPath);
         String s = in.readUTF();
@@ -500,18 +541,14 @@ public class TestCopyFiles extends TestCase {
         assertTrue("Dest got over written even with skip crc",
             s.equalsIgnoreCase(destData));
         in.close();
-        
         deldir(hdfs, "/logs");
-
-        // Run without the option        
-        ToolRunner.run(new DistCp(conf), new String[] {
-          "-p",
-          "-update",
-          "-log",
-          namenode+"/logs",
-          namenode+"/srcdat",
-          namenode+"/destdat"});
-        
+        argList.clear();
+        // Run without the option
+        addToArgList(argList, "-p", "-update", "-log", namenode + "/logs");
+        addSrcDstToArgList(argList, skipTmp, namenode + "/destdat", namenode
+            + "/srcdat");
+        ToolRunner.run(new DistCp(conf),
+            argList.toArray(new String[argList.size()]));
         // File should be overwritten
         in = hdfs.open(destPath);
         s = in.readUTF();
@@ -524,18 +561,29 @@ public class TestCopyFiles extends TestCase {
         deldir(hdfs, "/destdat");
         deldir(hdfs, "/srcdat");
         deldir(hdfs, "/logs");
-       }
+      }
     } finally {
-      if (cluster != null) { cluster.shutdown(); }
+      if (cluster != null) {
+        cluster.shutdown();
+      }
     }
   }
 
+  public void testCopyDfsToDfsUpdateWithSkipCRC() throws Exception {
+    testCopyDfsToDfsUpdateWithSkipCRC(false);
+  }
+
+  public void testCopyDfsToDfsUpdateWithSkipCRCSkiptmp() throws Exception {
+    testCopyDfsToDfsUpdateWithSkipCRC(true);
+  }
+
   /**
    * A helper function to test copying files between local file system and dfs
-   * file system, with staging area set to local file system. 
+   * file system, with staging area set to local file system.
    */
   private void stagingAreaTest(final FileSystem srcFs, final FileSystem destFs,
-      MiniDFSCluster cluster, Configuration conf) throws Exception {
+      MiniDFSCluster cluster, Configuration conf, boolean skipTmp)
+      throws Exception {
     try {
       final String fileDir = "/files";
       final String srcParent = "/srcdat";
@@ -548,45 +596,44 @@ public class TestCopyFiles extends TestCase {
       URI srcUri = srcFs.getUri();
       URI destUri = destFs.getUri();
 
-      final boolean isSrcLocalFs = srcUri.getScheme().equals(LOCAL_FS_URI.getScheme());
+      final boolean isSrcLocalFs =
+          srcUri.getScheme().equals(LOCAL_FS_URI.getScheme());
 
       final FileSystem localFs = FileSystem.get(LOCAL_FS_URI, conf);
       String prevStagingArea =
           conf.get(JT_STAGING_AREA_ROOT, JT_STAGING_AREA_ROOT_DEFAULT);
-      String newStagingArea = (isSrcLocalFs? source : destination);
+      String newStagingArea = (isSrcLocalFs ? source : destination);
       newStagingArea += "/STAGING";
       conf.set(JT_STAGING_AREA_ROOT, TEST_ROOT_DIR + newStagingArea);
-        
-      final String srcParentPrefix = isSrcLocalFs? TEST_ROOT_DIR : "";
-      final String destParentPrefix = isSrcLocalFs? "" : TEST_ROOT_DIR;
- 
+
+      final String srcParentPrefix = isSrcLocalFs ? TEST_ROOT_DIR : "";
+      final String destParentPrefix = isSrcLocalFs ? "" : TEST_ROOT_DIR;
+
       String createDelSrcParent = srcParentPrefix + srcParent;
       String createDelDestParent = destParentPrefix + destParent;
       String createDelSrc = createDelSrcParent + fileDir;
       String createDelDest = createDelDestParent + fileDir;
-      
+
       MyFile[] srcFiles = createFiles(srcUri, createDelSrc);
       createFiles(destUri, createDelDest);
 
-      String distcpSrc = String.valueOf(srcUri) + createDelSrc;     
+      String distcpSrc = String.valueOf(srcUri) + createDelSrc;
       String distcpDest = String.valueOf(destUri) + createDelDest;
-      
-      ToolRunner.run(new DistCp(conf), new String[] {
-        "-log",
-        LOCAL_FS_STR + logDir,
-        "-update",
-        "-delete",
-        distcpSrc,
-        distcpDest});
-          
-        assertTrue("Source and destination directories do not match.",
-            checkFiles(destFs, createDelDest, srcFiles));
 
-        deldir(localFs, logDir);
-        deldir(srcFs, createDelSrcParent);
-        deldir(destFs, createDelDestParent);
+      List<String> argList = new ArrayList<String>();
+      addToArgList(argList, "-log", LOCAL_FS_STR + logDir, "-update", "-delete");
+      addSrcDstToArgList(argList, skipTmp, distcpDest, distcpSrc);
+      ToolRunner.run(new DistCp(conf),
+          argList.toArray(new String[argList.size()]));
+
+      assertTrue("Source and destination directories do not match.",
+          checkFiles(destFs, createDelDest, srcFiles));
+
+      deldir(localFs, logDir);
+      deldir(srcFs, createDelSrcParent);
+      deldir(destFs, createDelDestParent);
 
-        conf.set(JT_STAGING_AREA_ROOT, prevStagingArea); 
+      conf.set(JT_STAGING_AREA_ROOT, prevStagingArea);
     } finally {
       if (cluster != null) {
         cluster.shutdown();
@@ -598,37 +645,108 @@ public class TestCopyFiles extends TestCase {
    * test copying files from local file system to dfs file system with staging
    * area in src
    */
-  public void testCopyFromLocalToDfsWithStagingAreaInSrc() throws Exception {
+  private void testCopyFromLocalToDfsWithStagingAreaInSrc(boolean skipTmp)
+      throws Exception {
     Configuration conf = new Configuration();
     MiniDFSCluster cluster = new MiniDFSCluster(conf, 2, true, null);
 
     String namenode = FileSystem.getDefaultUri(conf).toString();
-    assertTrue("Name node doesn't start with hdfs://", namenode.startsWith("hdfs://"));
-    
+    assertTrue("Name node doesn't start with hdfs://",
+        namenode.startsWith("hdfs://"));
+
     final FileSystem srcFs = FileSystem.get(LOCAL_FS_URI, conf);
     final FileSystem destFs = cluster.getFileSystem();
-    
-    stagingAreaTest(srcFs, destFs, cluster, conf);
+
+    stagingAreaTest(srcFs, destFs, cluster, conf, skipTmp);
+  }
+
+  public void testCopyFromLocalToDfsWithStagingAreaInSrcSkiptmp()
+      throws Exception {
+    testCopyFromLocalToDfsWithStagingAreaInSrc(true);
+  }
+
+  public void testCopyFromLocalToDfsWithStagingAreaInSrc() throws Exception {
+    testCopyFromLocalToDfsWithStagingAreaInSrc(false);
   }
 
   /**
    * test copying files from dfs file system to local file system with staging
-   * area in dest
+   * area in dest and setting skiptmp flag as needed
    */
-  public void testCopyFromDfsToLocalWithStagingAreaInDest() throws Exception {
+  public void testCopyFromDfsToLocalWithStagingAreaInDest(boolean skipTmp)
+      throws Exception {
     Configuration conf = new Configuration();
     MiniDFSCluster cluster = new MiniDFSCluster(conf, 2, true, null);
-    
+
     String namenode = FileSystem.getDefaultUri(conf).toString();
-    assertTrue("Name node doesn't start with hdfs://", namenode.startsWith("hdfs://"));
-    
+    assertTrue("Name node doesn't start with hdfs://",
+        namenode.startsWith("hdfs://"));
+
     final FileSystem srcFs = cluster.getFileSystem();
     final FileSystem destFs = FileSystem.get(LOCAL_FS_URI, conf);
-    
-    stagingAreaTest(srcFs, destFs, cluster, conf);
+
+    stagingAreaTest(srcFs, destFs, cluster, conf, skipTmp);
+  }
+  
+  
+  /**
+   * test copying files from dfs file system to local file system with staging
+   * area in dest and skiptmp set
+   */
+  public void testCopyFromDfsToLocalWithStagingAreaInDestSkiptmp()
+    throws Exception {
+	testCopyFromDfsToLocalWithStagingAreaInDest(true);
+  }
+
+  /**
+   * test copying files from dfs file system to local file system with staging
+   * area in dest
+   */
+  public void testCopyFromDfsToLocalWithStagingAreaInDest() throws Exception {
+	testCopyFromDfsToLocalWithStagingAreaInDest(false);
+  }
+
+  /**
+   * test copying files from dfs file system to local file system with staging
+   * area in dest. Optionally set skiptmp flag
+   */
+  private void testCopyDuplication(boolean skipTmp) throws Exception {
+    final FileSystem localfs =
+        FileSystem.get(LOCAL_FS_URI, new Configuration());
+    try {
+      MyFile[] files = createFiles(localfs, TEST_ROOT_DIR + "/srcdat");
+      List<String> argList = new ArrayList<String>();
+      addSrcDstToArgList(argList, skipTmp, LOCAL_FS_STR + TEST_ROOT_DIR
+          + "/src2/srcdat", LOCAL_FS_STR + TEST_ROOT_DIR + "/srcdat");
+      ToolRunner.run(new DistCp(new Configuration()),
+          argList.toArray(new String[argList.size()]));
+      assertTrue("Source and destination directories do not match.",
+          checkFiles(localfs, TEST_ROOT_DIR + "/src2/srcdat", files));
+      argList.clear();
+      
+      addSrcDstToArgList(argList, skipTmp, LOCAL_FS_STR + TEST_ROOT_DIR
+          + "/destdat", LOCAL_FS_STR + TEST_ROOT_DIR + "/srcdat",
+          LOCAL_FS_STR + TEST_ROOT_DIR + "/src2/srcdat");
+      assertEquals(
+          DistCp.DuplicationException.ERROR_CODE,
+          ToolRunner.run(new DistCp(new Configuration()),
+              argList.toArray(new String[argList.size()])));
+    } finally {
+      deldir(localfs, TEST_ROOT_DIR + "/destdat");
+      deldir(localfs, TEST_ROOT_DIR + "/srcdat");
+      deldir(localfs, TEST_ROOT_DIR + "/src2");
+    }
   }
 
   public void testCopyDuplication() throws Exception {
+    testCopyDuplication(false);
+  }
+
+  public void testCopyDuplicationSkiptmp() throws Exception {
+    testCopyDuplication(true);
+  }
+
+  public void oldtestCopyDuplication() throws Exception {
     final FileSystem localfs = FileSystem.get(LOCAL_FS_URI, new Configuration());
     try {    
       MyFile[] files = createFiles(localfs, TEST_ROOT_DIR+"/srcdat");
@@ -651,54 +769,70 @@ public class TestCopyFiles extends TestCase {
     }
   }
 
-  public void testCopySingleFile() throws Exception {
+  private void testCopySingleFile(boolean skipTmp) throws Exception {
     FileSystem fs = FileSystem.get(LOCAL_FS_URI, new Configuration());
-    Path root = new Path(TEST_ROOT_DIR+"/srcdat");
-    try {    
-      MyFile[] files = {createFile(root, fs)};
-      //copy a dir with a single file
+    Path root = new Path(TEST_ROOT_DIR + "/srcdat");
+    try {
+      MyFile[] files = { createFile(root, fs) };
+      List<String> argList = new ArrayList<String>();
+      // copy a dir with a single file
+      addSrcDstToArgList(argList, skipTmp, LOCAL_FS_STR + TEST_ROOT_DIR
+          + "/destdat", LOCAL_FS_STR + TEST_ROOT_DIR + "/srcdat");
       ToolRunner.run(new DistCp(new Configuration()),
-          new String[] {LOCAL_FS_STR+TEST_ROOT_DIR+"/srcdat",
-                        LOCAL_FS_STR+TEST_ROOT_DIR+"/destdat"});
+          argList.toArray(new String[argList.size()]));
       assertTrue("Source and destination directories do not match.",
-                 checkFiles(fs, TEST_ROOT_DIR+"/destdat", files));
-      
-      //copy a single file
+          checkFiles(fs, TEST_ROOT_DIR + "/destdat", files));
+      argList.clear();
+      // copy a single file
       String fname = files[0].getName();
       Path p = new Path(root, fname);
       FileSystem.LOG.info("fname=" + fname + ", exists? " + fs.exists(p));
+      addSrcDstToArgList(argList, skipTmp, LOCAL_FS_STR + TEST_ROOT_DIR
+          + "/dest2/" + fname, LOCAL_FS_STR + TEST_ROOT_DIR + "/srcdat/"
+          + fname);
       ToolRunner.run(new DistCp(new Configuration()),
-          new String[] {LOCAL_FS_STR+TEST_ROOT_DIR+"/srcdat/"+fname,
-                        LOCAL_FS_STR+TEST_ROOT_DIR+"/dest2/"+fname});
+          argList.toArray(new String[argList.size()]));
       assertTrue("Source and destination directories do not match.",
-          checkFiles(fs, TEST_ROOT_DIR+"/dest2", files));     
-      //copy single file to existing dir
-      deldir(fs, TEST_ROOT_DIR+"/dest2");
-      fs.mkdirs(new Path(TEST_ROOT_DIR+"/dest2"));
-      MyFile[] files2 = {createFile(root, fs, 0)};
+          checkFiles(fs, TEST_ROOT_DIR + "/dest2", files));
+      argList.clear();
+      // copy single file to existing dir
+      deldir(fs, TEST_ROOT_DIR + "/dest2");
+      fs.mkdirs(new Path(TEST_ROOT_DIR + "/dest2"));
+      MyFile[] files2 = { createFile(root, fs, 0) };
       String sname = files2[0].getName();
+      addToArgList(argList, "-update");
+      addSrcDstToArgList(argList, skipTmp, LOCAL_FS_STR + TEST_ROOT_DIR
+          + "/dest2/", LOCAL_FS_STR + TEST_ROOT_DIR + "/srcdat/" + sname);
       ToolRunner.run(new DistCp(new Configuration()),
-          new String[] {"-update",
-                        LOCAL_FS_STR+TEST_ROOT_DIR+"/srcdat/"+sname,
-                        LOCAL_FS_STR+TEST_ROOT_DIR+"/dest2/"});
+          argList.toArray(new String[argList.size()]));
       assertTrue("Source and destination directories do not match.",
-          checkFiles(fs, TEST_ROOT_DIR+"/dest2", files2));     
-      updateFiles(fs, TEST_ROOT_DIR+"/srcdat", files2, 1);
-      //copy single file to existing dir w/ dst name conflict
+          checkFiles(fs, TEST_ROOT_DIR + "/dest2", files2));
+      updateFiles(fs, TEST_ROOT_DIR + "/srcdat", files2, 1);
+      argList.clear();
+      // copy single file to existing dir w/ dst name conflict
+      addToArgList(argList, "-update");
+      addSrcDstToArgList(argList, skipTmp, LOCAL_FS_STR + TEST_ROOT_DIR
+          + "/dest2/", LOCAL_FS_STR + TEST_ROOT_DIR + "/srcdat/" + sname);
       ToolRunner.run(new DistCp(new Configuration()),
-          new String[] {"-update",
-                        LOCAL_FS_STR+TEST_ROOT_DIR+"/srcdat/"+sname,
-                        LOCAL_FS_STR+TEST_ROOT_DIR+"/dest2/"});
+          argList.toArray(new String[argList.size()]));
       assertTrue("Source and destination directories do not match.",
-          checkFiles(fs, TEST_ROOT_DIR+"/dest2", files2));     
-    }
-    finally {
-      deldir(fs, TEST_ROOT_DIR+"/destdat");
-      deldir(fs, TEST_ROOT_DIR+"/dest2");
-      deldir(fs, TEST_ROOT_DIR+"/srcdat");
+          checkFiles(fs, TEST_ROOT_DIR + "/dest2", files2));
+    } finally {
+      deldir(fs, TEST_ROOT_DIR + "/destdat");
+      deldir(fs, TEST_ROOT_DIR + "/dest2");
+      deldir(fs, TEST_ROOT_DIR + "/srcdat");
     }
   }
 
+  public void testCopySingleFile() throws Exception {
+    testCopySingleFile(false);
+  }
+
+  public void testCopySingleFileWithSkiptmp() throws Exception {
+    testCopySingleFile(true);
+  }
+  
+  
   public void testPreserveOption() throws Exception {
     Configuration conf = new Configuration();
     MiniDFSCluster cluster = null;

src/tools/org/apache/hadoop/tools/DistCp.java

@@ -94,6 +94,8 @@ public class DistCp implements Tool {
     "\n                       -p alone is equivalent to -prbugp" +
     "\n-i                     Ignore failures" +
     "\n-log <logdir>          Write logs to <logdir>" +
+    "\n-skiptmp               Do not copy files to tmp directory and rename." +
+    "\n                       Instead, copy files directly to the final destination." +
     "\n-m <num_maps>          Maximum number of simultaneous copies" +
     "\n-overwrite             Overwrite destination" +
     "\n-update                Overwrite if src size different from dst size" +
@@ -118,7 +120,14 @@ public class DistCp implements Tool {
     "\n     specified with symbolic representation.  For examples," +
     "\n       1230k = 1230 * 1024 = 1259520" +
     "\n       891g = 891 * 1024^3 = 956703965184" +
-    
+
+    "\n\nNOTE 3: By default, distcp copies files to temporary area first, " + 
+    "\n     then renames to the final destination. Using -skiptmp " +
+    "\n     switch means that distcp copies files directly to the " +
+    "\n     destination. Recommend to use it only when you really need to " +
+    "\n     (such as to avoid copy/rename overhead in s3, where rename is " +
+    "\n     not natively supported), because it may cause damage to " +
+    "\n     existing destination file if distcp fails for some reason. " +
     "\n";
   
   private static final long BYTES_PER_MAP =  256 * 1024 * 1024;
@@ -134,7 +143,8 @@ public class DistCp implements Tool {
     PRESERVE_STATUS("-p", NAME + ".preserve.status"),
     OVERWRITE("-overwrite", NAME + ".overwrite.always"),
     UPDATE("-update", NAME + ".overwrite.ifnewer"),
-    SKIPCRC("-skipcrccheck", NAME + ".skip.crc.check");
+    SKIPCRC("-skipcrccheck", NAME + ".skip.crc.check"),
+    SKIPTMP("-skiptmp", NAME + ".skip.tmp");
 
     final String cmd, propertyname;
 
@@ -327,6 +337,7 @@ public class DistCp implements Tool {
     private byte[] buffer = null;
     private JobConf job;
     private boolean skipCRCCheck = false;
+    private boolean skipTmp = false;
 
     // stats
     private int failcount = 0;
@@ -382,10 +393,17 @@ public class DistCp implements Tool {
     private void copy(FileStatus srcstat, Path relativedst,
         OutputCollector<WritableComparable<?>, Text> outc, Reporter reporter)
         throws IOException {
-      Path absdst = new Path(destPath, relativedst);
       int totfiles = job.getInt(SRC_COUNT_LABEL, -1);
       assert totfiles >= 0 : "Invalid file count " + totfiles;
 
+      // if we are copying a single file and the dest doesn't exist, we
+      // treat it as a copy/rename. The relativedst becomes the new
+      // filename and the destPath becomes its parent directory.
+      if (totfiles == 1 && !destFileSys.exists(destPath)) {
+        relativedst = new Path(destPath.getName());
+        destPath = destPath.getParent();
+      }
+      Path  absdst = new Path(destPath, relativedst);
       // if a directory, ensure created even if empty
       if (srcstat.isDir()) {
         if (destFileSys.exists(absdst)) {
@@ -444,23 +462,16 @@ public class DistCp implements Tool {
             + " from " + srcstat.getPath());        
       }
       else {
-        if (totfiles == 1) {
-          // Copying a single file; use dst path provided by user as destination
-          // rather than destination directory, if a file
-          Path dstparent = absdst.getParent();
-          if (!(destFileSys.exists(dstparent) &&
-                destFileSys.getFileStatus(dstparent).isDir())) {
-            absdst = dstparent;
-          }
-        }
         if (destFileSys.exists(absdst) &&
             destFileSys.getFileStatus(absdst).isDir()) {
           throw new IOException(absdst + " is a directory");
         }
         if (!destFileSys.mkdirs(absdst.getParent())) {
-          throw new IOException("Failed to craete parent dir: " + absdst.getParent());
+          throw new IOException("Failed to create parent dir: " + absdst.getParent());
+        }
+        if (!skipTmp){
+            rename(tmpfile, absdst);
         }
-        rename(tmpfile, absdst);
 
         FileStatus dststat = destFileSys.getFileStatus(absdst);
         if (dststat.getLen() != srcstat.getLen()) {
@@ -530,6 +541,7 @@ public class DistCp implements Tool {
       update = job.getBoolean(Options.UPDATE.propertyname, false);
       overwrite = !update && job.getBoolean(Options.OVERWRITE.propertyname, false);
       skipCRCCheck = job.getBoolean(Options.SKIPCRC.propertyname, false);
+      skipTmp = job.getBoolean(Options.SKIPTMP.propertyname, false);
       this.job = job;
     }
 
@@ -654,7 +666,6 @@ public class DistCp implements Tool {
     LOG.info("destPath=" + args.dst);
 
     JobConf job = createJobConf(conf);
-    
     checkSrcPath(job, args.srcs);
     if (args.preservedAttributes != null) {
       job.set(PRESERVE_STATUS_LABEL, args.preservedAttributes);
@@ -671,7 +682,9 @@ public class DistCp implements Tool {
       finalize(conf, job, args.dst, args.preservedAttributes);
     } finally {
       //delete tmp
-      fullyDelete(job.get(TMP_DIR_LABEL), job);
+      if(!args.flags.contains(Options.SKIPTMP)) {
+        fullyDelete(job.get(TMP_DIR_LABEL), job);
+      }
       //delete jobDirectory
       fullyDelete(job.get(JOB_DIR_LABEL), job);
     }
@@ -945,7 +958,7 @@ public class DistCp implements Tool {
    * command line) and at most (distcp.max.map.tasks, default
    * MAX_MAPS_PER_NODE * nodes in the cluster).
    * @param totalBytes Count of total bytes for job
-   * @param job The job to configure
+   * @param job The job configuration
    * @return Count of maps to run.
    */
   private static void setMapCount(long totalBytes, JobConf job) 
@@ -962,7 +975,9 @@ public class DistCp implements Tool {
   static void fullyDelete(String dir, Configuration conf) throws IOException {
     if (dir != null) {
       Path tmp = new Path(dir);
-      tmp.getFileSystem(conf).delete(tmp, true);
+      if (tmp.getFileSystem(conf).exists(tmp)){
+        tmp.getFileSystem(conf).delete(tmp, true);
+      }
     }
   }
 
@@ -1013,9 +1028,11 @@ public class DistCp implements Tool {
     //set boolean values
     final boolean update = args.flags.contains(Options.UPDATE);
     final boolean skipCRCCheck = args.flags.contains(Options.SKIPCRC);
+    final boolean skipTmp = args.flags.contains(Options.SKIPTMP);
     final boolean overwrite = !update && args.flags.contains(Options.OVERWRITE);
     jobConf.setBoolean(Options.UPDATE.propertyname, update);
     jobConf.setBoolean(Options.SKIPCRC.propertyname, skipCRCCheck);
+    jobConf.setBoolean(Options.SKIPTMP.propertyname, skipTmp);
     jobConf.setBoolean(Options.OVERWRITE.propertyname, overwrite);
     jobConf.setBoolean(Options.IGNORE_READ_FAILURES.propertyname,
         args.flags.contains(Options.IGNORE_READ_FAILURES));
@@ -1205,9 +1222,10 @@ public class DistCp implements Tool {
           jobfs, jobDirectory, jobConf, conf);
     }
 
-    Path tmpDir = new Path(
-        (dstExists && !dstIsDir) || (!dstExists && srcCount == 1)?
-        args.dst.getParent(): args.dst, "_distcp_tmp_" + randomId);
+    String tmpDirPrefix = (dstExists && !dstIsDir) || (!dstExists && srcCount == 1) ? 
+    		args.dst.getParent().toString() : args.dst.toString();
+    Path tmpDir = new Path(tmpDirPrefix + (skipTmp? "" : "/_distcp_tmp_" + randomId));
+
     jobConf.set(TMP_DIR_LABEL, tmpDir.toUri().toString());
     LOG.info("sourcePathsCount=" + srcCount);
     LOG.info("filesToCopyCount=" + fileCount);