HDFS-9621. Consolidate FSDirStatAndListingOp#createFileStatus to let its INodesInPath parameter always include the target INode. Contributed by Jing Zhao.

(cherry picked from commit 313f03bfdab32cf365bc3470c5f9b6928a24f099)

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

(cherry picked from commit c2708bc5fbf212ae6c48084442748c3265ddccd7)

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

@@ -171,12 +171,14 @@ class FSDirStatAndListingOp {
           .ID_UNSPECIFIED;
 
       if (!targetNode.isDirectory()) {
+        // return the file's status. note that the iip already includes the
+        // target INode
         INodeAttributes nodeAttrs = getINodeAttributes(
             fsd, src, HdfsFileStatus.EMPTY_NAME, targetNode,
             snapshot);
         return new DirectoryListing(
             new HdfsFileStatus[]{ createFileStatus(
-                fsd, HdfsFileStatus.EMPTY_NAME, targetNode, nodeAttrs,
+                fsd, HdfsFileStatus.EMPTY_NAME, nodeAttrs,
                 needLocation, parentStoragePolicy, snapshot, isRawPath, iip)
             }, 0);
       }
@@ -190,7 +192,7 @@ class FSDirStatAndListingOp {
       int locationBudget = fsd.getLsLimit();
       int listingCnt = 0;
       HdfsFileStatus listing[] = new HdfsFileStatus[numOfListing];
-      for (int i=0; i<numOfListing && locationBudget>0; i++) {
+      for (int i = 0; i < numOfListing && locationBudget > 0; i++) {
         INode cur = contents.get(startChild+i);
         byte curPolicy = isSuperUser && !cur.isSymlink()?
             cur.getLocalStoragePolicyID():
@@ -198,9 +200,11 @@ class FSDirStatAndListingOp {
         INodeAttributes nodeAttrs = getINodeAttributes(
             fsd, src, cur.getLocalNameBytes(), cur,
             snapshot);
-        listing[i] = createFileStatus(fsd, cur.getLocalNameBytes(),
-            cur, nodeAttrs, needLocation, getStoragePolicyID(curPolicy,
-                parentStoragePolicy), snapshot, isRawPath, iip);
+        final INodesInPath iipWithChild = INodesInPath.append(iip, cur,
+            cur.getLocalNameBytes());
+        listing[i] = createFileStatus(fsd, cur.getLocalNameBytes(), nodeAttrs,
+            needLocation, getStoragePolicyID(curPolicy, parentStoragePolicy),
+            snapshot, isRawPath, iipWithChild);
         listingCnt++;
         if (needLocation) {
             // Once we  hit lsLimit locations, stop.
@@ -255,9 +259,8 @@ class FSDirStatAndListingOp {
           fsd, src, sRoot.getLocalNameBytes(),
           node, Snapshot.CURRENT_STATE_ID);
       listing[i] = createFileStatus(
-          fsd, sRoot.getLocalNameBytes(),
-          sRoot, nodeAttrs,
-          BlockStoragePolicySuite.ID_UNSPECIFIED,
+          fsd, sRoot.getLocalNameBytes(), nodeAttrs,
+              BlockStoragePolicySuite.ID_UNSPECIFIED,
           Snapshot.CURRENT_STATE_ID, false,
           INodesInPath.fromINode(sRoot));
     }
@@ -267,32 +270,31 @@ class FSDirStatAndListingOp {
 
   /** Get the file info for a specific file.
    * @param fsd FSDirectory
-   * @param src The string representation of the path to the file
+   * @param iip 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 src, boolean isRawPath,
+      FSDirectory fsd, String path, INodesInPath iip, boolean isRawPath,
       boolean includeStoragePolicy)
       throws IOException {
     fsd.readLock();
     try {
-      final INode i = src.getLastINode();
-      if (i == null) {
+      final INode node = iip.getLastINode();
+      if (node == null) {
         return null;
       }
 
-      byte policyId = includeStoragePolicy && !i.isSymlink() ?
-          i.getStoragePolicyID() : BlockStoragePolicySuite.ID_UNSPECIFIED;
-      INodeAttributes nodeAttrs = getINodeAttributes(
-          fsd, path, HdfsFileStatus.EMPTY_NAME, i, src.getPathSnapshotId());
-      return createFileStatus(
-          fsd, HdfsFileStatus.EMPTY_NAME,
-          i, nodeAttrs, policyId,
-          src.getPathSnapshotId(),
-          isRawPath, src);
+      byte policyId = includeStoragePolicy && !node.isSymlink() ?
+          node.getStoragePolicyID() :
+          BlockStoragePolicySuite.ID_UNSPECIFIED;
+      INodeAttributes nodeAttrs = getINodeAttributes(fsd, path,
+                                                     HdfsFileStatus.EMPTY_NAME,
+                                                     node, iip.getPathSnapshotId());
+      return createFileStatus(fsd, HdfsFileStatus.EMPTY_NAME, nodeAttrs,
+                              policyId, iip.getPathSnapshotId(), isRawPath, iip);
     } finally {
       fsd.readUnlock();
     }
@@ -326,51 +328,54 @@ class FSDirStatAndListingOp {
    *
    * @param fsd FSDirectory
    * @param path the local name
-   * @param node inode
    * @param needLocation if block locations need to be included or not
    * @param isRawPath true if this is being called on behalf of a path in
    *                  /.reserved/raw
+   * @param iip the INodesInPath containing the target INode and its ancestors
    * @return a file status
    * @throws java.io.IOException if any error occurs
    */
   private static HdfsFileStatus createFileStatus(
-      FSDirectory fsd, byte[] path, INode node, INodeAttributes nodeAttrs,
+      FSDirectory fsd, byte[] path, INodeAttributes nodeAttrs,
       boolean needLocation, byte storagePolicy, int snapshot, boolean isRawPath,
       INodesInPath iip)
       throws IOException {
     if (needLocation) {
-      return createLocatedFileStatus(fsd, path, node, nodeAttrs, storagePolicy,
+      return createLocatedFileStatus(fsd, path, nodeAttrs, storagePolicy,
                                      snapshot, isRawPath, iip);
     } else {
-      return createFileStatus(fsd, path, node, nodeAttrs, storagePolicy,
+      return createFileStatus(fsd, path, nodeAttrs, storagePolicy,
                               snapshot, isRawPath, iip);
     }
   }
 
   /**
-   * Create FileStatus by file INode
+   * Create FileStatus for an given INodeFile.
+   * @param iip The INodesInPath containing the INodeFile and its ancestors
    */
   static HdfsFileStatus createFileStatusForEditLog(
-      FSDirectory fsd, String fullPath, byte[] path, INode node,
+      FSDirectory fsd, String fullPath, byte[] path,
       byte storagePolicy, int snapshot, boolean isRawPath,
       INodesInPath iip) throws IOException {
     INodeAttributes nodeAttrs = getINodeAttributes(
-        fsd, fullPath, path, node, snapshot);
-    return createFileStatus(fsd, path, node, nodeAttrs,
-                            storagePolicy, snapshot, isRawPath, iip);
+        fsd, fullPath, path, iip.getLastINode(), snapshot);
+    return createFileStatus(fsd, path, nodeAttrs, storagePolicy,
+                            snapshot, isRawPath, iip);
   }
 
   /**
-   * Create FileStatus by file INode
+   * create file status for a given INode
+   * @param iip the INodesInPath containing the target INode and its ancestors
    */
   static HdfsFileStatus createFileStatus(
-      FSDirectory fsd, byte[] path, INode node,
+      FSDirectory fsd, byte[] path,
       INodeAttributes nodeAttrs, byte storagePolicy, int snapshot,
       boolean isRawPath, INodesInPath iip) throws IOException {
     long size = 0;     // length is zero for directories
     short replication = 0;
     long blocksize = 0;
     final boolean isEncrypted;
+    final INode node = iip.getLastINode();
 
     final FileEncryptionInfo feInfo = isRawPath ? null :
         fsd.getFileEncryptionInfo(node, snapshot, iip);
@@ -381,9 +386,9 @@ class FSDirStatAndListingOp {
       replication = fileNode.getFileReplication(snapshot);
       blocksize = fileNode.getPreferredBlockSize();
       isEncrypted = (feInfo != null) ||
-          (isRawPath && fsd.isInAnEZ(INodesInPath.fromINode(node)));
+          (isRawPath && fsd.isInAnEZ(iip));
     } else {
-      isEncrypted = fsd.isInAnEZ(INodesInPath.fromINode(node));
+      isEncrypted = fsd.isInAnEZ(iip);
     }
 
     int childrenNum = node.isDirectory() ?
@@ -414,9 +419,10 @@ class FSDirStatAndListingOp {
 
   /**
    * Create FileStatus with location info by file INode
+   * @param iip the INodesInPath containing the target INode and its ancestors
    */
   private static HdfsLocatedFileStatus createLocatedFileStatus(
-      FSDirectory fsd, byte[] path, INode node, INodeAttributes nodeAttrs,
+      FSDirectory fsd, byte[] path, INodeAttributes nodeAttrs,
       byte storagePolicy, int snapshot,
       boolean isRawPath, INodesInPath iip) throws IOException {
     assert fsd.hasReadLock();
@@ -425,6 +431,8 @@ class FSDirStatAndListingOp {
     long blocksize = 0;
     LocatedBlocks loc = null;
     final boolean isEncrypted;
+    final INode node = iip.getLastINode();
+
     final FileEncryptionInfo feInfo = isRawPath ? null :
         fsd.getFileEncryptionInfo(node, snapshot, iip);
     if (node.isFile()) {
@@ -445,9 +453,9 @@ class FSDirStatAndListingOp {
         loc = new LocatedBlocks();
       }
       isEncrypted = (feInfo != null) ||
-          (isRawPath && fsd.isInAnEZ(INodesInPath.fromINode(node)));
+          (isRawPath && fsd.isInAnEZ(iip));
     } else {
-      isEncrypted = fsd.isInAnEZ(INodesInPath.fromINode(node));
+      isEncrypted = fsd.isInAnEZ(iip);
     }
     int childrenNum = node.isDirectory() ?
         node.asDirectory().getChildrenNum(snapshot) : 0;

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

@@ -1722,8 +1722,8 @@ public class FSDirectory implements Closeable {
       INode node, int snapshot) {
     INodeAttributes nodeAttrs = node.getSnapshotINode(snapshot);
     if (attributeProvider != null) {
-      fullPath = fullPath + (fullPath.endsWith(Path.SEPARATOR) ? ""
-                                                               : Path.SEPARATOR)
+      fullPath = fullPath
+          + (fullPath.endsWith(Path.SEPARATOR) ? "" : Path.SEPARATOR)
           + DFSUtil.bytes2String(path);
       nodeAttrs = attributeProvider.getAttributes(fullPath, nodeAttrs);
     }

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

@@ -374,14 +374,15 @@ public class FSEditLogLoader {
             addCloseOp.clientName,
             addCloseOp.clientMachine,
             addCloseOp.storagePolicyId);
+        assert newFile != null;
         iip = INodesInPath.replace(iip, iip.length() - 1, newFile);
         fsNamesys.leaseManager.addLease(addCloseOp.clientName, path);
 
         // add the op into retry cache if necessary
         if (toAddRetryCache) {
           HdfsFileStatus stat = FSDirStatAndListingOp.createFileStatusForEditLog(
-              fsNamesys.dir, path, HdfsFileStatus.EMPTY_NAME, newFile,
-              BlockStoragePolicySuite.ID_UNSPECIFIED, Snapshot.CURRENT_STATE_ID,
+              fsNamesys.dir, path, HdfsFileStatus.EMPTY_NAME,
+                  BlockStoragePolicySuite.ID_UNSPECIFIED, Snapshot.CURRENT_STATE_ID,
               false, iip);
           fsNamesys.addCacheEntryWithPayload(addCloseOp.rpcClientId,
               addCloseOp.rpcCallId, stat);
@@ -399,9 +400,8 @@ public class FSEditLogLoader {
           // add the op into retry cache if necessary
           if (toAddRetryCache) {
             HdfsFileStatus stat = FSDirStatAndListingOp.createFileStatusForEditLog(
-                fsNamesys.dir, path,
-                HdfsFileStatus.EMPTY_NAME, newFile,
-                BlockStoragePolicySuite.ID_UNSPECIFIED,
+                fsNamesys.dir, path, HdfsFileStatus.EMPTY_NAME,
+                    BlockStoragePolicySuite.ID_UNSPECIFIED,
                 Snapshot.CURRENT_STATE_ID, false, iip);
             fsNamesys.addCacheEntryWithPayload(addCloseOp.rpcClientId,
                 addCloseOp.rpcCallId, new LastBlockWithStatus(lb, stat));
@@ -473,8 +473,8 @@ public class FSEditLogLoader {
         // add the op into retry cache if necessary
         if (toAddRetryCache) {
           HdfsFileStatus stat = FSDirStatAndListingOp.createFileStatusForEditLog(
-              fsNamesys.dir, path, HdfsFileStatus.EMPTY_NAME, file,
-              BlockStoragePolicySuite.ID_UNSPECIFIED,
+              fsNamesys.dir, path, HdfsFileStatus.EMPTY_NAME,
+                  BlockStoragePolicySuite.ID_UNSPECIFIED,
               Snapshot.CURRENT_STATE_ID, false, iip);
           fsNamesys.addCacheEntryWithPayload(appendOp.rpcClientId,
               appendOp.rpcCallId, new LastBlockWithStatus(lb, stat));

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

@@ -253,7 +253,7 @@ public class INodesInPath {
    */
   public static INodesInPath append(INodesInPath iip, INode child,
       byte[] childName) {
-    Preconditions.checkArgument(!iip.isSnapshot && iip.length() > 0);
+    Preconditions.checkArgument(iip.length() > 0);
     Preconditions.checkArgument(iip.getLastINode() != null && iip
         .getLastINode().isDirectory());
     INode[] inodes = new INode[iip.length() + 1];
@@ -262,7 +262,7 @@ public class INodesInPath {
     byte[][] path = new byte[iip.path.length + 1][];
     System.arraycopy(iip.path, 0, path, 0, path.length - 1);
     path[path.length - 1] = childName;
-    return new INodesInPath(inodes, path, false, iip.snapshotId);
+    return new INodesInPath(inodes, path, iip.isSnapshot, iip.snapshotId);
   }
 
   private final byte[][] path;