HADOOP-5472. Change DistCp to support globbing of input paths. Contributed by Dhruba Borthakur and Rodrigo Schmidt

git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/trunk@778811 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -392,6 +392,9 @@ Trunk (unreleased changes)
     HADOOP-5438. Provide a single FileSystem method to create or open-for-append 
     to a file.  (He Yongqiang via dhruba)
 
+    HADOOP-5472. Change DistCp to support globbing of input paths.  (Dhruba
+    Borthakur and Rodrigo Schmidt via szetszwo)
+
   OPTIMIZATIONS
 
     HADOOP-5595. NameNode does not need to run a replicator to choose a

src/test/hdfs-with-mr/org/apache/hadoop/fs/TestCopyFiles.java

@@ -818,6 +818,36 @@ public class TestCopyFiles extends TestCase {
     }
   }
 
+  /** test globbing  */
+  public void testGlobbing() throws Exception {
+    String namenode = null;
+    MiniDFSCluster cluster = null;
+    try {
+      Configuration conf = new Configuration();
+      cluster = new MiniDFSCluster(conf, 2, true, null);
+      final FileSystem hdfs = cluster.getFileSystem();
+      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"});
+        assertTrue("Source and destination directories do not match.",
+                   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")));
+        deldir(hdfs, "/destdat");
+        deldir(hdfs, "/srcdat");
+        deldir(hdfs, "/logs");
+      }
+    } finally {
+      if (cluster != null) { cluster.shutdown(); }
+    }
+  }
+  
   static void create(FileSystem fs, Path f) throws IOException {
     FSDataOutputStream out = fs.create(f);
     try {

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

@@ -27,6 +27,7 @@ import java.io.InputStreamReader;
 import java.util.ArrayList;
 import java.util.EnumSet;
 import java.util.Iterator;
+import java.util.LinkedList;
 import java.util.List;
 import java.util.Random;
 import java.util.Stack;
@@ -616,15 +617,24 @@ public class DistCp implements Tool {
   private static void checkSrcPath(Configuration conf, List<Path> srcPaths
       ) throws IOException {
     List<IOException> rslt = new ArrayList<IOException>();
+    List<Path> unglobbed = new LinkedList<Path>(); 
     for (Path p : srcPaths) {
       FileSystem fs = p.getFileSystem(conf);
-      if (!fs.exists(p)) {
+      FileStatus[] inputs = fs.globStatus(p);
+      
+      if(inputs.length > 0) {
+        for (FileStatus onePath: inputs) {
+          unglobbed.add(onePath.getPath());
+        }
+      } else {
         rslt.add(new IOException("Input source " + p + " does not exist."));
       }
     }
     if (!rslt.isEmpty()) {
       throw new InvalidInputException(rslt);
     }
+    srcPaths.clear();
+    srcPaths.addAll(unglobbed);
   }
 
   /**