001/**
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.fs;
019
020import java.io.FileNotFoundException;
021import java.io.IOException;
022import java.io.InputStream;
023import java.io.OutputStream;
024import java.net.URI;
025import java.security.PrivilegedExceptionAction;
026import java.util.ArrayList;
027import java.util.EnumSet;
028import java.util.HashSet;
029import java.util.IdentityHashMap;
030import java.util.List;
031import java.util.Map;
032import java.util.Set;
033import java.util.Stack;
034import java.util.TreeSet;
035import java.util.Map.Entry;
036
037import org.apache.commons.logging.Log;
038import org.apache.commons.logging.LogFactory;
039import org.apache.hadoop.HadoopIllegalArgumentException;
040import org.apache.hadoop.classification.InterfaceAudience;
041import org.apache.hadoop.classification.InterfaceStability;
042import org.apache.hadoop.conf.Configuration;
043import org.apache.hadoop.fs.FileSystem.Statistics;
044import org.apache.hadoop.fs.Options.CreateOpts;
045import org.apache.hadoop.fs.permission.AclEntry;
046import org.apache.hadoop.fs.permission.AclStatus;
047import org.apache.hadoop.fs.permission.FsAction;
048import org.apache.hadoop.fs.permission.FsPermission;
049import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_DEFAULT_NAME_KEY;
050import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_DEFAULT_NAME_DEFAULT;
051import org.apache.hadoop.io.IOUtils;
052import org.apache.hadoop.ipc.RpcClientException;
053import org.apache.hadoop.ipc.RpcServerException;
054import org.apache.hadoop.ipc.UnexpectedServerException;
055import org.apache.hadoop.fs.InvalidPathException;
056import org.apache.hadoop.security.AccessControlException;
057import org.apache.hadoop.security.UserGroupInformation;
058import org.apache.hadoop.security.token.Token;
059import org.apache.hadoop.util.ShutdownHookManager;
060
061/**
062 * The FileContext class provides an interface to the application writer for
063 * using the Hadoop file system.
064 * It provides a set of methods for the usual operation: create, open, 
065 * list, etc 
066 * 
067 * <p>
068 * <b> *** Path Names *** </b>
069 * <p>
070 * 
071 * The Hadoop file system supports a URI name space and URI names.
072 * It offers a forest of file systems that can be referenced using fully
073 * qualified URIs.
074 * Two common Hadoop file systems implementations are
075 * <ul>
076 * <li> the local file system: file:///path
077 * <li> the hdfs file system hdfs://nnAddress:nnPort/path
078 * </ul>
079 * 
080 * While URI names are very flexible, it requires knowing the name or address
081 * of the server. For convenience one often wants to access the default system
082 * in one's environment without knowing its name/address. This has an
083 * additional benefit that it allows one to change one's default fs
084 *  (e.g. admin moves application from cluster1 to cluster2).
085 * <p>
086 * 
087 * To facilitate this, Hadoop supports a notion of a default file system.
088 * The user can set his default file system, although this is
089 * typically set up for you in your environment via your default config.
090 * A default file system implies a default scheme and authority; slash-relative
091 * names (such as /for/bar) are resolved relative to that default FS.
092 * Similarly a user can also have working-directory-relative names (i.e. names
093 * not starting with a slash). While the working directory is generally in the
094 * same default FS, the wd can be in a different FS.
095 * <p>
096 *  Hence Hadoop path names can be one of:
097 *  <ul>
098 *  <li> fully qualified URI: scheme://authority/path
099 *  <li> slash relative names: /path relative to the default file system
100 *  <li> wd-relative names: path  relative to the working dir
101 *  </ul>   
102 *  Relative paths with scheme (scheme:foo/bar) are illegal.
103 *  
104 *  <p>
105 *  <b>****The Role of the FileContext and configuration defaults****</b>
106 *  <p>
107 *  The FileContext provides file namespace context for resolving file names;
108 *  it also contains the umask for permissions, In that sense it is like the
109 *  per-process file-related state in Unix system.
110 *  These two properties
111 *  <ul> 
112 *  <li> default file system i.e your slash)
113 *  <li> umask
114 *  </ul>
115 *  in general, are obtained from the default configuration file
116 *  in your environment,  (@see {@link Configuration}).
117 *  
118 *  No other configuration parameters are obtained from the default config as 
119 *  far as the file context layer is concerned. All file system instances
120 *  (i.e. deployments of file systems) have default properties; we call these
121 *  server side (SS) defaults. Operation like create allow one to select many 
122 *  properties: either pass them in as explicit parameters or use
123 *  the SS properties.
124 *  <p>
125 *  The file system related SS defaults are
126 *  <ul>
127 *  <li> the home directory (default is "/user/userName")
128 *  <li> the initial wd (only for local fs)
129 *  <li> replication factor
130 *  <li> block size
131 *  <li> buffer size
132 *  <li> encryptDataTransfer 
133 *  <li> checksum option. (checksumType and  bytesPerChecksum)
134 *  </ul>
135 *
136 * <p>
137 * <b> *** Usage Model for the FileContext class *** </b>
138 * <p>
139 * Example 1: use the default config read from the $HADOOP_CONFIG/core.xml.
140 *   Unspecified values come from core-defaults.xml in the release jar.
141 *  <ul>  
142 *  <li> myFContext = FileContext.getFileContext(); // uses the default config
143 *                                                // which has your default FS 
144 *  <li>  myFContext.create(path, ...);
145 *  <li>  myFContext.setWorkingDir(path)
146 *  <li>  myFContext.open (path, ...);  
147 *  </ul>  
148 * Example 2: Get a FileContext with a specific URI as the default FS
149 *  <ul>  
150 *  <li> myFContext = FileContext.getFileContext(URI)
151 *  <li> myFContext.create(path, ...);
152 *   ...
153 * </ul> 
154 * Example 3: FileContext with local file system as the default
155 *  <ul> 
156 *  <li> myFContext = FileContext.getLocalFSFileContext()
157 *  <li> myFContext.create(path, ...);
158 *  <li> ...
159 *  </ul> 
160 * Example 4: Use a specific config, ignoring $HADOOP_CONFIG
161 *  Generally you should not need use a config unless you are doing
162 *   <ul> 
163 *   <li> configX = someConfigSomeOnePassedToYou.
164 *   <li> myFContext = getFileContext(configX); // configX is not changed,
165 *                                              // is passed down 
166 *   <li> myFContext.create(path, ...);
167 *   <li>...
168 *  </ul>                                          
169 *    
170 */
171
172@InterfaceAudience.Public
173@InterfaceStability.Evolving /*Evolving for a release,to be changed to Stable */
174public class FileContext {
175  
176  public static final Log LOG = LogFactory.getLog(FileContext.class);
177  /**
178   * Default permission for directory and symlink
179   * In previous versions, this default permission was also used to
180   * create files, so files created end up with ugo+x permission.
181   * See HADOOP-9155 for detail. 
182   * Two new constants are added to solve this, please use 
183   * {@link FileContext#DIR_DEFAULT_PERM} for directory, and use
184   * {@link FileContext#FILE_DEFAULT_PERM} for file.
185   * This constant is kept for compatibility.
186   */
187  public static final FsPermission DEFAULT_PERM = FsPermission.getDefault();
188  /**
189   * Default permission for directory
190   */
191  public static final FsPermission DIR_DEFAULT_PERM = FsPermission.getDirDefault();
192  /**
193   * Default permission for file
194   */
195  public static final FsPermission FILE_DEFAULT_PERM = FsPermission.getFileDefault();
196
197  /**
198   * Priority of the FileContext shutdown hook.
199   */
200  public static final int SHUTDOWN_HOOK_PRIORITY = 20;
201
202  /**
203   * List of files that should be deleted on JVM shutdown.
204   */
205  static final Map<FileContext, Set<Path>> DELETE_ON_EXIT = 
206    new IdentityHashMap<FileContext, Set<Path>>();
207
208  /** JVM shutdown hook thread. */
209  static final FileContextFinalizer FINALIZER = 
210    new FileContextFinalizer();
211  
212  private static final PathFilter DEFAULT_FILTER = new PathFilter() {
213    @Override
214    public boolean accept(final Path file) {
215      return true;
216    }
217  };
218  
219  /**
220   * The FileContext is defined by.
221   *  1) defaultFS (slash)
222   *  2) wd
223   *  3) umask
224   */   
225  private final AbstractFileSystem defaultFS; //default FS for this FileContext.
226  private Path workingDir;          // Fully qualified
227  private FsPermission umask;
228  private final Configuration conf;
229  private final UserGroupInformation ugi;
230  final boolean resolveSymlinks;
231
232  private FileContext(final AbstractFileSystem defFs,
233    final FsPermission theUmask, final Configuration aConf) {
234    defaultFS = defFs;
235    umask = FsPermission.getUMask(aConf);
236    conf = aConf;
237    try {
238      ugi = UserGroupInformation.getCurrentUser();
239    } catch (IOException e) {
240      LOG.error("Exception in getCurrentUser: ",e);
241      throw new RuntimeException("Failed to get the current user " +
242                "while creating a FileContext", e);
243    }
244    /*
245     * Init the wd.
246     * WorkingDir is implemented at the FileContext layer 
247     * NOT at the AbstractFileSystem layer. 
248     * If the DefaultFS, such as localFilesystem has a notion of
249     *  builtin WD, we use that as the initial WD.
250     *  Otherwise the WD is initialized to the home directory.
251     */
252    workingDir = defaultFS.getInitialWorkingDirectory();
253    if (workingDir == null) {
254      workingDir = defaultFS.getHomeDirectory();
255    }
256    resolveSymlinks = conf.getBoolean(
257        CommonConfigurationKeys.FS_CLIENT_RESOLVE_REMOTE_SYMLINKS_KEY,
258        CommonConfigurationKeys.FS_CLIENT_RESOLVE_REMOTE_SYMLINKS_DEFAULT);
259    util = new Util(); // for the inner class
260  }
261
262  /* 
263   * Remove relative part - return "absolute":
264   * If input is relative path ("foo/bar") add wd: ie "/<workingDir>/foo/bar"
265   * A fully qualified uri ("hdfs://nn:p/foo/bar") or a slash-relative path
266   * ("/foo/bar") are returned unchanged.
267   * 
268   * Applications that use FileContext should use #makeQualified() since
269   * they really want a fully qualified URI.
270   * Hence this method is not called makeAbsolute() and 
271   * has been deliberately declared private.
272   */
273  Path fixRelativePart(Path p) {
274    if (p.isUriPathAbsolute()) {
275      return p;
276    } else {
277      return new Path(workingDir, p);
278    }
279  }
280
281  /**
282   * Delete all the paths that were marked as delete-on-exit.
283   */
284  static void processDeleteOnExit() {
285    synchronized (DELETE_ON_EXIT) {
286      Set<Entry<FileContext, Set<Path>>> set = DELETE_ON_EXIT.entrySet();
287      for (Entry<FileContext, Set<Path>> entry : set) {
288        FileContext fc = entry.getKey();
289        Set<Path> paths = entry.getValue();
290        for (Path path : paths) {
291          try {
292            fc.delete(path, true);
293          } catch (IOException e) {
294            LOG.warn("Ignoring failure to deleteOnExit for path " + path);
295          }
296        }
297      }
298      DELETE_ON_EXIT.clear();
299    }
300  }
301
302  /**
303   * Get the file system of supplied path.
304   * 
305   * @param absOrFqPath - absolute or fully qualified path
306   * @return the file system of the path
307   * 
308   * @throws UnsupportedFileSystemException If the file system for
309   *           <code>absOrFqPath</code> is not supported.
310   * @throws IOExcepton If the file system for <code>absOrFqPath</code> could
311   *         not be instantiated.
312   */
313  protected AbstractFileSystem getFSofPath(final Path absOrFqPath)
314      throws UnsupportedFileSystemException, IOException {
315    absOrFqPath.checkNotSchemeWithRelative();
316    absOrFqPath.checkNotRelative();
317
318    try { 
319      // Is it the default FS for this FileContext?
320      defaultFS.checkPath(absOrFqPath);
321      return defaultFS;
322    } catch (Exception e) { // it is different FileSystem
323      return getAbstractFileSystem(ugi, absOrFqPath.toUri(), conf);
324    }
325  }
326  
327  private static AbstractFileSystem getAbstractFileSystem(
328      UserGroupInformation user, final URI uri, final Configuration conf)
329      throws UnsupportedFileSystemException, IOException {
330    try {
331      return user.doAs(new PrivilegedExceptionAction<AbstractFileSystem>() {
332        @Override
333        public AbstractFileSystem run() throws UnsupportedFileSystemException {
334          return AbstractFileSystem.get(uri, conf);
335        }
336      });
337    } catch (InterruptedException ex) {
338      LOG.error(ex);
339      throw new IOException("Failed to get the AbstractFileSystem for path: "
340          + uri, ex);
341    }
342  }
343  
344  /**
345   * Protected Static Factory methods for getting a FileContexts
346   * that take a AbstractFileSystem as input. To be used for testing.
347   */
348
349  /**
350   * Create a FileContext with specified FS as default using the specified
351   * config.
352   * 
353   * @param defFS
354   * @param aConf
355   * @return new FileContext with specifed FS as default.
356   */
357  public static FileContext getFileContext(final AbstractFileSystem defFS,
358                    final Configuration aConf) {
359    return new FileContext(defFS, FsPermission.getUMask(aConf), aConf);
360  }
361  
362  /**
363   * Create a FileContext for specified file system using the default config.
364   * 
365   * @param defaultFS
366   * @return a FileContext with the specified AbstractFileSystem
367   *                 as the default FS.
368   */
369  protected static FileContext getFileContext(
370    final AbstractFileSystem defaultFS) {
371    return getFileContext(defaultFS, new Configuration());
372  }
373 
374  /**
375   * Static Factory methods for getting a FileContext.
376   * Note new file contexts are created for each call.
377   * The only singleton is the local FS context using the default config.
378   * 
379   * Methods that use the default config: the default config read from the
380   * $HADOOP_CONFIG/core.xml,
381   * Unspecified key-values for config are defaulted from core-defaults.xml
382   * in the release jar.
383   * 
384   * The keys relevant to the FileContext layer are extracted at time of
385   * construction. Changes to the config after the call are ignore
386   * by the FileContext layer. 
387   * The conf is passed to lower layers like AbstractFileSystem and HDFS which
388   * pick up their own config variables.
389   */
390
391  /**
392   * Create a FileContext using the default config read from the
393   * $HADOOP_CONFIG/core.xml, Unspecified key-values for config are defaulted
394   * from core-defaults.xml in the release jar.
395   * 
396   * @throws UnsupportedFileSystemException If the file system from the default
397   *           configuration is not supported
398   */
399  public static FileContext getFileContext()
400      throws UnsupportedFileSystemException {
401    return getFileContext(new Configuration());
402  }
403
404  /**
405   * @return a FileContext for the local file system using the default config.
406   * @throws UnsupportedFileSystemException If the file system for
407   *           {@link FsConstants#LOCAL_FS_URI} is not supported.
408   */
409  public static FileContext getLocalFSFileContext()
410      throws UnsupportedFileSystemException {
411    return getFileContext(FsConstants.LOCAL_FS_URI);
412  }
413
414  /**
415   * Create a FileContext for specified URI using the default config.
416   * 
417   * @param defaultFsUri
418   * @return a FileContext with the specified URI as the default FS.
419   * 
420   * @throws UnsupportedFileSystemException If the file system for
421   *           <code>defaultFsUri</code> is not supported
422   */
423  public static FileContext getFileContext(final URI defaultFsUri)
424      throws UnsupportedFileSystemException {
425    return getFileContext(defaultFsUri, new Configuration());
426  }
427
428  /**
429   * Create a FileContext for specified default URI using the specified config.
430   * 
431   * @param defaultFsUri
432   * @param aConf
433   * @return new FileContext for specified uri
434   * @throws UnsupportedFileSystemException If the file system with specified is
435   *           not supported
436   * @throws RuntimeException If the file system specified is supported but
437   *         could not be instantiated, or if login fails.
438   */
439  public static FileContext getFileContext(final URI defaultFsUri,
440      final Configuration aConf) throws UnsupportedFileSystemException {
441    UserGroupInformation currentUser = null;
442    AbstractFileSystem defaultAfs = null;
443    if (defaultFsUri.getScheme() == null) {
444      return getFileContext(aConf);
445    }
446    try {
447      currentUser = UserGroupInformation.getCurrentUser();
448      defaultAfs = getAbstractFileSystem(currentUser, defaultFsUri, aConf);
449    } catch (UnsupportedFileSystemException ex) {
450      throw ex;
451    } catch (IOException ex) {
452      LOG.error(ex);
453      throw new RuntimeException(ex);
454    }
455    return getFileContext(defaultAfs, aConf);
456  }
457
458  /**
459   * Create a FileContext using the passed config. Generally it is better to use
460   * {@link #getFileContext(URI, Configuration)} instead of this one.
461   * 
462   * 
463   * @param aConf
464   * @return new FileContext
465   * @throws UnsupportedFileSystemException If file system in the config
466   *           is not supported
467   */
468  public static FileContext getFileContext(final Configuration aConf)
469      throws UnsupportedFileSystemException {
470    return getFileContext(
471      URI.create(aConf.get(FS_DEFAULT_NAME_KEY, FS_DEFAULT_NAME_DEFAULT)), 
472      aConf);
473  }
474
475  /**
476   * @param aConf - from which the FileContext is configured
477   * @return a FileContext for the local file system using the specified config.
478   * 
479   * @throws UnsupportedFileSystemException If default file system in the config
480   *           is not supported
481   * 
482   */
483  public static FileContext getLocalFSFileContext(final Configuration aConf)
484      throws UnsupportedFileSystemException {
485    return getFileContext(FsConstants.LOCAL_FS_URI, aConf);
486  }
487
488  /* This method is needed for tests. */
489  @InterfaceAudience.Private
490  @InterfaceStability.Unstable /* return type will change to AFS once
491                                  HADOOP-6223 is completed */
492  public AbstractFileSystem getDefaultFileSystem() {
493    return defaultFS;
494  }
495  
496  /**
497   * Set the working directory for wd-relative names (such a "foo/bar"). Working
498   * directory feature is provided by simply prefixing relative names with the
499   * working dir. Note this is different from Unix where the wd is actually set
500   * to the inode. Hence setWorkingDir does not follow symlinks etc. This works
501   * better in a distributed environment that has multiple independent roots.
502   * {@link #getWorkingDirectory()} should return what setWorkingDir() set.
503   * 
504   * @param newWDir new working directory
505   * @throws IOException 
506   * <br>
507   *           NewWdir can be one of:
508   *           <ul>
509   *           <li>relative path: "foo/bar";</li>
510   *           <li>absolute without scheme: "/foo/bar"</li>
511   *           <li>fully qualified with scheme: "xx://auth/foo/bar"</li>
512   *           </ul>
513   * <br>
514   *           Illegal WDs:
515   *           <ul>
516   *           <li>relative with scheme: "xx:foo/bar"</li>
517   *           <li>non existent directory</li>
518   *           </ul>
519   */
520  public void setWorkingDirectory(final Path newWDir) throws IOException {
521    newWDir.checkNotSchemeWithRelative();
522    /* wd is stored as a fully qualified path. We check if the given 
523     * path is not relative first since resolve requires and returns 
524     * an absolute path.
525     */  
526    final Path newWorkingDir = new Path(workingDir, newWDir);
527    FileStatus status = getFileStatus(newWorkingDir);
528    if (status.isFile()) {
529      throw new FileNotFoundException("Cannot setWD to a file");
530    }
531    workingDir = newWorkingDir;
532  }
533  
534  /**
535   * Gets the working directory for wd-relative names (such a "foo/bar").
536   */
537  public Path getWorkingDirectory() {
538    return workingDir;
539  }
540  
541  /**
542   * Gets the ugi in the file-context
543   * @return UserGroupInformation
544   */
545  public UserGroupInformation getUgi() {
546    return ugi;
547  }
548  
549  /**
550   * Return the current user's home directory in this file system.
551   * The default implementation returns "/user/$USER/".
552   * @return the home directory
553   */
554  public Path getHomeDirectory() {
555    return defaultFS.getHomeDirectory();
556  }
557  
558  /**
559   * 
560   * @return the umask of this FileContext
561   */
562  public FsPermission getUMask() {
563    return umask;
564  }
565  
566  /**
567   * Set umask to the supplied parameter.
568   * @param newUmask  the new umask
569   */
570  public void setUMask(final FsPermission newUmask) {
571    umask = newUmask;
572  }
573  
574  
575  /**
576   * Resolve the path following any symlinks or mount points
577   * @param f to be resolved
578   * @return fully qualified resolved path
579   * 
580   * @throws FileNotFoundException  If <code>f</code> does not exist
581   * @throws AccessControlException if access denied
582   * @throws IOException If an IO Error occurred
583   * 
584   * Exceptions applicable to file systems accessed over RPC:
585   * @throws RpcClientException If an exception occurred in the RPC client
586   * @throws RpcServerException If an exception occurred in the RPC server
587   * @throws UnexpectedServerException If server implementation throws
588   *           undeclared exception to RPC server
589   * 
590   * RuntimeExceptions:
591   * @throws InvalidPathException If path <code>f</code> is not valid
592   */
593  public Path resolvePath(final Path f) throws FileNotFoundException,
594      UnresolvedLinkException, AccessControlException, IOException {
595    return resolve(f);
596  }
597  
598  /**
599   * Make the path fully qualified if it is isn't. 
600   * A Fully-qualified path has scheme and authority specified and an absolute
601   * path.
602   * Use the default file system and working dir in this FileContext to qualify.
603   * @param path
604   * @return qualified path
605   */
606  public Path makeQualified(final Path path) {
607    return path.makeQualified(defaultFS.getUri(), getWorkingDirectory());
608  }
609
610  /**
611   * Create or overwrite file on indicated path and returns an output stream for
612   * writing into the file.
613   * 
614   * @param f the file name to open
615   * @param createFlag gives the semantics of create; see {@link CreateFlag}
616   * @param opts file creation options; see {@link Options.CreateOpts}.
617   *          <ul>
618   *          <li>Progress - to report progress on the operation - default null
619   *          <li>Permission - umask is applied against permisssion: default is
620   *          FsPermissions:getDefault()
621   * 
622   *          <li>CreateParent - create missing parent path; default is to not
623   *          to create parents
624   *          <li>The defaults for the following are SS defaults of the file
625   *          server implementing the target path. Not all parameters make sense
626   *          for all kinds of file system - eg. localFS ignores Blocksize,
627   *          replication, checksum
628   *          <ul>
629   *          <li>BufferSize - buffersize used in FSDataOutputStream
630   *          <li>Blocksize - block size for file blocks
631   *          <li>ReplicationFactor - replication for blocks
632   *          <li>ChecksumParam - Checksum parameters. server default is used
633   *          if not specified.
634   *          </ul>
635   *          </ul>
636   * 
637   * @return {@link FSDataOutputStream} for created file
638   * 
639   * @throws AccessControlException If access is denied
640   * @throws FileAlreadyExistsException If file <code>f</code> already exists
641   * @throws FileNotFoundException If parent of <code>f</code> does not exist
642   *           and <code>createParent</code> is false
643   * @throws ParentNotDirectoryException If parent of <code>f</code> is not a
644   *           directory.
645   * @throws UnsupportedFileSystemException If file system for <code>f</code> is
646   *           not supported
647   * @throws IOException If an I/O error occurred
648   * 
649   * Exceptions applicable to file systems accessed over RPC:
650   * @throws RpcClientException If an exception occurred in the RPC client
651   * @throws RpcServerException If an exception occurred in the RPC server
652   * @throws UnexpectedServerException If server implementation throws
653   *           undeclared exception to RPC server
654   * 
655   * RuntimeExceptions:
656   * @throws InvalidPathException If path <code>f</code> is not valid
657   */
658  public FSDataOutputStream create(final Path f,
659      final EnumSet<CreateFlag> createFlag, Options.CreateOpts... opts)
660      throws AccessControlException, FileAlreadyExistsException,
661      FileNotFoundException, ParentNotDirectoryException,
662      UnsupportedFileSystemException, IOException {
663    Path absF = fixRelativePart(f);
664
665    // If one of the options is a permission, extract it & apply umask
666    // If not, add a default Perms and apply umask;
667    // AbstractFileSystem#create
668
669    CreateOpts.Perms permOpt = CreateOpts.getOpt(CreateOpts.Perms.class, opts);
670    FsPermission permission = (permOpt != null) ? permOpt.getValue() :
671                                      FILE_DEFAULT_PERM;
672    permission = permission.applyUMask(umask);
673
674    final CreateOpts[] updatedOpts = 
675                      CreateOpts.setOpt(CreateOpts.perms(permission), opts);
676    return new FSLinkResolver<FSDataOutputStream>() {
677      @Override
678      public FSDataOutputStream next(final AbstractFileSystem fs, final Path p) 
679        throws IOException {
680        return fs.create(p, createFlag, updatedOpts);
681      }
682    }.resolve(this, absF);
683  }
684
685  /**
686   * Make(create) a directory and all the non-existent parents.
687   * 
688   * @param dir - the dir to make
689   * @param permission - permissions is set permission&~umask
690   * @param createParent - if true then missing parent dirs are created if false
691   *          then parent must exist
692   * 
693   * @throws AccessControlException If access is denied
694   * @throws FileAlreadyExistsException If directory <code>dir</code> already
695   *           exists
696   * @throws FileNotFoundException If parent of <code>dir</code> does not exist
697   *           and <code>createParent</code> is false
698   * @throws ParentNotDirectoryException If parent of <code>dir</code> is not a
699   *           directory
700   * @throws UnsupportedFileSystemException If file system for <code>dir</code>
701   *         is not supported
702   * @throws IOException If an I/O error occurred
703   * 
704   * Exceptions applicable to file systems accessed over RPC:
705   * @throws RpcClientException If an exception occurred in the RPC client
706   * @throws UnexpectedServerException If server implementation throws 
707   *           undeclared exception to RPC server
708   * 
709   * RuntimeExceptions:
710   * @throws InvalidPathException If path <code>dir</code> is not valid
711   */
712  public void mkdir(final Path dir, final FsPermission permission,
713      final boolean createParent) throws AccessControlException,
714      FileAlreadyExistsException, FileNotFoundException,
715      ParentNotDirectoryException, UnsupportedFileSystemException, 
716      IOException {
717    final Path absDir = fixRelativePart(dir);
718    final FsPermission absFerms = (permission == null ? 
719          FsPermission.getDirDefault() : permission).applyUMask(umask);
720    new FSLinkResolver<Void>() {
721      @Override
722      public Void next(final AbstractFileSystem fs, final Path p) 
723        throws IOException, UnresolvedLinkException {
724        fs.mkdir(p, absFerms, createParent);
725        return null;
726      }
727    }.resolve(this, absDir);
728  }
729
730  /**
731   * Delete a file.
732   * @param f the path to delete.
733   * @param recursive if path is a directory and set to 
734   * true, the directory is deleted else throws an exception. In
735   * case of a file the recursive can be set to either true or false.
736   *
737   * @throws AccessControlException If access is denied
738   * @throws FileNotFoundException If <code>f</code> does not exist
739   * @throws UnsupportedFileSystemException If file system for <code>f</code> is
740   *           not supported
741   * @throws IOException If an I/O error occurred
742   * 
743   * Exceptions applicable to file systems accessed over RPC:
744   * @throws RpcClientException If an exception occurred in the RPC client
745   * @throws RpcServerException If an exception occurred in the RPC server
746   * @throws UnexpectedServerException If server implementation throws 
747   *           undeclared exception to RPC server
748   * 
749   * RuntimeExceptions:
750   * @throws InvalidPathException If path <code>f</code> is invalid
751   */
752  public boolean delete(final Path f, final boolean recursive)
753      throws AccessControlException, FileNotFoundException,
754      UnsupportedFileSystemException, IOException {
755    Path absF = fixRelativePart(f);
756    return new FSLinkResolver<Boolean>() {
757      @Override
758      public Boolean next(final AbstractFileSystem fs, final Path p) 
759        throws IOException, UnresolvedLinkException {
760        return Boolean.valueOf(fs.delete(p, recursive));
761      }
762    }.resolve(this, absF);
763  }
764 
765  /**
766   * Opens an FSDataInputStream at the indicated Path using
767   * default buffersize.
768   * @param f the file name to open
769   *
770   * @throws AccessControlException If access is denied
771   * @throws FileNotFoundException If file <code>f</code> does not exist
772   * @throws UnsupportedFileSystemException If file system for <code>f</code>
773   *         is not supported
774   * @throws IOException If an I/O error occurred
775   * 
776   * Exceptions applicable to file systems accessed over RPC:
777   * @throws RpcClientException If an exception occurred in the RPC client
778   * @throws RpcServerException If an exception occurred in the RPC server
779   * @throws UnexpectedServerException If server implementation throws 
780   *           undeclared exception to RPC server
781   */
782  public FSDataInputStream open(final Path f) throws AccessControlException,
783      FileNotFoundException, UnsupportedFileSystemException, IOException {
784    final Path absF = fixRelativePart(f);
785    return new FSLinkResolver<FSDataInputStream>() {
786      @Override
787      public FSDataInputStream next(final AbstractFileSystem fs, final Path p) 
788        throws IOException, UnresolvedLinkException {
789        return fs.open(p);
790      }
791    }.resolve(this, absF);
792  }
793
794  /**
795   * Opens an FSDataInputStream at the indicated Path.
796   * 
797   * @param f the file name to open
798   * @param bufferSize the size of the buffer to be used.
799   * 
800   * @throws AccessControlException If access is denied
801   * @throws FileNotFoundException If file <code>f</code> does not exist
802   * @throws UnsupportedFileSystemException If file system for <code>f</code> is
803   *           not supported
804   * @throws IOException If an I/O error occurred
805   * 
806   * Exceptions applicable to file systems accessed over RPC:
807   * @throws RpcClientException If an exception occurred in the RPC client
808   * @throws RpcServerException If an exception occurred in the RPC server
809   * @throws UnexpectedServerException If server implementation throws 
810   *           undeclared exception to RPC server
811   */
812  public FSDataInputStream open(final Path f, final int bufferSize)
813      throws AccessControlException, FileNotFoundException,
814      UnsupportedFileSystemException, IOException {
815    final Path absF = fixRelativePart(f);
816    return new FSLinkResolver<FSDataInputStream>() {
817      @Override
818      public FSDataInputStream next(final AbstractFileSystem fs, final Path p) 
819        throws IOException, UnresolvedLinkException {
820        return fs.open(p, bufferSize);
821      }
822    }.resolve(this, absF);
823  }
824
825  /**
826   * Set replication for an existing file.
827   * 
828   * @param f file name
829   * @param replication new replication
830   *
831   * @return true if successful
832   *
833   * @throws AccessControlException If access is denied
834   * @throws FileNotFoundException If file <code>f</code> does not exist
835   * @throws IOException If an I/O error occurred
836   * 
837   * Exceptions applicable to file systems accessed over RPC:
838   * @throws RpcClientException If an exception occurred in the RPC client
839   * @throws RpcServerException If an exception occurred in the RPC server
840   * @throws UnexpectedServerException If server implementation throws 
841   *           undeclared exception to RPC server
842   */
843  public boolean setReplication(final Path f, final short replication)
844      throws AccessControlException, FileNotFoundException,
845      IOException {
846    final Path absF = fixRelativePart(f);
847    return new FSLinkResolver<Boolean>() {
848      @Override
849      public Boolean next(final AbstractFileSystem fs, final Path p) 
850        throws IOException, UnresolvedLinkException {
851        return Boolean.valueOf(fs.setReplication(p, replication));
852      }
853    }.resolve(this, absF);
854  }
855
856  /**
857   * Renames Path src to Path dst
858   * <ul>
859   * <li
860   * <li>Fails if src is a file and dst is a directory.
861   * <li>Fails if src is a directory and dst is a file.
862   * <li>Fails if the parent of dst does not exist or is a file.
863   * </ul>
864   * <p>
865   * If OVERWRITE option is not passed as an argument, rename fails if the dst
866   * already exists.
867   * <p>
868   * If OVERWRITE option is passed as an argument, rename overwrites the dst if
869   * it is a file or an empty directory. Rename fails if dst is a non-empty
870   * directory.
871   * <p>
872   * Note that atomicity of rename is dependent on the file system
873   * implementation. Please refer to the file system documentation for details
874   * <p>
875   * 
876   * @param src path to be renamed
877   * @param dst new path after rename
878   * 
879   * @throws AccessControlException If access is denied
880   * @throws FileAlreadyExistsException If <code>dst</code> already exists and
881   *           <code>options</options> has {@link Options.Rename#OVERWRITE} 
882   *           option false.
883   * @throws FileNotFoundException If <code>src</code> does not exist
884   * @throws ParentNotDirectoryException If parent of <code>dst</code> is not a
885   *           directory
886   * @throws UnsupportedFileSystemException If file system for <code>src</code>
887   *           and <code>dst</code> is not supported
888   * @throws IOException If an I/O error occurred
889   * 
890   * Exceptions applicable to file systems accessed over RPC:
891   * @throws RpcClientException If an exception occurred in the RPC client
892   * @throws RpcServerException If an exception occurred in the RPC server
893   * @throws UnexpectedServerException If server implementation throws
894   *           undeclared exception to RPC server
895   */
896  public void rename(final Path src, final Path dst,
897      final Options.Rename... options) throws AccessControlException,
898      FileAlreadyExistsException, FileNotFoundException,
899      ParentNotDirectoryException, UnsupportedFileSystemException,
900      IOException {
901    final Path absSrc = fixRelativePart(src);
902    final Path absDst = fixRelativePart(dst);
903    AbstractFileSystem srcFS = getFSofPath(absSrc);
904    AbstractFileSystem dstFS = getFSofPath(absDst);
905    if(!srcFS.getUri().equals(dstFS.getUri())) {
906      throw new IOException("Renames across AbstractFileSystems not supported");
907    }
908    try {
909      srcFS.rename(absSrc, absDst, options);
910    } catch (UnresolvedLinkException e) {
911      /* We do not know whether the source or the destination path
912       * was unresolved. Resolve the source path up until the final
913       * path component, then fully resolve the destination. 
914       */
915      final Path source = resolveIntermediate(absSrc);    
916      new FSLinkResolver<Void>() {
917        @Override
918        public Void next(final AbstractFileSystem fs, final Path p) 
919          throws IOException, UnresolvedLinkException {
920          fs.rename(source, p, options);
921          return null;
922        }
923      }.resolve(this, absDst);
924    }
925  }
926  
927  /**
928   * Set permission of a path.
929   * @param f
930   * @param permission - the new absolute permission (umask is not applied)
931   *
932   * @throws AccessControlException If access is denied
933   * @throws FileNotFoundException If <code>f</code> does not exist
934   * @throws UnsupportedFileSystemException If file system for <code>f</code>
935   *         is not supported
936   * @throws IOException If an I/O error occurred
937   * 
938   * Exceptions applicable to file systems accessed over RPC:
939   * @throws RpcClientException If an exception occurred in the RPC client
940   * @throws RpcServerException If an exception occurred in the RPC server
941   * @throws UnexpectedServerException If server implementation throws 
942   *           undeclared exception to RPC server
943   */
944  public void setPermission(final Path f, final FsPermission permission)
945      throws AccessControlException, FileNotFoundException,
946      UnsupportedFileSystemException, IOException {
947    final Path absF = fixRelativePart(f);
948    new FSLinkResolver<Void>() {
949      @Override
950      public Void next(final AbstractFileSystem fs, final Path p) 
951        throws IOException, UnresolvedLinkException {
952        fs.setPermission(p, permission);
953        return null;
954      }
955    }.resolve(this, absF);
956  }
957
958  /**
959   * Set owner of a path (i.e. a file or a directory). The parameters username
960   * and groupname cannot both be null.
961   * 
962   * @param f The path
963   * @param username If it is null, the original username remains unchanged.
964   * @param groupname If it is null, the original groupname remains unchanged.
965   * 
966   * @throws AccessControlException If access is denied
967   * @throws FileNotFoundException If <code>f</code> does not exist
968   * @throws UnsupportedFileSystemException If file system for <code>f</code> is
969   *           not supported
970   * @throws IOException If an I/O error occurred
971   * 
972   * Exceptions applicable to file systems accessed over RPC:
973   * @throws RpcClientException If an exception occurred in the RPC client
974   * @throws RpcServerException If an exception occurred in the RPC server
975   * @throws UnexpectedServerException If server implementation throws 
976   *           undeclared exception to RPC server
977   * 
978   * RuntimeExceptions:
979   * @throws HadoopIllegalArgumentException If <code>username</code> or
980   *           <code>groupname</code> is invalid.
981   */
982  public void setOwner(final Path f, final String username,
983      final String groupname) throws AccessControlException,
984      UnsupportedFileSystemException, FileNotFoundException,
985      IOException {
986    if ((username == null) && (groupname == null)) {
987      throw new HadoopIllegalArgumentException(
988          "username and groupname cannot both be null");
989    }
990    final Path absF = fixRelativePart(f);
991    new FSLinkResolver<Void>() {
992      @Override
993      public Void next(final AbstractFileSystem fs, final Path p) 
994        throws IOException, UnresolvedLinkException {
995        fs.setOwner(p, username, groupname);
996        return null;
997      }
998    }.resolve(this, absF);
999  }
1000
1001  /**
1002   * Set access time of a file.
1003   * @param f The path
1004   * @param mtime Set the modification time of this file.
1005   *        The number of milliseconds since epoch (Jan 1, 1970). 
1006   *        A value of -1 means that this call should not set modification time.
1007   * @param atime Set the access time of this file.
1008   *        The number of milliseconds since Jan 1, 1970. 
1009   *        A value of -1 means that this call should not set access time.
1010   *
1011   * @throws AccessControlException If access is denied
1012   * @throws FileNotFoundException If <code>f</code> does not exist
1013   * @throws UnsupportedFileSystemException If file system for <code>f</code> is
1014   *           not supported
1015   * @throws IOException If an I/O error occurred
1016   * 
1017   * Exceptions applicable to file systems accessed over RPC:
1018   * @throws RpcClientException If an exception occurred in the RPC client
1019   * @throws RpcServerException If an exception occurred in the RPC server
1020   * @throws UnexpectedServerException If server implementation throws 
1021   *           undeclared exception to RPC server
1022   */
1023  public void setTimes(final Path f, final long mtime, final long atime)
1024      throws AccessControlException, FileNotFoundException,
1025      UnsupportedFileSystemException, IOException {
1026    final Path absF = fixRelativePart(f);
1027    new FSLinkResolver<Void>() {
1028      @Override
1029      public Void next(final AbstractFileSystem fs, final Path p) 
1030        throws IOException, UnresolvedLinkException {
1031        fs.setTimes(p, mtime, atime);
1032        return null;
1033      }
1034    }.resolve(this, absF);
1035  }
1036
1037  /**
1038   * Get the checksum of a file.
1039   *
1040   * @param f file path
1041   *
1042   * @return The file checksum.  The default return value is null,
1043   *  which indicates that no checksum algorithm is implemented
1044   *  in the corresponding FileSystem.
1045   *
1046   * @throws AccessControlException If access is denied
1047   * @throws FileNotFoundException If <code>f</code> does not exist
1048   * @throws IOException If an I/O error occurred
1049   * 
1050   * Exceptions applicable to file systems accessed over RPC:
1051   * @throws RpcClientException If an exception occurred in the RPC client
1052   * @throws RpcServerException If an exception occurred in the RPC server
1053   * @throws UnexpectedServerException If server implementation throws 
1054   *           undeclared exception to RPC server
1055   */
1056  public FileChecksum getFileChecksum(final Path f)
1057      throws AccessControlException, FileNotFoundException,
1058      IOException {
1059    final Path absF = fixRelativePart(f);
1060    return new FSLinkResolver<FileChecksum>() {
1061      @Override
1062      public FileChecksum next(final AbstractFileSystem fs, final Path p) 
1063        throws IOException, UnresolvedLinkException {
1064        return fs.getFileChecksum(p);
1065      }
1066    }.resolve(this, absF);
1067  }
1068
1069  /**
1070   * Set the verify checksum flag for the  file system denoted by the path.
1071   * This is only applicable if the 
1072   * corresponding FileSystem supports checksum. By default doesn't do anything.
1073   * @param verifyChecksum
1074   * @param f set the verifyChecksum for the Filesystem containing this path
1075   *
1076   * @throws AccessControlException If access is denied
1077   * @throws FileNotFoundException If <code>f</code> does not exist
1078   * @throws UnsupportedFileSystemException If file system for <code>f</code> is
1079   *           not supported
1080   * @throws IOException If an I/O error occurred
1081   * 
1082   * Exceptions applicable to file systems accessed over RPC:
1083   * @throws RpcClientException If an exception occurred in the RPC client
1084   * @throws RpcServerException If an exception occurred in the RPC server
1085   * @throws UnexpectedServerException If server implementation throws 
1086   *           undeclared exception to RPC server
1087   */
1088  public void setVerifyChecksum(final boolean verifyChecksum, final Path f)
1089      throws AccessControlException, FileNotFoundException,
1090      UnsupportedFileSystemException, IOException {
1091    final Path absF = resolve(fixRelativePart(f));
1092    getFSofPath(absF).setVerifyChecksum(verifyChecksum);
1093  }
1094
1095  /**
1096   * Return a file status object that represents the path.
1097   * @param f The path we want information from
1098   *
1099   * @return a FileStatus object
1100   *
1101   * @throws AccessControlException If access is denied
1102   * @throws FileNotFoundException If <code>f</code> does not exist
1103   * @throws UnsupportedFileSystemException If file system for <code>f</code> is
1104   *           not supported
1105   * @throws IOException If an I/O error occurred
1106   * 
1107   * Exceptions applicable to file systems accessed over RPC:
1108   * @throws RpcClientException If an exception occurred in the RPC client
1109   * @throws RpcServerException If an exception occurred in the RPC server
1110   * @throws UnexpectedServerException If server implementation throws 
1111   *           undeclared exception to RPC server
1112   */
1113  public FileStatus getFileStatus(final Path f) throws AccessControlException,
1114      FileNotFoundException, UnsupportedFileSystemException, IOException {
1115    final Path absF = fixRelativePart(f);
1116    return new FSLinkResolver<FileStatus>() {
1117      @Override
1118      public FileStatus next(final AbstractFileSystem fs, final Path p) 
1119        throws IOException, UnresolvedLinkException {
1120        return fs.getFileStatus(p);
1121      }
1122    }.resolve(this, absF);
1123  }
1124
1125  /**
1126   * Checks if the user can access a path.  The mode specifies which access
1127   * checks to perform.  If the requested permissions are granted, then the
1128   * method returns normally.  If access is denied, then the method throws an
1129   * {@link AccessControlException}.
1130   * <p/>
1131   * The default implementation of this method calls {@link #getFileStatus(Path)}
1132   * and checks the returned permissions against the requested permissions.
1133   * Note that the getFileStatus call will be subject to authorization checks.
1134   * Typically, this requires search (execute) permissions on each directory in
1135   * the path's prefix, but this is implementation-defined.  Any file system
1136   * that provides a richer authorization model (such as ACLs) may override the
1137   * default implementation so that it checks against that model instead.
1138   * <p>
1139   * In general, applications should avoid using this method, due to the risk of
1140   * time-of-check/time-of-use race conditions.  The permissions on a file may
1141   * change immediately after the access call returns.  Most applications should
1142   * prefer running specific file system actions as the desired user represented
1143   * by a {@link UserGroupInformation}.
1144   *
1145   * @param path Path to check
1146   * @param mode type of access to check
1147   * @throws AccessControlException if access is denied
1148   * @throws FileNotFoundException if the path does not exist
1149   * @throws UnsupportedFileSystemException if file system for <code>path</code>
1150   *   is not supported
1151   * @throws IOException see specific implementation
1152   * 
1153   * Exceptions applicable to file systems accessed over RPC:
1154   * @throws RpcClientException If an exception occurred in the RPC client
1155   * @throws RpcServerException If an exception occurred in the RPC server
1156   * @throws UnexpectedServerException If server implementation throws 
1157   *           undeclared exception to RPC server
1158   */
1159  @InterfaceAudience.LimitedPrivate({"HDFS", "Hive"})
1160  public void access(final Path path, final FsAction mode)
1161      throws AccessControlException, FileNotFoundException,
1162      UnsupportedFileSystemException, IOException {
1163    final Path absPath = fixRelativePart(path);
1164    new FSLinkResolver<Void>() {
1165      @Override
1166      public Void next(AbstractFileSystem fs, Path p) throws IOException,
1167          UnresolvedLinkException {
1168        fs.access(p, mode);
1169        return null;
1170      }
1171    }.resolve(this, absPath);
1172  }
1173
1174  /**
1175   * Return a file status object that represents the path. If the path 
1176   * refers to a symlink then the FileStatus of the symlink is returned.
1177   * The behavior is equivalent to #getFileStatus() if the underlying
1178   * file system does not support symbolic links.
1179   * @param  f The path we want information from.
1180   * @return A FileStatus object
1181   * 
1182   * @throws AccessControlException If access is denied
1183   * @throws FileNotFoundException If <code>f</code> does not exist
1184   * @throws UnsupportedFileSystemException If file system for <code>f</code> is
1185   *           not supported
1186   * @throws IOException If an I/O error occurred
1187   */
1188  public FileStatus getFileLinkStatus(final Path f)
1189      throws AccessControlException, FileNotFoundException,
1190      UnsupportedFileSystemException, IOException {
1191    final Path absF = fixRelativePart(f);
1192    return new FSLinkResolver<FileStatus>() {
1193      @Override
1194      public FileStatus next(final AbstractFileSystem fs, final Path p) 
1195        throws IOException, UnresolvedLinkException {
1196        FileStatus fi = fs.getFileLinkStatus(p);
1197        if (fi.isSymlink()) {
1198          fi.setSymlink(FSLinkResolver.qualifySymlinkTarget(fs.getUri(), p,
1199              fi.getSymlink()));
1200        }
1201        return fi;
1202      }
1203    }.resolve(this, absF);
1204  }
1205  
1206  /**
1207   * Returns the target of the given symbolic link as it was specified
1208   * when the link was created.  Links in the path leading up to the
1209   * final path component are resolved transparently.
1210   *
1211   * @param f the path to return the target of
1212   * @return The un-interpreted target of the symbolic link.
1213   * 
1214   * @throws AccessControlException If access is denied
1215   * @throws FileNotFoundException If path <code>f</code> does not exist
1216   * @throws UnsupportedFileSystemException If file system for <code>f</code> is
1217   *           not supported
1218   * @throws IOException If the given path does not refer to a symlink
1219   *           or an I/O error occurred
1220   */
1221  public Path getLinkTarget(final Path f) throws AccessControlException,
1222      FileNotFoundException, UnsupportedFileSystemException, IOException {
1223    final Path absF = fixRelativePart(f);
1224    return new FSLinkResolver<Path>() {
1225      @Override
1226      public Path next(final AbstractFileSystem fs, final Path p) 
1227        throws IOException, UnresolvedLinkException {
1228        FileStatus fi = fs.getFileLinkStatus(p);
1229        return fi.getSymlink();
1230      }
1231    }.resolve(this, absF);
1232  }
1233  
1234  /**
1235   * Return blockLocation of the given file for the given offset and len.
1236   *  For a nonexistent file or regions, null will be returned.
1237   *
1238   * This call is most helpful with DFS, where it returns 
1239   * hostnames of machines that contain the given file.
1240   * 
1241   * @param f - get blocklocations of this file
1242   * @param start position (byte offset)
1243   * @param len (in bytes)
1244   *
1245   * @return block locations for given file at specified offset of len
1246   *
1247   * @throws AccessControlException If access is denied
1248   * @throws FileNotFoundException If <code>f</code> does not exist
1249   * @throws UnsupportedFileSystemException If file system for <code>f</code> is
1250   *           not supported
1251   * @throws IOException If an I/O error occurred
1252   * 
1253   * Exceptions applicable to file systems accessed over RPC:
1254   * @throws RpcClientException If an exception occurred in the RPC client
1255   * @throws RpcServerException If an exception occurred in the RPC server
1256   * @throws UnexpectedServerException If server implementation throws 
1257   *           undeclared exception to RPC server
1258   * 
1259   * RuntimeExceptions:
1260   * @throws InvalidPathException If path <code>f</code> is invalid
1261   */
1262  @InterfaceAudience.LimitedPrivate({"HDFS", "MapReduce"})
1263  @InterfaceStability.Evolving
1264  public BlockLocation[] getFileBlockLocations(final Path f, final long start,
1265      final long len) throws AccessControlException, FileNotFoundException,
1266      UnsupportedFileSystemException, IOException {
1267    final Path absF = fixRelativePart(f);
1268    return new FSLinkResolver<BlockLocation[]>() {
1269      @Override
1270      public BlockLocation[] next(final AbstractFileSystem fs, final Path p) 
1271        throws IOException, UnresolvedLinkException {
1272        return fs.getFileBlockLocations(p, start, len);
1273      }
1274    }.resolve(this, absF);
1275  }
1276  
1277  /**
1278   * Returns a status object describing the use and capacity of the
1279   * file system denoted by the Parh argument p.
1280   * If the file system has multiple partitions, the
1281   * use and capacity of the partition pointed to by the specified
1282   * path is reflected.
1283   * 
1284   * @param f Path for which status should be obtained. null means the
1285   * root partition of the default file system. 
1286   *
1287   * @return a FsStatus object
1288   *
1289   * @throws AccessControlException If access is denied
1290   * @throws FileNotFoundException If <code>f</code> does not exist
1291   * @throws UnsupportedFileSystemException If file system for <code>f</code> is
1292   *           not supported
1293   * @throws IOException If an I/O error occurred
1294   * 
1295   * Exceptions applicable to file systems accessed over RPC:
1296   * @throws RpcClientException If an exception occurred in the RPC client
1297   * @throws RpcServerException If an exception occurred in the RPC server
1298   * @throws UnexpectedServerException If server implementation throws 
1299   *           undeclared exception to RPC server
1300   */
1301  public FsStatus getFsStatus(final Path f) throws AccessControlException,
1302      FileNotFoundException, UnsupportedFileSystemException, IOException {
1303    if (f == null) {
1304      return defaultFS.getFsStatus();
1305    }
1306    final Path absF = fixRelativePart(f);
1307    return new FSLinkResolver<FsStatus>() {
1308      @Override
1309      public FsStatus next(final AbstractFileSystem fs, final Path p) 
1310        throws IOException, UnresolvedLinkException {
1311        return fs.getFsStatus(p);
1312      }
1313    }.resolve(this, absF);
1314  }
1315
1316  /**
1317   * Creates a symbolic link to an existing file. An exception is thrown if 
1318   * the symlink exits, the user does not have permission to create symlink,
1319   * or the underlying file system does not support symlinks.
1320   * 
1321   * Symlink permissions are ignored, access to a symlink is determined by
1322   * the permissions of the symlink target.
1323   * 
1324   * Symlinks in paths leading up to the final path component are resolved 
1325   * transparently. If the final path component refers to a symlink some 
1326   * functions operate on the symlink itself, these are:
1327   * - delete(f) and deleteOnExit(f) - Deletes the symlink.
1328   * - rename(src, dst) - If src refers to a symlink, the symlink is 
1329   *   renamed. If dst refers to a symlink, the symlink is over-written.
1330   * - getLinkTarget(f) - Returns the target of the symlink. 
1331   * - getFileLinkStatus(f) - Returns a FileStatus object describing
1332   *   the symlink.
1333   * Some functions, create() and mkdir(), expect the final path component
1334   * does not exist. If they are given a path that refers to a symlink that 
1335   * does exist they behave as if the path referred to an existing file or 
1336   * directory. All other functions fully resolve, ie follow, the symlink. 
1337   * These are: open, setReplication, setOwner, setTimes, setWorkingDirectory,
1338   * setPermission, getFileChecksum, setVerifyChecksum, getFileBlockLocations,
1339   * getFsStatus, getFileStatus, exists, and listStatus.
1340   * 
1341   * Symlink targets are stored as given to createSymlink, assuming the 
1342   * underlying file system is capable of storing a fully qualified URI.
1343   * Dangling symlinks are permitted. FileContext supports four types of 
1344   * symlink targets, and resolves them as follows
1345   * <pre>
1346   * Given a path referring to a symlink of form:
1347   * 
1348   *   <---X---> 
1349   *   fs://host/A/B/link 
1350   *   <-----Y----->
1351   * 
1352   * In this path X is the scheme and authority that identify the file system,
1353   * and Y is the path leading up to the final path component "link". If Y is
1354   * a symlink  itself then let Y' be the target of Y and X' be the scheme and
1355   * authority of Y'. Symlink targets may:
1356   * 
1357   * 1. Fully qualified URIs
1358   * 
1359   * fs://hostX/A/B/file  Resolved according to the target file system.
1360   * 
1361   * 2. Partially qualified URIs (eg scheme but no host)
1362   * 
1363   * fs:///A/B/file  Resolved according to the target file system. Eg resolving
1364   *                 a symlink to hdfs:///A results in an exception because
1365   *                 HDFS URIs must be fully qualified, while a symlink to 
1366   *                 file:///A will not since Hadoop's local file systems 
1367   *                 require partially qualified URIs.
1368   * 
1369   * 3. Relative paths
1370   * 
1371   * path  Resolves to [Y'][path]. Eg if Y resolves to hdfs://host/A and path 
1372   *       is "../B/file" then [Y'][path] is hdfs://host/B/file
1373   * 
1374   * 4. Absolute paths
1375   * 
1376   * path  Resolves to [X'][path]. Eg if Y resolves hdfs://host/A/B and path
1377   *       is "/file" then [X][path] is hdfs://host/file
1378   * </pre>
1379   * 
1380   * @param target the target of the symbolic link
1381   * @param link the path to be created that points to target
1382   * @param createParent if true then missing parent dirs are created if 
1383   *                     false then parent must exist
1384   *
1385   *
1386   * @throws AccessControlException If access is denied
1387   * @throws FileAlreadyExistsException If file <code>linkcode> already exists
1388   * @throws FileNotFoundException If <code>target</code> does not exist
1389   * @throws ParentNotDirectoryException If parent of <code>link</code> is not a
1390   *           directory.
1391   * @throws UnsupportedFileSystemException If file system for 
1392   *           <code>target</code> or <code>link</code> is not supported
1393   * @throws IOException If an I/O error occurred
1394   */
1395  @SuppressWarnings("deprecation")
1396  public void createSymlink(final Path target, final Path link,
1397      final boolean createParent) throws AccessControlException,
1398      FileAlreadyExistsException, FileNotFoundException,
1399      ParentNotDirectoryException, UnsupportedFileSystemException, 
1400      IOException { 
1401    if (!FileSystem.areSymlinksEnabled()) {
1402      throw new UnsupportedOperationException("Symlinks not supported");
1403    }
1404    final Path nonRelLink = fixRelativePart(link);
1405    new FSLinkResolver<Void>() {
1406      @Override
1407      public Void next(final AbstractFileSystem fs, final Path p) 
1408        throws IOException, UnresolvedLinkException {
1409        fs.createSymlink(target, p, createParent);
1410        return null;
1411      }
1412    }.resolve(this, nonRelLink);
1413  }
1414  
1415  /**
1416   * List the statuses of the files/directories in the given path if the path is
1417   * a directory.
1418   * 
1419   * @param f is the path
1420   *
1421   * @return an iterator that traverses statuses of the files/directories 
1422   *         in the given path
1423   *
1424   * @throws AccessControlException If access is denied
1425   * @throws FileNotFoundException If <code>f</code> does not exist
1426   * @throws UnsupportedFileSystemException If file system for <code>f</code> is
1427   *           not supported
1428   * @throws IOException If an I/O error occurred
1429   * 
1430   * Exceptions applicable to file systems accessed over RPC:
1431   * @throws RpcClientException If an exception occurred in the RPC client
1432   * @throws RpcServerException If an exception occurred in the RPC server
1433   * @throws UnexpectedServerException If server implementation throws 
1434   *           undeclared exception to RPC server
1435   */
1436  public RemoteIterator<FileStatus> listStatus(final Path f) throws
1437      AccessControlException, FileNotFoundException,
1438      UnsupportedFileSystemException, IOException {
1439    final Path absF = fixRelativePart(f);
1440    return new FSLinkResolver<RemoteIterator<FileStatus>>() {
1441      @Override
1442      public RemoteIterator<FileStatus> next(
1443          final AbstractFileSystem fs, final Path p) 
1444        throws IOException, UnresolvedLinkException {
1445        return fs.listStatusIterator(p);
1446      }
1447    }.resolve(this, absF);
1448  }
1449
1450  /**
1451   * @return an iterator over the corrupt files under the given path
1452   * (may contain duplicates if a file has more than one corrupt block)
1453   * @throws IOException
1454   */
1455  public RemoteIterator<Path> listCorruptFileBlocks(Path path)
1456    throws IOException {
1457    final Path absF = fixRelativePart(path);
1458    return new FSLinkResolver<RemoteIterator<Path>>() {
1459      @Override
1460      public RemoteIterator<Path> next(final AbstractFileSystem fs,
1461                                       final Path p) 
1462        throws IOException, UnresolvedLinkException {
1463        return fs.listCorruptFileBlocks(p);
1464      }
1465    }.resolve(this, absF);
1466  }
1467  
1468  /**
1469   * List the statuses of the files/directories in the given path if the path is
1470   * a directory. 
1471   * Return the file's status and block locations If the path is a file.
1472   * 
1473   * If a returned status is a file, it contains the file's block locations.
1474   * 
1475   * @param f is the path
1476   *
1477   * @return an iterator that traverses statuses of the files/directories 
1478   *         in the given path
1479   * If any IO exception (for example the input directory gets deleted while
1480   * listing is being executed), next() or hasNext() of the returned iterator
1481   * may throw a RuntimeException with the io exception as the cause.
1482   *
1483   * @throws AccessControlException If access is denied
1484   * @throws FileNotFoundException If <code>f</code> does not exist
1485   * @throws UnsupportedFileSystemException If file system for <code>f</code> is
1486   *           not supported
1487   * @throws IOException If an I/O error occurred
1488   * 
1489   * Exceptions applicable to file systems accessed over RPC:
1490   * @throws RpcClientException If an exception occurred in the RPC client
1491   * @throws RpcServerException If an exception occurred in the RPC server
1492   * @throws UnexpectedServerException If server implementation throws 
1493   *           undeclared exception to RPC server
1494   */
1495  public RemoteIterator<LocatedFileStatus> listLocatedStatus(
1496      final Path f) throws
1497      AccessControlException, FileNotFoundException,
1498      UnsupportedFileSystemException, IOException {
1499    final Path absF = fixRelativePart(f);
1500    return new FSLinkResolver<RemoteIterator<LocatedFileStatus>>() {
1501      @Override
1502      public RemoteIterator<LocatedFileStatus> next(
1503          final AbstractFileSystem fs, final Path p) 
1504        throws IOException, UnresolvedLinkException {
1505        return fs.listLocatedStatus(p);
1506      }
1507    }.resolve(this, absF);
1508  }
1509
1510  /**
1511   * Mark a path to be deleted on JVM shutdown.
1512   * 
1513   * @param f the existing path to delete.
1514   *
1515   * @return  true if deleteOnExit is successful, otherwise false.
1516   *
1517   * @throws AccessControlException If access is denied
1518   * @throws UnsupportedFileSystemException If file system for <code>f</code> is
1519   *           not supported
1520   * @throws IOException If an I/O error occurred
1521   * 
1522   * Exceptions applicable to file systems accessed over RPC:
1523   * @throws RpcClientException If an exception occurred in the RPC client
1524   * @throws RpcServerException If an exception occurred in the RPC server
1525   * @throws UnexpectedServerException If server implementation throws 
1526   *           undeclared exception to RPC server
1527   */
1528  public boolean deleteOnExit(Path f) throws AccessControlException,
1529      IOException {
1530    if (!this.util().exists(f)) {
1531      return false;
1532    }
1533    synchronized (DELETE_ON_EXIT) {
1534      if (DELETE_ON_EXIT.isEmpty()) {
1535        ShutdownHookManager.get().addShutdownHook(FINALIZER, SHUTDOWN_HOOK_PRIORITY);
1536      }
1537      
1538      Set<Path> set = DELETE_ON_EXIT.get(this);
1539      if (set == null) {
1540        set = new TreeSet<Path>();
1541        DELETE_ON_EXIT.put(this, set);
1542      }
1543      set.add(f);
1544    }
1545    return true;
1546  }
1547  
1548  private final Util util;
1549  public Util util() {
1550    return util;
1551  }
1552  
1553  
1554  /**
1555   * Utility/library methods built over the basic FileContext methods.
1556   * Since this are library functions, the oprtation are not atomic
1557   * and some of them may partially complete if other threads are making
1558   * changes to the same part of the name space.
1559   */
1560  public class Util {
1561    /**
1562     * Does the file exist?
1563     * Note: Avoid using this method if you already have FileStatus in hand.
1564     * Instead reuse the FileStatus 
1565     * @param f the  file or dir to be checked
1566     *
1567     * @throws AccessControlException If access is denied
1568     * @throws IOException If an I/O error occurred
1569     * @throws UnsupportedFileSystemException If file system for <code>f</code> is
1570     *           not supported
1571     * 
1572     * Exceptions applicable to file systems accessed over RPC:
1573     * @throws RpcClientException If an exception occurred in the RPC client
1574     * @throws RpcServerException If an exception occurred in the RPC server
1575     * @throws UnexpectedServerException If server implementation throws 
1576     *           undeclared exception to RPC server
1577     */
1578    public boolean exists(final Path f) throws AccessControlException,
1579      UnsupportedFileSystemException, IOException {
1580      try {
1581        FileStatus fs = FileContext.this.getFileStatus(f);
1582        assert fs != null;
1583        return true;
1584      } catch (FileNotFoundException e) {
1585        return false;
1586      }
1587    }
1588    
1589    /**
1590     * Return the {@link ContentSummary} of path f.
1591     * @param f path
1592     *
1593     * @return the {@link ContentSummary} of path f.
1594     *
1595     * @throws AccessControlException If access is denied
1596     * @throws FileNotFoundException If <code>f</code> does not exist
1597     * @throws UnsupportedFileSystemException If file system for 
1598     *         <code>f</code> is not supported
1599     * @throws IOException If an I/O error occurred
1600     * 
1601     * Exceptions applicable to file systems accessed over RPC:
1602     * @throws RpcClientException If an exception occurred in the RPC client
1603     * @throws RpcServerException If an exception occurred in the RPC server
1604     * @throws UnexpectedServerException If server implementation throws 
1605     *           undeclared exception to RPC server
1606     */
1607    public ContentSummary getContentSummary(Path f)
1608        throws AccessControlException, FileNotFoundException,
1609        UnsupportedFileSystemException, IOException {
1610      FileStatus status = FileContext.this.getFileStatus(f);
1611      if (status.isFile()) {
1612        return new ContentSummary(status.getLen(), 1, 0);
1613      }
1614      long[] summary = {0, 0, 1};
1615      RemoteIterator<FileStatus> statusIterator = 
1616        FileContext.this.listStatus(f);
1617      while(statusIterator.hasNext()) {
1618        FileStatus s = statusIterator.next();
1619        ContentSummary c = s.isDirectory() ? getContentSummary(s.getPath()) :
1620                                       new ContentSummary(s.getLen(), 1, 0);
1621        summary[0] += c.getLength();
1622        summary[1] += c.getFileCount();
1623        summary[2] += c.getDirectoryCount();
1624      }
1625      return new ContentSummary(summary[0], summary[1], summary[2]);
1626    }
1627    
1628    /**
1629     * See {@link #listStatus(Path[], PathFilter)}
1630     */
1631    public FileStatus[] listStatus(Path[] files) throws AccessControlException,
1632        FileNotFoundException, IOException {
1633      return listStatus(files, DEFAULT_FILTER);
1634    }
1635     
1636    /**
1637     * Filter files/directories in the given path using the user-supplied path
1638     * filter.
1639     * 
1640     * @param f is the path name
1641     * @param filter is the user-supplied path filter
1642     *
1643     * @return an array of FileStatus objects for the files under the given path
1644     *         after applying the filter
1645     *
1646     * @throws AccessControlException If access is denied
1647     * @throws FileNotFoundException If <code>f</code> does not exist
1648     * @throws UnsupportedFileSystemException If file system for 
1649     *         <code>pathPattern</code> is not supported
1650     * @throws IOException If an I/O error occurred
1651     * 
1652     * Exceptions applicable to file systems accessed over RPC:
1653     * @throws RpcClientException If an exception occurred in the RPC client
1654     * @throws RpcServerException If an exception occurred in the RPC server
1655     * @throws UnexpectedServerException If server implementation throws 
1656     *           undeclared exception to RPC server
1657     */
1658    public FileStatus[] listStatus(Path f, PathFilter filter)
1659        throws AccessControlException, FileNotFoundException,
1660        UnsupportedFileSystemException, IOException {
1661      ArrayList<FileStatus> results = new ArrayList<FileStatus>();
1662      listStatus(results, f, filter);
1663      return results.toArray(new FileStatus[results.size()]);
1664    }
1665    
1666    /**
1667     * Filter files/directories in the given list of paths using user-supplied
1668     * path filter.
1669     * 
1670     * @param files is a list of paths
1671     * @param filter is the filter
1672     *
1673     * @return a list of statuses for the files under the given paths after
1674     *         applying the filter
1675     *
1676     * @throws AccessControlException If access is denied
1677     * @throws FileNotFoundException If a file in <code>files</code> does not 
1678     *           exist
1679     * @throws IOException If an I/O error occurred
1680     * 
1681     * Exceptions applicable to file systems accessed over RPC:
1682     * @throws RpcClientException If an exception occurred in the RPC client
1683     * @throws RpcServerException If an exception occurred in the RPC server
1684     * @throws UnexpectedServerException If server implementation throws 
1685     *           undeclared exception to RPC server
1686     */
1687    public FileStatus[] listStatus(Path[] files, PathFilter filter)
1688        throws AccessControlException, FileNotFoundException, IOException {
1689      ArrayList<FileStatus> results = new ArrayList<FileStatus>();
1690      for (int i = 0; i < files.length; i++) {
1691        listStatus(results, files[i], filter);
1692      }
1693      return results.toArray(new FileStatus[results.size()]);
1694    }
1695  
1696    /*
1697     * Filter files/directories in the given path using the user-supplied path
1698     * filter. Results are added to the given array <code>results</code>.
1699     */
1700    private void listStatus(ArrayList<FileStatus> results, Path f,
1701        PathFilter filter) throws AccessControlException,
1702        FileNotFoundException, IOException {
1703      FileStatus[] listing = listStatus(f);
1704      if (listing != null) {
1705        for (int i = 0; i < listing.length; i++) {
1706          if (filter.accept(listing[i].getPath())) {
1707            results.add(listing[i]);
1708          }
1709        }
1710      }
1711    }
1712
1713    /**
1714     * List the statuses of the files/directories in the given path 
1715     * if the path is a directory.
1716     * 
1717     * @param f is the path
1718     *
1719     * @return an array that contains statuses of the files/directories 
1720     *         in the given path
1721     *
1722     * @throws AccessControlException If access is denied
1723     * @throws FileNotFoundException If <code>f</code> does not exist
1724     * @throws UnsupportedFileSystemException If file system for <code>f</code> is
1725     *           not supported
1726     * @throws IOException If an I/O error occurred
1727     * 
1728     * Exceptions applicable to file systems accessed over RPC:
1729     * @throws RpcClientException If an exception occurred in the RPC client
1730     * @throws RpcServerException If an exception occurred in the RPC server
1731     * @throws UnexpectedServerException If server implementation throws 
1732     *           undeclared exception to RPC server
1733     */
1734    public FileStatus[] listStatus(final Path f) throws AccessControlException,
1735        FileNotFoundException, UnsupportedFileSystemException,
1736        IOException {
1737      final Path absF = fixRelativePart(f);
1738      return new FSLinkResolver<FileStatus[]>() {
1739        @Override
1740        public FileStatus[] next(final AbstractFileSystem fs, final Path p) 
1741          throws IOException, UnresolvedLinkException {
1742          return fs.listStatus(p);
1743        }
1744      }.resolve(FileContext.this, absF);
1745    }
1746
1747    /**
1748     * List the statuses and block locations of the files in the given path.
1749     * 
1750     * If the path is a directory, 
1751     *   if recursive is false, returns files in the directory;
1752     *   if recursive is true, return files in the subtree rooted at the path.
1753     *   The subtree is traversed in the depth-first order.
1754     * If the path is a file, return the file's status and block locations.
1755     * Files across symbolic links are also returned.
1756     * 
1757     * @param f is the path
1758     * @param recursive if the subdirectories need to be traversed recursively
1759     *
1760     * @return an iterator that traverses statuses of the files
1761     * If any IO exception (for example a sub-directory gets deleted while
1762     * listing is being executed), next() or hasNext() of the returned iterator
1763     * may throw a RuntimeException with the IO exception as the cause.
1764     *
1765     * @throws AccessControlException If access is denied
1766     * @throws FileNotFoundException If <code>f</code> does not exist
1767     * @throws UnsupportedFileSystemException If file system for <code>f</code>
1768     *         is not supported
1769     * @throws IOException If an I/O error occurred
1770     * 
1771     * Exceptions applicable to file systems accessed over RPC:
1772     * @throws RpcClientException If an exception occurred in the RPC client
1773     * @throws RpcServerException If an exception occurred in the RPC server
1774     * @throws UnexpectedServerException If server implementation throws 
1775     *           undeclared exception to RPC server
1776     */
1777    public RemoteIterator<LocatedFileStatus> listFiles(
1778        final Path f, final boolean recursive) throws AccessControlException,
1779        FileNotFoundException, UnsupportedFileSystemException, 
1780        IOException {
1781      return new RemoteIterator<LocatedFileStatus>() {
1782        private Stack<RemoteIterator<LocatedFileStatus>> itors = 
1783          new Stack<RemoteIterator<LocatedFileStatus>>();
1784        RemoteIterator<LocatedFileStatus> curItor = listLocatedStatus(f);
1785        LocatedFileStatus curFile;
1786
1787        /**
1788         * Returns <tt>true</tt> if the iterator has more files.
1789         *
1790         * @return <tt>true</tt> if the iterator has more files.
1791         * @throws AccessControlException if not allowed to access next
1792         *                                file's status or locations
1793         * @throws FileNotFoundException if next file does not exist any more
1794         * @throws UnsupportedFileSystemException if next file's 
1795         *                                        fs is unsupported
1796         * @throws IOException for all other IO errors
1797         *                     for example, NameNode is not avaialbe or
1798         *                     NameNode throws IOException due to an error
1799         *                     while getting the status or block locations
1800         */
1801        @Override
1802        public boolean hasNext() throws IOException {
1803          while (curFile == null) {
1804            if (curItor.hasNext()) {
1805              handleFileStat(curItor.next());
1806            } else if (!itors.empty()) {
1807              curItor = itors.pop();
1808            } else {
1809              return false;
1810            }
1811          }
1812          return true;
1813        }
1814
1815        /**
1816         * Process the input stat.
1817         * If it is a file, return the file stat.
1818         * If it is a directory, traverse the directory if recursive is true;
1819         * ignore it if recursive is false.
1820         * If it is a symlink, resolve the symlink first and then process it
1821         * depending on if it is a file or directory.
1822         * @param stat input status
1823         * @throws AccessControlException if access is denied
1824         * @throws FileNotFoundException if file is not found
1825         * @throws UnsupportedFileSystemException if fs is not supported
1826         * @throws IOException for all other IO errors
1827         */
1828        private void handleFileStat(LocatedFileStatus stat)
1829        throws IOException {
1830          if (stat.isFile()) { // file
1831            curFile = stat;
1832          } else if (stat.isSymlink()) { // symbolic link
1833            // resolve symbolic link
1834            FileStatus symstat = FileContext.this.getFileStatus(
1835                stat.getSymlink());
1836            if (symstat.isFile() || (recursive && symstat.isDirectory())) {
1837              itors.push(curItor);
1838              curItor = listLocatedStatus(stat.getPath());
1839            }
1840          } else if (recursive) { // directory
1841            itors.push(curItor);
1842            curItor = listLocatedStatus(stat.getPath());
1843          }
1844        }
1845
1846        /**
1847         * Returns the next file's status with its block locations
1848         *
1849         * @throws AccessControlException if not allowed to access next
1850         *                                file's status or locations
1851         * @throws FileNotFoundException if next file does not exist any more
1852         * @throws UnsupportedFileSystemException if next file's 
1853         *                                        fs is unsupported
1854         * @throws IOException for all other IO errors
1855         *                     for example, NameNode is not avaialbe or
1856         *                     NameNode throws IOException due to an error
1857         *                     while getting the status or block locations
1858         */
1859        @Override
1860        public LocatedFileStatus next() throws IOException {
1861          if (hasNext()) {
1862            LocatedFileStatus result = curFile;
1863            curFile = null;
1864            return result;
1865          } 
1866          throw new java.util.NoSuchElementException("No more entry in " + f);
1867        }
1868      };
1869    }
1870
1871    /**
1872     * <p>Return all the files that match filePattern and are not checksum
1873     * files. Results are sorted by their names.
1874     * 
1875     * <p>
1876     * A filename pattern is composed of <i>regular</i> characters and
1877     * <i>special pattern matching</i> characters, which are:
1878     *
1879     * <dl>
1880     *  <dd>
1881     *   <dl>
1882     *    <p>
1883     *    <dt> <tt> ? </tt>
1884     *    <dd> Matches any single character.
1885     *
1886     *    <p>
1887     *    <dt> <tt> * </tt>
1888     *    <dd> Matches zero or more characters.
1889     *
1890     *    <p>
1891     *    <dt> <tt> [<i>abc</i>] </tt>
1892     *    <dd> Matches a single character from character set
1893     *     <tt>{<i>a,b,c</i>}</tt>.
1894     *
1895     *    <p>
1896     *    <dt> <tt> [<i>a</i>-<i>b</i>] </tt>
1897     *    <dd> Matches a single character from the character range
1898     *     <tt>{<i>a...b</i>}</tt>. Note: character <tt><i>a</i></tt> must be
1899     *     lexicographically less than or equal to character <tt><i>b</i></tt>.
1900     *
1901     *    <p>
1902     *    <dt> <tt> [^<i>a</i>] </tt>
1903     *    <dd> Matches a single char that is not from character set or range
1904     *     <tt>{<i>a</i>}</tt>.  Note that the <tt>^</tt> character must occur
1905     *     immediately to the right of the opening bracket.
1906     *
1907     *    <p>
1908     *    <dt> <tt> \<i>c</i> </tt>
1909     *    <dd> Removes (escapes) any special meaning of character <i>c</i>.
1910     *
1911     *    <p>
1912     *    <dt> <tt> {ab,cd} </tt>
1913     *    <dd> Matches a string from the string set <tt>{<i>ab, cd</i>} </tt>
1914     *    
1915     *    <p>
1916     *    <dt> <tt> {ab,c{de,fh}} </tt>
1917     *    <dd> Matches a string from string set <tt>{<i>ab, cde, cfh</i>}</tt>
1918     *
1919     *   </dl>
1920     *  </dd>
1921     * </dl>
1922     *
1923     * @param pathPattern a regular expression specifying a pth pattern
1924     *
1925     * @return an array of paths that match the path pattern
1926     *
1927     * @throws AccessControlException If access is denied
1928     * @throws UnsupportedFileSystemException If file system for 
1929     *         <code>pathPattern</code> is not supported
1930     * @throws IOException If an I/O error occurred
1931     * 
1932     * Exceptions applicable to file systems accessed over RPC:
1933     * @throws RpcClientException If an exception occurred in the RPC client
1934     * @throws RpcServerException If an exception occurred in the RPC server
1935     * @throws UnexpectedServerException If server implementation throws 
1936     *           undeclared exception to RPC server
1937     */
1938    public FileStatus[] globStatus(Path pathPattern)
1939        throws AccessControlException, UnsupportedFileSystemException,
1940        IOException {
1941      return new Globber(FileContext.this, pathPattern, DEFAULT_FILTER).glob();
1942    }
1943    
1944    /**
1945     * Return an array of FileStatus objects whose path names match pathPattern
1946     * and is accepted by the user-supplied path filter. Results are sorted by
1947     * their path names.
1948     * Return null if pathPattern has no glob and the path does not exist.
1949     * Return an empty array if pathPattern has a glob and no path matches it. 
1950     * 
1951     * @param pathPattern regular expression specifying the path pattern
1952     * @param filter user-supplied path filter
1953     *
1954     * @return an array of FileStatus objects
1955     *
1956     * @throws AccessControlException If access is denied
1957     * @throws UnsupportedFileSystemException If file system for 
1958     *         <code>pathPattern</code> is not supported
1959     * @throws IOException If an I/O error occurred
1960     * 
1961     * Exceptions applicable to file systems accessed over RPC:
1962     * @throws RpcClientException If an exception occurred in the RPC client
1963     * @throws RpcServerException If an exception occurred in the RPC server
1964     * @throws UnexpectedServerException If server implementation throws 
1965     *           undeclared exception to RPC server
1966     */
1967    public FileStatus[] globStatus(final Path pathPattern,
1968        final PathFilter filter) throws AccessControlException,
1969        UnsupportedFileSystemException, IOException {
1970      return new Globber(FileContext.this, pathPattern, filter).glob();
1971    }
1972
1973    /**
1974     * Copy file from src to dest. See
1975     * {@link #copy(Path, Path, boolean, boolean)}
1976     */
1977    public boolean copy(final Path src, final Path dst)
1978        throws AccessControlException, FileAlreadyExistsException,
1979        FileNotFoundException, ParentNotDirectoryException,
1980        UnsupportedFileSystemException, IOException {
1981      return copy(src, dst, false, false);
1982    }
1983    
1984    /**
1985     * Copy from src to dst, optionally deleting src and overwriting dst.
1986     * @param src
1987     * @param dst
1988     * @param deleteSource - delete src if true
1989     * @param overwrite  overwrite dst if true; throw IOException if dst exists
1990     *         and overwrite is false.
1991     *
1992     * @return true if copy is successful
1993     *
1994     * @throws AccessControlException If access is denied
1995     * @throws FileAlreadyExistsException If <code>dst</code> already exists
1996     * @throws FileNotFoundException If <code>src</code> does not exist
1997     * @throws ParentNotDirectoryException If parent of <code>dst</code> is not
1998     *           a directory
1999     * @throws UnsupportedFileSystemException If file system for 
2000     *         <code>src</code> or <code>dst</code> is not supported
2001     * @throws IOException If an I/O error occurred
2002     * 
2003     * Exceptions applicable to file systems accessed over RPC:
2004     * @throws RpcClientException If an exception occurred in the RPC client
2005     * @throws RpcServerException If an exception occurred in the RPC server
2006     * @throws UnexpectedServerException If server implementation throws 
2007     *           undeclared exception to RPC server
2008     * 
2009     * RuntimeExceptions:
2010     * @throws InvalidPathException If path <code>dst</code> is invalid
2011     */
2012    public boolean copy(final Path src, final Path dst, boolean deleteSource,
2013        boolean overwrite) throws AccessControlException,
2014        FileAlreadyExistsException, FileNotFoundException,
2015        ParentNotDirectoryException, UnsupportedFileSystemException, 
2016        IOException {
2017      src.checkNotSchemeWithRelative();
2018      dst.checkNotSchemeWithRelative();
2019      Path qSrc = makeQualified(src);
2020      Path qDst = makeQualified(dst);
2021      checkDest(qSrc.getName(), qDst, overwrite);
2022      FileStatus fs = FileContext.this.getFileStatus(qSrc);
2023      if (fs.isDirectory()) {
2024        checkDependencies(qSrc, qDst);
2025        mkdir(qDst, FsPermission.getDirDefault(), true);
2026        FileStatus[] contents = listStatus(qSrc);
2027        for (FileStatus content : contents) {
2028          copy(makeQualified(content.getPath()), makeQualified(new Path(qDst,
2029              content.getPath().getName())), deleteSource, overwrite);
2030        }
2031      } else {
2032        InputStream in=null;
2033        OutputStream out = null;
2034        try {
2035          in = open(qSrc);
2036          EnumSet<CreateFlag> createFlag = overwrite ? EnumSet.of(
2037              CreateFlag.CREATE, CreateFlag.OVERWRITE) : 
2038                EnumSet.of(CreateFlag.CREATE);
2039          out = create(qDst, createFlag);
2040          IOUtils.copyBytes(in, out, conf, true);
2041        } finally {
2042          IOUtils.closeStream(out);
2043          IOUtils.closeStream(in);
2044        }
2045      }
2046      if (deleteSource) {
2047        return delete(qSrc, true);
2048      } else {
2049        return true;
2050      }
2051    }
2052  }
2053
2054  /**
2055   * Check if copying srcName to dst would overwrite an existing 
2056   * file or directory.
2057   * @param srcName File or directory to be copied.
2058   * @param dst Destination to copy srcName to.
2059   * @param overwrite Whether it's ok to overwrite an existing file. 
2060   * @throws AccessControlException If access is denied.
2061   * @throws IOException If dst is an existing directory, or dst is an 
2062   * existing file and the overwrite option is not passed.
2063   */
2064  private void checkDest(String srcName, Path dst, boolean overwrite)
2065      throws AccessControlException, IOException {
2066    try {
2067      FileStatus dstFs = getFileStatus(dst);
2068      if (dstFs.isDirectory()) {
2069        if (null == srcName) {
2070          throw new IOException("Target " + dst + " is a directory");
2071        }
2072        // Recurse to check if dst/srcName exists.
2073        checkDest(null, new Path(dst, srcName), overwrite);
2074      } else if (!overwrite) {
2075        throw new IOException("Target " + new Path(dst, srcName)
2076            + " already exists");
2077      }
2078    } catch (FileNotFoundException e) {
2079      // dst does not exist - OK to copy.
2080    }
2081  }
2082   
2083  //
2084  // If the destination is a subdirectory of the source, then
2085  // generate exception
2086  //
2087  private static void checkDependencies(Path qualSrc, Path qualDst)
2088    throws IOException {
2089    if (isSameFS(qualSrc, qualDst)) {
2090      String srcq = qualSrc.toString() + Path.SEPARATOR;
2091      String dstq = qualDst.toString() + Path.SEPARATOR;
2092      if (dstq.startsWith(srcq)) {
2093        if (srcq.length() == dstq.length()) {
2094          throw new IOException("Cannot copy " + qualSrc + " to itself.");
2095        } else {
2096          throw new IOException("Cannot copy " + qualSrc +
2097                             " to its subdirectory " + qualDst);
2098        }
2099      }
2100    }
2101  }
2102  
2103  /**
2104   * Are qualSrc and qualDst of the same file system?
2105   * @param qualPath1 - fully qualified path
2106   * @param qualPath2 - fully qualified path
2107   * @return
2108   */
2109  private static boolean isSameFS(Path qualPath1, Path qualPath2) {
2110    URI srcUri = qualPath1.toUri();
2111    URI dstUri = qualPath2.toUri();
2112    return (srcUri.getScheme().equals(dstUri.getScheme()) && 
2113        !(srcUri.getAuthority() != null && dstUri.getAuthority() != null && srcUri
2114        .getAuthority().equals(dstUri.getAuthority())));
2115  }
2116
2117  /**
2118   * Deletes all the paths in deleteOnExit on JVM shutdown.
2119   */
2120  static class FileContextFinalizer implements Runnable {
2121    @Override
2122    public synchronized void run() {
2123      processDeleteOnExit();
2124    }
2125  }
2126
2127  /**
2128   * Resolves all symbolic links in the specified path.
2129   * Returns the new path object.
2130   */
2131  protected Path resolve(final Path f) throws FileNotFoundException,
2132      UnresolvedLinkException, AccessControlException, IOException {
2133    return new FSLinkResolver<Path>() {
2134      @Override
2135      public Path next(final AbstractFileSystem fs, final Path p) 
2136        throws IOException, UnresolvedLinkException {
2137        return fs.resolvePath(p);
2138      }
2139    }.resolve(this, f);
2140  }
2141
2142  /**
2143   * Resolves all symbolic links in the specified path leading up 
2144   * to, but not including the final path component.
2145   * @param f path to resolve
2146   * @return the new path object.
2147   */
2148  protected Path resolveIntermediate(final Path f) throws IOException {
2149    return new FSLinkResolver<FileStatus>() {
2150      @Override
2151      public FileStatus next(final AbstractFileSystem fs, final Path p) 
2152        throws IOException, UnresolvedLinkException {
2153        return fs.getFileLinkStatus(p);
2154      }
2155    }.resolve(this, f).getPath();
2156  }
2157
2158  /**
2159   * Returns the list of AbstractFileSystems accessed in the path. The list may
2160   * contain more than one AbstractFileSystems objects in case of symlinks.
2161   * 
2162   * @param f
2163   *          Path which needs to be resolved
2164   * @return List of AbstractFileSystems accessed in the path
2165   * @throws IOException
2166   */
2167  Set<AbstractFileSystem> resolveAbstractFileSystems(final Path f)
2168      throws IOException {
2169    final Path absF = fixRelativePart(f);
2170    final HashSet<AbstractFileSystem> result 
2171      = new HashSet<AbstractFileSystem>();
2172    new FSLinkResolver<Void>() {
2173      @Override
2174      public Void next(final AbstractFileSystem fs, final Path p)
2175          throws IOException, UnresolvedLinkException {
2176        result.add(fs);
2177        fs.getFileStatus(p);
2178        return null;
2179      }
2180    }.resolve(this, absF);
2181    return result;
2182  }
2183
2184  /**
2185   * Get the statistics for a particular file system
2186   * 
2187   * @param uri
2188   *          the uri to lookup the statistics. Only scheme and authority part
2189   *          of the uri are used as the key to store and lookup.
2190   * @return a statistics object
2191   */
2192  public static Statistics getStatistics(URI uri) {
2193    return AbstractFileSystem.getStatistics(uri);
2194  }
2195
2196  /**
2197   * Clears all the statistics stored in AbstractFileSystem, for all the file
2198   * systems.
2199   */
2200  public static void clearStatistics() {
2201    AbstractFileSystem.clearStatistics();
2202  }
2203
2204  /**
2205   * Prints the statistics to standard output. File System is identified by the
2206   * scheme and authority.
2207   */
2208  public static void printStatistics() {
2209    AbstractFileSystem.printStatistics();
2210  }
2211
2212  /**
2213   * @return Map of uri and statistics for each filesystem instantiated. The uri
2214   *         consists of scheme and authority for the filesystem.
2215   */
2216  public static Map<URI, Statistics> getAllStatistics() {
2217    return AbstractFileSystem.getAllStatistics();
2218  }
2219  
2220  /**
2221   * Get delegation tokens for the file systems accessed for a given
2222   * path.
2223   * @param p Path for which delegations tokens are requested.
2224   * @param renewer the account name that is allowed to renew the token.
2225   * @return List of delegation tokens.
2226   * @throws IOException
2227   */
2228  @InterfaceAudience.LimitedPrivate( { "HDFS", "MapReduce" })
2229  public List<Token<?>> getDelegationTokens(
2230      Path p, String renewer) throws IOException {
2231    Set<AbstractFileSystem> afsSet = resolveAbstractFileSystems(p);
2232    List<Token<?>> tokenList = 
2233        new ArrayList<Token<?>>();
2234    for (AbstractFileSystem afs : afsSet) {
2235      List<Token<?>> afsTokens = afs.getDelegationTokens(renewer);
2236      tokenList.addAll(afsTokens);
2237    }
2238    return tokenList;
2239  }
2240
2241  /**
2242   * Modifies ACL entries of files and directories.  This method can add new ACL
2243   * entries or modify the permissions on existing ACL entries.  All existing
2244   * ACL entries that are not specified in this call are retained without
2245   * changes.  (Modifications are merged into the current ACL.)
2246   *
2247   * @param path Path to modify
2248   * @param aclSpec List<AclEntry> describing modifications
2249   * @throws IOException if an ACL could not be modified
2250   */
2251  public void modifyAclEntries(final Path path, final List<AclEntry> aclSpec)
2252      throws IOException {
2253    Path absF = fixRelativePart(path);
2254    new FSLinkResolver<Void>() {
2255      @Override
2256      public Void next(final AbstractFileSystem fs, final Path p)
2257          throws IOException {
2258        fs.modifyAclEntries(p, aclSpec);
2259        return null;
2260      }
2261    }.resolve(this, absF);
2262  }
2263
2264  /**
2265   * Removes ACL entries from files and directories.  Other ACL entries are
2266   * retained.
2267   *
2268   * @param path Path to modify
2269   * @param aclSpec List<AclEntry> describing entries to remove
2270   * @throws IOException if an ACL could not be modified
2271   */
2272  public void removeAclEntries(final Path path, final List<AclEntry> aclSpec)
2273      throws IOException {
2274    Path absF = fixRelativePart(path);
2275    new FSLinkResolver<Void>() {
2276      @Override
2277      public Void next(final AbstractFileSystem fs, final Path p)
2278          throws IOException {
2279        fs.removeAclEntries(p, aclSpec);
2280        return null;
2281      }
2282    }.resolve(this, absF);
2283  }
2284
2285  /**
2286   * Removes all default ACL entries from files and directories.
2287   *
2288   * @param path Path to modify
2289   * @throws IOException if an ACL could not be modified
2290   */
2291  public void removeDefaultAcl(Path path)
2292      throws IOException {
2293    Path absF = fixRelativePart(path);
2294    new FSLinkResolver<Void>() {
2295      @Override
2296      public Void next(final AbstractFileSystem fs, final Path p)
2297          throws IOException {
2298        fs.removeDefaultAcl(p);
2299        return null;
2300      }
2301    }.resolve(this, absF);
2302  }
2303
2304  /**
2305   * Removes all but the base ACL entries of files and directories.  The entries
2306   * for user, group, and others are retained for compatibility with permission
2307   * bits.
2308   *
2309   * @param path Path to modify
2310   * @throws IOException if an ACL could not be removed
2311   */
2312  public void removeAcl(Path path) throws IOException {
2313    Path absF = fixRelativePart(path);
2314    new FSLinkResolver<Void>() {
2315      @Override
2316      public Void next(final AbstractFileSystem fs, final Path p)
2317          throws IOException {
2318        fs.removeAcl(p);
2319        return null;
2320      }
2321    }.resolve(this, absF);
2322  }
2323
2324  /**
2325   * Fully replaces ACL of files and directories, discarding all existing
2326   * entries.
2327   *
2328   * @param path Path to modify
2329   * @param aclSpec List<AclEntry> describing modifications, must include entries
2330   *   for user, group, and others for compatibility with permission bits.
2331   * @throws IOException if an ACL could not be modified
2332   */
2333  public void setAcl(Path path, final List<AclEntry> aclSpec)
2334      throws IOException {
2335    Path absF = fixRelativePart(path);
2336    new FSLinkResolver<Void>() {
2337      @Override
2338      public Void next(final AbstractFileSystem fs, final Path p)
2339          throws IOException {
2340        fs.setAcl(p, aclSpec);
2341        return null;
2342      }
2343    }.resolve(this, absF);
2344  }
2345
2346  /**
2347   * Gets the ACLs of files and directories.
2348   *
2349   * @param path Path to get
2350   * @return RemoteIterator<AclStatus> which returns each AclStatus
2351   * @throws IOException if an ACL could not be read
2352   */
2353  public AclStatus getAclStatus(Path path) throws IOException {
2354    Path absF = fixRelativePart(path);
2355    return new FSLinkResolver<AclStatus>() {
2356      @Override
2357      public AclStatus next(final AbstractFileSystem fs, final Path p)
2358          throws IOException {
2359        return fs.getAclStatus(p);
2360      }
2361    }.resolve(this, absF);
2362  }
2363
2364  /**
2365   * Set an xattr of a file or directory.
2366   * The name must be prefixed with the namespace followed by ".". For example,
2367   * "user.attr".
2368   * <p/>
2369   * Refer to the HDFS extended attributes user documentation for details.
2370   *
2371   * @param path Path to modify
2372   * @param name xattr name.
2373   * @param value xattr value.
2374   * @throws IOException
2375   */
2376  public void setXAttr(Path path, String name, byte[] value)
2377      throws IOException {
2378    setXAttr(path, name, value, EnumSet.of(XAttrSetFlag.CREATE,
2379        XAttrSetFlag.REPLACE));
2380  }
2381
2382  /**
2383   * Set an xattr of a file or directory.
2384   * The name must be prefixed with the namespace followed by ".". For example,
2385   * "user.attr".
2386   * <p/>
2387   * Refer to the HDFS extended attributes user documentation for details.
2388   *
2389   * @param path Path to modify
2390   * @param name xattr name.
2391   * @param value xattr value.
2392   * @param flag xattr set flag
2393   * @throws IOException
2394   */
2395  public void setXAttr(Path path, final String name, final byte[] value,
2396      final EnumSet<XAttrSetFlag> flag) throws IOException {
2397    final Path absF = fixRelativePart(path);
2398    new FSLinkResolver<Void>() {
2399      @Override
2400      public Void next(final AbstractFileSystem fs, final Path p)
2401          throws IOException {
2402        fs.setXAttr(p, name, value, flag);
2403        return null;
2404      }
2405    }.resolve(this, absF);
2406  }
2407
2408  /**
2409   * Get an xattr for a file or directory.
2410   * The name must be prefixed with the namespace followed by ".". For example,
2411   * "user.attr".
2412   * <p/>
2413   * Refer to the HDFS extended attributes user documentation for details.
2414   *
2415   * @param path Path to get extended attribute
2416   * @param name xattr name.
2417   * @return byte[] xattr value.
2418   * @throws IOException
2419   */
2420  public byte[] getXAttr(Path path, final String name) throws IOException {
2421    final Path absF = fixRelativePart(path);
2422    return new FSLinkResolver<byte[]>() {
2423      @Override
2424      public byte[] next(final AbstractFileSystem fs, final Path p)
2425          throws IOException {
2426        return fs.getXAttr(p, name);
2427      }
2428    }.resolve(this, absF);
2429  }
2430
2431  /**
2432   * Get all of the xattrs for a file or directory.
2433   * Only those xattrs for which the logged-in user has permissions to view
2434   * are returned.
2435   * <p/>
2436   * Refer to the HDFS extended attributes user documentation for details.
2437   *
2438   * @param path Path to get extended attributes
2439   * @return Map<String, byte[]> describing the XAttrs of the file or directory
2440   * @throws IOException
2441   */
2442  public Map<String, byte[]> getXAttrs(Path path) throws IOException {
2443    final Path absF = fixRelativePart(path);
2444    return new FSLinkResolver<Map<String, byte[]>>() {
2445      @Override
2446      public Map<String, byte[]> next(final AbstractFileSystem fs, final Path p)
2447          throws IOException {
2448        return fs.getXAttrs(p);
2449      }
2450    }.resolve(this, absF);
2451  }
2452
2453  /**
2454   * Get all of the xattrs for a file or directory.
2455   * Only those xattrs for which the logged-in user has permissions to view
2456   * are returned.
2457   * <p/>
2458   * Refer to the HDFS extended attributes user documentation for details.
2459   *
2460   * @param path Path to get extended attributes
2461   * @param names XAttr names.
2462   * @return Map<String, byte[]> describing the XAttrs of the file or directory
2463   * @throws IOException
2464   */
2465  public Map<String, byte[]> getXAttrs(Path path, final List<String> names)
2466      throws IOException {
2467    final Path absF = fixRelativePart(path);
2468    return new FSLinkResolver<Map<String, byte[]>>() {
2469      @Override
2470      public Map<String, byte[]> next(final AbstractFileSystem fs, final Path p)
2471          throws IOException {
2472        return fs.getXAttrs(p, names);
2473      }
2474    }.resolve(this, absF);
2475  }
2476
2477  /**
2478   * Remove an xattr of a file or directory.
2479   * The name must be prefixed with the namespace followed by ".". For example,
2480   * "user.attr".
2481   * <p/>
2482   * Refer to the HDFS extended attributes user documentation for details.
2483   *
2484   * @param path Path to remove extended attribute
2485   * @param name xattr name
2486   * @throws IOException
2487   */
2488  public void removeXAttr(Path path, final String name) throws IOException {
2489    final Path absF = fixRelativePart(path);
2490    new FSLinkResolver<Void>() {
2491      @Override
2492      public Void next(final AbstractFileSystem fs, final Path p)
2493          throws IOException {
2494        fs.removeXAttr(p, name);
2495        return null;
2496      }
2497    }.resolve(this, absF);
2498  }
2499
2500  /**
2501   * Get all of the xattr names for a file or directory.
2502   * Only those xattr names which the logged-in user has permissions to view
2503   * are returned.
2504   * <p/>
2505   * Refer to the HDFS extended attributes user documentation for details.
2506   *
2507   * @param path Path to get extended attributes
2508   * @return List<String> of the XAttr names of the file or directory
2509   * @throws IOException
2510   */
2511  public List<String> listXAttrs(Path path) throws IOException {
2512    final Path absF = fixRelativePart(path);
2513    return new FSLinkResolver<List<String>>() {
2514      @Override
2515      public List<String> next(final AbstractFileSystem fs, final Path p)
2516          throws IOException {
2517        return fs.listXAttrs(p);
2518      }
2519    }.resolve(this, absF);
2520  }
2521}