HDFS-4170. Add snapshot information to INodesInPath.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/HDFS-2802@1407703 13f79535-47bb-0310-9956-ffa450edef68

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

@@ -62,3 +62,5 @@ Branch-2802 Snapshot (Unreleased)
 
   HDFS-4159. Rename should fail when the destination directory is snapshottable
   and has snapshots. (Jing Zhao via szetszwo)
+
+  HDFS-4170. Add snapshot information to INodesInPath.  (szetszwo)

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

@@ -32,6 +32,7 @@ import org.apache.hadoop.hdfs.protocol.HdfsConstants;
 import org.apache.hadoop.hdfs.protocol.UnresolvedPathException;
 import org.apache.hadoop.hdfs.server.namenode.snapshot.INodeDirectorySnapshottable;
 import org.apache.hadoop.hdfs.server.namenode.snapshot.INodeDirectoryWithSnapshot;
+import org.apache.hadoop.hdfs.server.namenode.snapshot.Snapshot;
 
 import com.google.common.annotations.VisibleForTesting;
 
@@ -212,6 +213,13 @@ public class INodeDirectory extends INode {
       if (index >= 0) {
         existing.addNode(curNode);
       }
+      if (curNode instanceof INodeDirectorySnapshottable) {
+        //if the path is a non-snapshot path, update the latest snapshot.
+        if (!existing.isSnapshot()) {
+          existing.updateLatestSnapshot(
+              ((INodeDirectorySnapshottable)curNode).getLastSnapshot());
+        }
+      }
       if (curNode.isSymlink() && (!lastComp || (lastComp && resolveLink))) {
         final String path = constructPath(components, 0, components.length);
         final String preceding = constructPath(components, 0, count);
@@ -247,10 +255,17 @@ public class INodeDirectory extends INode {
           return existing;
         }
         // Resolve snapshot root
-        curNode = ((INodeDirectorySnapshottable) parentDir)
+        final Snapshot s = ((INodeDirectorySnapshottable)parentDir).getSnapshot(
-            .getSnapshotRoot(components[count + 1]);
+            components[count + 1]);
+        if (s == null) {
+          //snapshot not found
+          curNode = null;
+        } else {
+          curNode = s.getRoot();
+          existing.setSnapshot(s);
+        }
         if (index >= -1) {
-          existing.snapshotRootIndex = existing.size;
+          existing.snapshotRootIndex = existing.numNonNull;
         }
       } else {
         // normal case, and also for resolving file/dir under snapshot root
@@ -498,7 +513,7 @@ public class INodeDirectory extends INode {
     /**
      * Indicate the number of non-null elements in {@link #inodes}
      */
-    private int size;
+    private int numNonNull;
     /**
      * The path for a snapshot file/dir contains the .snapshot thus makes the
      * length of the path components larger the number of inodes. We use
@@ -513,16 +528,40 @@ public class INodeDirectory extends INode {
      * Index of {@link INodeDirectoryWithSnapshot} for snapshot path, else -1
      */
     private int snapshotRootIndex;
+    /**
+     * For snapshot paths, it is the reference to the snapshot; or null if the
+     * snapshot does not exist. For non-snapshot paths, it is the reference to
+     * the latest snapshot found in the path; or null if no snapshot is found.
+     */
+    private Snapshot snapshot = null; 
 
     INodesInPath(int number) {
       assert (number >= 0);
       inodes = new INode[number];
       capacity = number;
-      size = 0;
+      numNonNull = 0;
       isSnapshot = false;
       snapshotRootIndex = -1;
     }
 
+    /**
+     * @return the snapshot associated to the path.
+     * @see #snapshot
+     */
+    public Snapshot getSnapshot() {
+      return snapshot;
+    }
+
+    private void setSnapshot(Snapshot s) {
+      snapshot = s;
+    }
+    
+    private void updateLatestSnapshot(Snapshot s) {
+      if (snapshot == null || snapshot.compareTo(s) < 0) {
+        snapshot = s;
+      }
+    }
+
     /**
      * @return the whole inodes array including the null elements.
      */
@@ -556,8 +595,7 @@ public class INodeDirectory extends INode {
      * Add an INode at the end of the array
      */
     private void addNode(INode node) {
-      assert size < inodes.length;
+      inodes[numNonNull++] = node;
-      inodes[size++] = node;
     }
     
     void setINode(int i, INode inode) {
@@ -567,8 +605,35 @@ public class INodeDirectory extends INode {
     /**
      * @return The number of non-null elements
      */
-    int getSize() {
+    int getNumNonNull() {
-      return size;
+      return numNonNull;
+    }
+    
+    static String toString(INode inode) {
+      return inode == null? null: inode.getLocalName();
+    }
+
+    @Override
+    public String toString() {
+      final StringBuilder b = new StringBuilder(getClass().getSimpleName())
+          .append(":\n  inodes = ");
+      if (inodes == null) {
+        b.append("null");
+      } else if (inodes.length == 0) {
+        b.append("[]");
+      } else {
+        b.append("[").append(toString(inodes[0]));
+        for(int i = 1; i < inodes.length; i++) {
+          b.append(", ").append(toString(inodes[i]));
+        }
+        b.append("]");
+      }
+      b.append("\n  numNonNull = ").append(numNonNull)
+       .append("\n  capacity   = ").append(capacity)
+       .append("\n  isSnapshot        = ").append(isSnapshot)
+       .append("\n  snapshotRootIndex = ").append(snapshotRootIndex)
+       .append("\n  snapshot          = ").append(snapshot);
+      return b.toString();
     }
   }
 

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

@@ -65,6 +65,8 @@ public class INodeDirectorySnapshottable extends INodeDirectoryWithQuota {
 
   /** Snapshots of this directory in ascending order of snapshot id. */
   private final List<Snapshot> snapshots = new ArrayList<Snapshot>();
+  /** Snapshots of this directory in ascending order of snapshot names. */
+  private final List<Snapshot> snapshotsByNames = new ArrayList<Snapshot>();
 
   /** Number of snapshots allowed. */
   private int snapshotQuota;
@@ -79,16 +81,20 @@ public class INodeDirectorySnapshottable extends INodeDirectoryWithQuota {
     return snapshots.size();
   }
   
-  /** @return the root directory of a snapshot. */
+  private int searchSnapshot(byte[] snapshotName) {
-  public INodeDirectory getSnapshotRoot(byte[] snapshotName) {
+    return Collections.binarySearch(snapshotsByNames, snapshotName);
-    if (snapshots == null || snapshots.size() == 0) {
-      return null;
   }
-    int low = Collections.binarySearch(snapshots, snapshotName);
+
-    if (low >= 0) {
+  /** @return the snapshot with the given name. */
-      return snapshots.get(low).getRoot();
+  public Snapshot getSnapshot(byte[] snapshotName) {
+    final int i = searchSnapshot(snapshotName);
+    return i < 0? null: snapshotsByNames.get(i);
   }
-    return null;
+
+  /** @return the last snapshot. */
+  public Snapshot getLastSnapshot() {
+    final int n = snapshots.size();
+    return n == 0? null: snapshots.get(n - 1);
   }
 
   public int getSnapshotQuota() {
@@ -108,21 +114,30 @@ public class INodeDirectorySnapshottable extends INodeDirectoryWithQuota {
     return true;
   }
 
-  /** Add a snapshot root under this directory. */
+  /** Add a snapshot. */
-  void addSnapshot(final Snapshot s) throws SnapshotException {
+  Snapshot addSnapshot(int id, String name) throws SnapshotException {
     //check snapshot quota
     if (snapshots.size() + 1 > snapshotQuota) {
       throw new SnapshotException("Failed to add snapshot: there are already "
           + snapshots.size() + " snapshot(s) and the snapshot quota is "
           + snapshotQuota);
     }
+    final Snapshot s = new Snapshot(id, name, this);
+    final byte[] nameBytes = s.getRoot().getLocalNameBytes();
+    final int i = searchSnapshot(nameBytes);
+    if (i >= 0) {
+      throw new SnapshotException("Failed to add snapshot: there is already a "
+          + "snapshot with the same name \"" + name + "\".");
+    }
 
     snapshots.add(s);
+    snapshotsByNames.add(-i - 1, s);
 
     //set modification time
     final long timestamp = Time.now();
     s.getRoot().setModificationTime(timestamp);
     setModificationTime(timestamp);
+    return s;
   }
   
   @Override

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

@@ -32,7 +32,8 @@ public class Snapshot implements Comparable<byte[]> {
     this.root = new INodeDirectoryWithSnapshot(name, dir);
   }
 
-  INodeDirectoryWithSnapshot getRoot() {
+  /** @return the root directory of the snapshot. */
+  public INodeDirectoryWithSnapshot getRoot() {
     return root;
   }
 
@@ -40,4 +41,14 @@ public class Snapshot implements Comparable<byte[]> {
   public int compareTo(byte[] bytes) {
     return root.compareTo(bytes);
   }
+
+  /** Compare snapshot IDs. */
+  public int compareTo(Snapshot s) {
+    return id - s.id;
+  }
+  
+  @Override
+  public String toString() {
+    return getClass().getSimpleName() + ":" + root.getLocalName();
+  }
 }

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

@@ -105,8 +105,7 @@ public class SnapshotManager implements SnapshotStats {
         = INodeDirectorySnapshottable.valueOf(fsdir.getINode(path), path);
 
     synchronized(this) {
-      final Snapshot s = new Snapshot(snapshotID, snapshotName, srcRoot); 
+      final Snapshot s = srcRoot.addSnapshot(snapshotID, snapshotName);
-      srcRoot.addSnapshot(s);
       new SnapshotCreation().processRecursively(srcRoot, s.getRoot());
       
       //create success, update id

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestSnapshotPathINodes.java

@@ -25,15 +25,17 @@ import static org.junit.Assert.assertTrue;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hdfs.DFSTestUtil;
+import org.apache.hadoop.hdfs.DFSUtil;
 import org.apache.hadoop.hdfs.DistributedFileSystem;
 import org.apache.hadoop.hdfs.MiniDFSCluster;
 import org.apache.hadoop.hdfs.server.namenode.INodeDirectory.INodesInPath;
-import org.apache.hadoop.hdfs.server.namenode.snapshot.INodeDirectoryWithSnapshot;
 import org.apache.hadoop.hdfs.server.namenode.snapshot.INodeDirectorySnapshottable;
+import org.apache.hadoop.hdfs.server.namenode.snapshot.INodeDirectoryWithSnapshot;
 import org.apache.hadoop.hdfs.server.namenode.snapshot.INodeFileSnapshot;
-import org.junit.After;
+import org.apache.hadoop.hdfs.server.namenode.snapshot.Snapshot;
+import org.junit.AfterClass;
 import org.junit.Assert;
-import org.junit.Before;
+import org.junit.BeforeClass;
 import org.junit.Test;
 
 /** Test snapshot related operations. */
@@ -41,21 +43,21 @@ public class TestSnapshotPathINodes {
   private static final long seed = 0;
   private static final short REPLICATION = 3;
 
-  private final Path dir = new Path("/TestSnapshot");
+  static private final Path dir = new Path("/TestSnapshot");
   
-  private final Path sub1 = new Path(dir, "sub1");
+  static private final Path sub1 = new Path(dir, "sub1");
-  private final Path file1 = new Path(sub1, "file1");
+  static private final Path file1 = new Path(sub1, "file1");
-  private final Path file2 = new Path(sub1, "file2");
+  static private final Path file2 = new Path(sub1, "file2");
 
-  private Configuration conf;
+  static private Configuration conf;
-  private MiniDFSCluster cluster;
+  static private MiniDFSCluster cluster;
-  private FSNamesystem fsn;
+  static private FSNamesystem fsn;
-  private FSDirectory fsdir;
+  static private FSDirectory fsdir;
 
-  private DistributedFileSystem hdfs;
+  static private DistributedFileSystem hdfs;
 
-  @Before
+  @BeforeClass
-  public void setUp() throws Exception {
+  static public void setUp() throws Exception {
     conf = new Configuration();
     cluster = new MiniDFSCluster.Builder(conf)
       .numDataNodes(REPLICATION)
@@ -70,8 +72,8 @@ public class TestSnapshotPathINodes {
     DFSTestUtil.createFile(hdfs, file2, 1024, REPLICATION, seed);
   }
 
-  @After
+  @AfterClass
-  public void tearDown() throws Exception {
+  static public void tearDown() throws Exception {
     if (cluster != null) {
       cluster.shutdown();
     }
@@ -102,6 +104,23 @@ public class TestSnapshotPathINodes {
     }
   }
   
+  static Snapshot getSnapshot(INodesInPath inodesInPath, String name) {
+    if (name == null) {
+      return null;
+    }
+    final int i = inodesInPath.getSnapshotRootIndex() - 1;
+    final INode inode = inodesInPath.getINodes()[i];
+    return ((INodeDirectorySnapshottable)inode).getSnapshot(
+        DFSUtil.string2Bytes(name)); 
+  }
+
+  static void assertSnapshot(INodesInPath inodesInPath, boolean isSnapshot,
+      final Snapshot snapshot, int index) {
+    assertEquals(isSnapshot, inodesInPath.isSnapshot());
+    assertEquals(index, inodesInPath.getSnapshotRootIndex());
+    assertEquals(snapshot, inodesInPath.getSnapshot());
+  }
+
   /** 
    * Test {@link INodeDirectory#getExistingPathINodes(byte[][], int, boolean)} 
    * for normal (non-snapshot) file.
@@ -117,8 +136,8 @@ public class TestSnapshotPathINodes {
     // The number of inodes should be equal to components.length
     assertEquals(inodes.length, components.length);
     // The returned nodesInPath should be non-snapshot
-    assertFalse(nodesInPath.isSnapshot());
+    assertSnapshot(nodesInPath, false, null, -1);
-    assertEquals(nodesInPath.getSnapshotRootIndex(), -1);
+
     // The last INode should be associated with file1
     assertEquals(inodes[components.length - 1].getFullPathName(),
         file1.toString());
@@ -132,8 +151,7 @@ public class TestSnapshotPathINodes {
     nodesInPath = fsdir.rootDir.getExistingPathINodes(components, 1, false);
     inodes = nodesInPath.getINodes();
     assertEquals(inodes.length, 1);
-    assertFalse(nodesInPath.isSnapshot());
+    assertSnapshot(nodesInPath, false, null, -1);
-    assertEquals(nodesInPath.getSnapshotRootIndex(), -1);
     assertEquals(inodes[0].getFullPathName(), file1.toString());
     
     // Call getExistingPathINodes and request 2 INodes. This is usually used
@@ -141,8 +159,7 @@ public class TestSnapshotPathINodes {
     nodesInPath = fsdir.rootDir.getExistingPathINodes(components, 2, false);
     inodes = nodesInPath.getINodes();
     assertEquals(inodes.length, 2);
-    assertFalse(nodesInPath.isSnapshot());
+    assertSnapshot(nodesInPath, false, null, -1);
-    assertEquals(nodesInPath.getSnapshotRootIndex(), -1);
     assertEquals(inodes[1].getFullPathName(), file1.toString());
     assertEquals(inodes[0].getFullPathName(), sub1.toString());
   }
@@ -168,9 +185,9 @@ public class TestSnapshotPathINodes {
     // Length of inodes should be (components.length - 1), since we will ignore
     // ".snapshot" 
     assertEquals(inodes.length, components.length - 1);
-    assertTrue(nodesInPath.isSnapshot());
     // SnapshotRootIndex should be 3: {root, Testsnapshot, sub1, s1, file1}
-    assertEquals(nodesInPath.getSnapshotRootIndex(), 3);
+    final Snapshot snapshot = getSnapshot(nodesInPath, "s1");
+    assertSnapshot(nodesInPath, true, snapshot, 3);
     assertTrue(inodes[nodesInPath.getSnapshotRootIndex()] instanceof 
         INodeDirectoryWithSnapshot);
     // Check the INode for file1 (snapshot file)
@@ -184,10 +201,9 @@ public class TestSnapshotPathINodes {
     nodesInPath = fsdir.rootDir.getExistingPathINodes(components, 1, false);
     inodes = nodesInPath.getINodes();
     assertEquals(inodes.length, 1);
-    assertTrue(nodesInPath.isSnapshot());
     // The snapshotroot (s1) is not included in inodes. Thus the
     // snapshotRootIndex should be -1.
-    assertEquals(nodesInPath.getSnapshotRootIndex(), -1);
+    assertSnapshot(nodesInPath, true, snapshot, -1);
     // Check the INode for file1 (snapshot file)
     snapshotFileNode = inodes[inodes.length - 1]; 
     assertEquals(snapshotFileNode.getLocalName(), file1.getName());
@@ -197,10 +213,9 @@ public class TestSnapshotPathINodes {
     nodesInPath = fsdir.rootDir.getExistingPathINodes(components, 2, false);
     inodes = nodesInPath.getINodes();
     assertEquals(inodes.length, 2);
-    assertTrue(nodesInPath.isSnapshot());
     // There should be two INodes in inodes: s1 and snapshot of file1. Thus the
     // SnapshotRootIndex should be 0.
-    assertEquals(nodesInPath.getSnapshotRootIndex(), 0);
+    assertSnapshot(nodesInPath, true, snapshot, 0);
     snapshotFileNode = inodes[inodes.length - 1];
     // Check the INode for snapshot of file1
     assertEquals(snapshotFileNode.getLocalName(), file1.getName());
@@ -216,9 +231,9 @@ public class TestSnapshotPathINodes {
     // The number of INodes returned should be components.length - 1 since we
     // will ignore ".snapshot"
     assertEquals(inodes.length, components.length - 1);
-    assertTrue(nodesInPath.isSnapshot());
+
     // No SnapshotRoot dir is included in the resolved inodes  
-    assertEquals(nodesInPath.getSnapshotRootIndex(), -1);
+    assertSnapshot(nodesInPath, true, snapshot, -1);
     // The last INode should be the INode for sub1
     assertEquals(inodes[inodes.length - 1].getFullPathName(), sub1.toString());
     assertFalse(inodes[inodes.length - 1] instanceof INodeFileSnapshot);
@@ -233,11 +248,38 @@ public class TestSnapshotPathINodes {
     // Create a snapshot for the dir, and check the inodes for the path
     // pointing to a snapshot file
     hdfs.allowSnapshot(sub1.toString());
-    hdfs.createSnapshot("s1", sub1.toString());
+    hdfs.createSnapshot("s2", sub1.toString());
     
     // Delete the original file /TestSnapshot/sub1/file1
     hdfs.delete(file1, false);
     
+    final Snapshot snapshot;
+    {
+      // Resolve the path for the snapshot file
+      // /TestSnapshot/sub1/.snapshot/s2/file1
+      String snapshotPath = sub1.toString() + "/.snapshot/s2/file1";
+      String[] names = INode.getPathNames(snapshotPath);
+      byte[][] components = INode.getPathComponents(names);
+      INodesInPath nodesInPath = fsdir.rootDir.getExistingPathINodes(components,
+          components.length, false);
+      INode[] inodes = nodesInPath.getINodes();
+      // Length of inodes should be (components.length - 1), since we will ignore
+      // ".snapshot" 
+      assertEquals(inodes.length, components.length - 1);
+      // SnapshotRootIndex should be 3: {root, Testsnapshot, sub1, s2, file1}
+      snapshot = getSnapshot(nodesInPath, "s2");
+      assertSnapshot(nodesInPath, true, snapshot, 3);
+  
+      assertTrue(inodes[nodesInPath.getSnapshotRootIndex()] instanceof 
+          INodeDirectoryWithSnapshot);
+      // Check the INode for file1 (snapshot file)
+      INode snapshotFileNode = inodes[inodes.length - 1]; 
+      assertEquals(snapshotFileNode.getLocalName(), file1.getName());
+      assertTrue(snapshotFileNode instanceof INodeFileSnapshot);
+      assertTrue(snapshotFileNode.getParent() instanceof 
+          INodeDirectoryWithSnapshot);
+    }
+
     // Check the INodes for path /TestSnapshot/sub1/file1
     String[] names = INode.getPathNames(file1.toString());
     byte[][] components = INode.getPathComponents(names);
@@ -248,10 +290,9 @@ public class TestSnapshotPathINodes {
     assertEquals(inodes.length, components.length);
     // The number of non-null elements should be components.length - 1 since
     // file1 has been deleted
-    assertEquals(nodesInPath.getSize(), components.length - 1);
+    assertEquals(nodesInPath.getNumNonNull(), components.length - 1);
     // The returned nodesInPath should be non-snapshot
-    assertFalse(nodesInPath.isSnapshot());
+    assertSnapshot(nodesInPath, false, snapshot, -1);
-    assertEquals(nodesInPath.getSnapshotRootIndex(), -1);
     // The last INode should be null, and the one before should be associated
     // with sub1
     assertNull(inodes[components.length - 1]);
@@ -259,31 +300,10 @@ public class TestSnapshotPathINodes {
         sub1.toString());
     assertEquals(inodes[components.length - 3].getFullPathName(),
         dir.toString());
-    
-    // Resolve the path for the snapshot file
-    // /TestSnapshot/sub1/.snapshot/s1/file1
-    String snapshotPath = sub1.toString() + "/.snapshot/s1/file1";
-    names = INode.getPathNames(snapshotPath);
-    components = INode.getPathComponents(names);
-    nodesInPath = fsdir.rootDir.getExistingPathINodes(components,
-        components.length, false);
-    inodes = nodesInPath.getINodes();
-    // Length of inodes should be (components.length - 1), since we will ignore
-    // ".snapshot" 
-    assertEquals(inodes.length, components.length - 1);
-    assertTrue(nodesInPath.isSnapshot());
-    // SnapshotRootIndex should be 3: {root, Testsnapshot, sub1, s1, file1}
-    assertEquals(nodesInPath.getSnapshotRootIndex(), 3);
-    assertTrue(inodes[nodesInPath.getSnapshotRootIndex()] instanceof 
-        INodeDirectoryWithSnapshot);
-    // Check the INode for file1 (snapshot file)
-    INode snapshotFileNode = inodes[inodes.length - 1]; 
-    assertEquals(snapshotFileNode.getLocalName(), file1.getName());
-    assertTrue(snapshotFileNode instanceof INodeFileSnapshot);
-    assertTrue(snapshotFileNode.getParent() instanceof 
-        INodeDirectoryWithSnapshot);
   }
   
+  static private Snapshot s4;
+
   /** 
    * Test {@link INodeDirectory#getExistingPathINodes(byte[][], int, boolean)} 
    * for snapshot file while adding a new file after snapshot.
@@ -293,12 +313,39 @@ public class TestSnapshotPathINodes {
     // Create a snapshot for the dir, and check the inodes for the path
     // pointing to a snapshot file
     hdfs.allowSnapshot(sub1.toString());
-    hdfs.createSnapshot("s1", sub1.toString());
+    hdfs.createSnapshot("s4", sub1.toString());
     
     // Add a new file /TestSnapshot/sub1/file3
     final Path file3 = new Path(sub1, "file3");
     DFSTestUtil.createFile(hdfs, file3, 1024, REPLICATION, seed);
   
+    {
+      // Check the inodes for /TestSnapshot/sub1/.snapshot/s4/file3
+      String snapshotPath = sub1.toString() + "/.snapshot/s4/file3";
+      String[] names = INode.getPathNames(snapshotPath);
+      byte[][] components = INode.getPathComponents(names);
+      INodesInPath nodesInPath = fsdir.rootDir.getExistingPathINodes(components,
+          components.length, false);
+      INode[] inodes = nodesInPath.getINodes();
+      // Length of inodes should be (components.length - 1), since we will ignore
+      // ".snapshot" 
+      assertEquals(inodes.length, components.length - 1);
+      // The number of non-null inodes should be components.length - 2, since
+      // snapshot of file3 does not exist
+      assertEquals(nodesInPath.getNumNonNull(), components.length - 2);
+      s4 = getSnapshot(nodesInPath, "s4");
+
+      // SnapshotRootIndex should still be 3: {root, Testsnapshot, sub1, s4, null}
+      assertSnapshot(nodesInPath, true, s4, 3);
+  
+      assertTrue(inodes[nodesInPath.getSnapshotRootIndex()] instanceof 
+          INodeDirectoryWithSnapshot);
+      // Check the last INode in inodes, which should be null
+      assertNull(inodes[inodes.length - 1]);
+      assertTrue(inodes[inodes.length - 2] instanceof 
+          INodeDirectoryWithSnapshot);
+    }
+
     // Check the inodes for /TestSnapshot/sub1/file3
     String[] names = INode.getPathNames(file3.toString());
     byte[][] components = INode.getPathComponents(names);
@@ -307,9 +354,10 @@ public class TestSnapshotPathINodes {
     INode[] inodes = nodesInPath.getINodes();
     // The number of inodes should be equal to components.length
     assertEquals(inodes.length, components.length);
+
     // The returned nodesInPath should be non-snapshot
-    assertFalse(nodesInPath.isSnapshot());
+    assertSnapshot(nodesInPath, false, s4, -1);
-    assertEquals(nodesInPath.getSnapshotRootIndex(), -1);
+
     // The last INode should be associated with file3
     assertEquals(inodes[components.length - 1].getFullPathName(),
         file3.toString());
@@ -317,29 +365,6 @@ public class TestSnapshotPathINodes {
         sub1.toString());
     assertEquals(inodes[components.length - 3].getFullPathName(),
         dir.toString());
-    
-    // Check the inodes for /TestSnapshot/sub1/.snapshot/s1/file3
-    String snapshotPath = sub1.toString() + "/.snapshot/s1/file3";
-    names = INode.getPathNames(snapshotPath);
-    components = INode.getPathComponents(names);
-    nodesInPath = fsdir.rootDir.getExistingPathINodes(components,
-        components.length, false);
-    inodes = nodesInPath.getINodes();
-    // Length of inodes should be (components.length - 1), since we will ignore
-    // ".snapshot" 
-    assertEquals(inodes.length, components.length - 1);
-    // The number of non-null inodes should be components.length - 2, since
-    // snapshot of file3 does not exist
-    assertEquals(nodesInPath.getSize(), components.length - 2);
-    assertTrue(nodesInPath.isSnapshot());
-    // SnapshotRootIndex should still be 3: {root, Testsnapshot, sub1, s1, null}
-    assertEquals(nodesInPath.getSnapshotRootIndex(), 3);
-    assertTrue(inodes[nodesInPath.getSnapshotRootIndex()] instanceof 
-        INodeDirectoryWithSnapshot);
-    // Check the last INode in inodes, which should be null
-    assertNull(inodes[inodes.length - 1]);
-    assertTrue(inodes[inodes.length - 2] instanceof 
-        INodeDirectoryWithSnapshot);
   }
   
   /** 
@@ -348,6 +373,9 @@ public class TestSnapshotPathINodes {
    */
   @Test
   public void testSnapshotPathINodesAfterModification() throws Exception {
+    //file1 was deleted, create it again.
+    DFSTestUtil.createFile(hdfs, file1, 1024, REPLICATION, seed);
+
     // First check the INode for /TestSnapshot/sub1/file1
     String[] names = INode.getPathNames(file1.toString());
     byte[][] components = INode.getPathComponents(names);
@@ -356,6 +384,8 @@ public class TestSnapshotPathINodes {
     INode[] inodes = nodesInPath.getINodes();
     // The number of inodes should be equal to components.length
     assertEquals(inodes.length, components.length);
+    assertSnapshot(nodesInPath, false, s4, -1);
+
     // The last INode should be associated with file1
     assertEquals(inodes[components.length - 1].getFullPathName(),
         file1.toString());
@@ -363,13 +393,38 @@ public class TestSnapshotPathINodes {
     // Create a snapshot for the dir, and check the inodes for the path
     // pointing to a snapshot file
     hdfs.allowSnapshot(sub1.toString());
-    hdfs.createSnapshot("s1", sub1.toString());
+    hdfs.createSnapshot("s3", sub1.toString());
     
     // Modify file1
     DFSTestUtil.appendFile(hdfs, file1, "the content for appending");
+
+    // Check the INodes for snapshot of file1
+    String snapshotPath = sub1.toString() + "/.snapshot/s3/file1";
+    names = INode.getPathNames(snapshotPath);
+    components = INode.getPathComponents(names);
+    INodesInPath ssNodesInPath = fsdir.rootDir.getExistingPathINodes(
+        components, components.length, false);
+    INode[] ssInodes = ssNodesInPath.getINodes();
+    // Length of ssInodes should be (components.length - 1), since we will
+    // ignore ".snapshot" 
+    assertEquals(ssInodes.length, components.length - 1);
+    final Snapshot s3 = getSnapshot(ssNodesInPath, "s3");
+    assertSnapshot(ssNodesInPath, true, s3, 3);
+    // Check the INode for snapshot of file1
+    INode snapshotFileNode = ssInodes[ssInodes.length - 1]; 
+    assertEquals(snapshotFileNode.getLocalName(), file1.getName());
+    assertTrue(snapshotFileNode instanceof INodeFileSnapshot);
+    // The modification time of the snapshot INode should be the same with the
+    // original INode before modification
+    assertEquals(inodes[inodes.length - 1].getModificationTime(),
+        ssInodes[ssInodes.length - 1].getModificationTime());
+
     // Check the INode for /TestSnapshot/sub1/file1 again
+    names = INode.getPathNames(file1.toString());
+    components = INode.getPathComponents(names);
     INodesInPath newNodesInPath = fsdir.rootDir
         .getExistingPathINodes(components, components.length, false);
+    assertSnapshot(newNodesInPath, false, s3, -1);
     INode[] newInodes = newNodesInPath.getINodes();
     // The number of inodes should be equal to components.length
     assertEquals(newInodes.length, components.length);
@@ -379,25 +434,5 @@ public class TestSnapshotPathINodes {
     // The modification time of the INode for file3 should have been changed
     Assert.assertFalse(inodes[components.length - 1].getModificationTime() ==
         newInodes[components.length - 1].getModificationTime());
-    
-    // Check the INodes for snapshot of file1
-    String snapshotPath = sub1.toString() + "/.snapshot/s1/file1";
-    names = INode.getPathNames(snapshotPath);
-    components = INode.getPathComponents(names);
-    INodesInPath ssNodesInPath = fsdir.rootDir.getExistingPathINodes(
-        components, components.length, false);
-    INode[] ssInodes = ssNodesInPath.getINodes();
-    // Length of ssInodes should be (components.length - 1), since we will
-    // ignore ".snapshot" 
-    assertEquals(ssInodes.length, components.length - 1);
-    assertTrue(ssNodesInPath.isSnapshot());
-    // Check the INode for snapshot of file1
-    INode snapshotFileNode = ssInodes[ssInodes.length - 1]; 
-    assertEquals(snapshotFileNode.getLocalName(), file1.getName());
-    assertTrue(snapshotFileNode instanceof INodeFileSnapshot);
-    // The modification time of the snapshot INode should be the same with the
-    // original INode before modification
-    assertEquals(inodes[inodes.length - 1].getModificationTime(),
-        ssInodes[ssInodes.length - 1].getModificationTime());
   }
 }