HDFS-10745. Directly resolve paths into INodesInPath. Contributed by Daryn Sharp..

hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt

@@ -47,6 +47,9 @@ Release 2.7.4 - UNRELEASED
     HDFS-9145. Tracking methods that hold FSNamesytemLock for too long.
     (Mingliang Liu via Haohui Mai)
 
+    HDFS-10745. Directly resolve paths into INodesInPath.
+    (Daryn Sharp via kihwal)
+
   OPTIMIZATIONS
 
     HDFS-10896. Move lock logging logic from FSNamesystem into FSNamesystemLock.

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirAclOp.java

@@ -25,7 +25,6 @@ import org.apache.hadoop.fs.permission.FsAction;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.hdfs.DFSConfigKeys;
 import org.apache.hadoop.hdfs.protocol.AclException;
-import org.apache.hadoop.hdfs.protocol.HdfsConstants;
 import org.apache.hadoop.hdfs.protocol.HdfsFileStatus;
 
 import java.io.IOException;
@@ -39,11 +38,11 @@ class FSDirAclOp {
     String src = srcArg;
     checkAclsConfigFlag(fsd);
     FSPermissionChecker pc = fsd.getPermissionChecker();
-    src = fsd.resolvePath(pc, src);
     INodesInPath iip;
     fsd.writeLock();
     try {
-      iip = fsd.getINodesInPath4Write(FSDirectory.normalizePath(src), true);
+      iip = fsd.resolvePathForWrite(pc, src);
+      src = iip.getPath();
       fsd.checkOwner(pc, iip);
       INode inode = FSDirectory.resolveLastINode(iip);
       int snapshotId = iip.getLatestSnapshotId();
@@ -64,11 +63,11 @@ class FSDirAclOp {
     String src = srcArg;
     checkAclsConfigFlag(fsd);
     FSPermissionChecker pc = fsd.getPermissionChecker();
-    src = fsd.resolvePath(pc, src);
     INodesInPath iip;
     fsd.writeLock();
     try {
-      iip = fsd.getINodesInPath4Write(FSDirectory.normalizePath(src), true);
+      iip = fsd.resolvePathForWrite(pc, src);
+      src = iip.getPath();
       fsd.checkOwner(pc, iip);
       INode inode = FSDirectory.resolveLastINode(iip);
       int snapshotId = iip.getLatestSnapshotId();
@@ -88,11 +87,11 @@ class FSDirAclOp {
     String src = srcArg;
     checkAclsConfigFlag(fsd);
     FSPermissionChecker pc = fsd.getPermissionChecker();
-    src = fsd.resolvePath(pc, src);
     INodesInPath iip;
     fsd.writeLock();
     try {
-      iip = fsd.getINodesInPath4Write(FSDirectory.normalizePath(src), true);
+      iip = fsd.resolvePathForWrite(pc, src);
+      src = iip.getPath();
       fsd.checkOwner(pc, iip);
       INode inode = FSDirectory.resolveLastINode(iip);
       int snapshotId = iip.getLatestSnapshotId();
@@ -112,11 +111,11 @@ class FSDirAclOp {
     String src = srcArg;
     checkAclsConfigFlag(fsd);
     FSPermissionChecker pc = fsd.getPermissionChecker();
-    src = fsd.resolvePath(pc, src);
     INodesInPath iip;
     fsd.writeLock();
     try {
-      iip = fsd.getINodesInPath4Write(src);
+      iip = fsd.resolvePathForWrite(pc, src);
+      src = iip.getPath();
       fsd.checkOwner(pc, iip);
       unprotectedRemoveAcl(fsd, iip);
     } finally {
@@ -132,11 +131,11 @@ class FSDirAclOp {
     String src = srcArg;
     checkAclsConfigFlag(fsd);
     FSPermissionChecker pc = fsd.getPermissionChecker();
-    src = fsd.resolvePath(pc, src);
     INodesInPath iip;
     fsd.writeLock();
     try {
-      iip = fsd.getINodesInPath4Write(src);
+      iip = fsd.resolvePathForWrite(pc, src);
+      src = iip.getPath();
       fsd.checkOwner(pc, iip);
       List<AclEntry> newAcl = unprotectedSetAcl(fsd, src, aclSpec, false);
       fsd.getEditLog().logSetAcl(src, newAcl);
@@ -150,17 +149,15 @@ class FSDirAclOp {
       FSDirectory fsd, String src) throws IOException {
     checkAclsConfigFlag(fsd);
     FSPermissionChecker pc = fsd.getPermissionChecker();
-    src = fsd.resolvePath(pc, src);
-    String srcs = FSDirectory.normalizePath(src);
     fsd.readLock();
     try {
+      INodesInPath iip = fsd.resolvePath(pc, src);
+      src = iip.getPath();
       // There is no real inode for the path ending in ".snapshot", so return a
       // non-null, unpopulated AclStatus.  This is similar to getFileInfo.
-      if (srcs.endsWith(HdfsConstants.SEPARATOR_DOT_SNAPSHOT_DIR) &&
-          fsd.getINode4DotSnapshot(srcs) != null) {
+      if (iip.isDotSnapshotDir() && fsd.getINode4DotSnapshot(iip) != null) {
         return new AclStatus.Builder().owner("").group("").build();
       }
-      INodesInPath iip = fsd.getINodesInPath(srcs, true);
       if (fsd.isPermissionEnabled()) {
         fsd.checkTraverse(pc, iip);
       }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirAttrOp.java

@@ -55,8 +55,8 @@ public class FSDirAttrOp {
     INodesInPath iip;
     fsd.writeLock();
     try {
-      src = fsd.resolvePath(pc, src);
-      iip = fsd.getINodesInPath4Write(src);
+      iip = fsd.resolvePathForWrite(pc, src);
+      src = iip.getPath();
       fsd.checkOwner(pc, iip);
       unprotectedSetPermission(fsd, src, permission);
     } finally {
@@ -73,8 +73,8 @@ public class FSDirAttrOp {
     INodesInPath iip;
     fsd.writeLock();
     try {
-      src = fsd.resolvePath(pc, src);
-      iip = fsd.getINodesInPath4Write(src);
+      iip = fsd.resolvePathForWrite(pc, src);
+      src = iip.getPath();
       fsd.checkOwner(pc, iip);
       if (!pc.isSuperUser()) {
         if (username != null && !pc.getUser().equals(username)) {
@@ -107,8 +107,8 @@ public class FSDirAttrOp {
     INodesInPath iip;
     fsd.writeLock();
     try {
-      src = fsd.resolvePath(pc, src);
-      iip = fsd.getINodesInPath4Write(src);
+      iip = fsd.resolvePathForWrite(pc, src);
+      src = iip.getPath();
       // Write access is required to set access and modification times
       if (fsd.isPermissionEnabled()) {
         fsd.checkPathAccess(pc, iip, FsAction.WRITE);
@@ -137,8 +137,8 @@ public class FSDirAttrOp {
     FSPermissionChecker pc = fsd.getPermissionChecker();
     fsd.writeLock();
     try {
-      src = fsd.resolvePath(pc, src);
-      final INodesInPath iip = fsd.getINodesInPath4Write(src);
+      final INodesInPath iip = fsd.resolvePathForWrite(pc, src);
+      src = iip.getPath();
       if (fsd.isPermissionEnabled()) {
         fsd.checkPathAccess(pc, iip, FsAction.WRITE);
       }
@@ -200,8 +200,8 @@ public class FSDirAttrOp {
     FSPermissionChecker pc = fsd.getPermissionChecker();
     fsd.readLock();
     try {
-      src = fsd.resolvePath(pc, src);
-      final INodesInPath iip = fsd.getINodesInPath(src, false);
+      final INodesInPath iip = fsd.resolvePath(pc, src, false);
+      src = iip.getPath();
       if (fsd.isPermissionEnabled()) {
         fsd.checkTraverse(pc, iip);
       }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirDeleteOp.java

@@ -78,8 +78,8 @@ class FSDirDeleteOp {
     FSDirectory fsd = fsn.getFSDirectory();
     FSPermissionChecker pc = fsd.getPermissionChecker();
 
-    src = fsd.resolvePath(pc, src);
-    final INodesInPath iip = fsd.getINodesInPath4Write(src, false);
+    final INodesInPath iip = fsd.resolvePathForWrite(pc, src, false);
+    src = iip.getPath();
     if (!recursive && fsd.isNonEmptyDirectory(iip)) {
       throw new PathIsNotEmptyDirectoryException(src + " is non empty");
     }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirMkdirOp.java

@@ -52,8 +52,8 @@ class FSDirMkdirOp {
     FSPermissionChecker pc = fsd.getPermissionChecker();
     fsd.writeLock();
     try {
-      src = fsd.resolvePath(pc, src);
-      INodesInPath iip = fsd.getINodesInPath4Write(src);
+      INodesInPath iip = fsd.resolvePathForWrite(pc, src);
+      src = iip.getPath();
       if (fsd.isPermissionEnabled()) {
         fsd.checkTraverse(pc, iip);
       }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirRenameOp.java

@@ -65,12 +65,14 @@ class FSDirRenameOp {
     FSPermissionChecker pc = fsd.getPermissionChecker();
 
     HdfsFileStatus resultingStat = null;
-    src = fsd.resolvePath(pc, src);
-    dst = fsd.resolvePath(pc, dst);
+    // Rename does not operate on link targets
+    // Do not resolveLink when checking permissions of src and dst
+    INodesInPath srcIIP = fsd.resolvePathForWrite(pc, src, false);
+    INodesInPath dstIIP = fsd.resolvePathForWrite(pc, dst, false);
     @SuppressWarnings("deprecation")
-    final boolean status = renameTo(fsd, pc, src, dst, logRetryCache);
+    final boolean status = renameTo(fsd, pc, srcIIP, dstIIP, logRetryCache);
     if (status) {
-      INodesInPath dstIIP = fsd.getINodesInPath(dst, false);
+      dstIIP = fsd.getINodesInPath(dstIIP.getPath(), false);
       resultingStat = fsd.getAuditFileInfo(dstIIP);
     }
     return new RenameOldResult(status, resultingStat);
@@ -238,9 +240,8 @@ class FSDirRenameOp {
     final FSPermissionChecker pc = fsd.getPermissionChecker();
 
     BlocksMapUpdateInfo collectedBlocks = new BlocksMapUpdateInfo();
-    src = fsd.resolvePath(pc, src);
-    dst = fsd.resolvePath(pc, dst);
-    renameTo(fsd, pc, src, dst, collectedBlocks, logRetryCache, options);
+    // returns resolved path
+    dst = renameTo(fsd, pc, src, dst, collectedBlocks, logRetryCache, options);
     INodesInPath dstIIP = fsd.getINodesInPath(dst, false);
     HdfsFileStatus resultingStat = fsd.getAuditFileInfo(dstIIP);
 
@@ -252,11 +253,13 @@ class FSDirRenameOp {
    * @see {@link #unprotectedRenameTo(FSDirectory, String, String, INodesInPath,
    * INodesInPath, long, BlocksMapUpdateInfo, Options.Rename...)}
    */
-  static void renameTo(FSDirectory fsd, FSPermissionChecker pc, String src,
+  static String renameTo(FSDirectory fsd, FSPermissionChecker pc, String src,
       String dst, BlocksMapUpdateInfo collectedBlocks, boolean logRetryCache,
       Options.Rename... options) throws IOException {
-    final INodesInPath srcIIP = fsd.getINodesInPath4Write(src, false);
-    final INodesInPath dstIIP = fsd.getINodesInPath4Write(dst, false);
+    final INodesInPath srcIIP = fsd.resolvePathForWrite(pc, src, false);
+    final INodesInPath dstIIP = fsd.resolvePathForWrite(pc, dst, false);
+    src = srcIIP.getPath();
+    dst = dstIIP.getPath();
     if (fsd.isPermissionEnabled()) {
       // Rename does not operate on link targets
       // Do not resolveLink when checking permissions of src and dst
@@ -283,6 +286,7 @@ class FSDirRenameOp {
       fsd.writeUnlock();
     }
     fsd.getEditLog().logRename(src, dst, mtime, logRetryCache, options);
+    return dst;
   }
 
   /**
@@ -442,16 +446,17 @@ class FSDirRenameOp {
   @Deprecated
   @SuppressWarnings("deprecation")
   private static boolean renameTo(FSDirectory fsd, FSPermissionChecker pc,
-      String src, String dst, boolean logRetryCache) throws IOException {
-    // Rename does not operate on link targets
-    // Do not resolveLink when checking permissions of src and dst
-    // Check write access to parent of src
-    final INodesInPath srcIIP = fsd.getINodesInPath4Write(src, false);
+      INodesInPath srcIIP, INodesInPath dstIIP, boolean logRetryCache)
+          throws IOException {
+    String src = srcIIP.getPath();
+    String dst = dstIIP.getPath();
     // Note: We should not be doing this.  This is move() not renameTo().
-    final String actualDst = fsd.isDir(dst) ?
-        dst + Path.SEPARATOR + new Path(src).getName() : dst;
-    final INodesInPath dstIIP = fsd.getINodesInPath4Write(actualDst, false);
+    if (fsd.isDir(dst)) {
+      dstIIP = INodesInPath.append(dstIIP, null, srcIIP.getLastLocalName());
+    }
+    final String actualDst = dstIIP.getPath();
     if (fsd.isPermissionEnabled()) {
+      // Check write access to parent of src
       fsd.checkPermission(pc, srcIIP, false, null, FsAction.WRITE, null, null,
           false);
       // Check write access to ancestor of dst

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirStatAndListingOp.java

@@ -48,8 +48,8 @@ class FSDirStatAndListingOp {
       byte[] startAfter, boolean needLocation) throws IOException {
     FSPermissionChecker pc = fsd.getPermissionChecker();
     final String startAfterString = DFSUtil.bytes2String(startAfter);
-    final String src = fsd.resolvePath(pc, srcArg);
-    final INodesInPath iip = fsd.getINodesInPath(src, true);
+    final INodesInPath iip = fsd.resolvePath(pc, srcArg);
+    final String src = iip.getPath();
 
     // Get file name when startAfter is an INodePath
     if (FSDirectory.isReservedName(startAfterString)) {
@@ -93,16 +93,16 @@ class FSDirStatAndListingOp {
     if (!DFSUtil.isValidName(src)) {
       throw new InvalidPathException("Invalid file name: " + src);
     }
-    FSPermissionChecker pc = fsd.getPermissionChecker();
-    src = fsd.resolvePath(pc, srcArg);
-    final INodesInPath iip = fsd.getINodesInPath(src, resolveLink);
-    boolean isSuperUser = true;
     if (fsd.isPermissionEnabled()) {
+      FSPermissionChecker pc = fsd.getPermissionChecker();
+      final INodesInPath iip = fsd.resolvePath(pc, srcArg, resolveLink);
+      src = iip.getPath();
       fsd.checkPermission(pc, iip, false, null, null, null, null, false);
-      isSuperUser = pc.isSuperUser();
+    } else {
+      src = FSDirectory.resolvePath(srcArg, fsd);
     }
-    return getFileInfo(fsd, src, resolveLink,
-        FSDirectory.isReservedRawName(srcArg), isSuperUser);
+    return getFileInfo(fsd, src, FSDirectory.isReservedRawName(srcArg),
+        resolveLink);
   }
 
   /**
@@ -110,8 +110,8 @@ class FSDirStatAndListingOp {
    */
   static boolean isFileClosed(FSDirectory fsd, String src) throws IOException {
     FSPermissionChecker pc = fsd.getPermissionChecker();
-    src = fsd.resolvePath(pc, src);
-    final INodesInPath iip = fsd.getINodesInPath(src, true);
+    final INodesInPath iip = fsd.resolvePath(pc, src);
+    src = iip.getPath();
     if (fsd.isPermissionEnabled()) {
       fsd.checkTraverse(pc, iip);
     }
@@ -121,8 +121,7 @@ class FSDirStatAndListingOp {
   static ContentSummary getContentSummary(
       FSDirectory fsd, String src) throws IOException {
     FSPermissionChecker pc = fsd.getPermissionChecker();
-    src = fsd.resolvePath(pc, src);
-    final INodesInPath iip = fsd.getINodesInPath(src, false);
+    final INodesInPath iip = fsd.resolvePath(pc, src);
     if (fsd.isPermissionEnabled()) {
       fsd.checkPermission(pc, iip, false, null, null, null,
           FsAction.READ_EXECUTE);
@@ -270,54 +269,55 @@ class FSDirStatAndListingOp {
 
   /** Get the file info for a specific file.
    * @param fsd FSDirectory
-   * @param iip The path to the file, the file is included
+   * @param src The path to the file, the file is included
    * @param isRawPath true if a /.reserved/raw pathname was passed by the user
    * @param includeStoragePolicy whether to include storage policy
    * @return object containing information regarding the file
    *         or null if file not found
    */
   static HdfsFileStatus getFileInfo(
-      FSDirectory fsd, String path, INodesInPath iip, boolean isRawPath,
+      FSDirectory fsd, String path, INodesInPath src, boolean isRawPath,
       boolean includeStoragePolicy)
       throws IOException {
     fsd.readLock();
     try {
-      final INode node = iip.getLastINode();
-      if (node == null) {
+      final INode i = src.getLastINode();
+      if (i == null) {
         return null;
       }
 
-      byte policyId = includeStoragePolicy && !node.isSymlink() ?
-          node.getStoragePolicyID() :
+      byte policyId = includeStoragePolicy && !i.isSymlink() ?
+          i.getStoragePolicyID() :
           BlockStoragePolicySuite.ID_UNSPECIFIED;
       INodeAttributes nodeAttrs = getINodeAttributes(fsd, path,
                                                      HdfsFileStatus.EMPTY_NAME,
-                                                     node, iip.getPathSnapshotId());
+                                                     i, src.getPathSnapshotId());
       return createFileStatus(fsd, HdfsFileStatus.EMPTY_NAME, nodeAttrs,
-                              policyId, iip.getPathSnapshotId(), isRawPath, iip);
+                              policyId, src.getPathSnapshotId(), isRawPath, src);
     } finally {
       fsd.readUnlock();
     }
   }
 
   static HdfsFileStatus getFileInfo(
-      FSDirectory fsd, String src, boolean resolveLink, boolean isRawPath,
-      boolean includeStoragePolicy)
-    throws IOException {
-    String srcs = FSDirectory.normalizePath(src);
-    if (srcs.endsWith(HdfsConstants.SEPARATOR_DOT_SNAPSHOT_DIR)) {
-      if (fsd.getINode4DotSnapshot(srcs) != null) {
-        return new HdfsFileStatus(0, true, 0, 0, 0, 0, null, null, null, null,
-            HdfsFileStatus.EMPTY_NAME, -1L, 0, null,
-            BlockStoragePolicySuite.ID_UNSPECIFIED);
-      }
-      return null;
-    }
-
+      FSDirectory fsd, String src, boolean resolveLink, boolean isRawPath)
+      throws IOException {
     fsd.readLock();
     try {
-      final INodesInPath iip = fsd.getINodesInPath(srcs, resolveLink);
-      return getFileInfo(fsd, src, iip, isRawPath, includeStoragePolicy);
+      HdfsFileStatus status = null;
+      final INodesInPath iip = fsd.getINodesInPath(src, resolveLink);
+      if (FSDirectory.isExactReservedName(iip.getPathComponents())) {
+        status = new HdfsFileStatus(0, true, 0, 0, 0, 0, new FsPermission((short) 01770),
+            null, null, null, HdfsFileStatus.EMPTY_NAME, -1L, 0, null,
+            BlockStoragePolicySuite.ID_UNSPECIFIED);
+      } else if (iip.isDotSnapshotDir()) {
+        if (fsd.getINode4DotSnapshot(iip) != null) {
+          status = FSDirectory.DOT_SNAPSHOT_DIR_STATUS;
+        }
+      } else {
+        status = getFileInfo(fsd, src, iip, isRawPath, true);
+      }
+      return status;
     } finally {
       fsd.readUnlock();
     }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirSymlinkOp.java

@@ -55,8 +55,8 @@ class FSDirSymlinkOp {
     INodesInPath iip;
     fsd.writeLock();
     try {
-      link = fsd.resolvePath(pc, link);
-      iip = fsd.getINodesInPath4Write(link, false);
+      iip = fsd.resolvePathForWrite(pc, link, false);
+      link = iip.getPath();
       if (!createParent) {
         fsd.verifyParentDir(iip, link);
       }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirXAttrOp.java

@@ -67,13 +67,13 @@ class FSDirXAttrOp {
     FSPermissionChecker pc = fsd.getPermissionChecker();
     XAttrPermissionFilter.checkPermissionForApi(
         pc, xAttr, FSDirectory.isReservedRawName(src));
-    src = fsd.resolvePath(pc, src);
     List<XAttr> xAttrs = Lists.newArrayListWithCapacity(1);
     xAttrs.add(xAttr);
     INodesInPath iip;
     fsd.writeLock();
     try {
-      iip = fsd.getINodesInPath4Write(src);
+      iip = fsd.resolvePathForWrite(pc, src);
+      src = iip.getPath();
       checkXAttrChangeAccess(fsd, iip, xAttr, pc);
       unprotectedSetXAttrs(fsd, src, xAttrs, flag);
     } finally {
@@ -94,12 +94,11 @@ class FSDirXAttrOp {
     if (!getAll) {
       XAttrPermissionFilter.checkPermissionForApi(pc, xAttrs, isRawPath);
     }
-    src = fsd.resolvePath(pc, src);
-    final INodesInPath iip = fsd.getINodesInPath(src, true);
+    final INodesInPath iip = fsd.resolvePath(pc, src);
     if (fsd.isPermissionEnabled()) {
       fsd.checkPathAccess(pc, iip, FsAction.READ);
     }
-    List<XAttr> all = FSDirXAttrOp.getXAttrs(fsd, src);
+    List<XAttr> all = FSDirXAttrOp.getXAttrs(fsd, iip);
     List<XAttr> filteredAll = XAttrPermissionFilter.
         filterXAttrsForApi(pc, all, isRawPath);
 
@@ -133,13 +132,12 @@ class FSDirXAttrOp {
     FSDirXAttrOp.checkXAttrsConfigFlag(fsd);
     final FSPermissionChecker pc = fsd.getPermissionChecker();
     final boolean isRawPath = FSDirectory.isReservedRawName(src);
-    src = fsd.resolvePath(pc, src);
-    final INodesInPath iip = fsd.getINodesInPath(src, true);
+    final INodesInPath iip = fsd.resolvePath(pc, src);
     if (fsd.isPermissionEnabled()) {
       /* To access xattr names, you need EXECUTE in the owning directory. */
       fsd.checkParentAccess(pc, iip, FsAction.EXECUTE);
     }
-    final List<XAttr> all = FSDirXAttrOp.getXAttrs(fsd, src);
+    final List<XAttr> all = FSDirXAttrOp.getXAttrs(fsd, iip);
     return XAttrPermissionFilter.
         filterXAttrsForApi(pc, all, isRawPath);
   }
@@ -166,8 +164,8 @@ class FSDirXAttrOp {
     INodesInPath iip;
     fsd.writeLock();
     try {
-      src = fsd.resolvePath(pc, src);
-      iip = fsd.getINodesInPath4Write(src);
+      iip = fsd.resolvePathForWrite(pc, src);
+      src = iip.getPath();
       checkXAttrChangeAccess(fsd, iip, xAttr, pc);
 
       List<XAttr> removedXAttrs = unprotectedRemoveXAttrs(fsd, src, xAttrs);
@@ -438,12 +436,11 @@ class FSDirXAttrOp {
     }
   }
 
-  private static List<XAttr> getXAttrs(FSDirectory fsd,
-                                String src) throws IOException {
-    String srcs = FSDirectory.normalizePath(src);
+  private static List<XAttr> getXAttrs(FSDirectory fsd, INodesInPath iip)
+      throws IOException {
     fsd.readLock();
     try {
-      INodesInPath iip = fsd.getINodesInPath(srcs, true);
+      String src = iip.getPath();
       INode inode = FSDirectory.resolveLastINode(iip);
       int snapshotId = iip.getPathSnapshotId();
       return XAttrStorage.readINodeXAttrs(fsd.getAttributes(src,

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirectory.java

@@ -130,6 +130,11 @@ public class FSDirectory implements Closeable {
   private final static byte[] DOT_DOT =
       DFSUtil.string2Bytes("..");
 
+  public final static HdfsFileStatus DOT_SNAPSHOT_DIR_STATUS =
+      new HdfsFileStatus(0, true, 0, 0, 0, 0, null, null, null, null,
+          HdfsFileStatus.EMPTY_NAME, -1L, 0, null,
+          BlockStoragePolicySuite.ID_UNSPECIFIED);
+
   INodeDirectory rootDir;
   private final FSNamesystem namesystem;
   private volatile boolean skipQuotaCheck = false; //skip while consuming edits
@@ -565,12 +570,66 @@ public class FSDirectory implements Closeable {
    * @throws FileNotFoundException
    * @throws AccessControlException
    */
-  String resolvePath(FSPermissionChecker pc, String path)
-      throws FileNotFoundException, AccessControlException {
-    if (isReservedRawName(path) && isPermissionEnabled) {
+  INodesInPath resolvePath(FSPermissionChecker pc, String src)
+      throws UnresolvedLinkException, FileNotFoundException,
+      AccessControlException {
+    return resolvePath(pc, src, true);
+  }
+
+  INodesInPath resolvePath(FSPermissionChecker pc, String src,
+      boolean resolveLink) throws UnresolvedLinkException,
+  FileNotFoundException, AccessControlException {
+    byte[][] components = INode.getPathComponents(src);
+    if (isPermissionEnabled && pc != null && isReservedRawName(components)) {
       pc.checkSuperuserPrivilege();
     }
-    return resolvePath(path, this);
+    components = resolveComponents(components, this);
+    return INodesInPath.resolve(rootDir, components, resolveLink);
+  }
+
+  INodesInPath resolvePathForWrite(FSPermissionChecker pc, String src)
+      throws UnresolvedLinkException, FileNotFoundException,
+      AccessControlException {
+    return resolvePathForWrite(pc, src, true);
+  }
+
+  INodesInPath resolvePathForWrite(FSPermissionChecker pc, String src,
+      boolean resolveLink) throws UnresolvedLinkException,
+  FileNotFoundException, AccessControlException {
+    INodesInPath iip = resolvePath(pc, src, resolveLink);
+    if (iip.isSnapshot()) {
+      throw new SnapshotAccessControlException(
+          "Modification on a read-only snapshot is disallowed");
+    }
+    return iip;
+  }
+
+  INodesInPath resolvePath(FSPermissionChecker pc, String src, long fileId)
+      throws UnresolvedLinkException, FileNotFoundException,
+      AccessControlException {
+    // Older clients may not have given us an inode ID to work with.
+    // In this case, we have to try to resolve the path and hope it
+    // hasn't changed or been deleted since the file was opened for write.
+    INodesInPath iip;
+    if (fileId == INodeId.GRANDFATHER_INODE_ID) {
+      iip = resolvePath(pc, src);
+    } else {
+      INode inode = getInode(fileId);
+      if (inode == null) {
+        iip = INodesInPath.fromComponents(INode.getPathComponents(src));
+      } else {
+        iip = INodesInPath.fromINode(inode);
+      }
+    }
+    return iip;
+  }
+
+  // this method can be removed after IIP is used more extensively
+  static String resolvePath(String src,
+      FSDirectory fsd) throws FileNotFoundException {
+    byte[][] pathComponents = INode.getPathComponents(src);
+    pathComponents = resolveComponents(pathComponents, fsd);
+    return DFSUtil.byteArray2PathString(pathComponents);
   }
 
   /**
@@ -1426,17 +1485,29 @@ public class FSDirectory implements Closeable {
     return src.startsWith(DOT_RESERVED_PATH_PREFIX + Path.SEPARATOR);
   }
 
+  public static boolean isExactReservedName(byte[][] components) {
+    return CHECK_RESERVED_FILE_NAMES &&
+           (components.length == 2) &&
+           isReservedName(components);
+  }
+
   static boolean isReservedRawName(String src) {
     return src.startsWith(DOT_RESERVED_PATH_PREFIX +
         Path.SEPARATOR + RAW_STRING);
   }
 
   static boolean isReservedName(byte[][] components) {
-    return (components.length > 2) &&
+    return (components.length > 1) &&
         Arrays.equals(INodeDirectory.ROOT_NAME, components[0]) &&
         Arrays.equals(DOT_RESERVED, components[1]);
   }
 
+  static boolean isReservedRawName(byte[][] components) {
+    return (components.length > 2) &&
+           isReservedName(components) &&
+           Arrays.equals(RAW, components[2]);
+  }
+
   /**
    * Resolve a /.reserved/... path to a non-reserved path.
    * <p/>
@@ -1455,19 +1526,18 @@ public class FSDirectory implements Closeable {
    * /.reserved/raw/a/b/c is equivalent (they both refer to the same
    * unencrypted file).
    * 
-   * @param src path that is being processed
+   * @param pathComponents to be resolved
    * @param fsd FSDirectory
    * @return if the path indicates an inode, return path after replacing up to
    *         <inodeid> with the corresponding path of the inode, else the path
-   *         in {@code src} as is. If the path refers to a path in the "raw"
-   *         directory, return the non-raw pathname.
+   *         in {@code pathComponents} as is. If the path refers to a path in
+   *         the "raw" directory, return the non-raw pathname.
    * @throws FileNotFoundException if inodeid is invalid
    */
-  static String resolvePath(String src,
+  static byte[][] resolveComponents(byte[][] pathComponents,
       FSDirectory fsd) throws FileNotFoundException {
-    byte[][] pathComponents = INode.getPathComponents(src);
     final int nComponents = pathComponents.length;
-    if (!isReservedName(pathComponents)) {
+    if (nComponents < 3 || !isReservedName(pathComponents)) {
       /* This is not a /.reserved/ path so do nothing. */
     } else if (Arrays.equals(DOT_INODES, pathComponents[2])) {
       /* It's a /.reserved/.inodes path. */
@@ -1483,9 +1553,7 @@ public class FSDirectory implements Closeable {
             new byte[][]{INodeDirectory.ROOT_NAME}, pathComponents, 3);
       }
     }
-    // this double conversion will be unnecessary when resolving returns
-    // INodesInPath (needs components byte[][])
-    return DFSUtil.byteArray2PathString(pathComponents);
+    return pathComponents;
   }
 
   private static byte[][] resolveDotInodesPath(
@@ -1539,15 +1607,12 @@ public class FSDirectory implements Closeable {
     return components;
   }
 
-  INode getINode4DotSnapshot(String src) throws UnresolvedLinkException {
+  INode getINode4DotSnapshot(INodesInPath iip) throws UnresolvedLinkException {
     Preconditions.checkArgument(
-        src.endsWith(HdfsConstants.SEPARATOR_DOT_SNAPSHOT_DIR),
-        "%s does not end with %s", src, HdfsConstants.SEPARATOR_DOT_SNAPSHOT_DIR);
-
-    final String dirPath = normalizePath(src.substring(0,
-        src.length() - HdfsConstants.DOT_SNAPSHOT_DIR.length()));
+        iip.isDotSnapshotDir(), "%s does not end with %s",
+        iip.getPath(), HdfsConstants.SEPARATOR_DOT_SNAPSHOT_DIR);
 
-    final INode node = this.getINode(dirPath);
+    final INode node = iip.getINode(-2);
     if (node != null && node.isDirectory()
         && node.asDirectory().isSnapshottable()) {
       return node;

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java

@@ -1740,8 +1740,8 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
          * HDFS-7463. A better fix is to change the edit log of SetTime to
          * use inode id instead of a path.
          */
-        src = dir.resolvePath(pc, srcArg);
-        final INodesInPath iip = dir.getINodesInPath(src, true);
+        final INodesInPath iip = dir.resolvePath(pc, src);
+        src = iip.getPath();
         INode inode = iip.getLastINode();
         boolean updateAccessTime = inode != null &&
             now > inode.getAccessTime() + getAccessTimePrecision();
@@ -1817,8 +1817,8 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
       boolean needBlockToken)
       throws IOException {
     String src = srcArg;
-    src = dir.resolvePath(pc, srcArg);
-    final INodesInPath iip = dir.getINodesInPath(src, true);
+    final INodesInPath iip = dir.resolvePath(pc, src);
+    src = iip.getPath();
     final INodeFile inode = INodeFile.valueOf(iip.getLastINode(), src);
     if (isPermissionEnabled) {
       dir.checkPathAccess(pc, iip, FsAction.READ);
@@ -2016,7 +2016,8 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
     try {
       checkOperation(OperationCategory.WRITE);
       checkNameNodeSafeMode("Cannot truncate for " + src);
-      src = dir.resolvePath(pc, src);
+      INodesInPath iip = dir.resolvePath(pc, src);
+      src = iip.getPath();
       res = truncateInternal(src, newLength, clientName,
           clientMachine, mtime, pc, toRemoveBlocks);
       stat = dir.getAuditFileInfo(dir.getINodesInPath4Write(src, false));
@@ -2399,8 +2400,8 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
     if (provider != null) {
       readLock();
       try {
-        src = dir.resolvePath(pc, src);
-        INodesInPath iip = dir.getINodesInPath4Write(src);
+        INodesInPath iip = dir.resolvePathForWrite(pc, src);
+        src = iip.getPath();
         // Nothing to do if the path is not within an EZ
         final EncryptionZone zone = dir.getEZForPath(iip);
         if (zone != null) {
@@ -2437,8 +2438,8 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
       checkNameNodeSafeMode("Cannot create file" + src);
       dir.writeLock();
       try {
-        src = dir.resolvePath(pc, src);
-        final INodesInPath iip = dir.getINodesInPath4Write(src);
+        final INodesInPath iip = dir.resolvePathForWrite(pc, src);
+        src = iip.getPath();
         toRemoveBlocks = startFileInternal(
             pc, iip, permissions, holder,
             clientMachine, create, overwrite,
@@ -2446,7 +2447,7 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
             isLazyPersist, suite, protocolVersion, edek,
             logRetryCache);
         stat = FSDirStatAndListingOp.getFileInfo(
-            dir, src, false, FSDirectory.isReservedRawName(srcArg), true);
+            dir, src, false, FSDirectory.isReservedRawName(srcArg));
       } finally {
         dir.writeUnlock();
       }
@@ -2815,8 +2816,8 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
     try {
       checkOperation(OperationCategory.WRITE);
       checkNameNodeSafeMode("Cannot recover the lease of " + src);
-      src = dir.resolvePath(pc, src);
-      final INodesInPath iip = dir.getINodesInPath4Write(src);
+      final INodesInPath iip = dir.resolvePathForWrite(pc, src);
+      src = iip.getPath();
       final INodeFile inode = INodeFile.valueOf(iip.getLastINode(), src);
       if (!inode.isUnderConstruction()) {
         return true;
@@ -2967,12 +2968,12 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
     try {
       checkOperation(OperationCategory.WRITE);
       checkNameNodeSafeMode("Cannot append to file" + src);
-      src = dir.resolvePath(pc, src);
-      final INodesInPath iip = dir.getINodesInPath4Write(src);
+      final INodesInPath iip = dir.resolvePathForWrite(pc, src);
+      src = iip.getPath();
       lb = appendFileInternal(pc, iip, holder, clientMachine, newBlock,
           logRetryCache);
       stat = FSDirStatAndListingOp.getFileInfo(dir, src, false,
-          FSDirectory.isReservedRawName(srcArg), true);
+          FSDirectory.isReservedRawName(srcArg));
     } catch (StandbyException se) {
       skipSync = true;
       throw se;
@@ -3056,9 +3057,10 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
     readLock();
     try {
       checkOperation(OperationCategory.READ);
-      src = dir.resolvePath(pc, src);
+      INodesInPath iip = dir.resolvePath(pc, src, fileId);
+      src = iip.getPath();
       FileState fileState = analyzeFileState(
-          src, fileId, clientName, previous, onRetryBlock);
+          iip, fileId, clientName, previous, onRetryBlock);
       if (onRetryBlock[0] != null && onRetryBlock[0].getLocations().length > 0) {
         // This is a retry. No need to generate new locations.
         // Use the last block if it has locations.
@@ -3117,8 +3119,9 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
       // Run the full analysis again, since things could have changed
       // while chooseTarget() was executing.
       LocatedBlock[] onRetryBlock = new LocatedBlock[1];
+      final INodesInPath iip = dir.resolvePath(null, src, fileId);
       FileState fileState = 
-          analyzeFileState(src, fileId, clientName, previous, onRetryBlock);
+          analyzeFileState(iip, fileId, clientName, previous, onRetryBlock);
       final INodeFile pendingFile = fileState.inode;
       src = fileState.path;
 
@@ -3186,14 +3189,12 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
     }
   }
 
-  FileState analyzeFileState(String src,
-                                long fileId,
-                                String clientName,
-                                ExtendedBlock previous,
-                                LocatedBlock[] onRetryBlock)
+  private FileState analyzeFileState(
+      INodesInPath iip, long fileId, String clientName,
+      ExtendedBlock previous, LocatedBlock[] onRetryBlock)
           throws IOException  {
     assert hasReadLock();
-
+    String src = iip.getPath();
     checkBlock(previous);
     onRetryBlock[0] = null;
     checkNameNodeSafeMode("Cannot add block to " + src);
@@ -3202,24 +3203,7 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
     checkFsObjectLimit();
 
     Block previousBlock = ExtendedBlock.getLocalBlock(previous);
-    final INode inode;
-    final INodesInPath iip;
-    if (fileId == INodeId.GRANDFATHER_INODE_ID) {
-      // Older clients may not have given us an inode ID to work with.
-      // In this case, we have to try to resolve the path and hope it
-      // hasn't changed or been deleted since the file was opened for write.
-      iip = dir.getINodesInPath4Write(src);
-      inode = iip.getLastINode();
-    } else {
-      // Newer clients pass the inode ID, so we can just get the inode
-      // directly.
-      inode = dir.getInode(fileId);
-      iip = INodesInPath.fromINode(inode);
-      if (inode != null) {
-        src = iip.getPath();
-      }
-    }
-    final INodeFile pendingFile = checkLease(src, clientName, inode, fileId);
+    final INodeFile pendingFile = checkLease(iip, clientName, fileId);
     BlockInfoContiguous lastBlockInFile = pendingFile.getLastBlock();
     if (!Block.matchingIdAndGenStamp(previousBlock, lastBlockInFile)) {
       // The block that the client claims is the current last block
@@ -3317,20 +3301,11 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
       checkOperation(OperationCategory.READ);
       //check safe mode
       checkNameNodeSafeMode("Cannot add datanode; src=" + src + ", blk=" + blk);
-      src = dir.resolvePath(pc, src);
+      final INodesInPath iip = dir.resolvePath(pc, src, fileId);
+      src = iip.getPath();
 
       //check lease
-      final INode inode;
-      if (fileId == INodeId.GRANDFATHER_INODE_ID) {
-        // Older clients may not have given us an inode ID to work with.
-        // In this case, we have to try to resolve the path and hope it
-        // hasn't changed or been deleted since the file was opened for write.
-        inode = dir.getINode(src);
-      } else {
-        inode = dir.getInode(fileId);
-        if (inode != null) src = inode.getFullPathName();
-      }
-      final INodeFile file = checkLease(src, clientName, inode, fileId);
+      final INodeFile file = checkLease(iip, clientName, fileId);
       clientMachine = file.getFileUnderConstructionFeature().getClientMachine();
       clientnode = blockManager.getDatanodeManager().getDatanodeByHost(clientMachine);
       preferredblocksize = file.getPreferredBlockSize();
@@ -3369,27 +3344,12 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
     FSPermissionChecker pc = getPermissionChecker();
     waitForLoadingFSImage();
     writeLock();
+    final INodesInPath iip = dir.resolvePath(pc, src, fileId);
+    src = iip.getPath();
     try {
       checkOperation(OperationCategory.WRITE);
       checkNameNodeSafeMode("Cannot abandon block " + b + " for file" + src);
-      src = dir.resolvePath(pc, src);
-
-      final INode inode;
-      final INodesInPath iip;
-      if (fileId == INodeId.GRANDFATHER_INODE_ID) {
-        // Older clients may not have given us an inode ID to work with.
-        // In this case, we have to try to resolve the path and hope it
-        // hasn't changed or been deleted since the file was opened for write.
-        iip = dir.getINodesInPath(src, true);
-        inode = iip.getLastINode();
-      } else {
-        inode = dir.getInode(fileId);
-        iip = INodesInPath.fromINode(inode);
-        if (inode != null) {
-          src = iip.getPath();
-        }
-      }
-      final INodeFile file = checkLease(src, holder, inode, fileId);
+      final INodeFile file = checkLease(iip, holder, fileId);
 
       // Remove the block from the pending creates list
       boolean removed = dir.removeBlock(src, iip, file,
@@ -3408,8 +3368,10 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
     return true;
   }
 
-  private INodeFile checkLease(String src, String holder, INode inode,
-      long fileId) throws LeaseExpiredException, FileNotFoundException {
+  private INodeFile checkLease(INodesInPath iip, String holder, long fileId)
+      throws LeaseExpiredException, FileNotFoundException {
+    String src = iip.getPath();
+    INode inode = iip.getLastINode();
     assert hasReadLock();
     final String ident = src + " (inode " + fileId + ")";
     if (inode == null) {
@@ -3463,13 +3425,11 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
     checkBlock(last);
     boolean success = false;
     checkOperation(OperationCategory.WRITE);
-    FSPermissionChecker pc = getPermissionChecker();
     waitForLoadingFSImage();
     writeLock();
     try {
       checkOperation(OperationCategory.WRITE);
       checkNameNodeSafeMode("Cannot complete file " + src);
-      src = dir.resolvePath(pc, src);
       success = completeFileInternal(src, holder,
         ExtendedBlock.getLocalBlock(last), fileId);
     } finally {
@@ -3487,23 +3447,13 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
       long fileId) throws IOException {
     assert hasWriteLock();
     final INodeFile pendingFile;
-    final INodesInPath iip;
+    FSPermissionChecker pc = getPermissionChecker();
+    final INodesInPath iip = dir.resolvePath(pc, src, fileId);
+    src = iip.getPath();
     INode inode = null;
     try {
-      if (fileId == INodeId.GRANDFATHER_INODE_ID) {
-        // Older clients may not have given us an inode ID to work with.
-        // In this case, we have to try to resolve the path and hope it
-        // hasn't changed or been deleted since the file was opened for write.
-        iip = dir.getINodesInPath(src, true);
-        inode = iip.getLastINode();
-      } else {
-        inode = dir.getInode(fileId);
-        iip = INodesInPath.fromINode(inode);
-        if (inode != null) {
-          src = iip.getPath();
-        }
-      }
-      pendingFile = checkLease(src, holder, inode, fileId);
+      inode = iip.getLastINode();
+      pendingFile = checkLease(iip, holder, fileId);
     } catch (LeaseExpiredException lee) {
       if (inode != null && inode.isFile() &&
           !inode.asFile().isUnderConstruction()) {
@@ -3967,18 +3917,9 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
     try {
       checkOperation(OperationCategory.WRITE);
       checkNameNodeSafeMode("Cannot fsync file " + src);
-      src = dir.resolvePath(pc, src);
-      final INode inode;
-      if (fileId == INodeId.GRANDFATHER_INODE_ID) {
-        // Older clients may not have given us an inode ID to work with.
-        // In this case, we have to try to resolve the path and hope it
-        // hasn't changed or been deleted since the file was opened for write.
-        inode = dir.getINode(src);
-      } else {
-        inode = dir.getInode(fileId);
-        if (inode != null) src = inode.getFullPathName();
-      }
-      final INodeFile pendingFile = checkLease(src, clientName, inode, fileId);
+      INodesInPath iip = dir.resolvePath(pc, src, fileId);
+      src = iip.getPath();
+      final INodeFile pendingFile = checkLease(iip, clientName, fileId);
       if (lastBlockLength > 0) {
         pendingFile.getFileUnderConstructionFeature().updateLengthOfLastBlock(
             pendingFile, lastBlockLength);
@@ -8021,7 +7962,8 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
       checkSuperuserPrivilege();
       checkOperation(OperationCategory.WRITE);
       checkNameNodeSafeMode("Cannot create encryption zone on " + src);
-      src = dir.resolvePath(pc, src);
+      final INodesInPath iip = dir.resolvePathForWrite(pc, src);
+      src = iip.getPath();
 
       final CipherSuite suite = CipherSuite.convert(cipher);
       // For now this is hardcoded, as we only support one method.
@@ -8032,7 +7974,6 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
       List<XAttr> xAttrs = Lists.newArrayListWithCapacity(1);
       xAttrs.add(ezXAttr);
       getEditLog().logSetXAttrs(src, xAttrs, logRetryCache);
-      final INodesInPath iip = dir.getINodesInPath4Write(src, false);
       resultingStat = dir.getAuditFileInfo(iip);
     } finally {
       writeUnlock();
@@ -8059,8 +8000,7 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
     readLock();
     try {
       checkOperation(OperationCategory.READ);
-      src = dir.resolvePath(pc, src);
-      final INodesInPath iip = dir.getINodesInPath(src, true);
+      INodesInPath iip = dir.resolvePath(pc, src);
       if (isPermissionEnabled) {
         dir.checkPathAccess(pc, iip, FsAction.READ);
       }
@@ -8161,17 +8101,17 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
 
   void checkAccess(String src, FsAction mode) throws IOException {
     checkOperation(OperationCategory.READ);
+    FSPermissionChecker pc = getPermissionChecker();
     readLock();
     try {
       checkOperation(OperationCategory.READ);
-      src = FSDirectory.resolvePath(src, dir);
-      final INodesInPath iip = dir.getINodesInPath(src, true);
+      final INodesInPath iip = dir.resolvePath(pc, src);
+      src = iip.getPath();
       INode inode = iip.getLastINode();
       if (inode == null) {
         throw new FileNotFoundException("Path not found");
       }
       if (isPermissionEnabled) {
-        FSPermissionChecker pc = getPermissionChecker();
         dir.checkPathAccess(pc, iip, mode);
       }
     } catch (AccessControlException e) {

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/INodesInPath.java

@@ -72,6 +72,10 @@ public class INodesInPath {
     return new INodesInPath(inodes, path);
   }
 
+  static INodesInPath fromComponents(byte[][] components) {
+    return new INodesInPath(new INode[components.length], components);
+  }
+
   /**
    * Given some components, create a path name.
    * @param components The path components
@@ -433,6 +437,10 @@ public class INodesInPath {
     return this.isSnapshot;
   }
 
+  boolean isDotSnapshotDir() {
+    return isDotSnapshotDir(getLastLocalName());
+  }
+
   private static String toString(INode inode) {
     return inode == null? null: inode.getLocalName();
   }