HADOOP-2219. A new command "df -count" that counts the number of

files and directories.  (Tsz Wo (Nicholas), SZE via dhruba)



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

CHANGES.txt

@@ -35,6 +35,9 @@ Trunk (unreleased changes)
     HADOOP-2063. A new parameter to dfs -get command to fetch a file 
     even if it is corrupted.  (Tsz Wo (Nicholas), SZE via dhruba)
 
+    HADOOP-2219. A new command "df -count" that counts the number of
+    files and directories.  (Tsz Wo (Nicholas), SZE via dhruba)
+
   IMPROVEMENTS
 
     HADOOP-2655. Copy on write for data and metadata files in the 

src/java/org/apache/hadoop/dfs/ClientProtocol.java

@@ -21,6 +21,7 @@ import java.io.*;
 import org.apache.hadoop.ipc.VersionedProtocol;
 import org.apache.hadoop.dfs.FSConstants.UpgradeAction;
 import org.apache.hadoop.fs.permission.*;
+import org.apache.hadoop.fs.ContentSummary;
 
 /**********************************************************************
  * ClientProtocol is used by user code via 
@@ -35,9 +36,9 @@ interface ClientProtocol extends VersionedProtocol {
    * Compared to the previous version the following changes have been introduced:
    * (Only the latest change is reflected.
    * The log of historical changes can be retrieved from the svn).
-   * 24 : added fsync
+   * 25 : added {@link #getContentSummary(String path)}
    */
-  public static final long versionID = 24L;
+  public static final long versionID = 25L;
   
   ///////////////////////////////////////
   // File contents
@@ -436,9 +437,17 @@ interface ClientProtocol extends VersionedProtocol {
    * the specified directory.
    * @param src The string representation of the path
    * @return size of directory subtree in bytes
+   * @deprecated use {@link #getContentSummary(String)}
    */
+  @Deprecated
   public long getContentLength(String src) throws IOException;
 
+  /**
+   * Get {@link ContentSummary} rooted at the specified directory.
+   * @param path The string representation of the path
+   */
+  public ContentSummary getContentSummary(String path) throws IOException;
+
   /**
    * Write all metadata for this file into persistent storage.
    * The file must be currently open for writing.

src/java/org/apache/hadoop/dfs/DFSClient.java

@@ -553,10 +553,16 @@ class DFSClient implements FSConstants {
    * @param src
    * @throws IOException
    * @return the number of bytes in the subtree rooted at src
+   * @deprecated use {@link #getContentSummary(String)}
    */
+  @Deprecated
   public long getContentLength(String src
                                ) throws IOException {
-    return namenode.getContentLength(src);
+    return getContentSummary(src).getLength();
+  }
+  
+  ContentSummary getContentSummary(String src) throws IOException {
+    return namenode.getContentSummary(src);
   }
 
   /**

src/java/org/apache/hadoop/dfs/DFSFileInfo.java

@@ -48,7 +48,7 @@ class DFSFileInfo extends FileStatus {
    */
   public DFSFileInfo(String path, INode node) {
     // length is zero for directories
-    super(node.isDirectory() ? 0 : node.computeContentsLength(), 
+    super(node.isDirectory() ? 0 : node.computeContentSummary().getLength(), 
           node.isDirectory(), 
           node.isDirectory() ? 0 : ((INodeFile)node).getReplication(), 
           node.isDirectory() ? 0 : ((INodeFile)node).getPreferredBlockSize(),

src/java/org/apache/hadoop/dfs/DistributedFileSystem.java

@@ -169,6 +169,8 @@ public class DistributedFileSystem extends FileSystem {
     return dfs.exists(getPathName(f));
   }
 
+  /** {@inheritDoc} */
+  @Deprecated
   public long getContentLength(Path f) throws IOException {
     // If it is a directory, then issue a getContentLength
     // RPC to find the size of the entire subtree in one call.
@@ -179,7 +181,12 @@ public class DistributedFileSystem extends FileSystem {
         return dfspath.getContentsLength();
       }
     }
-    return dfs.getContentLength(getPathName(f));
+    return getContentSummary(f).getLength();
+  }
+
+  /** {@inheritDoc} */
+  public ContentSummary getContentSummary(Path f) throws IOException {
+    return dfs.getContentSummary(getPathName(f));
   }
 
   public FileStatus[] listStatus(Path p) throws IOException {

src/java/org/apache/hadoop/dfs/FSDirectory.java

@@ -22,6 +22,7 @@ import java.util.*;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.ContentSummary;
 import org.apache.hadoop.fs.permission.*;
 import org.apache.hadoop.metrics.MetricsRecord;
 import org.apache.hadoop.metrics.MetricsUtil;
@@ -692,9 +693,7 @@ class FSDirectory implements FSConstants {
     return src;
   }
 
-  /* Get the size of the directory subtree.
-   */
-  long getContentLength(String src) throws IOException {
+  ContentSummary getContentSummary(String src) throws IOException {
     String srcs = normalizePath(src);
     synchronized (rootDir) {
       INode targetNode = rootDir.getNode(srcs);
@@ -702,7 +701,7 @@ class FSDirectory implements FSConstants {
         throw new IOException(src + " does not exist");
       }
       else {
-        return targetNode.computeContentsLength();
+        return targetNode.computeContentSummary();
       }
     }
   }

src/java/org/apache/hadoop/dfs/FSNamesystem.java

@@ -31,6 +31,7 @@ import org.apache.hadoop.net.DNSToSwitchMapping;
 import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.net.NetworkTopology;
 import org.apache.hadoop.net.ScriptBasedMapping;
+import org.apache.hadoop.fs.ContentSummary;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.permission.*;
 import org.apache.hadoop.ipc.Server;
@@ -1117,7 +1118,7 @@ class FSNamesystem implements FSConstants, FSNamesystemMBean {
       if (!checkFileProgress(pendingFile, false)) {
         throw new NotReplicatedYetException("Not replicated yet:" + src);
       }
-      fileLength = pendingFile.computeContentsLength();
+      fileLength = pendingFile.computeContentSummary().getLength();
       blockSize = pendingFile.getPreferredBlockSize();
       clientNode = pendingFile.getClientNode();
       replication = (int)pendingFile.getReplication();
@@ -1598,16 +1599,11 @@ class FSNamesystem implements FSConstants, FSNamesystemMBean {
     return true;
   }
 
-  /* Get the size of the specified directory subtree.
-   * @param src The string representation of the path
-   * @throws IOException if path does not exist
-   * @return size in bytes
-   */
-  long getContentLength(String src) throws IOException {
+  ContentSummary getContentSummary(String src) throws IOException {
     if (isPermissionEnabled) {
       checkPermission(src, false, null, null, null, FsAction.READ_EXECUTE);
     }
-    return dir.getContentLength(src);
+    return dir.getContentSummary(src);
   }
 
   /** Persist all metadata about this file.

src/java/org/apache/hadoop/dfs/INode.java

@@ -24,15 +24,11 @@ import java.util.Collections;
 import java.util.Arrays;
 import java.util.List;
 import java.io.IOException;
-import java.io.DataOutput;
-import java.io.DataInput;
-import java.io.DataOutputStream;
 
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.ContentSummary;
 import org.apache.hadoop.fs.permission.*;
 import org.apache.hadoop.dfs.BlocksMap.BlockInfo;
-import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.io.UTF8;
 
 /**
  * We keep an in-memory representation of the file/block hierarchy.
@@ -139,7 +135,17 @@ abstract class INode implements Comparable<byte[]> {
    */
   abstract boolean isDirectory();
   abstract int collectSubtreeBlocks(List<Block> v);
-  abstract long computeContentsLength();
+
+  /** Compute {@link ContentSummary}. */
+  final ContentSummary computeContentSummary() {
+    long[] a = computeContentSummary(new long[]{0,0,0});
+    return new ContentSummary(a[0], a[1], a[2]);
+  }
+  /**
+   * @return an array of three longs. 
+   * 0: length, 1: file count, 2: directory count
+   */
+  abstract long[] computeContentSummary(long[] summary);
 
   /**
    * Get local file name
@@ -550,17 +556,15 @@ class INodeDirectory extends INode {
     return total;
   }
 
-  /**
-   */
-  long computeContentsLength() {
-    long total = 0;
-    if (children == null) {
-      return total;
-    }
-    for (INode child : children) {
-      total += child.computeContentsLength();
+  /** {@inheritDoc} */
+  long[] computeContentSummary(long[] summary) {
+    if (children != null) {
+      for (INode child : children) {
+        child.computeContentSummary(summary);
+      }
     }
-    return total;
+    summary[2]++;
+    return summary;
   }
 
   /**
@@ -704,12 +708,15 @@ class INodeFile extends INode {
     return 1;
   }
 
-  long computeContentsLength() {
-    long total = 0;
-    for (Block blk : blocks) {
-      total += blk.getNumBytes();
-    }
-    return total;
+  /** {@inheritDoc} */
+  long[] computeContentSummary(long[] summary) {
+    long bytes = 0;
+    for(Block blk : blocks) {
+      bytes += blk.getNumBytes();
+    }
+    summary[0] += bytes;
+    summary[1]++;
+    return summary;
   }
 
   /**

src/java/org/apache/hadoop/dfs/LocatedBlocks.java

@@ -42,7 +42,7 @@ public class LocatedBlocks implements Writable {
   }
   
   LocatedBlocks(INodeFile inode, List<LocatedBlock> blks) {
-    fileLength = inode.computeContentsLength();
+    fileLength = inode.computeContentSummary().getLength();
     blocks = blks;
   }
   

src/java/org/apache/hadoop/dfs/NameNode.java

@@ -19,6 +19,7 @@ package org.apache.hadoop.dfs;
 
 import org.apache.commons.logging.*;
 
+import org.apache.hadoop.fs.ContentSummary;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.Trash;
 import org.apache.hadoop.fs.permission.*;
@@ -526,13 +527,15 @@ public class NameNode implements ClientProtocol, DatanodeProtocol,
     namesystem.metaSave(filename);
   }
 
-  /* Get the size of the directory subtree.
-   * @param src The string representation of the path to the file
-   * @throws IOException if path does not exist
-   * @return size in bytes of the directory subtree
-   */
+  /** {@inheritDoc} */
+  @Deprecated
   public long getContentLength(String src)  throws IOException {
-    return namesystem.getContentLength(src);
+    return getContentSummary(src).getLength();
+  }
+
+  /** {@inheritDoc} */
+  public ContentSummary getContentSummary(String path) throws IOException {
+    return namesystem.getContentSummary(path);
   }
 
   /** {@inheritDoc} */

src/java/org/apache/hadoop/fs/ContentSummary.java

@@ -0,0 +1,80 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.fs;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.hadoop.io.Writable;
+
+/** Store the summary of a content (a directory or a file). */
+public class ContentSummary implements Writable{
+  private long length;
+  private long fileCount;
+  private long directoryCount;
+
+  /** Constructor */
+  public ContentSummary() {}
+  
+  /** Constructor */
+  public ContentSummary(long length, long fileCount, long directoryCount) {
+    this.length = length;
+    this.fileCount = fileCount;
+    this.directoryCount = directoryCount;
+  }
+
+  /** @return the length */
+  public long getLength() {return length;}
+
+  /** @return the directory count */
+  public long getDirectoryCount() {return directoryCount;}
+
+  /** @return the file count */
+  public long getFileCount() {return fileCount;}
+  
+  /** {@inheritDoc} */
+  public void write(DataOutput out) throws IOException {
+    out.writeLong(length);
+    out.writeLong(fileCount);
+    out.writeLong(directoryCount);
+  }
+
+  /** {@inheritDoc} */
+  public void readFields(DataInput in) throws IOException {
+    this.length = in.readLong();
+    this.fileCount = in.readLong();
+    this.directoryCount = in.readLong();
+  }
+  
+  /** 
+   * Output format:
+   * <----12----> <----12----> <-------18------->
+   *    DIR_COUNT   FILE_COUNT       CONTENT_SIZE FILE_NAME    
+   */
+  private static final String STRING_FORMAT = "%12d %12d %18d ";
+
+  /** The header string */
+  static String HEADER = String.format(
+      STRING_FORMAT.replace('d', 's'), "directories", "files", "bytes");
+
+  /** {@inheritDoc} */
+  public String toString() {
+    return String.format(STRING_FORMAT, directoryCount, fileCount, length);
+  }
+}

src/java/org/apache/hadoop/fs/FileSystem.java

@@ -595,20 +595,29 @@ public abstract class FileSystem extends Configured {
   /** Return the number of bytes of the given path 
    * If <i>f</i> is a file, return the size of the file;
    * If <i>f</i> is a directory, return the size of the directory tree
+   * @deprecated Use {@link #getContentSummary(Path)}.
    */
+  @Deprecated
   public long getContentLength(Path f) throws IOException {
-    if (!isDirectory(f)) {
+    return getContentSummary(f).getLength();
+  }
+
+  /** Return the {@link ContentSummary} of a given {@link Path}. */
+  public ContentSummary getContentSummary(Path f) throws IOException {
+    FileStatus status = getFileStatus(f);
+    if (!status.isDir()) {
       // f is a file
-      return getLength(f);
+      return new ContentSummary(status.getLen(), 1, 0);
     }
-      
-    // f is a diretory
-    Path[] contents = listPaths(f);
-    long size = 0;
-    for(int i=0; i<contents.length; i++) {
-      size += getContentLength(contents[i]);
+    // f is a directory
+    long[] summary = {0, 0, 1};
+    for(FileStatus s : listStatus(f)) {
+      ContentSummary c = getContentSummary(s.getPath());
+      summary[0] += c.getLength();
+      summary[1] += c.getFileCount();
+      summary[2] += c.getDirectoryCount();
     }
-    return size;
+    return new ContentSummary(summary[0], summary[1], summary[2]);
   }
 
   final private static PathFilter DEFAULT_FILTER = new PathFilter() {

src/java/org/apache/hadoop/fs/FsShell.java

@@ -30,6 +30,7 @@ import java.util.zip.GZIPInputStream;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.conf.Configured;
+import org.apache.hadoop.fs.shell.Count;
 import org.apache.hadoop.io.DataInputBuffer;
 import org.apache.hadoop.io.DataOutputBuffer;
 import org.apache.hadoop.io.IOUtils;
@@ -1256,6 +1257,7 @@ public class FsShell extends Configured implements Tool {
       "[" + FsShellPermissions.CHMOD_USAGE + "]\n\t" +
       "[" + FsShellPermissions.CHOWN_USAGE + "]\n\t" +
       "[" + FsShellPermissions.CHGRP_USAGE + "]\n\t" +      
+      "[" + Count.USAGE + "]\n\t" +      
       "[-help [cmd]]\n";
 
     String conf ="-conf <configuration file>:  Specify an application configuration file.";
@@ -1451,6 +1453,8 @@ public class FsShell extends Configured implements Tool {
       System.out.println(chown);
     } else if ("chgrp".equals(cmd)) {
       System.out.println(chgrp);
+    } else if (Count.NAME.equals(cmd)) {
+      System.out.println(Count.DESCRIPTION);
     } else if ("help".equals(cmd)) {
       System.out.println(help);
     } else {
@@ -1477,6 +1481,7 @@ public class FsShell extends Configured implements Tool {
       System.out.println(chmod);
       System.out.println(chown);      
       System.out.println(chgrp);
+      System.out.println(Count.DESCRIPTION);
       System.out.println(help);
     }        
 
@@ -1511,6 +1516,8 @@ public class FsShell extends Configured implements Tool {
           du(argv[i]);
         } else if ("-dus".equals(cmd)) {
           dus(argv[i]);
+        } else if (Count.matches(cmd)) {
+          Count.count(argv[i], getConf(), System.out);
         } else if ("-ls".equals(cmd)) {
           ls(argv[i], false);
         } else if ("-lsr".equals(cmd)) {
@@ -1552,6 +1559,7 @@ public class FsShell extends Configured implements Tool {
    * 
    */
   void printUsage(String cmd) {
+    String prefix = "Usage: java " + FsShell.class.getSimpleName();
     if ("-fs".equals(cmd)) {
       System.err.println("Usage: java FsShell" + 
                          " [-fs <local | file system URI>]");
@@ -1568,6 +1576,8 @@ public class FsShell extends Configured implements Tool {
                "-text".equals(cmd)) {
       System.err.println("Usage: java FsShell" + 
                          " [" + cmd + " <path>]");
+    } else if (Count.matches(cmd)) {
+      System.err.println(prefix + " [" + Count.USAGE + "]");
     } else if ("-mv".equals(cmd) || "-cp".equals(cmd)) {
       System.err.println("Usage: java FsShell" + 
                          " [" + cmd + " <src> <dst>]");
@@ -1601,6 +1611,7 @@ public class FsShell extends Configured implements Tool {
       System.err.println("           [-lsr <path>]");
       System.err.println("           [-du <path>]");
       System.err.println("           [-dus <path>]");
+      System.err.println("           [" + Count.USAGE + "]");
       System.err.println("           [-mv <src> <dst>]");
       System.err.println("           [-cp <src> <dst>]");
       System.err.println("           [-rm <path>]");
@@ -1745,6 +1756,12 @@ public class FsShell extends Configured implements Tool {
         } else {
           dus(".");
         }         
+      } else if (Count.matches(cmd)) {
+        if (i < argv.length) {
+          exitCode = doall(cmd, argv, getConf(), i);
+        } else {
+          Count.count(".", getConf(), System.out);
+        }         
       } else if ("-mkdir".equals(cmd)) {
         exitCode = doall(cmd, argv, getConf(), i);
       } else if ("-touchz".equals(cmd)) {

src/java/org/apache/hadoop/fs/shell/CommandUtils.java

@@ -0,0 +1,28 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.fs.shell;
+
+final class CommandUtils {
+  static String formatDescription(String usage, String... desciptions) {
+    StringBuilder b = new StringBuilder(usage + ": " + desciptions[0]);
+    for(int i = 1; i < desciptions.length; i++) {
+      b.append("\n\t\t" + desciptions[i]);
+    }
+    return b.toString();
+  }
+}

src/java/org/apache/hadoop/fs/shell/Count.java

@@ -0,0 +1,57 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.fs.shell;
+
+import java.io.*;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+/**
+ * Count the number of directories, files and bytes.
+ */
+public class Count {
+  public static final String NAME = "count";
+  public static final String USAGE = "-" + NAME + " <path>";
+  public static final String DESCRIPTION = CommandUtils.formatDescription(USAGE, 
+      "Count the number of directories, files and bytes under the paths",
+      "that match the specified file pattern.  The output columns are:",
+      "DIR_COUNT FILE_COUNT CONTENT_SIZE FILE_NAME");
+
+  public static boolean matches(String cmd) {
+    return ("-" + NAME).equals(cmd); 
+  }
+
+  public static void count(String src, Configuration conf, PrintStream out
+      ) throws IOException {
+    Path srcPath = new Path(src);
+    FileSystem srcFs = srcPath.getFileSystem(conf);
+    FileStatus[] statuses = srcFs.globStatus(srcPath);
+    if (statuses == null || statuses.length == 0) {
+      throw new FileNotFoundException(src + " not found.");
+    }
+    for(FileStatus s : statuses) {
+      Path p = s.getPath();
+      String pathstr = p.toString();
+      out.println(srcFs.getContentSummary(p)
+          + ("".equals(pathstr)? ".": pathstr));
+    }
+  }
+}

src/test/org/apache/hadoop/dfs/TestDFSShell.java

@@ -25,6 +25,7 @@ import java.util.zip.GZIPOutputStream;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.*;
+import org.apache.hadoop.fs.shell.*;
 import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.util.ToolRunner;
 
@@ -465,32 +466,13 @@ public class TestDFSShell extends TestCase {
     shell.setConf(conf);
 
     try {
-      {
-        // create a tree
-        //   ROOT
-        //   |- f1
-        //   |- f2
-        //   + sub
-        //      |- f3
-        //      |- f4
-        //   ROOT2
-        //   |- f1
-        Path root = mkdir(dfs, new Path("/test/copyToLocal"));
-        Path sub = mkdir(dfs, new Path(root, "sub"));
-        Path root2 = mkdir(dfs, new Path("/test/copyToLocal2"));        
-
-        writeFile(fs, new Path(root, "f1"));
-        writeFile(fs, new Path(root, "f2"));
-        writeFile(fs, new Path(sub, "f3"));
-        writeFile(fs, new Path(sub, "f4"));
-        writeFile(fs, new Path(root2, "f1"));        
-      }
+      String root = createTree(dfs, "copyToLocal");
 
       // Verify copying the tree
       {
-        String[] args = {"-copyToLocal", "/test/copyToLocal*", TEST_ROOT_DIR};
         try {
-          assertEquals(0, shell.run(args));
+          assertEquals(0,
+              runCmd(shell, "-copyToLocal", root + "*", TEST_ROOT_DIR));
         } catch (Exception e) {
           System.err.println("Exception raised from DFSShell.run " +
                              e.getLocalizedMessage());
@@ -546,6 +528,67 @@ public class TestDFSShell extends TestCase {
     }
   }
 
+  static String createTree(FileSystem fs, String name) throws IOException {
+    // create a tree
+    //   ROOT
+    //   |- f1
+    //   |- f2
+    //   + sub
+    //      |- f3
+    //      |- f4
+    //   ROOT2
+    //   |- f1
+    String path = "/test/" + name;
+    Path root = mkdir(fs, new Path(path));
+    Path sub = mkdir(fs, new Path(root, "sub"));
+    Path root2 = mkdir(fs, new Path(path + "2"));        
+
+    writeFile(fs, new Path(root, "f1"));
+    writeFile(fs, new Path(root, "f2"));
+    writeFile(fs, new Path(sub, "f3"));
+    writeFile(fs, new Path(sub, "f4"));
+    writeFile(fs, new Path(root2, "f1"));
+    mkdir(fs, new Path(root2, "sub"));
+    return path;
+  }
+
+  public void testCount() throws Exception {
+    Configuration conf = new Configuration();
+    MiniDFSCluster cluster = new MiniDFSCluster(conf, 2, true, null);
+    DistributedFileSystem dfs = (DistributedFileSystem)cluster.getFileSystem();
+    FsShell shell = new FsShell();
+    shell.setConf(conf);
+
+    try {
+      String root = createTree(dfs, "count");
+
+      // Verify the counts
+      runCount(root, 2, 4, conf);
+      runCount(root + "2", 2, 1, conf);
+      runCount(root + "2/f1", 0, 1, conf);
+      runCount(root + "2/sub", 1, 0, conf);
+    } finally {
+      try {
+        dfs.close();
+      } catch (Exception e) {
+      }
+      cluster.shutdown();
+    }
+  }
+  private void runCount(String path, long dirs, long files, Configuration conf
+      ) throws IOException {
+    ByteArrayOutputStream bytes = new ByteArrayOutputStream(); 
+    PrintStream out = new PrintStream(bytes);
+    Count.count(path, conf, out);
+    String results = bytes.toString();
+    System.out.println(results);
+    Scanner in = new Scanner(results);
+    assertEquals(dirs, in.nextLong());
+    assertEquals(files, in.nextLong());
+    in.close();
+    out.close();
+  }
+
   //throws IOException instead of Exception as shell.run() does.
   private int runCmd(FsShell shell, String... args) throws IOException {
     try {