HADOOP-4952. Add new improved file system interface FileContext for the application writer. Contributed by Sanjay Radia.

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

CHANGES.txt

@@ -193,6 +193,9 @@ Trunk (unreleased changes)
     HADOOP-6246. Add backward compatibility support to use deprecated decimal 
     umask from old configuration. (Jakob Homan via suresh)
 
+    HADOOP-4952. Add new improved file system interface FileContext for the
+    application writer (Sanjay Radia via suresh)
+
   IMPROVEMENTS
 
     HADOOP-4565. Added CombineFileInputFormat to use data locality information

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

@@ -0,0 +1,1423 @@
+/**
+ * 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.FileNotFoundException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.EnumSet;
+import java.util.List;
+import java.util.regex.Pattern;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.util.Progressable;
+import org.apache.hadoop.classification.*;
+import org.apache.hadoop.classification.InterfaceAudience.LimitedPrivate.*;
+
+/**
+ * The FileContext class provides an interface to the application writer for
+ * using the Hadoop filesystem.
+ * It provides a set of methods for the usual operation: create, open, 
+ * list, etc 
+ * 
+ * *** Path Names ***
+ * 
+ * The Hadoop filesystem supports a URI name space and URI names.
+ * It offers a a forest of filesystems that can be referenced using fully
+ * qualified URIs.
+ * 
+ * Two common Hadoop filesystems implementations are
+ *   the local filesystem: file:///path
+ *   the hdfs filesystem hdfs://nnAddress:nnPort/path
+ *   
+ * While URI names are very flexible, it requires knowing the name or address
+ * of the server. For convenience one often wants to access the default system
+ * in your environment without knowing its name/address. This has an
+ * additional benefit that it allows one to change one's default fs
+ *  (say your admin moves you from cluster1 to cluster2).
+ *  
+ * Too facilitate this Hadoop supports a notion of a default filesystem.
+ * The user can set his default filesystem, although this is
+ * typically set up for you in your environment in your default config.
+ * A default filesystem implies a default scheme and authority; slash-relative
+ * names (such as /for/bar) are resolved relative to that default FS.
+ * Similarly a user can also have working-directory-relative names (i.e. names
+ * not starting with a slash). While the working directory is generally in the
+ * same default FS, the wd can be in a different FS; in particular, changing
+ * the default filesystem DOES NOT change the working directory,
+ * 
+ *  Hence Hadoop path names can be one of:
+ *       fully qualified URI:  scheme://authority/path
+ *       slash relative names: /path    - relative to the default filesystem
+ *       wd-relative names:    path        - relative to the working dir
+ *       
+ *       Relative paths with scheme (scheme:foo/bar) are illegal
+ *  
+ *  ****The Role of the FileContext and configuration defaults****
+ *  The FileContext provides file namespace context for resolving file names;
+ *  it also contains the umask for permissions, In that sense it is like the
+ *  per-process file-related state in Unix system.
+ *  These, in general, are obtained from the default configuration file
+ *  in your environment,  (@see {@link Configuration}
+ *  
+ *  No other configuration parameters are obtained from the default config as 
+ *  far as the file context layer is concerned. All filesystem instances
+ *  (i.e. deployments of filesystems) have default properties; we call these
+ *  server side (SS) defaults. Operation like create allow one to select many 
+ *  properties: either pass them in as explicit parameters or 
+ *  one can choose to used the SS properties.
+ *  
+ *  The filesystem related SS defaults are
+ *    - the home directory (default is "/user/<userName>")
+ *    - the initial wd (only for local fs)
+ *    - replication factor
+ *    - block size
+ *    - buffer size
+ *    - bytesPerChecksum (if used).
+ *
+ * 
+ * *** Usage Model for the FileContext class ***
+ * 
+ * Example 1: use the default config read from the $HADOOP_CONFIG/core.xml.
+ *            Unspecified values come from core-defaults.xml in the release jar.
+ *            
+ *     myFiles = getFileContext(); // uses the default config 
+ *     myFiles.create(path, ...);
+ *     myFiles.setWorkingDir(path)
+ *     myFiles.open (path, ...);   
+ *     
+ * Example 2: Use a specific config, ignoring $HADOOP_CONFIG
+ *    configX = someConfigSomeOnePassedToYou.
+ *    myFContext = getFileContext(configX); //configX not changed but passeddown
+ *    myFContext.create(path, ...);
+ *    myFContext.setWorkingDir(path)
+ *                                             
+ * Other ways of creating new FileContexts:
+ *   getLocalFSFileContext(...)  // local filesystem is the default FS
+ *   getLocalFileContext(URI, ...) // where specified URI is default FS.
+ *    
+ */
+
+@InterfaceAudience.Public
+@InterfaceStability.Evolving /*Evolving for a release,to be changed to Stable */
+
+public final class FileContext {
+  
+  /**
+   * The FileContext is defined by.
+   *  1) defaultFS (slash)
+   *  2) wd
+   *  3) umask
+   *
+   */   
+  private final FileSystem defaultFS; // the default FS for this FileContext.
+  private Path workingDir;          // Fully qualified
+  private FsPermission umask;
+  private final Configuration conf; // passed to the filesystem below
+                                  // When we move to new AbstractFileSystem
+                                  // then it is not really needed except for
+                                  // undocumented config vars;
+
+  private FileContext(final FileSystem defFs, final FsPermission theUmask, 
+      final Configuration aConf) {
+    defaultFS = defFs;
+    umask = FsPermission.getUMask(aConf);
+    conf = aConf;
+    /*
+     * Init the wd.
+     * WorkingDir is implemented at the FileContext layer 
+     * NOT at the FileSystem layer. 
+     * If the DefaultFS, such as localFilesystem has a notion of
+     *  builtin WD, we use that as the initial WD.
+     *  Otherwise the WD is initialized to the home directory.
+     */
+    workingDir = defaultFS.getInitialWorkingDirectory();
+    if (workingDir == null) {
+      workingDir = defaultFS.getHomeDirectory();
+    }
+    util = new Util(); // for the inner class
+  }
+ 
+  /* 
+   * Remove relative part - return "absolute":
+   * If input is relative path ("foo/bar") add wd: ie "/<workingDir>/foo/bar"
+   * A fully qualified uri ("hdfs://nn:p/foo/bar") or a slash-relative path
+   * ("/foo/bar") are returned unchanged.
+   * 
+   * Applications that use FileContext should use #makeQualified() since
+   * they really want a fully qualified URI.
+   * Hence this method os not called makeAbsolute() and 
+   * has been deliberately declared private.
+   */
+
+  private Path fixRelativePart(Path f) {
+    if (f.isUriPathAbsolute()) {
+      return f;
+    } else {
+      return new Path(workingDir, f);
+    }
+  }
+
+  /**
+   * Pathnames with scheme and relative path are illegal.
+   * @param path to be checked
+   * @throws IllegalArgumentException if of type scheme:foo/bar
+   */
+  private static void checkNotSchemeWithRelative(final Path path) {
+    if (path.toUri().isAbsolute() && !path.isUriPathAbsolute()) {
+      throw new IllegalArgumentException(
+          "Unsupported name: has scheme but relative path-part");
+    }
+  }
+  
+  /**
+   * Get the filesystem of supplied path.
+   * @param absOrFqPath - absolute or fully qualified path
+   * @return the filesystem of the path
+   * @throws IOException
+   */
+  private FileSystem getFSofPath(final Path absOrFqPath) throws IOException {
+    checkNotSchemeWithRelative(absOrFqPath);
+    if (!absOrFqPath.isAbsolute() && absOrFqPath.toUri().getScheme() == null) {
+      throw new IllegalArgumentException(
+          "FileContext Bug: path is relative");
+    }
+    
+    // TBD cleanup this impl once we create a new FileSystem to replace current
+    // one - see HADOOP-6223.
+    try { 
+      // Is it the default FS for this FileContext?
+      defaultFS.checkPath(absOrFqPath);
+      return defaultFS;
+    } catch (Exception e) { // it is different FileSystem
+      return FileSystem.get(absOrFqPath.toUri(), conf);
+    }
+  }
+  
+  
+  /**
+   * Protected Static Factory methods for getting a FileContexts
+   * that take a FileSystem as input. To be used for testing.
+   * Protected since new FileSystem will be protected.
+   * Note new file contexts are created for each call.
+   */
+  
+  /**
+   * Create a FileContext with specified FS as default 
+   * using the specified config.
+   * 
+   * @param defFS
+   * @param aConf
+   * @return new FileContext with specifed FS as default.
+   * @throws IOException if the filesystem with specified cannot be created
+   */
+  protected static FileContext getFileContext(final FileSystem defFS,
+                    final Configuration aConf) throws IOException {
+    return new FileContext(defFS, FsPermission.getUMask(aConf), aConf);
+  }
+  
+  /**
+   * Create a FileContext for specified FileSystem using the default config.
+   * 
+   * @param defaultFS
+   * @return a FileSystem for the specified URI
+   * @throws IOException if the filesysem with specified cannot be created
+   */
+  protected static FileContext getFileContext(final FileSystem defaultFS)
+    throws IOException {
+    return getFileContext(defaultFS, new Configuration());
+  }
+ 
+ 
+  public static final URI LOCAL_FS_URI = URI.create("file:///");
+  public static final FsPermission DEFAULT_PERM = FsPermission.getDefault();
+  
+  /**
+   * Static Factory methods for getting a FileContext.
+   * Note new file contexts are created for each call.
+   * The only singleton is the local FS context using the default config.
+   * 
+   * Methods that use the default config: the default config read from the
+   * $HADOOP_CONFIG/core.xml,
+   * Unspecified key-values for config are defaulted from core-defaults.xml
+   * in the release jar.
+   * 
+   * The keys relevant to the FileContext layer are extracted at time of
+   * construction. Changes to the config after the call are ignore
+   * by the FileContext layer. 
+   * The conf is passed to lower layers like FileSystem and HDFS which
+   * pick up their own config variables.
+   */
+   
+  /**
+   * Create a FileContext using the default config read from the
+   * $HADOOP_CONFIG/core.xml,
+   * Unspecified key-values for config are defaulted from core-defaults.xml
+   * in the release jar.
+   * 
+   * @throws IOException if default FileSystem in the config  cannot be created
+   */
+  public static FileContext getFileContext() throws IOException {
+    return getFileContext(new Configuration());
+  } 
+  
+  private static FileContext localFsSingleton = null;
+  /**
+   * 
+   * @return a FileContext for the local filesystem using the default config.
+   * @throws IOException 
+   */
+  public static FileContext getLocalFSFileContext() throws IOException {
+    if (localFsSingleton == null) {
+      localFsSingleton = getFileContext(LOCAL_FS_URI); 
+    }
+    return localFsSingleton;
+  }
+
+  
+  /**
+   * Create a FileContext for specified URI using the default config.
+   * 
+   * @param defaultFsUri
+   * @return a FileSystem for the specified URI
+   * @throws IOException if the filesysem with specified cannot be created
+   */
+  public static FileContext getFileContext(final URI defaultFsUri)
+    throws IOException {
+    return getFileContext(defaultFsUri, new Configuration());
+  }
+  
+  /**
+   * Create a FileContext for specified default URI using the specified config.
+   * 
+   * @param defaultFsUri
+   * @param aConf
+   * @return new FileContext for specified uri
+   * @throws IOException if the filesysem with specified cannot be created
+   */
+  public static FileContext getFileContext(final URI defaultFsUri,
+                    final Configuration aConf) throws IOException {
+    return getFileContext(FileSystem.get(defaultFsUri,  aConf), aConf);
+  }
+
+  /**
+   * Create a FileContext using the passed config.
+   * 
+   * @param aConf
+   * @return new FileContext
+   * @throws IOException  if default FileSystem in the config  cannot be created
+   */
+  public static FileContext getFileContext(final Configuration aConf)
+    throws IOException {
+    return getFileContext(URI.create(FsConfig.getDefaultFsURI(aConf)), aConf);
+  }
+  
+  
+  /**
+   * @param aConf - from which the FileContext is configured
+   * @return a FileContext for the local filesystem using the specified config.
+   * @throws IOException 
+   */
+  public static FileContext getLocalFSFileContext(final Configuration aConf)
+    throws IOException {
+    return getFileContext(LOCAL_FS_URI, aConf);
+  }
+
+  /* This method is needed for tests. */
+  @InterfaceAudience.Private
+  @InterfaceStability.Unstable /* return type will change to AFS once
+                                  HADOOP-6223 is completed */
+  protected FileSystem getDefaultFileSystem() {
+    return defaultFS;
+  }
+  
+  /**
+   * Set the working directory for wd-relative names (such a "foo/bar").
+   * @param newWDir
+   * @throws IOException
+   * 
+   * newWdir can be one of 
+   *     - relative path:  "foo/bar";
+   *     - absolute without scheme: "/foo/bar"
+   *     - fully qualified with scheme: "xx://auth/foo/bar"
+   *  Illegal WDs:
+   *      - relative with scheme: "xx:foo/bar" 
+   */
+  public void setWorkingDirectory(final Path newWDir) throws IOException {
+    checkNotSchemeWithRelative(newWDir);
+    // wd is stored as fully qualified path. 
+
+    final Path newWorkingDir =  new Path(workingDir, newWDir);
+    FileStatus status = getFileStatus(newWorkingDir);
+    if (!status.isDir()) {
+      throw new FileNotFoundException(" Cannot setWD to a file");
+    }
+    workingDir = newWorkingDir;
+  }
+  
+  /**
+   * Gets the working directory for wd-relative names (such a "foo/bar").
+   */
+  public Path getWorkingDirectory() {
+    return workingDir;
+  }
+  
+  /**
+   * 
+   * @return the umask of this FileContext
+   */
+  public FsPermission getUMask() {
+    return umask;
+  }
+  
+  /**
+   * Set umask to the supplied parameter.
+   * @param newUmask  the new umask
+   */
+  public void setUMask(final FsPermission newUmask) {
+    umask = newUmask;
+  }
+  
+  
+  /**
+   * Make the path fully qualified if it is isn't. 
+   * A Fully-qualified path has scheme and authority specified and an absolute
+   * path.
+   * Use the default filesystem and working dir in this FileContext to qualify.
+   * @param path
+   * @return qualified path
+   */
+  public Path makeQualified(final Path path) {
+    return path.makeQualified(defaultFS.getUri(), getWorkingDirectory());
+  } 
+
+  /**
+   * Class to support the varargs for create() options.
+   *
+   */
+  public static class CreateOpts {
+    private CreateOpts() { };
+    public static BlockSize blockSize(long bs) { 
+      return new BlockSize(bs);
+    }
+    public static BufferSize bufferSize(short bs) { 
+      return new BufferSize(bs);
+    }
+    public static ReplicationFactor repFac(short rf) { 
+      return new ReplicationFactor(rf);
+    }
+    public static BytesPerChecksum bytesPerChecksum(short crc) {
+      return new BytesPerChecksum(crc);
+    }
+    public static Perms perms(FsPermission perm) {
+      return new Perms(perm);
+    }
+    
+    static class BlockSize extends CreateOpts {
+      private final long blockSize;
+      protected BlockSize(long bs) {
+        if (bs <= 0) {
+          throw new IllegalArgumentException(
+                        "Block size must be greater than 0");
+        }
+        blockSize = bs; 
+      }
+      long getValue() { return blockSize; }
+    }
+    
+    static class ReplicationFactor extends CreateOpts {
+      private final short replication;
+      protected ReplicationFactor(short rf) { 
+        if (rf <= 0) {
+          throw new IllegalArgumentException(
+                      "Replication must be greater than 0");
+        }
+        replication = rf;
+      }
+      short getValue() { return replication; }
+    }
+    
+    static class BufferSize extends CreateOpts {
+      private final int bufferSize;
+      protected BufferSize(short bs) {
+        if (bs <= 0) {
+          throw new IllegalArgumentException(
+                        "Buffer size must be greater than 0");
+        }
+        bufferSize = bs; 
+      }
+      int getValue() { return bufferSize; }
+    }
+    
+    static class BytesPerChecksum extends CreateOpts {
+      private final int bytesPerChecksum;
+      protected BytesPerChecksum(short bpc) { 
+        if (bpc <= 0) {
+          throw new IllegalArgumentException(
+                        "Bytes per checksum must be greater than 0");
+        }
+        bytesPerChecksum = bpc; 
+      }
+      int getValue() { return bytesPerChecksum; }
+    }
+    
+    static class Perms extends CreateOpts {
+      private final FsPermission permissions;
+      protected Perms(FsPermission perm) { 
+        if(perm == null) {
+          throw new IllegalArgumentException("Permissions must not be null");
+        }
+        permissions = perm; 
+      }
+      FsPermission getValue() { return permissions; }
+    }
+    
+    static class Progress extends CreateOpts {
+      private final Progressable progress;
+      protected Progress(Progressable prog) { 
+        if(prog == null) {
+          throw new IllegalArgumentException("Progress must not be null");
+        }
+        progress = prog;
+      }
+      Progressable getValue() { return progress; }
+    }
+  }
+  
+  /**
+   * Create or overwrite file on indicated path and returns an output stream
+   * for writing into the file.
+   * @param f the file name to open
+   * @param createFlag gives the semantics  of create: overwrite, append etc.
+   * @param opts  - varargs of CreateOpt:
+   *     Progress - to report progress on the operation - default null
+   *     Permission - umask is applied against permisssion:
+   *                  default FsPermissions:getDefault()
+   *                  @see #setPermission(Path, FsPermission)
+   *                The defaults for the following are  SS defaults of the
+   *                file server implementing the tart path.
+   *                Not all parameters make sense for all kinds of filesystem
+   *                - eg. localFS ignores Blocksize, replication, checksum
+   *    BufferSize - buffersize used in FSDataOutputStream
+   *    Blocksize - block size for file blocks
+   *    ReplicationFactor - replication for blocks
+   *    BytesPerChecksum - bytes per checksum
+   *    
+   * @throws IOException
+   */
+  public FSDataOutputStream create(final Path f,
+                                    final EnumSet<CreateFlag> createFlag,
+                                    CreateOpts... opts)
+    throws IOException {
+    Path absF = fixRelativePart(f);
+    FileSystem fsOfAbsF = getFSofPath(absF);
+
+    int bufferSize = -1;
+    short replication = -1;
+    long blockSize = -1;
+    int bytesPerChecksum = -1;
+    FsPermission permission = null;
+    Progressable progress = null;
+ 
+    for (CreateOpts iOpt : opts) {
+      if (CreateOpts.BlockSize.class.isInstance(iOpt)) {
+        if (blockSize != -1) {
+          throw new IllegalArgumentException("multiple varargs of same kind");
+        }
+        blockSize = ((CreateOpts.BlockSize) iOpt).getValue();
+      } else if (CreateOpts.BufferSize.class.isInstance(iOpt)) {
+        if (bufferSize != -1) {
+          throw new IllegalArgumentException("multiple varargs of same kind");
+        }
+        bufferSize = ((CreateOpts.BufferSize) iOpt).getValue();
+      } else if (CreateOpts.ReplicationFactor.class.isInstance(iOpt)) {
+        if (replication != -1) {
+          throw new IllegalArgumentException("multiple varargs of same kind");
+        }
+        replication = ((CreateOpts.ReplicationFactor) iOpt).getValue();
+      } else if (CreateOpts.BytesPerChecksum.class.isInstance(iOpt)) {
+        if (bytesPerChecksum != -1) {
+          throw new IllegalArgumentException("multiple varargs of same kind");
+        }
+        bytesPerChecksum = ((CreateOpts.BytesPerChecksum) iOpt).getValue();
+      } else if (CreateOpts.Perms.class.isInstance(iOpt)) {
+        if (permission != null) {
+          throw new IllegalArgumentException("multiple varargs of same kind");
+        }
+        permission = ((CreateOpts.Perms) iOpt).getValue();
+      } else if (CreateOpts.Progress.class.isInstance(iOpt)) {
+        if (progress != null) {
+          throw new IllegalArgumentException("multiple varargs of same kind");
+        }
+        progress = ((CreateOpts.Progress) iOpt).getValue();
+      } else {
+        throw new IllegalArgumentException("Unkown CreateOpts of type " +
+            iOpt.getClass().getName());
+      }
+    }
+    if (blockSize % bytesPerChecksum != 0) {
+      throw new IllegalArgumentException(
+          "blockSize should be a multiple of checksumsize");
+    }
+    
+    FsServerDefaults ssDef = fsOfAbsF.getServerDefaults();
+    
+    if (blockSize == -1) {
+      blockSize = ssDef.getBlockSize();
+    }
+    if (bufferSize == -1) {
+      bufferSize = ssDef.getFileBufferSize();
+    }
+    if (replication == -1) {
+      replication = ssDef.getReplication();
+    }
+    if (bytesPerChecksum == -1) {
+      bytesPerChecksum = ssDef.getBytesPerChecksum();
+    }
+    if (permission == null) {
+      permission = FsPermission.getDefault();
+    }
+
+    FsPermission absPerms = (permission == null ? 
+        FsPermission.getDefault() : permission).applyUMask(umask);
+
+    return fsOfAbsF.primitiveCreate(absF, absPerms, createFlag,
+                bufferSize, replication, blockSize, progress, bytesPerChecksum);
+  }
+  
+  /**
+   * Make the given file and all non-existent parents into
+   * directories. Has the semantics of Unix 'mkdir -p'.
+   * Existence of the directory hierarchy is not an error.
+   * 
+   * @param dir - the dir to make
+   * @param permission - permissions is set permission&~umask
+   * @return true if the operation succeeds; false if dir already exists 
+   * @throws IOException when operation fails (e.g. permissions) etc.
+   */
+  public boolean mkdirs(final Path dir, final FsPermission permission)
+    throws IOException {
+    Path absDir = fixRelativePart(dir);
+    FsPermission absFerms = (permission == null ? 
+          FsPermission.getDefault() : permission).applyUMask(umask);
+    return getFSofPath(absDir).primitiveMkdir(absDir, absFerms);
+  }
+
+  /** Delete a file.
+   *
+   * @param f the path to delete.
+   * @param recursive if path is a directory and set to 
+   * true, the directory is deleted else throws an exception. In
+   * case of a file the recursive can be set to either true or false. 
+   * @return  true if delete is successful else false. 
+   * @throws IOException
+   */
+  public boolean delete(final Path f, final boolean recursive) 
+    throws IOException {
+    Path absF = fixRelativePart(f);
+    return getFSofPath(absF).delete(absF, recursive);
+  }
+ 
+  /**
+   * Opens an FSDataInputStream at the indicated Path using
+   * default buffersize.
+   * @param f the file name to open
+   */
+  public FSDataInputStream open(final Path f) throws IOException {
+    final Path absF = fixRelativePart(f);
+    return getFSofPath(absF).open(absF);
+  }
+
+  /**
+   * Opens an FSDataInputStream at the indicated Path.
+   * @param f the file name to open
+   * @param bufferSize the size of the buffer to be used.
+   */
+  public FSDataInputStream open(final Path f, int bufferSize)
+    throws IOException {
+    final Path absF = fixRelativePart(f);
+    return getFSofPath(absF).open(absF, bufferSize);
+  }
+
+ /**
+  * Set replication for an existing file.
+  * 
+  * @param f file name
+  * @param replication new replication
+  * @throws IOException
+  * @return true if successful;
+  *         false if file does not exist or is a directory
+  */
+  public boolean setReplication(final Path f, final short replication)
+    throws IOException {
+    final Path absF = fixRelativePart(f);
+    return getFSofPath(absF).setReplication(absF, replication);
+  }
+
+ /**
+  * Renames Path src to Path dst. 
+  *  
+  * @param src
+  * @param dst
+  * @throws IOException if a rename is attempted across URI filessystem or
+  * across volumes within a file system.
+  */
+  public void rename(final Path src, final Path dst)
+    throws IOException {
+    final Path absSrc  = fixRelativePart(src);
+    final Path absDst = fixRelativePart(dst);
+    FileSystem srcFS = getFSofPath(absSrc);
+    FileSystem dstFS = getFSofPath(absDst);
+    if(!srcFS.getUri().equals(dstFS.getUri())) {
+      throw new IOException("Renames across FileSystems not supported");
+    }
+    if(srcFS.rename(absSrc, absDst)) {
+      return;
+    }
+    throw new IOException("bug in underlying filesystem");
+  }
+  
+  /**
+   * Set permission of a path.
+   * @param f
+   * @param permission - the new absolute permission (umask is not applied)
+   */
+  public void setPermission(final Path f, final FsPermission permission)
+    throws IOException {
+    final Path absF = fixRelativePart(f);
+    getFSofPath(absF).setPermission(absF, permission);
+  }
+
+  /**
+   * Set owner of a path (i.e. a file or a directory).
+   * The parameters username and groupname cannot both be null.
+   * @param f The path
+   * @param username If it is null, the original username remains unchanged.
+   * @param groupname If it is null, the original groupname remains unchanged.
+   */
+  public void setOwner(final Path f, final String username,
+                        final String groupname) throws IOException {
+    if ((username == null) && (groupname == null)) {
+      throw new IllegalArgumentException(
+          "usernme and groupname cannot both be null");
+    }
+    final Path absF = fixRelativePart(f);
+    getFSofPath(absF).setOwner(absF, username, groupname);
+  }
+
+  /**
+   * Set access time of a file.
+   * @param f The path
+   * @param mtime Set the modification time of this file.
+   *        The number of milliseconds since epoch (Jan 1, 1970). 
+   *        A value of -1 means that this call should not set modification time.
+   * @param atime Set the access time of this file.
+   *        The number of milliseconds since Jan 1, 1970. 
+   *        A value of -1 means that this call should not set access time.
+   */
+  public void setTimes(final Path f, final long mtime, final long atime)
+    throws IOException {
+    final Path absF = fixRelativePart(f);
+    getFSofPath(absF).setTimes(absF, mtime, atime);
+  }
+
+  /**
+   * Get the checksum of a file.
+   *
+   * @param f The file path
+   * @return The file checksum.  The default return value is null,
+   *  which indicates that no checksum algorithm is implemented
+   *  in the corresponding FileSystem.
+   */
+  public FileChecksum getFileChecksum(final Path f) throws IOException {
+    final Path absF = fixRelativePart(f);
+    return getFSofPath(absF).getFileChecksum(absF);
+  }
+
+  /**
+   * Set the verify checksum flag for the  filesystem denoted by the path.
+   * This is only applicable if the 
+   * corresponding FileSystem supports checksum. By default doesn't do anything.
+   * @param verifyChecksum
+   * @param f - set the verifyChecksum for the Filesystem containing this path
+   * @throws IOException 
+   */
+
+  public void setVerifyChecksum(final boolean verifyChecksum, final Path f)
+    throws IOException {
+    final Path absF = fixRelativePart(f);
+    getFSofPath(absF).setVerifyChecksum(verifyChecksum);
+    
+    //TBD need to be changed when we add symlinks.
+  }
+
+  /**
+   * Return a file status object that represents the path.
+   * @param f The path we want information from
+   * @return a FileStatus object
+   * @throws FileNotFoundException when the path does not exist;
+   *         IOException see specific implementation
+   */
+  public FileStatus getFileStatus(final Path f) throws IOException {
+    final Path absF = fixRelativePart(f);
+    return getFSofPath(absF).getFileStatus(absF);
+  }
+  
+  /**
+   * Return blockLocation of the given file for the given offset and len.
+   *  For a nonexistent file or regions, null will be returned.
+   *
+   * This call is most helpful with DFS, where it returns 
+   * hostnames of machines that contain the given file.
+   * 
+   * @param p - get blocklocations of this file
+   * @param start position (byte offset)
+   * @param len (in bytes)
+   * @return block locations for given file at specified offset of len
+   * @throws IOException
+   */
+  
+  @InterfaceAudience.LimitedPrivate({Project.HDFS, Project.MAPREDUCE})
+  @InterfaceStability.Evolving
+  public BlockLocation[] getFileBlockLocations(final Path p, 
+    final long start, final long len) throws IOException {
+    return getFSofPath(p).getFileBlockLocations(p, start, len);
+  }
+  
+  /**
+   * Returns a status object describing the use and capacity of the
+   * filesystem denoted by the Parh argument p.
+   * If the filesystem has multiple partitions, the
+   * use and capacity of the partition pointed to by the specified
+   * path is reflected.
+   * 
+   * @param f Path for which status should be obtained. null means the
+   * root partition of the default filesystem. 
+   * @return a FsStatus object
+   * @throws IOException
+   *           see specific implementation
+   */
+  public FsStatus getFsStatus(final Path f) throws IOException {
+    if (f == null) {
+      return defaultFS.getStatus(null);
+    }
+    final Path absF = fixRelativePart(f);
+    return getFSofPath(absF).getStatus(absF);
+  }
+  
+  /**
+   * Does the file exist?
+   * Note: Avoid using this method if you already have FileStatus in hand.
+   * Instead reuse the FileStatus 
+   * @param f the  file or dir to be checked
+   */
+  public boolean exists(final Path f) throws IOException {
+    try {
+      return getFileStatus(f) != null;
+    } catch (FileNotFoundException e) {
+      return false;
+    }
+  }
+
+  /**
+   * Is a directory?
+   * Note: Avoid using this method if you already have FileStatus in hand.
+   * Instead reuse the FileStatus 
+   * returned by getFileStatus() or listStatus() methods.
+   * 
+   * @param f Path to evaluate
+   * @return True iff the named path is a directory.
+   * @throws IOException
+   */
+  public boolean isDirectory(final Path f) throws IOException {
+    try {
+      final Path absF = fixRelativePart(f);
+      return getFileStatus(absF).isDir();
+    } catch (FileNotFoundException e) {
+      return false;               // f does not exist
+    }
+  }
+
+  /** True iff the named path is a regular file.
+   * Note: Avoid using this method  if you already have FileStatus in hand
+   * Instead reuse the FileStatus 
+   * returned by getFileStatus() or listStatus() methods.
+   */
+  public boolean isFile(final Path f) throws IOException {
+    try {
+      final Path absF = fixRelativePart(f);
+      return !getFileStatus(absF).isDir();
+    } catch (FileNotFoundException e) {
+      return false;               // f does not exist
+    }
+  }
+  
+  /**
+   * List the statuses of the files/directories in the given path if the path is
+   * a directory.
+   * 
+   * @param f is the path
+   * @return the statuses of the files/directories in the given path
+   * @throws IOException
+   */
+  public FileStatus[] listStatus(final Path f) throws IOException {
+    final Path absF = fixRelativePart(f);
+    return getFSofPath(absF).listStatus(absF);
+  }
+
+  private final Util util;
+  public Util util() {
+    return util;
+  }
+  
+  
+  /**
+   * Utility/library methods built over the basic FileContext methods.
+   * Since this are library functions, the oprtation are not atomic
+   * and some of them may partially complete if other threads are making
+   * changes to the same part of the name space.
+   */
+  public class Util {
+    /**
+     * Return a list of file status objects that corresponds to supplied paths
+     * excluding those non-existent paths.
+     * 
+     * @param paths are the list of paths we want information from
+     * @return a list of FileStatus objects
+     * @throws IOException
+     */
+    private FileStatus[] getFileStatus(Path[] paths) throws IOException {
+      if (paths == null) {
+        return null;
+      }
+      ArrayList<FileStatus> results = new ArrayList<FileStatus>(paths.length);
+      for (int i = 0; i < paths.length; i++) {
+        results.add(FileContext.this.getFileStatus(paths[i]));
+      }
+      return results.toArray(new FileStatus[results.size()]);
+    }
+    
+    /**
+     * Filter files/directories in the given list of paths using default
+     * path filter.
+     * 
+     * @param files is the list of paths
+     * @return a list of statuses for the files under the given paths after
+     *         applying the filter default Path filter
+     * @exception IOException
+     */
+    public FileStatus[] listStatus(Path[] files)
+      throws IOException {
+      return listStatus(files, DEFAULT_FILTER);
+    }
+     
+    /**
+     * Filter files/directories in the given path using the user-supplied path
+     * filter.
+     * 
+     * @param f is the path name
+     * @param filter is the user-supplied path filter
+     * @return an array of FileStatus objects for the files under the given path
+     *         after applying the filter
+     * @throws IOException
+     *           if encounter any problem while fetching the status
+     */
+    public FileStatus[] listStatus(Path f, PathFilter filter)
+      throws IOException {
+      ArrayList<FileStatus> results = new ArrayList<FileStatus>();
+      listStatus(results, f, filter);
+      return results.toArray(new FileStatus[results.size()]);
+    }
+    
+    /**
+     * Filter files/directories in the given list of paths using user-supplied
+     * path filter.
+     * 
+     * @param files is a list of paths
+     * @param filter is the filter
+     * @return a list of statuses for the files under the given paths after
+     *         applying the filter
+     * @exception IOException
+     */
+    public FileStatus[] listStatus(Path[] files, PathFilter filter)
+      throws IOException {
+      ArrayList<FileStatus> results = new ArrayList<FileStatus>();
+      for (int i = 0; i < files.length; i++) {
+        listStatus(results, files[i], filter);
+      }
+      return results.toArray(new FileStatus[results.size()]);
+    }
+  
+    /*
+     * Filter files/directories in the given path using the user-supplied path
+     * filter. Results are added to the given array <code>results</code>.
+     */
+    private void listStatus(ArrayList<FileStatus> results, Path f,
+        PathFilter filter) throws IOException {
+      FileStatus[] listing = FileContext.this.listStatus(f);
+      if (listing != null) {
+        for (int i = 0; i < listing.length; i++) {
+          if (filter.accept(listing[i].getPath())) {
+            results.add(listing[i]);
+          }
+        }
+      }
+    }
+  
+    /**
+     * <p>Return all the files that match filePattern and are not checksum
+     * files. Results are sorted by their names.
+     * 
+     * <p>
+     * A filename pattern is composed of <i>regular</i> characters and
+     * <i>special pattern matching</i> characters, which are:
+     *
+     * <dl>
+     *  <dd>
+     *   <dl>
+     *    <p>
+     *    <dt> <tt> ? </tt>
+     *    <dd> Matches any single character.
+     *
+     *    <p>
+     *    <dt> <tt> * </tt>
+     *    <dd> Matches zero or more characters.
+     *
+     *    <p>
+     *    <dt> <tt> [<i>abc</i>] </tt>
+     *    <dd> Matches a single character from character set
+     *     <tt>{<i>a,b,c</i>}</tt>.
+     *
+     *    <p>
+     *    <dt> <tt> [<i>a</i>-<i>b</i>] </tt>
+     *    <dd> Matches a single character from the character range
+     *     <tt>{<i>a...b</i>}</tt>. Note: character <tt><i>a</i></tt> must be
+     *     lexicographically less than or equal to character <tt><i>b</i></tt>.
+     *
+     *    <p>
+     *    <dt> <tt> [^<i>a</i>] </tt>
+     *    <dd> Matches a single char that is not from character set or range
+     *     <tt>{<i>a</i>}</tt>.  Note that the <tt>^</tt> character must occur
+     *     immediately to the right of the opening bracket.
+     *
+     *    <p>
+     *    <dt> <tt> \<i>c</i> </tt>
+     *    <dd> Removes (escapes) any special meaning of character <i>c</i>.
+     *
+     *    <p>
+     *    <dt> <tt> {ab,cd} </tt>
+     *    <dd> Matches a string from the string set <tt>{<i>ab, cd</i>} </tt>
+     *    
+     *    <p>
+     *    <dt> <tt> {ab,c{de,fh}} </tt>
+     *    <dd> Matches a string from string set <tt>{<i>ab, cde, cfh</i>}</tt>
+     *
+     *   </dl>
+     *  </dd>
+     * </dl>
+     *
+     * @param pathPattern a regular expression specifying a pth pattern
+
+     * @return an array of paths that match the path pattern
+     * @throws IOException
+     */
+    public FileStatus[] globStatus(Path pathPattern) throws IOException {
+      return globStatus(pathPattern, DEFAULT_FILTER);
+    }
+    
+    /**
+     * Return an array of FileStatus objects whose path names match pathPattern
+     * and is accepted by the user-supplied path filter. Results are sorted by
+     * their path names.
+     * Return null if pathPattern has no glob and the path does not exist.
+     * Return an empty array if pathPattern has a glob and no path matches it. 
+     * 
+     * @param pathPattern
+     *          a regular expression specifying the path pattern
+     * @param filter
+     *          a user-supplied path filter
+     * @return an array of FileStatus objects
+     * @throws IOException if any I/O error occurs when fetching file status
+     */
+    public FileStatus[] globStatus(Path pathPattern, PathFilter filter)
+      throws IOException {
+      String filename = pathPattern.toUri().getPath();
+      List<String> filePatterns = GlobExpander.expand(filename);
+      if (filePatterns.size() == 1) {
+        return globStatusInternal(pathPattern, filter);
+      } else {
+        List<FileStatus> results = new ArrayList<FileStatus>();
+        for (String filePattern : filePatterns) {
+          FileStatus[] files = 
+                      globStatusInternal(new Path(filePattern), filter);
+          for (FileStatus file : files) {
+            results.add(file);
+          }
+        }
+        return results.toArray(new FileStatus[results.size()]);
+      }
+    }
+
+    private FileStatus[] globStatusInternal(Path pathPattern, PathFilter filter)
+      throws IOException {
+      Path[] parents = new Path[1];
+      int level = 0;
+      String filename = pathPattern.toUri().getPath();
+      
+      // path has only zero component
+      if ("".equals(filename) || Path.SEPARATOR.equals(filename)) {
+        return getFileStatus(new Path[]{pathPattern});
+      }
+
+      // path has at least one component
+      String[] components = filename.split(Path.SEPARATOR);
+      // get the first component
+      if (pathPattern.isAbsolute()) {
+        parents[0] = new Path(Path.SEPARATOR);
+        level = 1;
+      } else {
+        parents[0] = new Path(Path.CUR_DIR);
+      }
+
+      // glob the paths that match the parent path, ie. [0, components.length-1]
+      boolean[] hasGlob = new boolean[]{false};
+      Path[] parentPaths = globPathsLevel(parents, components, level, hasGlob);
+      FileStatus[] results;
+      if (parentPaths == null || parentPaths.length == 0) {
+        results = null;
+      } else {
+        // Now work on the last component of the path
+        GlobFilter fp = 
+                    new GlobFilter(components[components.length - 1], filter);
+        if (fp.hasPattern()) { // last component has a pattern
+          // list parent directories and then glob the results
+          results = listStatus(parentPaths, fp);
+          hasGlob[0] = true;
+        } else { // last component does not have a pattern
+          // get all the path names
+          ArrayList<Path> filteredPaths = 
+                                      new ArrayList<Path>(parentPaths.length);
+          for (int i = 0; i < parentPaths.length; i++) {
+            parentPaths[i] = new Path(parentPaths[i],
+              components[components.length - 1]);
+            if (fp.accept(parentPaths[i])) {
+              filteredPaths.add(parentPaths[i]);
+            }
+          }
+          // get all their statuses
+          results = getFileStatus(
+              filteredPaths.toArray(new Path[filteredPaths.size()]));
+        }
+      }
+
+      // Decide if the pathPattern contains a glob or not
+      if (results == null) {
+        if (hasGlob[0]) {
+          results = new FileStatus[0];
+        }
+      } else {
+        if (results.length == 0) {
+          if (!hasGlob[0]) {
+            results = null;
+          }
+        } else {
+          Arrays.sort(results);
+        }
+      }
+      return results;
+    }
+
+    /*
+     * For a path of N components, return a list of paths that match the
+     * components [<code>level</code>, <code>N-1</code>].
+     */
+    private Path[] globPathsLevel(Path[] parents, String[] filePattern,
+        int level, boolean[] hasGlob) throws IOException {
+      if (level == filePattern.length - 1) {
+        return parents;
+      }
+      if (parents == null || parents.length == 0) {
+        return null;
+      }
+      GlobFilter fp = new GlobFilter(filePattern[level]);
+      if (fp.hasPattern()) {
+        parents = FileUtil.stat2Paths(listStatus(parents, fp));
+        hasGlob[0] = true;
+      } else {
+        for (int i = 0; i < parents.length; i++) {
+          parents[i] = new Path(parents[i], filePattern[level]);
+        }
+      }
+      return globPathsLevel(parents, filePattern, level + 1, hasGlob);
+    }
+
+    /**
+     * Copy file from src to dest.
+     * @param src
+     * @param dst
+     * @return true if copy is successful
+     * @throws IOException
+     */
+    public boolean copy(final Path src, final Path dst)  throws IOException {
+      return copy(src, dst, false, false);
+    }
+    
+    /**
+     * Copy from src to dst, optionally deleting src and overwriting dst.
+     * @param src
+     * @param dst
+     * @param deleteSource - delete src if true
+     * @param overwrite  overwrite dst if true; throw IOException if dst exists
+     *         and overwrite is false.
+     * @return true if copy is successful
+     * @throws IOException
+     */
+    public boolean copy(final Path src,  final Path dst,
+        boolean deleteSource, boolean overwrite)
+      throws IOException {
+      checkNotSchemeWithRelative(src);
+      checkNotSchemeWithRelative(dst);
+      Path qSrc = makeQualified(src);
+      Path qDst = makeQualified(dst);
+      checkDest(qSrc.getName(), qDst, false);
+      if (isDirectory(qSrc)) {
+        checkDependencies(qSrc, qDst);
+        if (!mkdirs(qDst, FsPermission.getDefault())) {
+          throw new IOException("Failed to create destination directory `" +
+              qDst + "'");
+        }
+        FileStatus[] contents = FileContext.this.listStatus(qSrc);
+        for (FileStatus content : contents) {
+          copy(content.getPath(), new Path(qDst, content.getPath()),
+               deleteSource, overwrite);
+        }
+      } else {
+        InputStream in=null;
+        OutputStream out = null;
+        try {
+          in = open(qSrc);
+          out = create(qDst, EnumSet.of(CreateFlag.OVERWRITE));
+          IOUtils.copyBytes(in, out, conf, true);
+        } catch (IOException e) {
+          IOUtils.closeStream(out);
+          IOUtils.closeStream(in);
+          throw e;
+        }
+      }
+      if (deleteSource) {
+        return delete(qSrc, true);
+      } else {
+        return true;
+      }
+    }
+  }
+
+  private static final PathFilter DEFAULT_FILTER = new PathFilter() {
+    public boolean accept(final Path file) {
+      return true;
+    }
+  };
+  
+  /* A class that could decide if a string matches the glob or not */
+  private static class GlobFilter implements PathFilter {
+    private PathFilter userFilter = DEFAULT_FILTER;
+    private Pattern regex;
+    private boolean hasPattern = false;
+      
+    /** Default pattern character: Escape any special meaning. */
+    private static final char  PAT_ESCAPE = '\\';
+    /** Default pattern character: Any single character. */
+    private static final char  PAT_ANY = '.';
+    /** Default pattern character: Character set close. */
+    private static final char  PAT_SET_CLOSE = ']';
+      
+    GlobFilter() {
+    }
+      
+    GlobFilter(final String filePattern) throws IOException {
+      setRegex(filePattern);
+    }
+      
+    GlobFilter(final String filePattern, final PathFilter filter)
+      throws IOException {
+      userFilter = filter;
+      setRegex(filePattern);
+    }
+      
+    private boolean isJavaRegexSpecialChar(char pChar) {
+      return pChar == '.' || pChar == '$' || pChar == '(' || pChar == ')' ||
+             pChar == '|' || pChar == '+';
+    }
+    
+    void setRegex(String filePattern) throws IOException {
+      int len;
+      int setOpen;
+      int curlyOpen;
+      boolean setRange;
+
+      StringBuilder fileRegex = new StringBuilder();
+
+      // Validate the pattern
+      len = filePattern.length();
+      if (len == 0) {
+        return;
+      }
+
+      setOpen = 0;
+      setRange = false;
+      curlyOpen = 0;
+
+      for (int i = 0; i < len; i++) {
+        char pCh;
+          
+        // Examine a single pattern character
+        pCh = filePattern.charAt(i);
+        if (pCh == PAT_ESCAPE) {
+          fileRegex.append(pCh);
+          i++;
+          if (i >= len) {
+            error("An escaped character does not present", filePattern, i);
+          }
+          pCh = filePattern.charAt(i);
+        } else if (isJavaRegexSpecialChar(pCh)) {
+          fileRegex.append(PAT_ESCAPE);
+        } else if (pCh == '*') {
+          fileRegex.append(PAT_ANY);
+          hasPattern = true;
+        } else if (pCh == '?') {
+          pCh = PAT_ANY;
+          hasPattern = true;
+        } else if (pCh == '{') {
+          fileRegex.append('(');
+          pCh = '(';
+          curlyOpen++;
+          hasPattern = true;
+        } else if (pCh == ',' && curlyOpen > 0) {
+          fileRegex.append(")|");
+          pCh = '(';
+        } else if (pCh == '}' && curlyOpen > 0) {
+          // End of a group
+          curlyOpen--;
+          fileRegex.append(")");
+          pCh = ')';
+        } else if (pCh == '[' && setOpen == 0) {
+          setOpen++;
+          hasPattern = true;
+        } else if (pCh == '^' && setOpen > 0) {
+        } else if (pCh == '-' && setOpen > 0) {
+          // Character set range
+          setRange = true;
+        } else if (pCh == PAT_SET_CLOSE && setRange) {
+          // Incomplete character set range
+          error("Incomplete character set range", filePattern, i);
+        } else if (pCh == PAT_SET_CLOSE && setOpen > 0) {
+          // End of a character set
+          if (setOpen < 2) {
+            error("Unexpected end of set", filePattern, i);
+          }
+          setOpen = 0;
+        } else if (setOpen > 0) {
+          // Normal character, or the end of a character set range
+          setOpen++;
+          setRange = false;
+        }
+        fileRegex.append(pCh);
+      }
+        
+      // Check for a well-formed pattern
+      if (setOpen > 0 || setRange || curlyOpen > 0) {
+        // Incomplete character set or character range
+        error("Expecting set closure character or end of range, or }", 
+            filePattern, len);
+      }
+      regex = Pattern.compile(fileRegex.toString());
+    }
+      
+    boolean hasPattern() {
+      return hasPattern;
+    }
+      
+    public boolean accept(final Path path) {
+      return regex.matcher(path.getName()).matches() && userFilter.accept(path);
+    }
+      
+    private void error(final String s, final String pattern, final int pos)
+      throws IOException {
+      throw new IOException("Illegal file pattern: "
+                            +s+ " for glob "+ pattern + " at " + pos);
+    }
+  }
+  
+  //
+  // Check destionation. Throw IOException if destination already exists
+  // and overwrite is not true
+  //
+  private void checkDest(String srcName, Path dst, boolean overwrite)
+    throws IOException {
+    if (exists(dst)) {
+      if (isDirectory(dst)) {
+      // TBD not very clear
+        if (null == srcName) {
+          throw new IOException("Target " + dst + " is a directory");
+        }
+        checkDest(null, new Path(dst, srcName), overwrite);
+      } else if (!overwrite) {
+        throw new IOException("Target " + dst + " already exists");
+      }
+    }
+  }
+   
+  //
+  // If the destination is a subdirectory of the source, then
+  // generate exception
+  //
+  private static void checkDependencies(Path qualSrc, Path qualDst)
+    throws IOException {
+    if (isSameFS(qualSrc, qualDst)) {
+      String srcq = qualSrc.toString() + Path.SEPARATOR;
+      String dstq = qualDst.toString() + Path.SEPARATOR;
+      if (dstq.startsWith(srcq)) {
+        if (srcq.length() == dstq.length()) {
+          throw new IOException("Cannot copy " + qualSrc + " to itself.");
+        } else {
+          throw new IOException("Cannot copy " + qualSrc +
+                             " to its subdirectory " + qualDst);
+        }
+      }
+    }
+  }
+  
+  /**
+   * Are qualSrc and qualDst of the same file system?
+   * @param qualPath1 - fully qualified path
+   * @param qualPath2 - fully qualified path
+   * @return
+   */
+  private static boolean isSameFS(Path qualPath1, Path qualPath2) {
+    URI srcUri = qualPath1.toUri();
+    URI dstUri = qualPath2.toUri();
+    return (srcUri.getAuthority().equals(dstUri.getAuthority()) && srcUri
+        .getAuthority().equals(dstUri.getAuthority()));
+  }
+}

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

@@ -23,7 +23,6 @@ import java.io.IOException;
 import java.net.URI;
 import java.util.ArrayList;
 import java.util.Arrays;
-import java.util.Collection;
 import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -244,7 +243,7 @@ public abstract class FileSystem extends Configured implements Closeable {
   /** Make sure that a path specifies a FileSystem. */
   public Path makeQualified(Path path) {
     checkPath(path);
-    return path.makeQualified(this);
+    return path.makeQualified(this.getUri(), this.getWorkingDirectory());
   }
     
   /** create a file with the provided permission
@@ -363,6 +362,26 @@ public abstract class FileSystem extends Configured implements Closeable {
     String[] host = { "localhost" };
     return new BlockLocation[] { new BlockLocation(name, host, 0, file.getLen()) };
   }
+ 
+
+  /**
+   * Return an array containing hostnames, offset and size of 
+   * portions of the given file.  For a nonexistent 
+   * file or regions, null will be returned.
+   *
+   * This call is most helpful with DFS, where it returns 
+   * hostnames of machines that contain the given file.
+   *
+   * The FileSystem will simply return an elt containing 'localhost'.
+   */
+  public BlockLocation[] getFileBlockLocations(Path p, 
+      long start, long len) throws IOException {
+    if (p == null) {
+      throw new NullPointerException();
+    }
+    FileStatus file = getFileStatus(p);
+    return getFileBlockLocations(file, start, len);
+  }
   
   /**
    * Return a set of server default configuration values
@@ -555,7 +574,7 @@ public abstract class FileSystem extends Configured implements Closeable {
    * Opens an FSDataOutputStream at the indicated Path with write-progress
    * reporting.
    * @param f the file name to open.
-   * @param permission
+   * @param permission - applied against umask
    * @param flag determines the semantic of this create.
    * @param bufferSize the size of the buffer to be used.
    * @param replication required block replication for the file.
@@ -569,6 +588,46 @@ public abstract class FileSystem extends Configured implements Closeable {
       EnumSet<CreateFlag> flag, int bufferSize, short replication, long blockSize,
       Progressable progress) throws IOException ;
   
+  /*
+   * This version of the create method assumes that the permission 
+   * of create does not matter.
+   * It has been added to support the FileContext that processes the permission
+   * with umask before calling this method.
+   * This a temporary method added to support the transition from FileSystem
+   * to FileContext for user applications.
+   */
+  @Deprecated
+  protected FSDataOutputStream primitiveCreate(Path f,
+     FsPermission absolutePermission, EnumSet<CreateFlag> flag, int bufferSize,
+     short replication, long blockSize, Progressable progress,
+     int bytesPerChecksum) throws IOException {
+    
+    // Default impl  assumes that permissions do not matter and 
+    // nor does the bytesPerChecksum  hence
+    // calling the regular create is good enough.
+    // FSs that implement permissions should override this.
+    
+    return this.create(f, absolutePermission, flag, bufferSize, replication,
+        blockSize, progress);
+  }
+  
+
+  /**
+   * This version of the mkdirs method assumes that the permission.
+   * It has been added to support the FileContext that processes the the permission
+   * with umask before calling this method.
+   * This a temporary method added to support the transition from FileSystem
+   * to FileContext for user applications.
+   */
+  @Deprecated
+  protected boolean primitiveMkdir(Path f, FsPermission absolutePermission)
+    throws IOException {
+    // Default impl is to assume that permissions do not matter and hence
+    // calling the regular mkdirs is good enough.
+    // FSs that implement permissions should override this.
+   return this.mkdirs(f, absolutePermission);
+  }
+
 
   /**
    * Creates the given Path as a brand-new zero-length file.  If
@@ -1142,8 +1201,8 @@ public abstract class FileSystem extends Configured implements Closeable {
    * The default implementation returns "/user/$USER/".
    */
   public Path getHomeDirectory() {
-    return new Path("/user/"+System.getProperty("user.name"))
-      .makeQualified(this);
+    return this.makeQualified(
+        new Path("/user/"+System.getProperty("user.name")));
   }
 
 
@@ -1160,6 +1219,23 @@ public abstract class FileSystem extends Configured implements Closeable {
    * @return the directory pathname
    */
   public abstract Path getWorkingDirectory();
+  
+  
+  /**
+   * Note: with the new FilesContext class, getWorkingDirectory()
+   * will be removed. 
+   * The working directory is implemented in FilesContext.
+   * 
+   * Some file systems like LocalFileSystem have an initial workingDir
+   * that we use as the starting workingDir. For other file systems
+   * like HDFS there is no built in notion of an inital workingDir.
+   * 
+   * @return if there is built in notion of workingDir then it
+   * is returned; else a null is returned.
+   */
+  protected Path getInitialWorkingDirectory() {
+    return null;
+  }
 
   /**
    * Call {@link #mkdirs(Path, FsPermission)} with default permission.

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

@@ -167,7 +167,11 @@ public class FilterFileSystem extends FileSystem {
   public Path getWorkingDirectory() {
     return fs.getWorkingDirectory();
   }
-
+  
+  protected Path getInitialWorkingDirectory() {
+    return fs.getInitialWorkingDirectory();
+  }
+  
   /** {@inheritDoc} */
   @Override
   public FsStatus getStatus(Path p) throws IOException {
@@ -276,4 +280,19 @@ public class FilterFileSystem extends FileSystem {
       ) throws IOException {
     fs.setPermission(p, permission);
   }
+
+  @Override
+  protected FSDataOutputStream primitiveCreate(Path f,
+      FsPermission absolutePermission, EnumSet<CreateFlag> flag,
+      int bufferSize, short replication, long blockSize, Progressable progress, int bytesPerChecksum)
+      throws IOException {
+    return fs.primitiveCreate(f, absolutePermission, flag,
+        bufferSize, replication, blockSize, progress, bytesPerChecksum);
+  }
+
+  @Override
+  protected boolean primitiveMkdir(Path f, FsPermission abdolutePermission)
+      throws IOException {
+    return fs.primitiveMkdir(f, abdolutePermission);
+  }
 }

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

@@ -0,0 +1,113 @@
+/**
+ * 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.net.URI;
+
+import org.apache.hadoop.conf.Configuration;
+
+/** 
+ * This class is thin layer to manage the FS related keys in
+ * a configuration object.
+ * It provides convenience static method to set and get the keys from a 
+ * a configuration.
+ *
+ */
+
+final class FsConfig {
+  private FsConfig() {}
+  
+  // Configuration keys  and default values in the config file
+  // TBD note we should deprecate the keys constants elsewhere
+  
+  
+  // The Keys
+  static final String FS_DEFAULT_NAME_KEY = "fs.default.name";
+  static final String FS_HOME_DIR_ROOT_KEY = "fs.homeDir";
+  static final String FS_REPLICATION_FACTOR_KEY = "dfs.replication";
+  static final String FS_BLOCK_SIZE_KEY = "dfs.block.size";
+  static final String IO_BUFFER_SIZE_KEY ="io.file.buffer.size";
+
+
+  // The default values
+  // Default values of SERVER_DEFAULT(-1) implies use the ones from
+  // the target file system where files are created.
+  static final String FS_DEFAULT_NAME = "file:///";
+  static final String FS_HOME_DIR_ROOT = "/user"; // relative to FS_DEFAULT
+  static final short FS_DEFAULT_REPLICATION_FACTOR = 3;
+  static final long FS_DEFAULT_BLOCK_SIZE = 32 * 1024 * 1024;
+  static final int IO_BUFFER_SIZE =4096;
+  
+  
+  
+  public static String getDefaultFsURI(final Configuration conf) {
+    return conf.get(FS_DEFAULT_NAME_KEY, FS_DEFAULT_NAME);
+  }
+  
+  public static String getHomeDir(final Configuration conf) {
+    return conf.get(FS_HOME_DIR_ROOT_KEY, FS_HOME_DIR_ROOT);
+  }
+  
+  public static short getDefaultReplicationFactor(final Configuration conf) {
+    return (short) 
+        conf.getInt(FS_REPLICATION_FACTOR_KEY, FS_DEFAULT_REPLICATION_FACTOR);
+  }
+  
+  public static long getDefaultBlockSize(final Configuration conf) {
+    return conf.getLong(FS_BLOCK_SIZE_KEY, FS_DEFAULT_BLOCK_SIZE);
+  }
+
+  
+  public static int getDefaultIOBuffersize(final Configuration conf) {
+    return conf.getInt(IO_BUFFER_SIZE_KEY, IO_BUFFER_SIZE);
+  }
+  
+  public static Class<?> getImplClass(URI uri, Configuration conf) {
+    String scheme = uri.getScheme();
+    if (scheme == null) {
+      throw new IllegalArgumentException("No scheme");
+    }
+    return conf.getClass("fs." + uri.getScheme() + ".impl", null);
+  }
+
+  
+  /**
+   * The Setters: see the note on the javdoc for the class above.
+   */
+
+  public static void setDefaultFS(final Configuration conf, String uri) {
+    conf.set(FS_DEFAULT_NAME_KEY, uri);
+  }
+  
+  public static void setHomeDir(final Configuration conf, String path) {
+    conf.set(FS_HOME_DIR_ROOT_KEY, path);
+  }
+  
+  public static void setDefaultReplicationFactor(final Configuration conf,
+    short rf) {
+    conf.setInt(FS_REPLICATION_FACTOR_KEY, rf);
+  }
+  
+  public static void setDefaultBlockSize(final Configuration conf, long bs) {
+    conf.setLong(FS_BLOCK_SIZE_KEY, bs);
+  }
+  
+  public static void setDefaultIOBuffersize(final Configuration conf, int bs) {
+    conf.setInt(IO_BUFFER_SIZE_KEY, bs);
+  }
+}

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

@@ -126,6 +126,13 @@ public class Path implements Comparable {
     initialize(scheme, authority, path);
   }
 
+  /**
+   * Construct a path from a URI
+   */
+  public Path(URI aUri) {
+    uri = aUri;
+  }
+  
   /** Construct a Path from components. */
   public Path(String scheme, String authority, String path) {
     checkPathArg( path );
@@ -175,10 +182,23 @@ public class Path implements Comparable {
     return FileSystem.get(this.toUri(), conf);
   }
 
-  /** True if the directory of this path is absolute. */
-  public boolean isAbsolute() {
+  /**
+   *  True if the path component (i.e. directory) of this URI is absolute.
+   */
+  public boolean isUriPathAbsolute() {
     int start = hasWindowsDrive(uri.getPath(), true) ? 3 : 0;
     return uri.getPath().startsWith(SEPARATOR, start);
+   }
+  
+  /** True if the directory of this path is absolute. */
+  /**
+   * There is some ambiguity here. An absolute path is a slash
+   * relative name without a scheme or an authority.
+   * So either this method was incorrectly named or its
+   * implementation is incorrect.
+   */
+  public boolean isAbsolute() {
+     return isUriPathAbsolute();
   }
 
   /** Returns the final component of this path.*/
@@ -265,29 +285,41 @@ public class Path implements Comparable {
     return depth;
   }
 
-  /** Returns a qualified path object. */
+  
+  /**
+   *  Returns a qualified path object.
+   *  
+   *  Deprecated - use {@link #makeQualified(URI, Path)}
+   */
+ 
+  @Deprecated
   public Path makeQualified(FileSystem fs) {
+    return makeQualified(fs.getUri(), fs.getWorkingDirectory());
+  }
+  
+  
+  /** Returns a qualified path object. */
+  public Path makeQualified(URI defaultUri, Path workingDir ) {
     Path path = this;
     if (!isAbsolute()) {
-      path = new Path(fs.getWorkingDirectory(), this);
+      path = new Path(workingDir, this);
     }
 
     URI pathUri = path.toUri();
-    URI fsUri = fs.getUri();
       
     String scheme = pathUri.getScheme();
     String authority = pathUri.getAuthority();
 
     if (scheme != null &&
-        (authority != null || fsUri.getAuthority() == null))
+        (authority != null || defaultUri.getAuthority() == null))
       return path;
 
     if (scheme == null) {
-      scheme = fsUri.getScheme();
+      scheme = defaultUri.getScheme();
     }
 
     if (authority == null) {
-      authority = fsUri.getAuthority();
+      authority = defaultUri.getAuthority();
       if (authority == null) {
         authority = "";
       }

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

@@ -46,7 +46,7 @@ public class RawLocalFileSystem extends FileSystem {
   private Path workingDir;
   
   public RawLocalFileSystem() {
-    workingDir = new Path(System.getProperty("user.dir")).makeQualified(this);
+    workingDir = getInitialWorkingDirectory();
   }
   
   /** Convert a path to a File. */
@@ -96,10 +96,10 @@ public class RawLocalFileSystem extends FileSystem {
   }
 
   /*******************************************************
-   * For open()'s FSInputStream
+   * For open()'s FSInputStream.
    *******************************************************/
   class LocalFSFileInputStream extends FSInputStream {
-    FileInputStream fis;
+    private FileInputStream fis;
     private long position;
 
     public LocalFSFileInputStream(Path f) throws IOException {
@@ -181,7 +181,7 @@ public class RawLocalFileSystem extends FileSystem {
    * For create()'s FSOutputStream.
    *********************************************************/
   class LocalFSFileOutputStream extends OutputStream implements Syncable {
-    FileOutputStream fos;
+    private FileOutputStream fos;
     
     private LocalFSFileOutputStream(Path f, boolean append) throws IOException {
       this.fos = new FileOutputStream(pathToFile(f), append);
@@ -229,7 +229,7 @@ public class RawLocalFileSystem extends FileSystem {
 
   /** {@inheritDoc} */
   public FSDataOutputStream create(Path f, boolean overwrite, int bufferSize,
-                                   short replication, long blockSize, Progressable progress)
+    short replication, long blockSize, Progressable progress)
     throws IOException {
     if (exists(f) && !overwrite) {
       throw new IOException("File already exists:"+f);
@@ -245,23 +245,38 @@ public class RawLocalFileSystem extends FileSystem {
   /** {@inheritDoc} */
   @Override
   public FSDataOutputStream create(Path f, FsPermission permission,
-      EnumSet<CreateFlag> flag, int bufferSize, short replication, long blockSize,
-      Progressable progress) throws IOException {
+    EnumSet<CreateFlag> flag, int bufferSize, short replication, long blockSize,
+    Progressable progress) throws IOException {
+    return primitiveCreate(f,
+        permission.applyUMask(FsPermission.getUMask(getConf())), flag,
+        bufferSize,  replication,  blockSize,  progress,  -1);
+    
+    
+     
+  }
+  
+
+  @Override
+  protected FSDataOutputStream primitiveCreate(Path f,
+      FsPermission absolutePermission, EnumSet<CreateFlag> flag,
+      int bufferSize, short replication, long blockSize, Progressable progress,
+      int bytesPerChecksum) throws IOException {
     
-      if(flag.contains(CreateFlag.APPEND)){
-        if (!exists(f)){
-          if(flag.contains(CreateFlag.CREATE))
-            return create(f, false, bufferSize, replication, blockSize, progress);
+    if(flag.contains(CreateFlag.APPEND)){
+      if (!exists(f)){
+        if(flag.contains(CreateFlag.CREATE)) {
+          return create(f, false, bufferSize, replication, blockSize, null);
         }
-        return append(f, bufferSize, progress);
+      }
+      return append(f, bufferSize, null);
     }
-   
-    FSDataOutputStream out = create(f,
-        flag.contains(CreateFlag.OVERWRITE), bufferSize, replication, blockSize, progress);
-    setPermission(f, permission);
+ 
+    FSDataOutputStream out = create(f, flag.contains(CreateFlag.OVERWRITE),
+                                 bufferSize, replication, blockSize, progress);
+    setPermission(f, absolutePermission);
     return out;
   }
-  
+
   public boolean rename(Path src, Path dst) throws IOException {
     if (pathToFile(src).renameTo(pathToFile(dst))) {
       return true;
@@ -289,7 +304,7 @@ public class RawLocalFileSystem extends FileSystem {
     }
     if (localf.isFile()) {
       return new FileStatus[] {
-          new RawLocalFileStatus(localf, getDefaultBlockSize(), this) };
+        new RawLocalFileStatus(localf, getDefaultBlockSize(), this) };
     }
 
     String[] names = localf.list();
@@ -328,14 +343,25 @@ public class RawLocalFileSystem extends FileSystem {
   @Override
   public boolean mkdirs(Path f, FsPermission permission) throws IOException {
     boolean b = mkdirs(f);
-    if(b)
+    if(b) {
       setPermission(f, permission);
+    }
     return b;
   }
   
+
+  @Override
+  protected boolean primitiveMkdir(Path f, FsPermission absolutePermission)
+    throws IOException {
+    boolean b = mkdirs(f);
+    setPermission(f, absolutePermission);
+    return b;
+  }
+  
+  
   @Override
   public Path getHomeDirectory() {
-    return new Path(System.getProperty("user.home")).makeQualified(this);
+    return this.makeQualified(new Path(System.getProperty("user.home")));
   }
 
   /**
@@ -350,6 +376,11 @@ public class RawLocalFileSystem extends FileSystem {
   public Path getWorkingDirectory() {
     return workingDir;
   }
+  
+  @Override
+  protected Path getInitialWorkingDirectory() {
+    return this.makeQualified(new Path(System.getProperty("user.dir")));
+  }
 
   /** {@inheritDoc} */
   @Override
@@ -391,7 +422,7 @@ public class RawLocalFileSystem extends FileSystem {
     if (path.exists()) {
       return new RawLocalFileStatus(pathToFile(f), getDefaultBlockSize(), this);
     } else {
-      throw new FileNotFoundException( "File " + f + " does not exist.");
+      throw new FileNotFoundException("File " + f + " does not exist.");
     }
   }
 
@@ -406,7 +437,7 @@ public class RawLocalFileSystem extends FileSystem {
     
     RawLocalFileStatus(File f, long defaultBlockSize, FileSystem fs) {
       super(f.length(), f.isDirectory(), 1, defaultBlockSize,
-            f.lastModified(), new Path(f.getPath()).makeQualified(fs));
+            f.lastModified(), fs.makeQualified(new Path(f.getPath())));
     }
     
     @Override
@@ -482,8 +513,8 @@ public class RawLocalFileSystem extends FileSystem {
    * Use the command chown to set owner.
    */
   @Override
-  public void setOwner(Path p, String username, String groupname
-      ) throws IOException {
+  public void setOwner(Path p, String username, String groupname)
+    throws IOException {
     if (username == null && groupname == null) {
       throw new IOException("username == null && groupname == null");
     }
@@ -501,8 +532,8 @@ public class RawLocalFileSystem extends FileSystem {
    * Use the command chmod to set permission.
    */
   @Override
-  public void setPermission(Path p, FsPermission permission
-      ) throws IOException {
+  public void setPermission(Path p, FsPermission permission)
+    throws IOException {
     execCommand(pathToFile(p), Shell.SET_PERMISSION_COMMAND,
         String.format("%05o", permission.toShort()));
   }
@@ -514,4 +545,5 @@ public class RawLocalFileSystem extends FileSystem {
     String output = Shell.execCommand(args);
     return output;
   }
+
 }

src/test/core/org/apache/hadoop/fs/FileContextMainOperationsBaseTest.java

@@ -0,0 +1,572 @@
+/**
+ * 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.FileNotFoundException;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.EnumSet;
+import java.util.Random;
+
+import org.junit.*;
+
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.FileContext.CreateOpts;
+import org.apache.hadoop.fs.permission.FsPermission;
+
+/**
+ * <p>
+ * A collection of tests for the {@link FileContext}.
+ * This test should be used for testing an instance of FileContext
+ *  that has been initialized to a specific default FileSystem such a
+ *  LocalFileSystem, HDFS,S3, etc.
+ * </p>
+ * <p>
+ * To test a given {@link FileSystem} implementation create a subclass of this
+ * test and override {@link #setUp()} to initialize the <code>fc</code> 
+ * {@link FileContext} instance variable.
+ * 
+ * Since this a junit 4 you can also do a single setup before 
+ * the start of any tests.
+ * E.g.
+ *     @BeforeClass   public static void clusterSetupAtBegining()
+ *     @AfterClass    public static void ClusterShutdownAtEnd()
+ * </p>
+ */
+public abstract class FileContextMainOperationsBaseTest  {
+  
+  
+  private static String TEST_ROOT_DIR =
+    System.getProperty("test.build.data", "/tmp");
+  
+  protected Path getTestRootPath(String pathString) {
+    return fc.makeQualified(new Path(TEST_ROOT_DIR, pathString));
+  }
+
+  
+  protected static FileContext fc;
+  private static byte[] data = new byte[getBlockSize() * 2]; // two blocks of data
+  {
+    for (int i = 0; i < data.length; i++) {
+      data[i] = (byte) (i % 10);
+    }
+  }
+  
+  @Before
+  public void setUp() throws Exception {
+    fc.mkdirs(getTestRootPath("test"), FileContext.DEFAULT_PERM);
+  }
+  
+  @After
+  public void tearDown() throws Exception {
+    fc.delete(getTestRootPath("test"), true);
+  }
+  
+  protected static int getBlockSize() {
+    return 1024;
+  }
+  
+  protected Path getDefaultWorkingDirectory() throws IOException {
+    return getTestRootPath("/user/" + System.getProperty("user.name")).makeQualified(
+              fc.getDefaultFileSystem().getUri(), fc.getWorkingDirectory());
+  }
+
+  protected boolean renameSupported() {
+    return true;
+  }
+
+  @Test
+  public void testFsStatus() throws Exception {
+    FsStatus fsStatus = fc.getFsStatus(null);
+    Assert.assertNotNull(fsStatus);
+    //used, free and capacity are non-negative longs
+    Assert.assertTrue(fsStatus.getUsed() >= 0);
+    Assert.assertTrue(fsStatus.getRemaining() >= 0);
+    Assert.assertTrue(fsStatus.getCapacity() >= 0);
+  }
+  
+  @Test
+  public void testWorkingDirectory() throws Exception {
+
+    Path workDir = getDefaultWorkingDirectory();
+    Assert.assertEquals(workDir, fc.getWorkingDirectory());
+
+    fc.setWorkingDirectory(new Path("."));
+    Assert.assertEquals(workDir, fc.getWorkingDirectory());
+
+    fc.setWorkingDirectory(new Path(".."));
+    Assert.assertEquals(workDir.getParent(), fc.getWorkingDirectory());
+    
+    // cd using a relative path
+    Path relativeDir = new Path("existingDir1");
+    Path absoluteDir = new Path(workDir.getParent(),"existingDir1");
+    fc.mkdirs(absoluteDir, FileContext.DEFAULT_PERM);
+    fc.setWorkingDirectory(relativeDir);
+    Assert.assertEquals(absoluteDir,
+                                        fc.getWorkingDirectory());
+    // cd using a absolute path
+    absoluteDir = getTestRootPath("test/existingDir2");
+    fc.mkdirs(absoluteDir, FileContext.DEFAULT_PERM);
+    fc.setWorkingDirectory(absoluteDir);
+    Assert.assertEquals(absoluteDir, fc.getWorkingDirectory());
+    
+    // Now open a file relative to the wd we just set above.
+    Path absolutePath = new Path(absoluteDir, "foo");
+    fc.create(absolutePath, EnumSet.of(CreateFlag.CREATE)).close();
+    fc.open(new Path("foo")).close();
+
+    absoluteDir = getTestRootPath("nonexistingPath");
+    try {
+      fc.setWorkingDirectory(absoluteDir);
+      Assert.fail("cd to non existing dir should have failed");
+    } catch (Exception e) {
+      // Exception as expected
+    }
+    
+    // Try a URI
+    absoluteDir = new Path("file:///tmp/existingDir");
+    fc.mkdirs(absoluteDir, FileContext.DEFAULT_PERM);
+    fc.setWorkingDirectory(absoluteDir);
+    Assert.assertEquals(absoluteDir, fc.getWorkingDirectory());
+
+  }
+  
+  @Test
+  public void testMkdirs() throws Exception {
+    Path testDir = getTestRootPath("test/hadoop");
+    Assert.assertFalse(fc.exists(testDir));
+    Assert.assertFalse(fc.isFile(testDir));
+
+    Assert.assertTrue(fc.mkdirs(testDir, FsPermission.getDefault()));
+
+    Assert.assertTrue(fc.exists(testDir));
+    Assert.assertFalse(fc.isFile(testDir));
+    
+    Assert.assertTrue(fc.mkdirs(testDir, FsPermission.getDefault()));
+
+    Assert.assertTrue(fc.exists(testDir));
+    Assert.assertFalse(fc.isFile(testDir));
+
+    Path parentDir = testDir.getParent();
+    Assert.assertTrue(fc.exists(parentDir));
+    Assert.assertFalse(fc.isFile(parentDir));
+
+    Path grandparentDir = parentDir.getParent();
+    Assert.assertTrue(fc.exists(grandparentDir));
+    Assert.assertFalse(fc.isFile(grandparentDir));
+    
+  }
+  
+  @Test
+  public void testMkdirsFailsForSubdirectoryOfExistingFile() throws Exception {
+    Path testDir = getTestRootPath("test/hadoop");
+    Assert.assertFalse(fc.exists(testDir));
+    Assert.assertTrue(fc.mkdirs(testDir, FsPermission.getDefault()));
+    Assert.assertTrue(fc.exists(testDir));
+    
+    createFile(getTestRootPath("test/hadoop/file"));
+    
+    Path testSubDir = getTestRootPath("test/hadoop/file/subdir");
+    try {
+      fc.mkdirs(testSubDir, FsPermission.getDefault());
+      Assert.fail("Should throw IOException.");
+    } catch (IOException e) {
+      // expected
+    }
+    Assert.assertFalse(fc.exists(testSubDir));
+    
+    Path testDeepSubDir = getTestRootPath("test/hadoop/file/deep/sub/dir");
+    try {
+      fc.mkdirs(testDeepSubDir, FsPermission.getDefault());
+      Assert.fail("Should throw IOException.");
+    } catch (IOException e) {
+      // expected
+    }
+    Assert.assertFalse(fc.exists(testDeepSubDir));
+    
+  }
+  
+  @Test
+  public void testGetFileStatusThrowsExceptionForNonExistentFile() 
+    throws Exception {
+    try {
+      fc.getFileStatus(getTestRootPath("test/hadoop/file"));
+      Assert.fail("Should throw FileNotFoundException");
+    } catch (FileNotFoundException e) {
+      // expected
+    }
+  } 
+  
+  public void testListStatusThrowsExceptionForNonExistentFile()
+                                                    throws Exception {
+    try {
+      fc.listStatus(getTestRootPath("test/hadoop/file"));
+      Assert.fail("Should throw FileNotFoundException");
+    } catch (FileNotFoundException fnfe) {
+      // expected
+    }
+  }
+  
+  @Test
+  public void testListStatus() throws Exception {
+    Path[] testDirs = { getTestRootPath("test/hadoop/a"),
+                        getTestRootPath("test/hadoop/b"),
+                        getTestRootPath("test/hadoop/c/1"), };
+    Assert.assertFalse(fc.exists(testDirs[0]));
+
+    for (Path path : testDirs) {
+      Assert.assertTrue(fc.mkdirs(path, FsPermission.getDefault()));
+    }
+
+    FileStatus[] paths = fc.listStatus(getTestRootPath("test"));
+    Assert.assertEquals(1, paths.length);
+    Assert.assertEquals(getTestRootPath("test/hadoop"), paths[0].getPath());
+
+    paths = fc.listStatus(getTestRootPath("test/hadoop"));
+    Assert.assertEquals(3, paths.length);
+
+    
+    Assert.assertTrue(getTestRootPath("test/hadoop/a").equals(paths[0].getPath()) ||
+        getTestRootPath("test/hadoop/a").equals(paths[1].getPath()) ||
+        getTestRootPath("test/hadoop/a").equals(paths[2].getPath()));
+    Assert.assertTrue(getTestRootPath("test/hadoop/b").equals(paths[0].getPath()) ||
+        getTestRootPath("test/hadoop/b").equals(paths[1].getPath()) ||
+        getTestRootPath("test/hadoop/b").equals(paths[2].getPath()));
+    Assert.assertTrue(getTestRootPath("test/hadoop/c").equals(paths[0].getPath()) ||
+        getTestRootPath("test/hadoop/c").equals(paths[1].getPath()) ||
+        getTestRootPath("test/hadoop/c").equals(paths[2].getPath()));
+
+
+    paths = fc.listStatus(getTestRootPath("test/hadoop/a"));
+    Assert.assertEquals(0, paths.length);
+  }
+  
+  @Test
+  public void testWriteReadAndDeleteEmptyFile() throws Exception {
+    writeReadAndDelete(0);
+  }
+
+  @Test
+  public void testWriteReadAndDeleteHalfABlock() throws Exception {
+    writeReadAndDelete(getBlockSize() / 2);
+  }
+
+  @Test
+  public void testWriteReadAndDeleteOneBlock() throws Exception {
+    writeReadAndDelete(getBlockSize());
+  }
+  
+  @Test
+  public void testWriteReadAndDeleteOneAndAHalfBlocks() throws Exception {
+    writeReadAndDelete(getBlockSize() + (getBlockSize() / 2));
+  }
+  
+  @Test
+  public void testWriteReadAndDeleteTwoBlocks() throws Exception {
+    writeReadAndDelete(getBlockSize() * 2);
+  }
+  
+  private void writeReadAndDelete(int len) throws IOException {
+    Path path = getTestRootPath("test/hadoop/file");
+    
+    fc.mkdirs(path.getParent(), FsPermission.getDefault());
+
+    FSDataOutputStream out = fc.create(path, EnumSet.of(CreateFlag.CREATE), 
+        CreateOpts.repFac((short) 1), CreateOpts.blockSize(getBlockSize()));
+    out.write(data, 0, len);
+    out.close();
+
+    Assert.assertTrue("Exists", fc.exists(path));
+    Assert.assertEquals("Length", len, fc.getFileStatus(path).getLen());
+
+    FSDataInputStream in = fc.open(path);
+    byte[] buf = new byte[len];
+    in.readFully(0, buf);
+    in.close();
+
+    Assert.assertEquals(len, buf.length);
+    for (int i = 0; i < buf.length; i++) {
+      Assert.assertEquals("Position " + i, data[i], buf[i]);
+    }
+    
+    Assert.assertTrue("Deleted", fc.delete(path, false));
+    
+    Assert.assertFalse("No longer exists", fc.exists(path));
+
+  }
+  
+  @Test
+  public void testOverwrite() throws IOException {
+    Path path = getTestRootPath("test/hadoop/file");
+    
+    fc.mkdirs(path.getParent(), FsPermission.getDefault());
+
+    createFile(path);
+    
+    Assert.assertTrue("Exists", fc.exists(path));
+    Assert.assertEquals("Length", data.length, fc.getFileStatus(path).getLen());
+    
+    try {
+      fc.create(path, EnumSet.of(CreateFlag.CREATE));
+      Assert.fail("Should throw IOException.");
+    } catch (IOException e) {
+      // Expected
+    }
+    
+    FSDataOutputStream out = fc.create(path,EnumSet.of(CreateFlag.OVERWRITE));
+    out.write(data, 0, data.length);
+    out.close();
+    
+    Assert.assertTrue("Exists", fc.exists(path));
+    Assert.assertEquals("Length", data.length, fc.getFileStatus(path).getLen());
+    
+  }
+  
+  @Test
+  public void testWriteInNonExistentDirectory() throws IOException {
+    Path path = getTestRootPath("test/hadoop/file");
+    Assert.assertFalse("Parent doesn't exist", fc.exists(path.getParent()));
+    createFile(path);
+    
+    Assert.assertTrue("Exists", fc.exists(path));
+    Assert.assertEquals("Length", data.length, fc.getFileStatus(path).getLen());
+    Assert.assertTrue("Parent exists", fc.exists(path.getParent()));
+  }
+
+  @Test
+  public void testDeleteNonExistentFile() throws IOException {
+    Path path = getTestRootPath("test/hadoop/file");    
+    Assert.assertFalse("Doesn't exist", fc.exists(path));
+    Assert.assertFalse("No deletion", fc.delete(path, true));
+  }
+  
+  @Test
+  public void testDeleteRecursively() throws IOException {
+    Path dir = getTestRootPath("test/hadoop");
+    Path file = getTestRootPath("test/hadoop/file");
+    Path subdir = getTestRootPath("test/hadoop/subdir");
+    
+    createFile(file);
+    Assert.assertTrue("Created subdir", fc.mkdirs(subdir, FsPermission.getDefault()));
+    
+    Assert.assertTrue("File exists", fc.exists(file));
+    Assert.assertTrue("Dir exists", fc.exists(dir));
+    Assert.assertTrue("Subdir exists", fc.exists(subdir));
+    
+    try {
+      fc.delete(dir, false);
+      Assert.fail("Should throw IOException.");
+    } catch (IOException e) {
+      // expected
+    }
+    Assert.assertTrue("File still exists", fc.exists(file));
+    Assert.assertTrue("Dir still exists", fc.exists(dir));
+    Assert.assertTrue("Subdir still exists", fc.exists(subdir));
+    
+    Assert.assertTrue("Deleted", fc.delete(dir, true));
+    Assert.assertFalse("File doesn't exist", fc.exists(file));
+    Assert.assertFalse("Dir doesn't exist", fc.exists(dir));
+    Assert.assertFalse("Subdir doesn't exist", fc.exists(subdir));
+  }
+  
+  @Test
+  public void testDeleteEmptyDirectory() throws IOException {
+    Path dir = getTestRootPath("test/hadoop");
+    Assert.assertTrue(fc.mkdirs(dir, FsPermission.getDefault()));
+    Assert.assertTrue("Dir exists", fc.exists(dir));
+    Assert.assertTrue("Deleted", fc.delete(dir, false));
+    Assert.assertFalse("Dir doesn't exist", fc.exists(dir));
+  }
+  
+  @Test
+  public void testRenameNonExistentPath() throws Exception {
+    if (!renameSupported()) return;
+    Path src = getTestRootPath("test/hadoop/NonExistingPath");
+    Path dst = getTestRootPath("test/new/newpath");
+    try {
+      fc.rename(src, dst);
+      Assert.assertTrue("rename of non existing path should have Assert.failed",
+                                                    false);
+    } catch (Exception e) {
+      // expected
+    }
+  }
+
+  @Test
+  public void testRenameFileMoveToNonExistentDirectory() throws Exception {
+    if (!renameSupported()) return;
+    
+    Path src = getTestRootPath("test/hadoop/file");
+    createFile(src);
+    Path dst = getTestRootPath("test/NonExisting/foo");
+    rename(src, dst, false, true, false);
+  }
+
+  @Test
+  public void testRenameFileMoveToExistingDirectory() throws Exception {
+    if (!renameSupported()) return;
+    
+    Path src = getTestRootPath("test/hadoop/file");
+    createFile(src);
+    Path dst = getTestRootPath("test/Existing/newfile");
+    fc.mkdirs(dst.getParent(), FsPermission.getDefault());
+    rename(src, dst, true, false, true);
+  }
+
+  @Test
+  public void testRenameFileAsExistingFile() throws Exception {
+    if (!renameSupported()) return;
+    
+    Path src = getTestRootPath("test/hadoop/file");
+    createFile(src);
+    Path dst = getTestRootPath("test/existing/existingFile");
+    createFile(dst);
+    rename(src, dst, true, false, true);
+  }
+
+  @Test
+  public void testRenameFileAsExistingDirectory() throws Exception {
+    if (!renameSupported()) return;
+    
+    Path src = getTestRootPath("test/hadoop/file");
+    createFile(src);
+    Path dst = getTestRootPath("test/existing/existingDir");
+    fc.mkdirs(dst, FsPermission.getDefault());
+    rename(src, dst, true, false, true);
+    Assert.assertTrue("Destination changed",
+        fc.exists(getTestRootPath("test/existing/existingDir/file")));
+  }
+  
+  @Test
+  public void testRenameDirectoryMoveToNonExistentDirectory() 
+    throws Exception {
+    if (!renameSupported()) return;
+    
+    Path src = getTestRootPath("test/hadoop/dir");
+    fc.mkdirs(src, FsPermission.getDefault());
+    Path dst = getTestRootPath("test/nonExisting/newdir");
+    rename(src, dst, false, true, false);
+  }
+  
+  @Test
+  public void testRenameDirectoryMoveToExistingDirectory() throws Exception {
+    if (!renameSupported()) return;
+    
+    Path src = getTestRootPath("test/hadoop/dir");
+    fc.mkdirs(src, FsPermission.getDefault());
+    createFile(getTestRootPath("test/hadoop/dir/file1"));
+    createFile(getTestRootPath("test/hadoop/dir/subdir/file2"));
+    
+    Path dst = getTestRootPath("test/new/newdir");
+    fc.mkdirs(dst.getParent(), FsPermission.getDefault());
+    rename(src, dst, true, false, true);
+    
+    Assert.assertFalse("Nested file1 exists",
+        fc.exists(getTestRootPath("test/hadoop/dir/file1")));
+    Assert.assertFalse("Nested file2 exists",
+        fc.exists(getTestRootPath("test/hadoop/dir/subdir/file2")));
+    Assert.assertTrue("Renamed nested file1 exists",
+        fc.exists(getTestRootPath("test/new/newdir/file1")));
+    Assert.assertTrue("Renamed nested exists",
+        fc.exists(getTestRootPath("test/new/newdir/subdir/file2")));
+  }
+  
+  @Test
+  public void testRenameDirectoryAsExistingFile() throws Exception {
+    if (!renameSupported()) return;
+    
+    Path src = getTestRootPath("test/hadoop/dir");
+    fc.mkdirs(src, FsPermission.getDefault());
+    Path dst = getTestRootPath("test/new/newfile");
+    createFile(dst);
+    rename(src, dst, false, true, true);
+  }
+  
+  @Test
+  public void testRenameDirectoryAsExistingDirectory() throws Exception {
+    if (!renameSupported()) return;
+    
+    Path src = getTestRootPath("test/hadoop/dir");
+    fc.mkdirs(src, FsPermission.getDefault());
+    createFile(getTestRootPath("test/hadoop/dir/file1"));
+    createFile(getTestRootPath("test/hadoop/dir/subdir/file2"));
+    
+    Path dst = getTestRootPath("test/new/newdir");
+    fc.mkdirs(dst, FsPermission.getDefault());
+    rename(src, dst, true, false, true);
+    Assert.assertTrue("Destination changed",
+        fc.exists(getTestRootPath("test/new/newdir/dir")));    
+    Assert.assertFalse("Nested file1 exists",
+        fc.exists(getTestRootPath("test/hadoop/dir/file1")));
+    Assert.assertFalse("Nested file2 exists",
+        fc.exists(getTestRootPath("test/hadoop/dir/subdir/file2")));
+    Assert.assertTrue("Renamed nested file1 exists",
+        fc.exists(getTestRootPath("test/new/newdir/dir/file1")));
+    Assert.assertTrue("Renamed nested exists",
+        fc.exists(getTestRootPath("test/new/newdir/dir/subdir/file2")));
+  }
+
+  @Test
+  public void testInputStreamClosedTwice() throws IOException {
+    //HADOOP-4760 according to Closeable#close() closing already-closed 
+    //streams should have no effect. 
+    Path src = getTestRootPath("test/hadoop/file");
+    createFile(src);
+    FSDataInputStream in = fc.open(src);
+    in.close();
+    in.close();
+  }
+  
+  @Test
+  public void testOutputStreamClosedTwice() throws IOException {
+    //HADOOP-4760 according to Closeable#close() closing already-closed 
+    //streams should have no effect. 
+    Path src = getTestRootPath("test/hadoop/file");
+    FSDataOutputStream out = fc.create(src, EnumSet.of(CreateFlag.CREATE));
+    out.writeChar('H'); //write some data
+    out.close();
+    out.close();
+  }
+
+  protected void createFile(Path path) throws IOException {
+    FSDataOutputStream out = fc.create(path, EnumSet.of(CreateFlag.CREATE));
+    out.write(data, 0, data.length);
+    out.close();
+  }
+  
+  private void rename(Path src, Path dst, boolean renameShouldSucceed,
+      boolean srcExists, boolean dstExists) throws IOException {
+    try {
+      fc.rename(src, dst);
+      if (!renameShouldSucceed)
+        Assert.fail("rename should have thrown exception");
+    } catch (Exception e) {
+      if (renameShouldSucceed)
+        Assert.fail("rename should have suceeded, but threw exception");
+    }
+
+    Assert.assertEquals("Source exists", srcExists, fc.exists(src));
+    Assert.assertEquals("Destination exists", dstExists, fc.exists(dst));
+  }
+}

src/test/core/org/apache/hadoop/fs/TestLocalFSFileContextMainOperations.java

@@ -0,0 +1,59 @@
+/**
+ * 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.IOException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.junit.Before;
+import org.junit.Test;
+
+public class TestLocalFSFileContextMainOperations extends FileContextMainOperationsBaseTest {
+
+  @Before
+  public void setUp() throws Exception {
+    fc = FileContext.getLocalFSFileContext();
+    super.setUp();
+  }
+  
+  static Path wd = null;
+  protected Path getDefaultWorkingDirectory() throws IOException {
+    if (wd == null)
+      wd = FileSystem.getLocal(new Configuration()).getWorkingDirectory();
+    return wd;
+  }
+  
+  @Override
+  @Test
+  public void testRenameFileMoveToNonExistentDirectory() throws Exception {
+    // ignore base class test till hadoop-6240 is fixed
+  }
+    
+  @Override
+  @Test
+  public void testRenameDirectoryMoveToNonExistentDirectory() 
+    throws Exception {
+    // ignore base class test till hadoop-6240 is fixed
+  }
+  @Override
+  @Test
+  public void testRenameDirectoryAsExistingDirectory() throws Exception {
+    // ignore base class test till hadoop-6240 is fixed
+  }
+}