HDFS-6562. Refactor rename() in FSDirectory. Contributed by Haohui Mai.

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

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

@@ -467,6 +467,8 @@ Release 2.5.0 - UNRELEASED
     HDFS-6578. add toString method to DatanodeStorage for easier debugging.
     (Yongjun Zhang via Arpit Agarwal)
 
+    HDFS-6562. Refactor rename() in FSDirectory. (wheat9)
+
   OPTIMIZATIONS
 
     HDFS-6214. Webhdfs has poor throughput for files >2GB (daryn)

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

@@ -67,7 +67,6 @@ import org.apache.hadoop.hdfs.protocol.SnapshotException;
 import org.apache.hadoop.hdfs.server.blockmanagement.BlockInfo;
 import org.apache.hadoop.hdfs.server.blockmanagement.BlockInfoUnderConstruction;
 import org.apache.hadoop.hdfs.server.blockmanagement.BlockManager;
-import org.apache.hadoop.hdfs.server.blockmanagement.DatanodeDescriptor;
 import org.apache.hadoop.hdfs.server.blockmanagement.DatanodeStorageInfo;
 import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.BlockUCState;
 import org.apache.hadoop.hdfs.server.namenode.INode.BlocksMapUpdateInfo;
@@ -469,55 +468,30 @@ public class FSDirectory implements Closeable {
     assert hasWriteLock();
     INodesInPath srcIIP = getINodesInPath4Write(src, false);
     final INode srcInode = srcIIP.getLastINode();
-    
+    try {
-    // check the validation of the source
+      validateRenameSource(src, srcIIP);
-    if (srcInode == null) {
+    } catch (SnapshotException e) {
-      NameNode.stateChangeLog.warn("DIR* FSDirectory.unprotectedRenameTo: "
+      throw e;
-          + "failed to rename " + src + " to " + dst
+    } catch (IOException ignored) {
-          + " because source does not exist");
       return false;
     }
-    if (srcIIP.getINodes().length == 1) {
-      NameNode.stateChangeLog.warn("DIR* FSDirectory.unprotectedRenameTo: "
-          +"failed to rename "+src+" to "+dst+ " because source is the root");
-      return false;
-    }
-    
-    // srcInode and its subtree cannot contain snapshottable directories with
-    // snapshots
-    List<INodeDirectorySnapshottable> snapshottableDirs = 
-        new ArrayList<INodeDirectorySnapshottable>();
-    checkSnapshot(srcInode, snapshottableDirs);
 
     if (isDir(dst)) {
       dst += Path.SEPARATOR + new Path(src).getName();
     }
 
-    // check the validity of the destination
+    // validate the destination
     if (dst.equals(src)) {
       return true;
     }
-    if (srcInode.isSymlink() && 
-        dst.equals(srcInode.asSymlink().getSymlinkString())) {
-      throw new FileAlreadyExistsException(
-          "Cannot rename symlink "+src+" to its target "+dst);
-    }
 
-    // dst cannot be directory or a file under src
+    try {
-    if (dst.startsWith(src) && 
+      validateRenameDestination(src, dst, srcInode);
-        dst.charAt(src.length()) == Path.SEPARATOR_CHAR) {
+    } catch (IOException ignored) {
-      NameNode.stateChangeLog.warn("DIR* FSDirectory.unprotectedRenameTo: "
-          + "failed to rename " + src + " to " + dst
-          + " because destination starts with src");
       return false;
     }
 
-    byte[][] dstComponents = INode.getPathComponents(dst);
+    INodesInPath dstIIP = getINodesInPath4Write(dst, false);
-    INodesInPath dstIIP = getExistingPathINodes(dstComponents);
-    if (dstIIP.isSnapshot()) {
-      throw new SnapshotAccessControlException(
-          "Modification on RO snapshot is disallowed");
-    }
     if (dstIIP.getLastINode() != null) {
       NameNode.stateChangeLog.warn("DIR* FSDirectory.unprotectedRenameTo: "
                                    +"failed to rename "+src+" to "+dst+ 
@@ -536,41 +510,9 @@ public class FSDirectory implements Closeable {
     verifyFsLimitsForRename(srcIIP, dstIIP);
     verifyQuotaForRename(srcIIP.getINodes(), dstIIP.getINodes());
 
+    RenameOperation tx = new RenameOperation(src, dst, srcIIP, dstIIP);
+
     boolean added = false;
-    INode srcChild = srcIIP.getLastINode();
-    final byte[] srcChildName = srcChild.getLocalNameBytes();
-    final boolean isSrcInSnapshot = srcChild.isInLatestSnapshot(
-        srcIIP.getLatestSnapshotId());
-    final boolean srcChildIsReference = srcChild.isReference();
-    
-    // Record the snapshot on srcChild. After the rename, before any new 
-    // snapshot is taken on the dst tree, changes will be recorded in the latest
-    // snapshot of the src tree.
-    if (isSrcInSnapshot) {
-      srcChild = srcChild.recordModification(srcIIP.getLatestSnapshotId());
-      srcIIP.setLastINode(srcChild);
-    }
-    
-    // check srcChild for reference
-    final INodeReference.WithCount withCount;
-    Quota.Counts oldSrcCounts = Quota.Counts.newInstance();
-    int srcRefDstSnapshot = srcChildIsReference ? srcChild.asReference()
-        .getDstSnapshotId() : Snapshot.CURRENT_STATE_ID;
-    if (isSrcInSnapshot) {
-      final INodeReference.WithName withName = 
-          srcIIP.getINode(-2).asDirectory().replaceChild4ReferenceWithName(
-              srcChild, srcIIP.getLatestSnapshotId()); 
-      withCount = (INodeReference.WithCount) withName.getReferredINode();
-      srcChild = withName;
-      srcIIP.setLastINode(srcChild);
-      // get the counts before rename
-      withCount.getReferredINode().computeQuotaUsage(oldSrcCounts, true);
-    } else if (srcChildIsReference) {
-      // srcChild is reference but srcChild is not in latest snapshot
-      withCount = (WithCount) srcChild.asReference().getReferredINode();
-    } else {
-      withCount = null;
-    }
 
     try {
       // remove src
@@ -582,83 +524,21 @@ public class FSDirectory implements Closeable {
         return false;
       }
 
-      if (dstParent.getParent() == null) {
+      added = tx.addSourceToDestination();
-        // src and dst file/dir are in the same directory, and the dstParent has
-        // been replaced when we removed the src. Refresh the dstIIP and
-        // dstParent.
-        dstIIP = getExistingPathINodes(dstComponents);
-        dstParent = dstIIP.getINode(-2);
-      }
-      
-      // add src to the destination
-      
-      srcChild = srcIIP.getLastINode();
-      final byte[] dstChildName = dstIIP.getLastLocalName();
-      final INode toDst;
-      if (withCount == null) {
-        srcChild.setLocalName(dstChildName);
-        toDst = srcChild;
-      } else {
-        withCount.getReferredINode().setLocalName(dstChildName);
-        int dstSnapshotId = dstIIP.getLatestSnapshotId();
-        toDst = new INodeReference.DstReference(
-            dstParent.asDirectory(), withCount, dstSnapshotId);
-      }
-      
-      added = addLastINodeNoQuotaCheck(dstIIP, toDst);
       if (added) {
         if (NameNode.stateChangeLog.isDebugEnabled()) {
           NameNode.stateChangeLog.debug("DIR* FSDirectory.unprotectedRenameTo: " 
               + src + " is renamed to " + dst);
         }
-        // update modification time of dst and the parent of src
-        final INode srcParent = srcIIP.getINode(-2);
-        srcParent.updateModificationTime(timestamp, srcIIP.getLatestSnapshotId());
-        dstParent = dstIIP.getINode(-2); // refresh dstParent
-        dstParent.updateModificationTime(timestamp, dstIIP.getLatestSnapshotId());
-        // update moved leases with new filename
-        getFSNamesystem().unprotectedChangeLease(src, dst);     
 
-        // update the quota usage in src tree
+        tx.updateMtimeAndLease(timestamp);
-        if (isSrcInSnapshot) {
+        tx.updateQuotasInSourceTree();
-          // get the counts after rename
-          Quota.Counts newSrcCounts = srcChild.computeQuotaUsage(
-              Quota.Counts.newInstance(), false);
-          newSrcCounts.subtract(oldSrcCounts);
-          srcParent.addSpaceConsumed(newSrcCounts.get(Quota.NAMESPACE),
-              newSrcCounts.get(Quota.DISKSPACE), false);
-        }
         
         return true;
       }
     } finally {
       if (!added) {
-        final INodeDirectory srcParent = srcIIP.getINode(-2).asDirectory();
+        tx.restoreSource();
-        final INode oldSrcChild = srcChild;
-        // put it back
-        if (withCount == null) {
-          srcChild.setLocalName(srcChildName);
-        } else if (!srcChildIsReference) { // src must be in snapshot
-          // the withCount node will no longer be used thus no need to update
-          // its reference number here
-          srcChild = withCount.getReferredINode();
-          srcChild.setLocalName(srcChildName);
-        } else {
-          withCount.removeReference(oldSrcChild.asReference());
-          srcChild = new INodeReference.DstReference(
-              srcParent, withCount, srcRefDstSnapshot);
-          withCount.getReferredINode().setLocalName(srcChildName);
-        }
-        
-        if (isSrcInSnapshot) {
-          // srcParent must have snapshot feature since isSrcInSnapshot is true
-          // and src node has been removed from srcParent 
-          srcParent.undoRename4ScrParent(oldSrcChild.asReference(), srcChild);
-        } else {
-          // original srcChild is not in latest snapshot, we only need to add
-          // the srcChild back
-          addLastINodeNoQuotaCheck(srcIIP, srcChild);
-        }
       }
     }
     NameNode.stateChangeLog.warn("DIR* FSDirectory.unprotectedRenameTo: "
@@ -681,53 +561,21 @@ public class FSDirectory implements Closeable {
       FileNotFoundException, ParentNotDirectoryException,
       QuotaExceededException, UnresolvedLinkException, IOException {
     assert hasWriteLock();
-    boolean overwrite = false;
+    boolean overwrite = options != null && Arrays.asList(options).contains
-    if (null != options) {
+            (Rename.OVERWRITE);
-      for (Rename option : options) {
+
-        if (option == Rename.OVERWRITE) {
-          overwrite = true;
-        }
-      }
-    }
     final String error;
     final INodesInPath srcIIP = getINodesInPath4Write(src, false);
     final INode srcInode = srcIIP.getLastINode();
-    // validate source
+    validateRenameSource(src, srcIIP);
-    if (srcInode == null) {
-      error = "rename source " + src + " is not found.";
-      NameNode.stateChangeLog.warn("DIR* FSDirectory.unprotectedRenameTo: "
-          + error);
-      throw new FileNotFoundException(error);
-    }
-    if (srcIIP.getINodes().length == 1) {
-      error = "rename source cannot be the root";
-      NameNode.stateChangeLog.warn("DIR* FSDirectory.unprotectedRenameTo: "
-          + error);
-      throw new IOException(error);
-    }
-    // srcInode and its subtree cannot contain snapshottable directories with
-    // snapshots
-    checkSnapshot(srcInode, null);
 
     // validate the destination
     if (dst.equals(src)) {
       throw new FileAlreadyExistsException(
           "The source "+src+" and destination "+dst+" are the same");
     }
-    if (srcInode.isSymlink() && 
+    validateRenameDestination(src, dst, srcInode);
-        dst.equals(srcInode.asSymlink().getSymlinkString())) {
+
-      throw new FileAlreadyExistsException(
-          "Cannot rename symlink "+src+" to its target "+dst);
-    }
-    // dst cannot be a directory or a file under src
-    if (dst.startsWith(src) && 
-        dst.charAt(src.length()) == Path.SEPARATOR_CHAR) {
-      error = "Rename destination " + dst
-          + " is a directory or file under source " + src;
-      NameNode.stateChangeLog.warn("DIR* FSDirectory.unprotectedRenameTo: "
-          + error);
-      throw new IOException(error);
-    }
     INodesInPath dstIIP = getINodesInPath4Write(dst, false);
     if (dstIIP.getINodes().length == 1) {
       error = "rename destination cannot be the root";
@@ -740,7 +588,117 @@ public class FSDirectory implements Closeable {
     List<INodeDirectorySnapshottable> snapshottableDirs = 
         new ArrayList<INodeDirectorySnapshottable>();
     if (dstInode != null) { // Destination exists
-      // It's OK to rename a file to a symlink and vice versa
+      validateRenameOverwrite(src, dst, overwrite, srcInode, dstInode);
+      checkSnapshot(dstInode, snapshottableDirs);
+    }
+
+    INode dstParent = dstIIP.getINode(-2);
+    if (dstParent == null) {
+      error = "rename destination parent " + dst + " not found.";
+      NameNode.stateChangeLog.warn("DIR* FSDirectory.unprotectedRenameTo: "
+          + error);
+      throw new FileNotFoundException(error);
+    }
+    if (!dstParent.isDirectory()) {
+      error = "rename destination parent " + dst + " is a file.";
+      NameNode.stateChangeLog.warn("DIR* FSDirectory.unprotectedRenameTo: "
+          + error);
+      throw new ParentNotDirectoryException(error);
+    }
+
+    // Ensure dst has quota to accommodate rename
+    verifyFsLimitsForRename(srcIIP, dstIIP);
+    verifyQuotaForRename(srcIIP.getINodes(), dstIIP.getINodes());
+
+    RenameOperation tx = new RenameOperation(src, dst, srcIIP, dstIIP);
+
+    boolean undoRemoveSrc = true;
+    final long removedSrc = removeLastINode(srcIIP);
+    if (removedSrc == -1) {
+      error = "Failed to rename " + src + " to " + dst
+          + " because the source can not be removed";
+      NameNode.stateChangeLog.warn("DIR* FSDirectory.unprotectedRenameTo: "
+          + error);
+      throw new IOException(error);
+    }
+    
+    boolean undoRemoveDst = false;
+    INode removedDst = null;
+    long removedNum = 0;
+    try {
+      if (dstInode != null) { // dst exists remove it
+        if ((removedNum = removeLastINode(dstIIP)) != -1) {
+          removedDst = dstIIP.getLastINode();
+          undoRemoveDst = true;
+        }
+      }
+
+      // add src as dst to complete rename
+      if (tx.addSourceToDestination()) {
+        undoRemoveSrc = false;
+        if (NameNode.stateChangeLog.isDebugEnabled()) {
+          NameNode.stateChangeLog.debug(
+              "DIR* FSDirectory.unprotectedRenameTo: " + src
+              + " is renamed to " + dst);
+        }
+
+        tx.updateMtimeAndLease(timestamp);
+
+        // Collect the blocks and remove the lease for previous dst
+        long filesDeleted = -1;
+        if (removedDst != null) {
+          undoRemoveDst = false;
+          if (removedNum > 0) {
+            BlocksMapUpdateInfo collectedBlocks = new BlocksMapUpdateInfo();
+            List<INode> removedINodes = new ChunkedArrayList<INode>();
+            filesDeleted = removedDst.cleanSubtree(Snapshot.CURRENT_STATE_ID,
+                dstIIP.getLatestSnapshotId(), collectedBlocks, removedINodes,
+                true).get(Quota.NAMESPACE);
+            getFSNamesystem().removePathAndBlocks(src, collectedBlocks,
+                removedINodes);
+          }
+        }
+
+        if (snapshottableDirs.size() > 0) {
+          // There are snapshottable directories (without snapshots) to be
+          // deleted. Need to update the SnapshotManager.
+          namesystem.removeSnapshottableDirs(snapshottableDirs);
+        }
+
+        tx.updateQuotasInSourceTree();
+        return filesDeleted >= 0;
+      }
+    } finally {
+      if (undoRemoveSrc) {
+        tx.restoreSource();
+      }
+
+      if (undoRemoveDst) {
+        // Rename failed - restore dst
+        if (dstParent.isDirectory() && dstParent.asDirectory().isWithSnapshot()) {
+          dstParent.asDirectory().undoRename4DstParent(removedDst,
+              dstIIP.getLatestSnapshotId());
+        } else {
+          addLastINodeNoQuotaCheck(dstIIP, removedDst);
+        }
+        if (removedDst.isReference()) {
+          final INodeReference removedDstRef = removedDst.asReference();
+          final INodeReference.WithCount wc = 
+              (WithCount) removedDstRef.getReferredINode().asReference();
+          wc.addReference(removedDstRef);
+        }
+      }
+    }
+    NameNode.stateChangeLog.warn("DIR* FSDirectory.unprotectedRenameTo: "
+        + "failed to rename " + src + " to " + dst);
+    throw new IOException("rename from " + src + " to " + dst + " failed.");
+  }
+
+  private static void validateRenameOverwrite(String src, String dst,
+                                              boolean overwrite,
+                                              INode srcInode, INode dstInode)
+          throws IOException {
+    String error;// It's OK to rename a file to a symlink and vice versa
     if (dstInode.isDirectory() != srcInode.isDirectory()) {
       error = "Source " + src + " and destination " + dst
           + " must both be directories";
@@ -764,46 +722,90 @@ public class FSDirectory implements Closeable {
         throw new IOException(error);
       }
     }
-      checkSnapshot(dstInode, snapshottableDirs);
   }
 
-    INode dstParent = dstIIP.getINode(-2);
+  private static void validateRenameDestination(String src, String dst, INode srcInode)
-    if (dstParent == null) {
+          throws IOException {
-      error = "rename destination parent " + dst + " not found.";
+    String error;
+    if (srcInode.isSymlink() &&
+        dst.equals(srcInode.asSymlink().getSymlinkString())) {
+      throw new FileAlreadyExistsException(
+          "Cannot rename symlink "+src+" to its target "+dst);
+    }
+    // dst cannot be a directory or a file under src
+    if (dst.startsWith(src) &&
+        dst.charAt(src.length()) == Path.SEPARATOR_CHAR) {
+      error = "Rename destination " + dst
+          + " is a directory or file under source " + src;
+      NameNode.stateChangeLog.warn("DIR* FSDirectory.unprotectedRenameTo: "
+          + error);
+      throw new IOException(error);
+    }
+  }
+
+  private static void validateRenameSource(String src, INodesInPath srcIIP)
+          throws IOException {
+    String error;
+    final INode srcInode = srcIIP.getLastINode();
+    // validate source
+    if (srcInode == null) {
+      error = "rename source " + src + " is not found.";
       NameNode.stateChangeLog.warn("DIR* FSDirectory.unprotectedRenameTo: "
           + error);
       throw new FileNotFoundException(error);
     }
-    if (!dstParent.isDirectory()) {
+    if (srcIIP.getINodes().length == 1) {
-      error = "rename destination parent " + dst + " is a file.";
+      error = "rename source cannot be the root";
       NameNode.stateChangeLog.warn("DIR* FSDirectory.unprotectedRenameTo: "
           + error);
-      throw new ParentNotDirectoryException(error);
+      throw new IOException(error);
+    }
+    // srcInode and its subtree cannot contain snapshottable directories with
+    // snapshots
+    checkSnapshot(srcInode, null);
   }
 
-    // Ensure dst has quota to accommodate rename
-    verifyFsLimitsForRename(srcIIP, dstIIP);
-    verifyQuotaForRename(srcIIP.getINodes(), dstIIP.getINodes());
 
-    INode srcChild = srcIIP.getLastINode();
+
-    final byte[] srcChildName = srcChild.getLocalNameBytes();
+  private class RenameOperation {
-    final boolean isSrcInSnapshot = srcChild.isInLatestSnapshot(
+    private final INodesInPath srcIIP;
+    private final INodesInPath dstIIP;
+    private final String src;
+    private final String dst;
+
+    private INode srcChild;
+    private final INodeReference.WithCount withCount;
+    private final int srcRefDstSnapshot;
+    private final INodeDirectory srcParent;
+    private final byte[] srcChildName;
+    private final boolean isSrcInSnapshot;
+    private final boolean srcChildIsReference;
+    private final Quota.Counts oldSrcCounts;
+
+    private RenameOperation(String src, String dst, INodesInPath srcIIP, INodesInPath dstIIP)
+            throws QuotaExceededException {
+      this.srcIIP = srcIIP;
+      this.dstIIP = dstIIP;
+      this.src = src;
+      this.dst = dst;
+      srcChild = srcIIP.getLastINode();
+      srcChildName = srcChild.getLocalNameBytes();
+      isSrcInSnapshot = srcChild.isInLatestSnapshot(
               srcIIP.getLatestSnapshotId());
-    final boolean srcChildIsReference = srcChild.isReference();
+      srcChildIsReference = srcChild.isReference();
+      srcParent = srcIIP.getINode(-2).asDirectory();
 
       // Record the snapshot on srcChild. After the rename, before any new
       // snapshot is taken on the dst tree, changes will be recorded in the latest
       // snapshot of the src tree.
       if (isSrcInSnapshot) {
         srcChild = srcChild.recordModification(srcIIP.getLatestSnapshotId());
-      srcIIP.setLastINode(srcChild);
       }
 
       // check srcChild for reference
-    final INodeReference.WithCount withCount;
+      srcRefDstSnapshot = srcChildIsReference ? srcChild.asReference()
-    int srcRefDstSnapshot = srcChildIsReference ? srcChild.asReference()
               .getDstSnapshotId() : Snapshot.CURRENT_STATE_ID;
-    Quota.Counts oldSrcCounts = Quota.Counts.newInstance();    
+      oldSrcCounts = Quota.Counts.newInstance();
       if (isSrcInSnapshot) {
         final INodeReference.WithName withName = srcIIP.getINode(-2).asDirectory()
                 .replaceChild4ReferenceWithName(srcChild, srcIIP.getLatestSnapshotId());
@@ -818,37 +820,11 @@ public class FSDirectory implements Closeable {
       } else {
         withCount = null;
       }
-    
-    boolean undoRemoveSrc = true;
-    final long removedSrc = removeLastINode(srcIIP);
-    if (removedSrc == -1) {
-      error = "Failed to rename " + src + " to " + dst
-          + " because the source can not be removed";
-      NameNode.stateChangeLog.warn("DIR* FSDirectory.unprotectedRenameTo: "
-          + error);
-      throw new IOException(error);
-    }
-    
-    if (dstParent.getParent() == null) {
-      // src and dst file/dir are in the same directory, and the dstParent has
-      // been replaced when we removed the src. Refresh the dstIIP and
-      // dstParent.
-      dstIIP = getINodesInPath4Write(dst, false);
-    }
-    
-    boolean undoRemoveDst = false;
-    INode removedDst = null;
-    long removedNum = 0;
-    try {
-      if (dstInode != null) { // dst exists remove it
-        if ((removedNum = removeLastINode(dstIIP)) != -1) {
-          removedDst = dstIIP.getLastINode();
-          undoRemoveDst = true;
-        }
     }
 
+    boolean addSourceToDestination() {
+      final INode dstParent = dstIIP.getINode(-2);
       srcChild = srcIIP.getLastINode();
-
       final byte[] dstChildName = dstIIP.getLastLocalName();
       final INode toDst;
       if (withCount == null) {
@@ -858,62 +834,21 @@ public class FSDirectory implements Closeable {
         withCount.getReferredINode().setLocalName(dstChildName);
         int dstSnapshotId = dstIIP.getLatestSnapshotId();
         toDst = new INodeReference.DstReference(
-            dstIIP.getINode(-2).asDirectory(), withCount, dstSnapshotId);
+                dstParent.asDirectory(), withCount, dstSnapshotId);
+      }
+      return addLastINodeNoQuotaCheck(dstIIP, toDst);
     }
 
-      // add src as dst to complete rename
+    void updateMtimeAndLease(long timestamp) throws QuotaExceededException {
-      if (addLastINodeNoQuotaCheck(dstIIP, toDst)) {
-        undoRemoveSrc = false;
-        if (NameNode.stateChangeLog.isDebugEnabled()) {
-          NameNode.stateChangeLog.debug(
-              "DIR* FSDirectory.unprotectedRenameTo: " + src
-              + " is renamed to " + dst);
-        }
-
-        final INode srcParent = srcIIP.getINode(-2);
       srcParent.updateModificationTime(timestamp, srcIIP.getLatestSnapshotId());
-        dstParent = dstIIP.getINode(-2);
+      final INode dstParent = dstIIP.getINode(-2);
       dstParent.updateModificationTime(timestamp, dstIIP.getLatestSnapshotId());
       // update moved lease with new filename
       getFSNamesystem().unprotectedChangeLease(src, dst);
-
-        // Collect the blocks and remove the lease for previous dst
-        long filesDeleted = -1;
-        if (removedDst != null) {
-          undoRemoveDst = false;
-          if (removedNum > 0) {
-            BlocksMapUpdateInfo collectedBlocks = new BlocksMapUpdateInfo();
-            List<INode> removedINodes = new ChunkedArrayList<INode>();
-            filesDeleted = removedDst.cleanSubtree(Snapshot.CURRENT_STATE_ID,
-                dstIIP.getLatestSnapshotId(), collectedBlocks, removedINodes,
-                true).get(Quota.NAMESPACE);
-            getFSNamesystem().removePathAndBlocks(src, collectedBlocks,
-                removedINodes);
-          }
     }
 
-        if (snapshottableDirs.size() > 0) {
+    void restoreSource() throws QuotaExceededException {
-          // There are snapshottable directories (without snapshots) to be
-          // deleted. Need to update the SnapshotManager.
-          namesystem.removeSnapshottableDirs(snapshottableDirs);
-        }
-        
-        // update the quota usage in src tree
-        if (isSrcInSnapshot) {
-          // get the counts after rename
-          Quota.Counts newSrcCounts = srcChild.computeQuotaUsage(
-              Quota.Counts.newInstance(), false);
-          newSrcCounts.subtract(oldSrcCounts);
-          srcParent.addSpaceConsumed(newSrcCounts.get(Quota.NAMESPACE),
-              newSrcCounts.get(Quota.DISKSPACE), false);
-        }
-        
-        return filesDeleted >= 0;
-      }
-    } finally {
-      if (undoRemoveSrc) {
       // Rename failed - restore src
-        final INodeDirectory srcParent = srcIIP.getINode(-2).asDirectory();
       final INode oldSrcChild = srcChild;
       // put it back
       if (withCount == null) {
@@ -930,7 +865,7 @@ public class FSDirectory implements Closeable {
         withCount.getReferredINode().setLocalName(srcChildName);
       }
 
-        if (srcParent.isWithSnapshot()) {
+      if (isSrcInSnapshot) {
         srcParent.undoRename4ScrParent(oldSrcChild.asReference(), srcChild);
       } else {
         // srcParent is not an INodeDirectoryWithSnapshot, we only need to add
@@ -938,26 +873,19 @@ public class FSDirectory implements Closeable {
         addLastINodeNoQuotaCheck(srcIIP, srcChild);
       }
     }
-      if (undoRemoveDst) {
+
-        // Rename failed - restore dst
+    void updateQuotasInSourceTree() throws QuotaExceededException {
-        if (dstParent.isDirectory() && dstParent.asDirectory().isWithSnapshot()) {
+      // update the quota usage in src tree
-          dstParent.asDirectory().undoRename4DstParent(removedDst,
+      if (isSrcInSnapshot) {
-              dstIIP.getLatestSnapshotId());
+        // get the counts after rename
-        } else {
+        Quota.Counts newSrcCounts = srcChild.computeQuotaUsage(
-          addLastINodeNoQuotaCheck(dstIIP, removedDst);
+                Quota.Counts.newInstance(), false);
-        }
+        newSrcCounts.subtract(oldSrcCounts);
-        if (removedDst.isReference()) {
+        srcParent.addSpaceConsumed(newSrcCounts.get(Quota.NAMESPACE),
-          final INodeReference removedDstRef = removedDst.asReference();
+                newSrcCounts.get(Quota.DISKSPACE), false);
-          final INodeReference.WithCount wc = 
-              (WithCount) removedDstRef.getReferredINode().asReference();
-          wc.addReference(removedDstRef);
       }
     }
   }
-    NameNode.stateChangeLog.warn("DIR* FSDirectory.unprotectedRenameTo: "
-        + "failed to rename " + src + " to " + dst);
-    throw new IOException("rename from " + src + " to " + dst + " failed.");
-  }
 
   /**
    * Set file replication
@@ -1332,14 +1260,14 @@ public class FSDirectory implements Closeable {
    *                          but do not have snapshots yet
    */
   private static void checkSnapshot(INode target,
-      List<INodeDirectorySnapshottable> snapshottableDirs) throws IOException {
+      List<INodeDirectorySnapshottable> snapshottableDirs) throws SnapshotException {
     if (target.isDirectory()) {
       INodeDirectory targetDir = target.asDirectory();
       if (targetDir.isSnapshottable()) {
         INodeDirectorySnapshottable ssTargetDir = 
             (INodeDirectorySnapshottable) targetDir;
         if (ssTargetDir.getNumSnapshots() > 0) {
-          throw new IOException("The directory " + ssTargetDir.getFullPathName()
+          throw new SnapshotException("The directory " + ssTargetDir.getFullPathName()
               + " cannot be deleted since " + ssTargetDir.getFullPathName()
               + " is snapshottable and already has snapshots");
         } else {
@@ -2053,10 +1981,6 @@ public class FSDirectory implements Closeable {
     if (!parent.removeChild(last, latestSnapshot)) {
       return -1;
     }
-    INodeDirectory newParent = last.getParent();
-    if (parent != newParent) {
-      iip.setINode(-2, newParent);
-    }
     
     if (!last.isInLatestSnapshot(latestSnapshot)) {
       final Quota.Counts counts = last.computeQuotaUsage();