HDFS-17839. Rename skips path without valid 'DirectoryWithQuotaFeature' in FSDirRenameOp

Author: huangzhaobo99

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

@@ -224,7 +224,7 @@ private static INodesInPath unprotectedMkdir(FSDirectory fsd, long inodeId,
         timestamp);
 
     INodesInPath iip = fsd.addLastINode(parent, dir, permission.getPermission(),
-        true, Optional.empty());
+        true, Optional.empty(), true);
     if (iip != null && aclEntries != null) {
       AclStorage.updateINodeAcl(dir, aclEntries, Snapshot.CURRENT_STATE_ID);
     }

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

@@ -17,7 +17,7 @@
  */
 package org.apache.hadoop.hdfs.server.namenode;
 
-import org.apache.commons.lang3.tuple.Pair;
+import org.apache.commons.lang3.tuple.Triple;
 import org.apache.hadoop.hdfs.protocol.HdfsConstants;
 import org.apache.hadoop.util.Preconditions;
 import org.apache.hadoop.fs.FileAlreadyExistsException;
@@ -72,18 +72,28 @@ static RenameResult renameToInt(
    * Verify quota for rename operation where srcInodes[srcInodes.length-1] moves
    * dstInodes[dstInodes.length-1]
    */
-  private static Pair<Optional<QuotaCounts>, Optional<QuotaCounts>> verifyQuotaForRename(
-      FSDirectory fsd, INodesInPath src, INodesInPath dst) throws QuotaExceededException {
+  private static Triple<Boolean, Optional<QuotaCounts>, Optional<QuotaCounts>> verifyQuotaForRename(
+      FSDirectory fsd, INodesInPath src, INodesInPath dst, boolean overwrite)
+      throws QuotaExceededException {
     Optional<QuotaCounts> srcDelta = Optional.empty();
     Optional<QuotaCounts> dstDelta = Optional.empty();
     if (!fsd.getFSNamesystem().isImageLoaded() || fsd.shouldSkipQuotaChecks()) {
       // Do not check quota if edits log is still being processed
-      return Pair.of(srcDelta, dstDelta);
+      return Triple.of(false, srcDelta, dstDelta);
     }
     int i = 0;
     while (src.getINode(i) == dst.getINode(i)) {
       i++;
     }
+
+    // Verify path without valid 'DirectoryWithQuotaFeature'
+    // Note: In overwrite scenarios, quota calculation is still required,
+    // overwrite operations delete existing content, which will affects the root directory's quota.
+    // (i - 1) is common ancestor inode index.
+    if (!overwrite && verifyPathWithoutValidQuotaFeature(src, dst, i - 1)) {
+      return Triple.of(false, srcDelta, dstDelta);
+    }
+
     // src[i - 1] is the last common ancestor.
     BlockStoragePolicySuite bsps = fsd.getBlockStoragePolicySuite();
     // Assume dstParent existence check done by callers.
@@ -108,7 +118,7 @@ private static Pair<Optional<QuotaCounts>, Optional<QuotaCounts>> verifyQuotaFor
       delta.subtract(counts);
     }
     FSDirectory.verifyQuota(dst, dst.length() - 1, delta, src.getINode(i - 1));
-    return Pair.of(srcDelta, dstDelta);
+    return Triple.of(true, srcDelta, dstDelta);
   }
 
   /**
@@ -127,6 +137,25 @@ static void verifyFsLimitsForRename(FSDirectory fsd, INodesInPath srcIIP,
     }
   }
 
+  /**
+   * Verify that the src and dst path does not contain valid quota feature.
+   *
+   * @param src   source path.
+   * @param dst   destination path.
+   * @param index common ancestor inode index.
+   * @return true if no valid quota feature, otherwise false.
+   */
+  static boolean verifyPathWithoutValidQuotaFeature(INodesInPath src, INodesInPath dst, int index) {
+    // Excluding root directory
+    if (!FSDirectory.verifyWithoutQuotaFeature(src, index, 1)) {
+      return false;
+    }
+    if (!FSDirectory.verifyWithoutQuotaFeature(src, src.length() - 2, index - 1)) {
+      return false;
+    }
+    return FSDirectory.verifyWithoutQuotaFeature(dst, dst.length() - 2, index - 1);
+  }
+
   /**
    * <br>
    * Note: This is to be used by {@link FSEditLogLoader} only.
@@ -216,10 +245,10 @@ static INodesInPath unprotectedRenameTo(FSDirectory fsd,
     fsd.ezManager.checkMoveValidity(srcIIP, dstIIP);
     // Ensure dst has quota to accommodate rename
     verifyFsLimitsForRename(fsd, srcIIP, dstIIP);
-    Pair<Optional<QuotaCounts>, Optional<QuotaCounts>> countPair =
-        verifyQuotaForRename(fsd, srcIIP, dstIIP);
+    Triple<Boolean, Optional<QuotaCounts>, Optional<QuotaCounts>> countTriple =
+        verifyQuotaForRename(fsd, srcIIP, dstIIP, false);
 
-    RenameOperation tx = new RenameOperation(fsd, srcIIP, dstIIP, countPair);
+    RenameOperation tx = new RenameOperation(fsd, srcIIP, dstIIP, countTriple);
 
     boolean added = false;
 
@@ -436,10 +465,10 @@ static RenameResult unprotectedRenameTo(FSDirectory fsd,
 
     // Ensure dst has quota to accommodate rename
     verifyFsLimitsForRename(fsd, srcIIP, dstIIP);
-    Pair<Optional<QuotaCounts>, Optional<QuotaCounts>> quotaPair =
-        verifyQuotaForRename(fsd, srcIIP, dstIIP);
+    Triple<Boolean, Optional<QuotaCounts>, Optional<QuotaCounts>> countTriple =
+        verifyQuotaForRename(fsd, srcIIP, dstIIP, overwrite);
 
-    RenameOperation tx = new RenameOperation(fsd, srcIIP, dstIIP, quotaPair);
+    RenameOperation tx = new RenameOperation(fsd, srcIIP, dstIIP, countTriple);
 
     boolean undoRemoveSrc = true;
     tx.removeSrc();
@@ -656,13 +685,14 @@ private static class RenameOperation {
     private final boolean srcChildIsReference;
     private final QuotaCounts oldSrcCountsInSnapshot;
     private final boolean sameStoragePolicy;
+    private final boolean updateQuota;
     private final Optional<QuotaCounts> srcSubTreeCount;
     private final Optional<QuotaCounts> dstSubTreeCount;
     private INode srcChild;
     private INode oldDstChild;
 
     RenameOperation(FSDirectory fsd, INodesInPath srcIIP, INodesInPath dstIIP,
-        Pair<Optional<QuotaCounts>, Optional<QuotaCounts>> quotaPair) {
+        Triple<Boolean, Optional<QuotaCounts>, Optional<QuotaCounts>> countTriple) {
       this.fsd = fsd;
       this.srcIIP = srcIIP;
       this.dstIIP = dstIIP;
@@ -712,9 +742,10 @@ private static class RenameOperation {
         withCount = null;
       }
       // Set quota for src and dst, ignore src is in Snapshot or is Reference
+      this.updateQuota = countTriple.getLeft() || withCount != null;
       this.srcSubTreeCount = withCount == null ?
-          quotaPair.getLeft() : Optional.empty();
-      this.dstSubTreeCount = quotaPair.getRight();
+          countTriple.getMiddle() : Optional.empty();
+      this.dstSubTreeCount = countTriple.getRight();
     }
 
     boolean isSameStoragePolicy() {
@@ -755,9 +786,11 @@ long removeSrc() throws IOException {
         throw new IOException(error);
       } else {
         // update the quota count if necessary
-        Optional<QuotaCounts> countOp = sameStoragePolicy ?
-            srcSubTreeCount : Optional.empty();
-        fsd.updateCountForDelete(srcChild, srcIIP, countOp);
+        if (updateQuota) {
+          Optional<QuotaCounts> countOp = sameStoragePolicy ?
+              srcSubTreeCount : Optional.empty();
+          fsd.updateCountForDelete(srcChild, srcIIP, countOp);
+        }
         srcIIP = INodesInPath.replace(srcIIP, srcIIP.length() - 1, null);
         return removedNum;
       }
@@ -772,9 +805,11 @@ boolean removeSrc4OldRename() {
         return false;
       } else {
         // update the quota count if necessary
-        Optional<QuotaCounts> countOp = sameStoragePolicy ?
-            srcSubTreeCount : Optional.empty();
-        fsd.updateCountForDelete(srcChild, srcIIP, countOp);
+        if (updateQuota) {
+          Optional<QuotaCounts> countOp = sameStoragePolicy ?
+              srcSubTreeCount : Optional.empty();
+          fsd.updateCountForDelete(srcChild, srcIIP, countOp);
+        }
         srcIIP = INodesInPath.replace(srcIIP, srcIIP.length() - 1, null);
         return true;
       }
@@ -785,7 +820,9 @@ long removeDst() {
       if (removedNum != -1) {
         oldDstChild = dstIIP.getLastINode();
         // update the quota count if necessary
-        fsd.updateCountForDelete(oldDstChild, dstIIP, dstSubTreeCount);
+        if (updateQuota) {
+          fsd.updateCountForDelete(oldDstChild, dstIIP, dstSubTreeCount);
+        }
         dstIIP = INodesInPath.replace(dstIIP, dstIIP.length() - 1, null);
       }
       return removedNum;
@@ -803,7 +840,7 @@ INodesInPath addSourceToDestination() {
         toDst = new INodeReference.DstReference(dstParent.asDirectory(),
             withCount, dstIIP.getLatestSnapshotId());
       }
-      return fsd.addLastINodeNoQuotaCheck(dstParentIIP, toDst, srcSubTreeCount);
+      return fsd.addLastINodeNoQuotaCheck(dstParentIIP, toDst, srcSubTreeCount, updateQuota);
     }
 
     void updateMtimeAndLease(long timestamp) {
@@ -837,7 +874,7 @@ void restoreSource() {
         // the srcChild back
         Optional<QuotaCounts> countOp = sameStoragePolicy ?
             srcSubTreeCount : Optional.empty();
-        fsd.addLastINodeNoQuotaCheck(srcParentIIP, srcChild, countOp);
+        fsd.addLastINodeNoQuotaCheck(srcParentIIP, srcChild, countOp, updateQuota);
       }
     }
 
@@ -847,7 +884,7 @@ void restoreDst(BlockStoragePolicySuite bsps) {
       if (dstParent.isWithSnapshot()) {
         dstParent.undoRename4DstParent(bsps, oldDstChild, dstIIP.getLatestSnapshotId());
       } else {
-        fsd.addLastINodeNoQuotaCheck(dstParentIIP, oldDstChild, dstSubTreeCount);
+        fsd.addLastINodeNoQuotaCheck(dstParentIIP, oldDstChild, dstSubTreeCount, updateQuota);
       }
       if (oldDstChild != null && oldDstChild.isReference()) {
         final INodeReference removedDstRef = oldDstChild.asReference();

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

@@ -1196,7 +1196,7 @@ INodesInPath addINode(INodesInPath existing, INode child,
     cacheName(child);
     writeLock();
     try {
-      return addLastINode(existing, child, modes, true, Optional.empty());
+      return addLastINode(existing, child, modes, true, Optional.empty(), true);
     } finally {
       writeUnlock();
     }
@@ -1242,6 +1242,25 @@ static void verifyQuota(INodesInPath iip, int pos, QuotaCounts deltas,
     }
   }
 
+  /**
+   * Verify that the iNodesInPath from the start position to the end position
+   * does not contain any valid quota feature.
+   *
+   * @param iip   the INodesInPath instance containing all the INodes for the path.
+   * @param start path inode start index.
+   * @param end   path inode end index.
+   * @return true if no valid quota feature, otherwise false.
+   */
+  static boolean verifyWithoutQuotaFeature(INodesInPath iip, int start, int end) {
+    for (int i = start; i >= end; i--) {
+      INodeDirectory d = iip.getINode(i).asDirectory();
+      if (d.isWithQuota()) {
+        return false;
+      }
+    }
+    return true;
+  }
+
   /** Verify if the inode name is legal. */
   void verifyINodeName(byte[] childName) throws HadoopIllegalArgumentException {
     if (Arrays.equals(HdfsServerConstants.DOT_SNAPSHOT_DIR_BYTES, childName)) {
@@ -1343,11 +1362,12 @@ private void copyINodeDefaultAcl(INode child, FsPermission modes) {
    * @param inode the new INode to add
    * @param modes create modes
    * @param checkQuota whether to check quota
+   * @param updateQuota whether to update quota
    * @return an INodesInPath instance containing the new INode
    */
   @VisibleForTesting
-  public INodesInPath addLastINode(INodesInPath existing, INode inode,
-      FsPermission modes, boolean checkQuota, Optional<QuotaCounts> quotaCount)
+  public INodesInPath addLastINode(INodesInPath existing, INode inode, FsPermission modes,
+      boolean checkQuota, Optional<QuotaCounts> quotaCount, boolean updateQuota)
       throws QuotaExceededException {
     assert existing.getLastINode() != null &&
         existing.getLastINode().isDirectory();
@@ -1379,17 +1399,23 @@ public INodesInPath addLastINode(INodesInPath existing, INode inode,
     // always verify inode name
     verifyINodeName(inode.getLocalNameBytes());
 
-    final QuotaCounts counts = quotaCount.orElseGet(() -> inode.
-        computeQuotaUsage(getBlockStoragePolicySuite(),
-            parent.getStoragePolicyID(), false,
-            Snapshot.CURRENT_STATE_ID));
-    updateCount(existing, pos, counts, checkQuota);
+    QuotaCounts counts = null;
+    if (updateQuota) {
+      counts = quotaCount.orElseGet(() -> inode.
+          computeQuotaUsage(getBlockStoragePolicySuite(),
+          parent.getStoragePolicyID(), false,
+          Snapshot.CURRENT_STATE_ID));
+      updateCount(existing, pos, counts, checkQuota);
+    }
 
     boolean isRename = (inode.getParent() != null);
     final boolean added = parent.addChild(inode, true,
         existing.getLatestSnapshotId());
     if (!added) {
-      updateCountNoQuotaCheck(existing, pos, counts.negation());
+      // When adding INode fails, if 'updateQuota' is true, rollback quota.
+      if (updateQuota) {
+        updateCountNoQuotaCheck(existing, pos, counts.negation());
+      }
       return null;
     } else {
       if (!isRename) {
@@ -1401,10 +1427,10 @@ public INodesInPath addLastINode(INodesInPath existing, INode inode,
   }
 
   INodesInPath addLastINodeNoQuotaCheck(INodesInPath existing, INode i,
-      Optional<QuotaCounts> quotaCount) {
+      Optional<QuotaCounts> quotaCount, boolean updateQuota) {
     try {
       // All callers do not have create modes to pass.
-      return addLastINode(existing, i, null, false, quotaCount);
+      return addLastINode(existing, i, null, false, quotaCount, updateQuota);
     } catch (QuotaExceededException e) {
       NameNode.LOG.warn("FSDirectory.addChildNoQuotaCheck - unexpected", e);
     }