HDFS-13173.  Replace ArrayList with DirectoryDiffList(SnapshotSkipList) to 
store DirectoryDiffs.  Contributed by Shashikant Banerjee


Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/ba82e5c4
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/ba82e5c4
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/ba82e5c4

Branch: refs/heads/HDFS-7240
Commit: ba82e5c488ca0081534c1e40810b3f9e7da9eaad
Parents: 90d2bdc
Author: Tsz-Wo Nicholas Sze <szets...@hortonworks.com>
Authored: Fri Mar 2 17:46:50 2018 -0800
Committer: Tsz-Wo Nicholas Sze <szets...@hortonworks.com>
Committed: Fri Mar 2 17:47:48 2018 -0800

----------------------------------------------------------------------
 .../org/apache/hadoop/hdfs/DFSConfigKeys.java   |   9 +
 .../snapshot/AbstractINodeDiffList.java         |   8 +-
 .../namenode/snapshot/DiffListBySkipList.java   | 545 +++++++++++++++++++
 .../namenode/snapshot/DirectoryDiffList.java    | 545 -------------------
 .../snapshot/DirectoryDiffListFactory.java      |  46 ++
 .../snapshot/DirectoryWithSnapshotFeature.java  |   6 +
 .../namenode/snapshot/SnapshotManager.java      |   8 +
 .../src/main/resources/hdfs-default.xml         |  19 +-
 .../snapshot/TestDiffListBySkipList.java        | 264 +++++++++
 .../snapshot/TestDirectoryDiffList.java         | 264 ---------
 10 files changed, 902 insertions(+), 812 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hadoop/blob/ba82e5c4/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
----------------------------------------------------------------------
diff --git 
a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
 
b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
index dc9197b..8ad9d65 100644
--- 
a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
+++ 
b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSConfigKeys.java
@@ -419,6 +419,15 @@ public class DFSConfigKeys extends CommonConfigurationKeys 
{
       "dfs.namenode.snapshot.max.limit";
 
   public static final int DFS_NAMENODE_SNAPSHOT_MAX_LIMIT_DEFAULT = 65536;
+  public static final String DFS_NAMENODE_SNAPSHOT_SKIPLIST_SKIP_INTERVAL =
+      "dfs.namenode.snapshot.skiplist.interval";
+  public static final int DFS_NAMENODE_SNAPSHOT_SKIPLIST_SKIP_INTERVAL_DEFAULT 
=
+      10;
+  public static final String DFS_NAMENODE_SNAPSHOT_SKIPLIST_MAX_LEVELS =
+      "dfs.namenode.snapshot.skiplist.max.levels";
+  public static final int
+      DFS_NAMENODE_SNAPSHOT_SKIPLIST_MAX_SKIP_LEVELS_DEFAULT = 0;
+
   // Whether to enable datanode's stale state detection and usage for reads
   public static final String DFS_NAMENODE_AVOID_STALE_DATANODE_FOR_READ_KEY = 
"dfs.namenode.avoid.read.stale.datanode";
   public static final boolean 
DFS_NAMENODE_AVOID_STALE_DATANODE_FOR_READ_DEFAULT = false;

http://git-wip-us.apache.org/repos/asf/hadoop/blob/ba82e5c4/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/snapshot/AbstractINodeDiffList.java
----------------------------------------------------------------------
diff --git 
a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/snapshot/AbstractINodeDiffList.java
 
b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/snapshot/AbstractINodeDiffList.java
index 8f2465a..5163e5e 100644
--- 
a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/snapshot/AbstractINodeDiffList.java
+++ 
b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/snapshot/AbstractINodeDiffList.java
@@ -138,10 +138,14 @@ abstract class AbstractINodeDiffList<N extends INode,
     return n == 0 ? null : diffs.get(n - 1);
   }
 
+  DiffList<D> newDiffs() {
+    return new DiffListByArrayList<>(
+        INodeDirectory.DEFAULT_FILES_PER_DIRECTORY);
+  }
+
   private void createDiffsIfNeeded() {
     if (diffs == null) {
-      diffs =
-          new 
DiffListByArrayList<>(INodeDirectory.DEFAULT_FILES_PER_DIRECTORY);
+      diffs = newDiffs();
     }
   }
 

http://git-wip-us.apache.org/repos/asf/hadoop/blob/ba82e5c4/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/snapshot/DiffListBySkipList.java
----------------------------------------------------------------------
diff --git 
a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/snapshot/DiffListBySkipList.java
 
b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/snapshot/DiffListBySkipList.java
new file mode 100644
index 0000000..0d24a32
--- /dev/null
+++ 
b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/snapshot/DiffListBySkipList.java
@@ -0,0 +1,545 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs.server.namenode.snapshot;
+
+import org.apache.hadoop.hdfs.server.namenode.INodeDirectory;
+import org.apache.hadoop.hdfs.server.namenode.snapshot.
+    DirectoryWithSnapshotFeature.DirectoryDiff;
+import org.apache.hadoop.hdfs.server.namenode.snapshot.
+    DirectoryWithSnapshotFeature.ChildrenDiff;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Random;
+import java.util.Objects;
+import java.util.concurrent.ThreadLocalRandom;
+
+/**
+ * SkipList is an implementation of a data structure for storing a sorted list
+ * of Directory Diff elements, using a hierarchy of linked lists that connect
+ * increasingly sparse subsequences(defined by skip interval here) of the 
diffs.
+ * The elements contained in the tree must be mutually comparable.
+ * <p>
+ * Consider  a case where we have 10 snapshots for a directory starting from s0
+ * to s9 each associated with certain change records in terms of inodes deleted
+ * and created after a particular snapshot and before the next snapshot. The
+ * sequence will look like this:
+ * <p>
+ * s0->s1->s2->s3->s4->s5->s6->s7->s8->s9.
+ * <p>
+ * Assuming a skip interval of 3, which means a new diff will be added at a
+ * level higher than the current level after we have  ore than 3 snapshots.
+ * Next level promotion happens after 9 snapshots and so on.
+ * <p>
+ * level 2:   s08------------------------------->s9
+ * level 1:   S02------->s35-------->s68-------->s9
+ * level 0:  s0->s1->s2->s3->s4->s5->s6->s7->s8->s9
+ * <p>
+ * s02 will be created by combining diffs for s0, s1, s2 once s3 gets created.
+ * Similarly, s08 will be created by combining s02, s35 and s68 once s9 gets
+ * created.So, for constructing the children list fot s0, we have  to combine
+ * s08, s9 and reverse apply to the live fs.
+ * <p>
+ * Similarly, for constructing the children list for s2, s2, s35, s68 and s9
+ * need to get combined(or added) and reverse applied to current fs.
+ * <p>
+ * This approach will improve the snapshot deletion and snapshot diff
+ * calculation.
+ * <p>
+ * Once a snapshot gets deleted, the list needs to be balanced.
+ */
+public class DiffListBySkipList implements DiffList<DirectoryDiff> {
+  public static final Logger LOG =
+      LoggerFactory.getLogger(DiffListBySkipList.class);
+
+  private static class SkipDiff {
+    /**
+     * The references to the subsequent nodes.
+     */
+    private SkipListNode skipTo;
+    /**
+     * combined diff over a skip Interval.
+     */
+    private ChildrenDiff diff;
+
+    SkipDiff(ChildrenDiff diff) {
+      this.diff = diff;
+    }
+
+    public ChildrenDiff getDiff() {
+      return diff;
+    }
+
+    public SkipListNode getSkipTo() {
+      return skipTo;
+    }
+
+    public void setSkipTo(SkipListNode node) {
+      skipTo = node;
+    }
+
+    public void setDiff(ChildrenDiff diff) {
+      this.diff = diff;
+    }
+
+    @Override
+    public String toString() {
+      return "->" + skipTo + (diff == null? " (diff==null)": "");
+    }
+  }
+
+  /**
+   * SkipListNode is an implementation of a DirectoryDiff List node,
+   * which stores a Directory Diff and references to subsequent nodes.
+   */
+  private final static class SkipListNode implements Comparable<Integer> {
+
+    /**
+     * The data element stored in this node.
+     */
+    private DirectoryDiff diff;
+
+    /**
+     * List containing combined children diffs over a skip interval.
+     */
+    private List<SkipDiff> skipDiffList;
+
+    /**
+     * Constructs a new instance of SkipListNode with the specified data 
element
+     * and level.
+     *
+     * @param diff The element to be stored in the node.
+     */
+    SkipListNode(DirectoryDiff diff, int level) {
+      this.diff = diff;
+      skipDiffList = new ArrayList<>(level + 1);
+    }
+
+    /**
+     * Returns the level of this SkipListNode.
+     */
+    public int level() {
+      return skipDiffList.size() - 1;
+    }
+
+    void trim() {
+      for (int level = level();
+           level > 0 && getSkipNode(level) == null; level--) {
+        skipDiffList.remove(level);
+      }
+    }
+
+    public DirectoryDiff getDiff() {
+      return diff;
+    }
+
+    /**
+     * Compare diffs with snapshot ID.
+     */
+    @Override
+    public int compareTo(Integer that) {
+      return diff.compareTo(that);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) {
+        return true;
+      }
+      if (o == null || getClass() != o.getClass()) {
+        return false;
+      }
+      SkipListNode that = (SkipListNode) o;
+      return Objects.equals(diff, that.diff);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(diff);
+    }
+
+    public void setSkipDiff(ChildrenDiff cDiff, int level) {
+      if (level < skipDiffList.size()) {
+        skipDiffList.get(level).setDiff(cDiff);
+      } else {
+        skipDiffList.add(new SkipDiff(cDiff));
+      }
+    }
+
+    public void setSkipTo(SkipListNode node, int level) {
+      for (int i = skipDiffList.size(); i <= level; i++) {
+        skipDiffList.add(new SkipDiff(null));
+      }
+      skipDiffList.get(level).setSkipTo(node);
+    }
+
+    public ChildrenDiff getChildrenDiff(int level) {
+      if (level == 0) {
+        return diff.getChildrenDiff();
+      } else {
+        return skipDiffList.get(level).getDiff();
+      }
+    }
+
+    SkipListNode getSkipNode(int level) {
+      if (level >= skipDiffList.size()) {
+        return null;
+      } else {
+        return skipDiffList.get(level).getSkipTo();
+      }
+    }
+
+    @Override
+    public String toString() {
+      return diff != null ? "" + diff.getSnapshotId() : "?";
+    }
+  }
+
+  /**
+   * The reference to the first node of the list.
+   * The list will grow linearly once a new Directory diff gets added.
+   * All the list inteface defined methods provide a linear view of the list.
+   */
+  private List<SkipListNode> skipNodeList;
+
+  /**
+   * The max no of skipLevels.
+   */
+  private final int maxSkipLevels;
+
+  /**
+   * The no of diffs after which the level promotion happens.
+   */
+  private final int skipInterval;
+
+  /**
+   * The head node to the list.
+   */
+  private SkipListNode head;
+
+  /**
+   * Constructs a new, empty instance of SkipList.
+   */
+  public DiffListBySkipList(int capacity, int interval, int skipLevel) {
+    skipNodeList = new ArrayList<>(capacity);
+    head = new SkipListNode(null, 0);
+    this.maxSkipLevels = skipLevel;
+    this.skipInterval = interval;
+  }
+
+  /**
+   * Adds the specified data element to the beginning of the SkipList,
+   * if the element is not already present.
+   * @param diff the element to be inserted
+   */
+  @Override
+  public void addFirst(DirectoryDiff diff) {
+    final int nodeLevel = randomLevel(skipInterval, maxSkipLevels);
+    final SkipListNode[] nodePath = new SkipListNode[nodeLevel + 1];
+
+    Arrays.fill(nodePath, head);
+    for (int level = head.level() + 1; level <= nodeLevel; level++) {
+      head.skipDiffList.add(new SkipDiff(null));
+    }
+
+    final SkipListNode newNode = new SkipListNode(diff, nodeLevel);
+    for (int level = 0; level <= nodeLevel; level++) {
+      if (level > 0) {
+        // Case : S0 is added at the beginning and it has 3 levels
+        //  suppose the list is like:
+        //  level 1: head ------------------->s5------------->NULL
+        //  level 0:head->    s1->s2->s3->s4->s5->s6->s7->s8->s9
+        //  in this case:
+        //  level 2: head -> s0 -------------------------------->NULL
+        //  level 1: head -> s0'---------------->s5------------->NULL
+        //  level 0:head->   s0->s1->s2->s3->s4->s5->s6->s7->s8->s9
+        //  At level 1, we need to combine s0, s1, s2, s3, s4 and s5 and store
+        //  as s0'. At level 2, s0 of next is pointing to null;
+        //  Note: in this case, the diff of element being added is included
+        //  while combining the diffs.
+        final SkipListNode nextNode = head.getSkipNode(level);
+        if (nextNode != null) {
+          ChildrenDiff combined = combineDiff(newNode, nextNode, level);
+          if (combined != null) {
+            newNode.setSkipDiff(combined, level);
+          }
+        }
+      }
+      //insert to the linked list
+      newNode.setSkipTo(nodePath[level].getSkipNode(level), level);
+      nodePath[level].setSkipTo(newNode, level);
+    }
+    skipNodeList.add(0, newNode);
+  }
+
+  private SkipListNode[] findPreviousNodes(SkipListNode node, int nodeLevel) {
+    final SkipListNode[] nodePath = new SkipListNode[nodeLevel + 1];
+    SkipListNode cur = head;
+    final int headLevel = head.level();
+    for (int level = headLevel < nodeLevel ? headLevel : nodeLevel;
+         level >= 0; level--) {
+      while (cur.getSkipNode(level) != node) {
+        cur = cur.getSkipNode(level);
+      }
+      nodePath[level] = cur;
+    }
+    for (int level = headLevel + 1; level <= nodeLevel; level++) {
+      nodePath[level] = head;
+    }
+    return nodePath;
+  }
+
+  /**
+   * Adds the specified data element to the end of the SkipList,
+   * if the element is not already present.
+   * @param diff the element to be inserted
+   */
+  @Override
+  public boolean addLast(DirectoryDiff diff) {
+    final int nodeLevel = randomLevel(skipInterval, maxSkipLevels);
+    final int headLevel = head.level();
+    final SkipListNode[] nodePath = findPreviousNodes(null, nodeLevel);
+    for (int level = headLevel + 1; level <= nodeLevel; level++) {
+      head.skipDiffList.add(new SkipDiff(null));
+      nodePath[level] = head;
+    }
+
+    final SkipListNode current = new SkipListNode(diff, nodeLevel);
+    for (int level = 0; level <= nodeLevel; level++) {
+      if (level > 0 && nodePath[level] != head) {
+        //  suppose the list is like:
+        //  level 2: head ->  s1----------------------------->NULL
+        //  level 1: head ->  s1---->s3'------>s5------------->NULL
+        //  level 0:head->    s1->s2->s3->s4->s5->s6->s7->s8->s9
+
+        // case : s10 is added at the end the let the level for this node = 4
+        //  in this case,
+        //  level 2: head ->  s1''------------------------------------>s10
+        //  level 1: head ->  s1'---->s3'------>s5'-------------------->s10
+        //  level 0:head->    s1->s2->s3->s4->s5->s6->s7->s8->s9---->s10
+        //  At level 1, we combine s5, s6, s7, s8, s9 and store as s5'
+        //  At level 2, we combine s1', s3', s5' and form s1'' and store at s1.
+        // Note : the last element(elemnt being added) diff is not added while
+        // combining the diffs.
+        ChildrenDiff combined = combineDiff(nodePath[level], current, level);
+        if (combined != null) {
+          nodePath[level].setSkipDiff(combined, level);
+        }
+      }
+      nodePath[level].setSkipTo(current, level);
+      current.setSkipTo(null, level);
+    }
+    return skipNodeList.add(current);
+  }
+
+  private static ChildrenDiff combineDiff(SkipListNode from, SkipListNode to,
+      int level) {
+    ChildrenDiff combined = null;
+    SkipListNode cur = from;
+    for (int i = level - 1; i >= 0; i--) {
+      while (cur != to) {
+        final SkipListNode next = cur.getSkipNode(i);
+        if (next == null) {
+          break;
+        }
+        if (combined == null) {
+          combined = new ChildrenDiff();
+        }
+        combined.combinePosterior(cur.getChildrenDiff(i), null);
+        cur = next;
+      }
+    }
+    return combined;
+  }
+
+  /**
+   * Returns the data element at the specified index in this SkipList.
+   *
+   * @param index The index of the element to be returned.
+   * @return The element at the specified index in this SkipList.
+   */
+  @Override
+  public DirectoryDiff get(int index) {
+    return skipNodeList.get(index).getDiff();
+  }
+
+  /**
+   * Removes the element at the specified position in this list.
+   *
+   * @param index the index of the element to be removed
+   * @return the removed DirectoryDiff
+   */
+  @Override
+  public DirectoryDiff remove(int index) {
+    SkipListNode node = getNode(index);
+    int headLevel = head.level();
+    int nodeLevel = node.level();
+    final SkipListNode[] nodePath = findPreviousNodes(node, nodeLevel);
+    for (int level = 0; level <= nodeLevel; level++) {
+      if (nodePath[level] != head && level > 0) {
+        // if the last snapshot is deleted, for all the skip level nodes
+        // pointing to the last one, the combined children diff at each level
+        // > 0 should be made null and skip pointers will be updated to null.
+        // if the snapshot being deleted is not the last one, we have to merge
+        // the diff of deleted node at each level to the previous skip level
+        // node at that level and the skip pointers will be updated to point to
+        // the skip nodes of the deleted node.
+        if (index == size() - 1) {
+          nodePath[level].setSkipDiff(null, level);
+        } else {
+          /* Ideally at level 0, the deleted diff will be combined with
+           * the previous diff , and deleted inodes will be cleaned up
+           * by passing a deleted processor here while combining the diffs.
+           * Level 0 merge with previous diff will be handled inside the
+           * {@link AbstractINodeDiffList#deleteSnapshotDiff} function.
+           */
+          if (node.getChildrenDiff(level) != null) {
+            nodePath[level].getChildrenDiff(level)
+                .combinePosterior(node.getChildrenDiff(level), null);
+          }
+        }
+      }
+      nodePath[level].setSkipTo(node.getSkipNode(level), level);
+    }
+    if (nodeLevel == headLevel) {
+      head.trim();
+    }
+    return skipNodeList.remove(index).getDiff();
+  }
+
+  /**
+   * Returns true if this SkipList contains no data elements. In other words,
+   * returns true if the size of this SkipList is zero.
+   *
+   * @return True if this SkipList contains no elements.
+   */
+  @Override
+  public boolean isEmpty() {
+    return skipNodeList.isEmpty();
+  }
+
+  /**
+   * Returns the number of data elements in this SkipList.
+   *
+   * @return The number of elements in this SkipList.
+   */
+  @Override
+  public int size() {
+    return skipNodeList.size();
+  }
+
+  /**
+   * Iterator is an iterator over the SkipList. This should
+   * always provide a linear view of the list.
+   */
+  @Override
+  public Iterator<DirectoryDiff> iterator() {
+    final Iterator<SkipListNode> i = skipNodeList.iterator();
+    return new Iterator<DirectoryDiff>() {
+
+      @Override
+      public boolean hasNext() {
+        return i.hasNext();
+      }
+
+      @Override
+      public DirectoryDiff next() {
+        return i.next().getDiff();
+      }
+    };
+  }
+
+  @Override
+  public int binarySearch(int key) {
+    return Collections.binarySearch(skipNodeList, key);
+  }
+
+  private SkipListNode getNode(int index) {
+    return skipNodeList.get(index);
+  }
+
+  /**
+   * Returns the level of the skipList node.
+   *
+   * @param skipInterval The max interval after which the next level promotion
+   *                     should happen.
+   * @param maxLevel     Maximum no of skip levels
+   * @return A value in the range 0 to maxLevel-1.
+   */
+  static int randomLevel(int skipInterval, int maxLevel) {
+    final Random r = ThreadLocalRandom.current();
+    for (int level = 0; level < maxLevel; level++) {
+      // skip to the next level with probability 1/skipInterval
+      if (r.nextInt(skipInterval) > 0) {
+        return level;
+      }
+    }
+    return maxLevel;
+  }
+
+  /**
+   * This function returns the minimal set of diffs required to combine in
+   * order to generate all the changes occurred between fromIndex and
+   * toIndex.
+   *
+   * @param fromIndex index from where the summation has to start(inclusive)
+   * @param toIndex   index till where the summation has to end(exclusive)
+   * @return list of Directory Diff
+   */
+  @Override
+  public List<DirectoryDiff> getMinListForRange(int fromIndex, int toIndex,
+      INodeDirectory dir) {
+    final List<DirectoryDiff> subList = new ArrayList<>();
+    final int toSnapshotId = get(toIndex - 1).getSnapshotId();
+    for (SkipListNode current = getNode(fromIndex); current != null;) {
+      SkipListNode next = null;
+      ChildrenDiff childrenDiff = null;
+      for (int level = current.level(); level >= 0; level--) {
+        next = current.getSkipNode(level);
+        if (next != null && next.getDiff().compareTo(toSnapshotId) <= 0) {
+          childrenDiff = current.getChildrenDiff(level);
+          break;
+        }
+      }
+      final DirectoryDiff curDiff = current.getDiff();
+      subList.add(childrenDiff == null ? curDiff :
+          new DirectoryDiff(curDiff.getSnapshotId(), dir, childrenDiff));
+
+      if (current.getDiff().compareTo(toSnapshotId) == 0) {
+        break;
+      }
+      current = next;
+    }
+    return subList;
+  }
+
+  @Override
+  public String toString() {
+    final StringBuilder b = new StringBuilder(getClass().getSimpleName());
+    b.append(" head: ").append(head).append(head.skipDiffList);
+    for (SkipListNode n : skipNodeList) {
+      b.append("\n  ").append(n).append(n.skipDiffList);
+    }
+    return b.toString();
+  }
+}

http://git-wip-us.apache.org/repos/asf/hadoop/blob/ba82e5c4/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/snapshot/DirectoryDiffList.java
----------------------------------------------------------------------
diff --git 
a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/snapshot/DirectoryDiffList.java
 
b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/snapshot/DirectoryDiffList.java
deleted file mode 100644
index 58930e2..0000000
--- 
a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/snapshot/DirectoryDiffList.java
+++ /dev/null
@@ -1,545 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.hadoop.hdfs.server.namenode.snapshot;
-
-import org.apache.hadoop.hdfs.server.namenode.INodeDirectory;
-import org.apache.hadoop.hdfs.server.namenode.snapshot.
-    DirectoryWithSnapshotFeature.DirectoryDiff;
-import org.apache.hadoop.hdfs.server.namenode.snapshot.
-    DirectoryWithSnapshotFeature.ChildrenDiff;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.List;
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.Random;
-import java.util.Objects;
-import java.util.concurrent.ThreadLocalRandom;
-
-/**
- * SkipList is an implementation of a data structure for storing a sorted list
- * of Directory Diff elements, using a hierarchy of linked lists that connect
- * increasingly sparse subsequences(defined by skip interval here) of the 
diffs.
- * The elements contained in the tree must be mutually comparable.
- * <p>
- * Consider  a case where we have 10 snapshots for a directory starting from s0
- * to s9 each associated with certain change records in terms of inodes deleted
- * and created after a particular snapshot and before the next snapshot. The
- * sequence will look like this:
- * <p>
- * s0->s1->s2->s3->s4->s5->s6->s7->s8->s9.
- * <p>
- * Assuming a skip interval of 3, which means a new diff will be added at a
- * level higher than the current level after we have  ore than 3 snapshots.
- * Next level promotion happens after 9 snapshots and so on.
- * <p>
- * level 2:   s08------------------------------->s9
- * level 1:   S02------->s35-------->s68-------->s9
- * level 0:  s0->s1->s2->s3->s4->s5->s6->s7->s8->s9
- * <p>
- * s02 will be created by combining diffs for s0, s1, s2 once s3 gets created.
- * Similarly, s08 will be created by combining s02, s35 and s68 once s9 gets
- * created.So, for constructing the children list fot s0, we have  to combine
- * s08, s9 and reverse apply to the live fs.
- * <p>
- * Similarly, for constructing the children list for s2, s2, s35, s68 and s9
- * need to get combined(or added) and reverse applied to current fs.
- * <p>
- * This approach will improve the snapshot deletion and snapshot diff
- * calculation.
- * <p>
- * Once a snapshot gets deleted, the list needs to be balanced.
- */
-public class DirectoryDiffList implements DiffList<DirectoryDiff> {
-  public static final Logger LOG =
-      LoggerFactory.getLogger(DirectoryDiffList.class);
-
-  private static class SkipDiff {
-    /**
-     * The references to the subsequent nodes.
-     */
-    private SkipListNode skipTo;
-    /**
-     * combined diff over a skip Interval.
-     */
-    private ChildrenDiff diff;
-
-    SkipDiff(ChildrenDiff diff) {
-      this.diff = diff;
-    }
-
-    public ChildrenDiff getDiff() {
-      return diff;
-    }
-
-    public SkipListNode getSkipTo() {
-      return skipTo;
-    }
-
-    public void setSkipTo(SkipListNode node) {
-      skipTo = node;
-    }
-
-    public void setDiff(ChildrenDiff diff) {
-      this.diff = diff;
-    }
-
-    @Override
-    public String toString() {
-      return "->" + skipTo + (diff == null? " (diff==null)": "");
-    }
-  }
-
-  /**
-   * SkipListNode is an implementation of a DirectoryDiff List node,
-   * which stores a Directory Diff and references to subsequent nodes.
-   */
-  private final static class SkipListNode implements Comparable<Integer> {
-
-    /**
-     * The data element stored in this node.
-     */
-    private DirectoryDiff diff;
-
-    /**
-     * List containing combined children diffs over a skip interval.
-     */
-    private List<SkipDiff> skipDiffList;
-
-    /**
-     * Constructs a new instance of SkipListNode with the specified data 
element
-     * and level.
-     *
-     * @param diff The element to be stored in the node.
-     */
-    SkipListNode(DirectoryDiff diff, int level) {
-      this.diff = diff;
-      skipDiffList = new ArrayList<>(level + 1);
-    }
-
-    /**
-     * Returns the level of this SkipListNode.
-     */
-    public int level() {
-      return skipDiffList.size() - 1;
-    }
-
-    void trim() {
-      for (int level = level();
-           level > 0 && getSkipNode(level) == null; level--) {
-        skipDiffList.remove(level);
-      }
-    }
-
-    public DirectoryDiff getDiff() {
-      return diff;
-    }
-
-    /**
-     * Compare diffs with snapshot ID.
-     */
-    @Override
-    public int compareTo(Integer that) {
-      return diff.compareTo(that);
-    }
-
-    @Override
-    public boolean equals(Object o) {
-      if (this == o) {
-        return true;
-      }
-      if (o == null || getClass() != o.getClass()) {
-        return false;
-      }
-      SkipListNode that = (SkipListNode) o;
-      return Objects.equals(diff, that.diff);
-    }
-
-    @Override
-    public int hashCode() {
-      return Objects.hash(diff);
-    }
-
-    public void setSkipDiff(ChildrenDiff cDiff, int level) {
-      if (level < skipDiffList.size()) {
-        skipDiffList.get(level).setDiff(cDiff);
-      } else {
-        skipDiffList.add(new SkipDiff(cDiff));
-      }
-    }
-
-    public void setSkipTo(SkipListNode node, int level) {
-      for (int i = skipDiffList.size(); i <= level; i++) {
-        skipDiffList.add(new SkipDiff(null));
-      }
-      skipDiffList.get(level).setSkipTo(node);
-    }
-
-    public ChildrenDiff getChildrenDiff(int level) {
-      if (level == 0) {
-        return diff.getChildrenDiff();
-      } else {
-        return skipDiffList.get(level).getDiff();
-      }
-    }
-
-    SkipListNode getSkipNode(int level) {
-      if (level >= skipDiffList.size()) {
-        return null;
-      } else {
-        return skipDiffList.get(level).getSkipTo();
-      }
-    }
-
-    @Override
-    public String toString() {
-      return diff != null ? "" + diff.getSnapshotId() : "?";
-    }
-  }
-
-  /**
-   * The reference to the first node of the list.
-   * The list will grow linearly once a new Directory diff gets added.
-   * All the list inteface defined methods provide a linear view of the list.
-   */
-  private List<SkipListNode> skipNodeList;
-
-  /**
-   * The max no of skipLevels.
-   */
-  private final int maxSkipLevels;
-
-  /**
-   * The no of diffs after which the level promotion happens.
-   */
-  private final int skipInterval;
-
-  /**
-   * The head node to the list.
-   */
-  private SkipListNode head;
-
-  /**
-   * Constructs a new, empty instance of SkipList.
-   */
-  public DirectoryDiffList(int capacity, int interval, int skipLevel) {
-    skipNodeList = new ArrayList<>(capacity);
-    head = new SkipListNode(null, 0);
-    this.maxSkipLevels = skipLevel;
-    this.skipInterval = interval;
-  }
-
-  /**
-   * Adds the specified data element to the beginning of the SkipList,
-   * if the element is not already present.
-   * @param diff the element to be inserted
-   */
-  @Override
-  public void addFirst(DirectoryDiff diff) {
-    final int nodeLevel = randomLevel(skipInterval, maxSkipLevels);
-    final SkipListNode[] nodePath = new SkipListNode[nodeLevel + 1];
-
-    Arrays.fill(nodePath, head);
-    for (int level = head.level() + 1; level <= nodeLevel; level++) {
-      head.skipDiffList.add(new SkipDiff(null));
-    }
-
-    final SkipListNode newNode = new SkipListNode(diff, nodeLevel);
-    for (int level = 0; level <= nodeLevel; level++) {
-      if (level > 0) {
-        // Case : S0 is added at the beginning and it has 3 levels
-        //  suppose the list is like:
-        //  level 1: head ------------------->s5------------->NULL
-        //  level 0:head->    s1->s2->s3->s4->s5->s6->s7->s8->s9
-        //  in this case:
-        //  level 2: head -> s0 -------------------------------->NULL
-        //  level 1: head -> s0'---------------->s5------------->NULL
-        //  level 0:head->   s0->s1->s2->s3->s4->s5->s6->s7->s8->s9
-        //  At level 1, we need to combine s0, s1, s2, s3, s4 and s5 and store
-        //  as s0'. At level 2, s0 of next is pointing to null;
-        //  Note: in this case, the diff of element being added is included
-        //  while combining the diffs.
-        final SkipListNode nextNode = head.getSkipNode(level);
-        if (nextNode != null) {
-          ChildrenDiff combined = combineDiff(newNode, nextNode, level);
-          if (combined != null) {
-            newNode.setSkipDiff(combined, level);
-          }
-        }
-      }
-      //insert to the linked list
-      newNode.setSkipTo(nodePath[level].getSkipNode(level), level);
-      nodePath[level].setSkipTo(newNode, level);
-    }
-    skipNodeList.add(0, newNode);
-  }
-
-  private SkipListNode[] findPreviousNodes(SkipListNode node, int nodeLevel) {
-    final SkipListNode[] nodePath = new SkipListNode[nodeLevel + 1];
-    SkipListNode cur = head;
-    final int headLevel = head.level();
-    for (int level = headLevel < nodeLevel ? headLevel : nodeLevel;
-         level >= 0; level--) {
-      while (cur.getSkipNode(level) != node) {
-        cur = cur.getSkipNode(level);
-      }
-      nodePath[level] = cur;
-    }
-    for (int level = headLevel + 1; level <= nodeLevel; level++) {
-      nodePath[level] = head;
-    }
-    return nodePath;
-  }
-
-  /**
-   * Adds the specified data element to the end of the SkipList,
-   * if the element is not already present.
-   * @param diff the element to be inserted
-   */
-  @Override
-  public boolean addLast(DirectoryDiff diff) {
-    final int nodeLevel = randomLevel(skipInterval, maxSkipLevels);
-    final int headLevel = head.level();
-    final SkipListNode[] nodePath = findPreviousNodes(null, nodeLevel);
-    for (int level = headLevel + 1; level <= nodeLevel; level++) {
-      head.skipDiffList.add(new SkipDiff(null));
-      nodePath[level] = head;
-    }
-
-    final SkipListNode current = new SkipListNode(diff, nodeLevel);
-    for (int level = 0; level <= nodeLevel; level++) {
-      if (level > 0 && nodePath[level] != head) {
-        //  suppose the list is like:
-        //  level 2: head ->  s1----------------------------->NULL
-        //  level 1: head ->  s1---->s3'------>s5------------->NULL
-        //  level 0:head->    s1->s2->s3->s4->s5->s6->s7->s8->s9
-
-        // case : s10 is added at the end the let the level for this node = 4
-        //  in this case,
-        //  level 2: head ->  s1''------------------------------------>s10
-        //  level 1: head ->  s1'---->s3'------>s5'-------------------->s10
-        //  level 0:head->    s1->s2->s3->s4->s5->s6->s7->s8->s9---->s10
-        //  At level 1, we combine s5, s6, s7, s8, s9 and store as s5'
-        //  At level 2, we combine s1', s3', s5' and form s1'' and store at s1.
-        // Note : the last element(elemnt being added) diff is not added while
-        // combining the diffs.
-        ChildrenDiff combined = combineDiff(nodePath[level], current, level);
-        if (combined != null) {
-          nodePath[level].setSkipDiff(combined, level);
-        }
-      }
-      nodePath[level].setSkipTo(current, level);
-      current.setSkipTo(null, level);
-    }
-    return skipNodeList.add(current);
-  }
-
-  private static ChildrenDiff combineDiff(SkipListNode from, SkipListNode to,
-      int level) {
-    ChildrenDiff combined = null;
-    SkipListNode cur = from;
-    for (int i = level - 1; i >= 0; i--) {
-      while (cur != to) {
-        final SkipListNode next = cur.getSkipNode(i);
-        if (next == null) {
-          break;
-        }
-        if (combined == null) {
-          combined = new ChildrenDiff();
-        }
-        combined.combinePosterior(cur.getChildrenDiff(i), null);
-        cur = next;
-      }
-    }
-    return combined;
-  }
-
-  /**
-   * Returns the data element at the specified index in this SkipList.
-   *
-   * @param index The index of the element to be returned.
-   * @return The element at the specified index in this SkipList.
-   */
-  @Override
-  public DirectoryDiff get(int index) {
-    return skipNodeList.get(index).getDiff();
-  }
-
-  /**
-   * Removes the element at the specified position in this list.
-   *
-   * @param index the index of the element to be removed
-   * @return the removed DirectoryDiff
-   */
-  @Override
-  public DirectoryDiff remove(int index) {
-    SkipListNode node = getNode(index);
-    int headLevel = head.level();
-    int nodeLevel = node.level();
-    final SkipListNode[] nodePath = findPreviousNodes(node, nodeLevel);
-    for (int level = 0; level <= nodeLevel; level++) {
-      if (nodePath[level] != head && level > 0) {
-        // if the last snapshot is deleted, for all the skip level nodes
-        // pointing to the last one, the combined children diff at each level
-        // > 0 should be made null and skip pointers will be updated to null.
-        // if the snapshot being deleted is not the last one, we have to merge
-        // the diff of deleted node at each level to the previous skip level
-        // node at that level and the skip pointers will be updated to point to
-        // the skip nodes of the deleted node.
-        if (index == size() - 1) {
-          nodePath[level].setSkipDiff(null, level);
-        } else {
-          /* Ideally at level 0, the deleted diff will be combined with
-           * the previous diff , and deleted inodes will be cleaned up
-           * by passing a deleted processor here while combining the diffs.
-           * Level 0 merge with previous diff will be handled inside the
-           * {@link AbstractINodeDiffList#deleteSnapshotDiff} function.
-           */
-          if (node.getChildrenDiff(level) != null) {
-            nodePath[level].getChildrenDiff(level)
-                .combinePosterior(node.getChildrenDiff(level), null);
-          }
-        }
-      }
-      nodePath[level].setSkipTo(node.getSkipNode(level), level);
-    }
-    if (nodeLevel == headLevel) {
-      head.trim();
-    }
-    return skipNodeList.remove(index).getDiff();
-  }
-
-  /**
-   * Returns true if this SkipList contains no data elements. In other words,
-   * returns true if the size of this SkipList is zero.
-   *
-   * @return True if this SkipList contains no elements.
-   */
-  @Override
-  public boolean isEmpty() {
-    return skipNodeList.isEmpty();
-  }
-
-  /**
-   * Returns the number of data elements in this SkipList.
-   *
-   * @return The number of elements in this SkipList.
-   */
-  @Override
-  public int size() {
-    return skipNodeList.size();
-  }
-
-  /**
-   * Iterator is an iterator over the SkipList. This should
-   * always provide a linear view of the list.
-   */
-  @Override
-  public Iterator<DirectoryDiff> iterator() {
-    final Iterator<SkipListNode> i = skipNodeList.iterator();
-    return new Iterator<DirectoryDiff>() {
-
-      @Override
-      public boolean hasNext() {
-        return i.hasNext();
-      }
-
-      @Override
-      public DirectoryDiff next() {
-        return i.next().getDiff();
-      }
-    };
-  }
-
-  @Override
-  public int binarySearch(int key) {
-    return Collections.binarySearch(skipNodeList, key);
-  }
-
-  private SkipListNode getNode(int index) {
-    return skipNodeList.get(index);
-  }
-
-  /**
-   * Returns the level of the skipList node.
-   *
-   * @param skipInterval The max interval after which the next level promotion
-   *                     should happen.
-   * @param maxLevel     Maximum no of skip levels
-   * @return A value in the range 0 to maxLevel-1.
-   */
-  static int randomLevel(int skipInterval, int maxLevel) {
-    final Random r = ThreadLocalRandom.current();
-    for (int level = 0; level < maxLevel; level++) {
-      // skip to the next level with probability 1/skipInterval
-      if (r.nextInt(skipInterval) > 0) {
-        return level;
-      }
-    }
-    return maxLevel;
-  }
-
-  /**
-   * This function returns the minimal set of diffs required to combine in
-   * order to generate all the changes occurred between fromIndex and
-   * toIndex.
-   *
-   * @param fromIndex index from where the summation has to start(inclusive)
-   * @param toIndex   index till where the summation has to end(exclusive)
-   * @return list of Directory Diff
-   */
-  @Override
-  public List<DirectoryDiff> getMinListForRange(int fromIndex, int toIndex,
-      INodeDirectory dir) {
-    final List<DirectoryDiff> subList = new ArrayList<>();
-    final int toSnapshotId = get(toIndex - 1).getSnapshotId();
-    for (SkipListNode current = getNode(fromIndex); current != null;) {
-      SkipListNode next = null;
-      ChildrenDiff childrenDiff = null;
-      for (int level = current.level(); level >= 0; level--) {
-        next = current.getSkipNode(level);
-        if (next != null && next.getDiff().compareTo(toSnapshotId) <= 0) {
-          childrenDiff = current.getChildrenDiff(level);
-          break;
-        }
-      }
-      final DirectoryDiff curDiff = current.getDiff();
-      subList.add(childrenDiff == null ? curDiff :
-          new DirectoryDiff(curDiff.getSnapshotId(), dir, childrenDiff));
-
-      if (current.getDiff().compareTo(toSnapshotId) == 0) {
-        break;
-      }
-      current = next;
-    }
-    return subList;
-  }
-
-  @Override
-  public String toString() {
-    final StringBuilder b = new StringBuilder(getClass().getSimpleName());
-    b.append(" head: ").append(head).append(head.skipDiffList);
-    for (SkipListNode n : skipNodeList) {
-      b.append("\n  ").append(n).append(n.skipDiffList);
-    }
-    return b.toString();
-  }
-}

http://git-wip-us.apache.org/repos/asf/hadoop/blob/ba82e5c4/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/snapshot/DirectoryDiffListFactory.java
----------------------------------------------------------------------
diff --git 
a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/snapshot/DirectoryDiffListFactory.java
 
b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/snapshot/DirectoryDiffListFactory.java
new file mode 100644
index 0000000..e77d468
--- /dev/null
+++ 
b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/snapshot/DirectoryDiffListFactory.java
@@ -0,0 +1,46 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs.server.namenode.snapshot;
+
+import org.apache.commons.logging.Log;
+import 
org.apache.hadoop.hdfs.server.namenode.snapshot.DirectoryWithSnapshotFeature.DirectoryDiff;
+
+import java.util.function.IntFunction;
+
+/** For creating {@link DiffList} for {@link DirectoryDiff}. */
+public abstract class DirectoryDiffListFactory {
+  public static DiffList<DirectoryDiff> createDiffList(int capacity) {
+    return constructor.apply(capacity);
+  }
+
+  public static void init(int skipInterval, int maxLevels, Log log) {
+    if (maxLevels > 0) {
+      constructor = c -> new DiffListBySkipList(c, skipInterval, maxLevels);
+      log.info("SkipList is enabled with skipInterval=" + skipInterval
+          + ", maxLevels=" + maxLevels);
+    } else {
+      constructor = c -> new DiffListByArrayList<>(c);
+      log.info("SkipList is disabled");
+    }
+  }
+
+  private static volatile IntFunction<DiffList<DirectoryDiff>> constructor
+      = c -> new DiffListByArrayList<>(c);
+
+  private DirectoryDiffListFactory() {}
+}

http://git-wip-us.apache.org/repos/asf/hadoop/blob/ba82e5c4/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/snapshot/DirectoryWithSnapshotFeature.java
----------------------------------------------------------------------
diff --git 
a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/snapshot/DirectoryWithSnapshotFeature.java
 
b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/snapshot/DirectoryWithSnapshotFeature.java
index eecbe11..b3ce602 100644
--- 
a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/snapshot/DirectoryWithSnapshotFeature.java
+++ 
b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/snapshot/DirectoryWithSnapshotFeature.java
@@ -334,6 +334,12 @@ public class DirectoryWithSnapshotFeature implements 
INode.Feature {
         : new INodeDirectoryAttributes.SnapshotCopy(currentDir);
     }
 
+    @Override
+    DiffList<DirectoryDiff> newDiffs() {
+      return DirectoryDiffListFactory
+          .createDiffList(INodeDirectory.DEFAULT_FILES_PER_DIRECTORY);
+    }
+
     /** Replace the given child in the created/deleted list, if there is any. 
*/
     public boolean replaceChild(final ListType type, final INode oldChild,
         final INode newChild) {

http://git-wip-us.apache.org/repos/asf/hadoop/blob/ba82e5c4/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/snapshot/SnapshotManager.java
----------------------------------------------------------------------
diff --git 
a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/snapshot/SnapshotManager.java
 
b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/snapshot/SnapshotManager.java
index 7cb45dc..038ad3c 100644
--- 
a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/snapshot/SnapshotManager.java
+++ 
b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/snapshot/SnapshotManager.java
@@ -127,6 +127,14 @@ public class SnapshotManager implements 
SnapshotStatsMXBean {
         + snapshotDiffAllowSnapRootDescendant
         + ", maxSnapshotLimit: "
         + maxSnapshotLimit);
+
+    final int maxLevels = conf.getInt(
+        DFSConfigKeys.DFS_NAMENODE_SNAPSHOT_SKIPLIST_MAX_LEVELS,
+        DFSConfigKeys.DFS_NAMENODE_SNAPSHOT_SKIPLIST_MAX_SKIP_LEVELS_DEFAULT);
+    final int skipInterval = conf.getInt(
+        DFSConfigKeys.DFS_NAMENODE_SNAPSHOT_SKIPLIST_SKIP_INTERVAL,
+        DFSConfigKeys.DFS_NAMENODE_SNAPSHOT_SKIPLIST_SKIP_INTERVAL_DEFAULT);
+    DirectoryDiffListFactory.init(skipInterval, maxLevels, LOG);
   }
 
   @VisibleForTesting

http://git-wip-us.apache.org/repos/asf/hadoop/blob/ba82e5c4/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml
----------------------------------------------------------------------
diff --git 
a/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml 
b/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml
index b2da5a0..2d3c5e7 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml
@@ -4362,7 +4362,6 @@
     across to the client within one rpc call.
   </description>
 </property>
-
 <property>
   <name>dfs.namenode.snapshot.max.limit</name>
   <value>65536</value>
@@ -4374,6 +4373,24 @@
 </property>
 
 <property>
+  <name>dfs.namenode.snapshot.skiplist.max.levels</name>
+  <value>0</value>
+  <description>
+    Maximum no of the skip levels to be maintained in the skip list for
+    storing directory snapshot diffs. By default, it is set to 0 and a linear
+    list will be used to store the directory snapshot diffs.
+  </description>
+</property>
+<property>
+  <name>dfs.namenode.snapshot.skiplist.interval</name>
+  <value>10</value>
+  <description>
+    The interval after which the skip levels will be formed in the skip list
+    for storing directory snapshot diffs. By default, value is set to 10.
+  </description>
+</property>
+
+<property>
   <name>dfs.pipeline.ecn</name>
   <value>false</value>
   <description>

http://git-wip-us.apache.org/repos/asf/hadoop/blob/ba82e5c4/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/snapshot/TestDiffListBySkipList.java
----------------------------------------------------------------------
diff --git 
a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/snapshot/TestDiffListBySkipList.java
 
b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/snapshot/TestDiffListBySkipList.java
new file mode 100644
index 0000000..8062951
--- /dev/null
+++ 
b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/snapshot/TestDiffListBySkipList.java
@@ -0,0 +1,264 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs.server.namenode.snapshot;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.DistributedFileSystem;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.hdfs.server.namenode.FSDirectory;
+import org.apache.hadoop.hdfs.server.namenode.FSNamesystem;
+import org.apache.hadoop.hdfs.server.namenode.INode;
+import org.apache.hadoop.hdfs.server.namenode.INodeDirectory;
+import 
org.apache.hadoop.hdfs.server.namenode.snapshot.DirectoryWithSnapshotFeature.ChildrenDiff;
+import 
org.apache.hadoop.hdfs.server.namenode.snapshot.DirectoryWithSnapshotFeature.DirectoryDiff;
+import org.apache.hadoop.hdfs.util.ReadOnlyList;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.function.IntFunction;
+
+/**
+ * This class tests the DirectoryDiffList API's.
+ */
+public class TestDiffListBySkipList {
+  static final int NUM_SNAPSHOTS = 100;
+  static {
+    SnapshotTestHelper.disableLogs();
+  }
+
+  private static final Configuration CONF = new Configuration();
+  private static MiniDFSCluster cluster;
+  private static FSNamesystem fsn;
+  private static FSDirectory fsdir;
+  private static DistributedFileSystem hdfs;
+
+  @Before
+  public void setUp() throws Exception {
+    cluster =
+        new MiniDFSCluster.Builder(CONF).numDataNodes(0).format(true).build();
+    cluster.waitActive();
+    fsn = cluster.getNamesystem();
+    fsdir = fsn.getFSDirectory();
+    hdfs = cluster.getFileSystem();
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    if (cluster != null) {
+      cluster.shutdown();
+      cluster = null;
+    }
+  }
+
+  static void assertList(List<INode> expected, List<INode> computed) {
+    Assert.assertEquals(expected.size(), computed.size());
+    for (int index = 0; index < expected.size(); index++) {
+      Assert.assertEquals(expected.get(index), computed.get(index));
+    }
+  }
+
+  static void verifyChildrenList(DiffListBySkipList skip, INodeDirectory dir) {
+    final int n = skip.size();
+    for (int i = 0; i < skip.size(); i++) {
+      final List<INode> expected = ReadOnlyList.Util.asList(
+          dir.getChildrenList(dir.getDiffs().asList().get(i).getSnapshotId()));
+      final List<INode> computed = getChildrenList(skip, i, n, dir);
+      try {
+        assertList(expected, computed);
+      } catch (AssertionError ae) {
+        throw new AssertionError(
+            "i = " + i + "\ncomputed = " + computed + "\nexpected = "
+                + expected + "\n" + skip, ae);
+      }
+    }
+  }
+
+  static void verifyChildrenList(
+      DiffList<DirectoryDiff> array, DiffListBySkipList skip,
+      INodeDirectory dir, List<INode> childrenList) {
+    final int n = array.size();
+    Assert.assertEquals(n, skip.size());
+    for (int i = 0; i < n - 1; i++) {
+      for (int j = i + 1; j < n - 1; j++) {
+        final List<INode> expected = getCombined(array, i, j, dir)
+            .apply2Previous(childrenList);
+        final List<INode> computed = getCombined(skip, i, j, dir)
+            .apply2Previous(childrenList);
+        try {
+          assertList(expected, computed);
+        } catch (AssertionError ae) {
+          throw new AssertionError(
+              "i = " + i + ", j = " + j + "\ncomputed = " + computed
+                  + "\nexpected = " + expected + "\n" + skip, ae);
+        }
+      }
+    }
+  }
+
+  private static ChildrenDiff getCombined(
+      DiffList<DirectoryDiff> list, int from, int to, INodeDirectory dir) {
+    final List<DirectoryDiff> minList = list.getMinListForRange(from, to, dir);
+    final ChildrenDiff combined = new ChildrenDiff();
+    for (DirectoryDiff d : minList) {
+      combined.combinePosterior(d.getChildrenDiff(), null);
+    }
+    return combined;
+  }
+
+  static List<INode> getChildrenList(
+      DiffList<DirectoryDiff> list, int from, int to, INodeDirectory dir) {
+    final ChildrenDiff combined = getCombined(list, from, to, dir);
+    return combined.apply2Current(ReadOnlyList.Util.asList(dir.getChildrenList(
+        Snapshot.CURRENT_STATE_ID)));
+  }
+
+  static Path getChildPath(Path parent, int i) {
+    return new Path(parent, "c" + i);
+  }
+
+  @Test
+  public void testAddLast() throws Exception {
+    testAddLast(NUM_SNAPSHOTS);
+  }
+
+  static void testAddLast(int n) throws Exception {
+    final Path root = new Path("/testAddLast" + n);
+    DiffListBySkipList.LOG.info("run " + root);
+
+    final DiffListBySkipList skipList = new DiffListBySkipList(0, 3, 5);
+    final DiffList<DirectoryDiff> arrayList = new DiffListByArrayList<>(0);
+    INodeDirectory dir = addDiff(n, skipList, arrayList, root);
+    // verify that the both the children list obtained from hdfs and
+    // DiffListBySkipList are same
+    verifyChildrenList(skipList, dir);
+    verifyChildrenList(arrayList, skipList, dir, Collections.emptyList());
+  }
+
+
+  @Test
+  public void testAddFirst() throws Exception {
+    testAddFirst(NUM_SNAPSHOTS);
+  }
+
+  static void testAddFirst(int n) throws Exception {
+    final Path root = new Path("/testAddFirst" + n);
+    DiffListBySkipList.LOG.info("run " + root);
+
+    hdfs.mkdirs(root);
+    for (int i = 1; i < n; i++) {
+      final Path child = getChildPath(root, i);
+      hdfs.mkdirs(child);
+    }
+    INodeDirectory dir = fsdir.getINode(root.toString()).asDirectory();
+    SnapshotTestHelper.createSnapshot(hdfs, root, "s0");
+    for (int i = 1; i < n; i++) {
+      final Path child = getChildPath(root, n - i);
+      hdfs.delete(child, false);
+      hdfs.createSnapshot(root, "s" + i);
+    }
+    DiffList<DirectoryDiff> diffs = dir.getDiffs().asList();
+    List<INode> childrenList = ReadOnlyList.Util.asList(dir.getChildrenList(
+        diffs.get(0).getSnapshotId()));
+    final DiffListBySkipList skipList = new DiffListBySkipList(0, 3, 5);
+    final DiffList<DirectoryDiff> arrayList = new DiffListByArrayList<>(0);
+    for (int i = diffs.size() - 1; i >= 0; i--) {
+      final DirectoryDiff d = diffs.get(i);
+      skipList.addFirst(d);
+      arrayList.addFirst(d);
+    }
+    // verify that the both the children list obtained from hdfs and
+    // DiffListBySkipList are same
+    verifyChildrenList(skipList, dir);
+    verifyChildrenList(arrayList, skipList, dir, childrenList);
+  }
+
+  static INodeDirectory addDiff(int n, DiffList skipList, DiffList arrayList,
+      final Path root) throws Exception {
+    hdfs.mkdirs(root);
+    SnapshotTestHelper.createSnapshot(hdfs, root, "s0");
+    for (int i = 1; i < n; i++) {
+      final Path child = getChildPath(root, i);
+      hdfs.mkdirs(child);
+      hdfs.createSnapshot(root, "s" + i);
+    }
+    INodeDirectory dir = fsdir.getINode(root.toString()).asDirectory();
+    DiffList<DirectoryDiff> diffs = dir.getDiffs().asList();
+    for (DirectoryDiff d : diffs) {
+      skipList.addLast(d);
+      arrayList.addLast(d);
+    }
+    return dir;
+  }
+
+  @Test
+  public void testRemoveFromTail() throws Exception {
+    final int n = NUM_SNAPSHOTS;
+    testRemove("FromTail", n, i -> n - 1 - i);
+  }
+
+  @Test
+  public void testReomveFromHead() throws Exception {
+    testRemove("FromHead", NUM_SNAPSHOTS, i -> 0);
+  }
+
+  @Test
+  public void testRemoveRandom() throws Exception {
+    final int n = NUM_SNAPSHOTS;
+    testRemove("Random", n, i -> ThreadLocalRandom.current().nextInt(n - i));
+  }
+
+  static void testRemove(String name, int n, IntFunction<Integer> 
indexFunction)
+      throws Exception {
+    final Path root = new Path("/testRemove" + name + n);
+    DiffListBySkipList.LOG.info("run " + root);
+
+    final DiffListBySkipList skipList = new DiffListBySkipList(0, 3, 5);
+    final DiffList<DirectoryDiff> arrayList = new DiffListByArrayList<>(0);
+    final INodeDirectory dir = addDiff(n, skipList, arrayList, root);
+    Assert.assertEquals(n, arrayList.size());
+    Assert.assertEquals(n, skipList.size());
+
+    for(int i = 0; i < n; i++) {
+      final int index = indexFunction.apply(i);
+      final DirectoryDiff diff = remove(index, skipList, arrayList);
+      hdfs.deleteSnapshot(root, "s" + diff.getSnapshotId());
+      verifyChildrenList(skipList, dir);
+      verifyChildrenList(arrayList, skipList, dir, Collections.emptyList());
+    }
+  }
+
+  static DirectoryDiff remove(int i, DiffListBySkipList skip,
+      DiffList<DirectoryDiff> array) {
+    DiffListBySkipList.LOG.info("remove " + i);
+    final DirectoryDiff expected = array.remove(i);
+    final DirectoryDiff computed = skip.remove(i);
+    assertDirectoryDiff(expected, computed);
+    return expected;
+  }
+
+  static void assertDirectoryDiff(DirectoryDiff expected,
+      DirectoryDiff computed) {
+    Assert.assertEquals(expected.getSnapshotId(), computed.getSnapshotId());
+  }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/hadoop/blob/ba82e5c4/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/snapshot/TestDirectoryDiffList.java
----------------------------------------------------------------------
diff --git 
a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/snapshot/TestDirectoryDiffList.java
 
b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/snapshot/TestDirectoryDiffList.java
deleted file mode 100644
index e493e4b..0000000
--- 
a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/snapshot/TestDirectoryDiffList.java
+++ /dev/null
@@ -1,264 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.hadoop.hdfs.server.namenode.snapshot;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hdfs.DistributedFileSystem;
-import org.apache.hadoop.hdfs.MiniDFSCluster;
-import org.apache.hadoop.hdfs.server.namenode.FSDirectory;
-import org.apache.hadoop.hdfs.server.namenode.FSNamesystem;
-import org.apache.hadoop.hdfs.server.namenode.INode;
-import org.apache.hadoop.hdfs.server.namenode.INodeDirectory;
-import 
org.apache.hadoop.hdfs.server.namenode.snapshot.DirectoryWithSnapshotFeature.ChildrenDiff;
-import 
org.apache.hadoop.hdfs.server.namenode.snapshot.DirectoryWithSnapshotFeature.DirectoryDiff;
-import org.apache.hadoop.hdfs.util.ReadOnlyList;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
-
-import java.util.Collections;
-import java.util.List;
-import java.util.concurrent.ThreadLocalRandom;
-import java.util.function.IntFunction;
-
-/**
- * This class tests the DirectoryDiffList API's.
- */
-public class TestDirectoryDiffList{
-  static final int NUM_SNAPSHOTS = 100;
-  static {
-    SnapshotTestHelper.disableLogs();
-  }
-
-  private static final Configuration CONF = new Configuration();
-  private static MiniDFSCluster cluster;
-  private static FSNamesystem fsn;
-  private static FSDirectory fsdir;
-  private static DistributedFileSystem hdfs;
-
-  @Before
-  public void setUp() throws Exception {
-    cluster =
-        new MiniDFSCluster.Builder(CONF).numDataNodes(0).format(true).build();
-    cluster.waitActive();
-    fsn = cluster.getNamesystem();
-    fsdir = fsn.getFSDirectory();
-    hdfs = cluster.getFileSystem();
-  }
-
-  @After
-  public void tearDown() throws Exception {
-    if (cluster != null) {
-      cluster.shutdown();
-      cluster = null;
-    }
-  }
-
-  static void assertList(List<INode> expected, List<INode> computed) {
-    Assert.assertEquals(expected.size(), computed.size());
-    for (int index = 0; index < expected.size(); index++) {
-      Assert.assertEquals(expected.get(index), computed.get(index));
-    }
-  }
-
-  static void verifyChildrenList(DirectoryDiffList skip, INodeDirectory dir) {
-    final int n = skip.size();
-    for (int i = 0; i < skip.size(); i++) {
-      final List<INode> expected = ReadOnlyList.Util.asList(
-          dir.getChildrenList(dir.getDiffs().asList().get(i).getSnapshotId()));
-      final List<INode> computed = getChildrenList(skip, i, n, dir);
-      try {
-        assertList(expected, computed);
-      } catch (AssertionError ae) {
-        throw new AssertionError(
-            "i = " + i + "\ncomputed = " + computed + "\nexpected = "
-                + expected + "\n" + skip, ae);
-      }
-    }
-  }
-
-  static void verifyChildrenList(
-      DiffList<DirectoryDiff> array, DirectoryDiffList skip,
-      INodeDirectory dir, List<INode> childrenList) {
-    final int n = array.size();
-    Assert.assertEquals(n, skip.size());
-    for (int i = 0; i < n - 1; i++) {
-      for (int j = i + 1; j < n - 1; j++) {
-        final List<INode> expected = getCombined(array, i, j, dir)
-            .apply2Previous(childrenList);
-        final List<INode> computed = getCombined(skip, i, j, dir)
-            .apply2Previous(childrenList);
-        try {
-          assertList(expected, computed);
-        } catch (AssertionError ae) {
-          throw new AssertionError(
-              "i = " + i + ", j = " + j + "\ncomputed = " + computed
-                  + "\nexpected = " + expected + "\n" + skip, ae);
-        }
-      }
-    }
-  }
-
-  private static ChildrenDiff getCombined(
-      DiffList<DirectoryDiff> list, int from, int to, INodeDirectory dir) {
-    final List<DirectoryDiff> minList = list.getMinListForRange(from, to, dir);
-    final ChildrenDiff combined = new ChildrenDiff();
-    for (DirectoryDiff d : minList) {
-      combined.combinePosterior(d.getChildrenDiff(), null);
-    }
-    return combined;
-  }
-
-  static List<INode> getChildrenList(
-      DiffList<DirectoryDiff> list, int from, int to, INodeDirectory dir) {
-    final ChildrenDiff combined = getCombined(list, from, to, dir);
-    return combined.apply2Current(ReadOnlyList.Util.asList(dir.getChildrenList(
-        Snapshot.CURRENT_STATE_ID)));
-  }
-
-  static Path getChildPath(Path parent, int i) {
-    return new Path(parent, "c" + i);
-  }
-
-  @Test
-  public void testAddLast() throws Exception {
-    testAddLast(NUM_SNAPSHOTS);
-  }
-
-  static void testAddLast(int n) throws Exception {
-    final Path root = new Path("/testAddLast" + n);
-    DirectoryDiffList.LOG.info("run " + root);
-
-    final DirectoryDiffList skipList = new DirectoryDiffList(0, 3, 5);
-    final DiffList<DirectoryDiff> arrayList = new DiffListByArrayList<>(0);
-    INodeDirectory dir = addDiff(n, skipList, arrayList, root);
-    // verify that the both the children list obtained from hdfs and
-    // DirectoryDiffList are same
-    verifyChildrenList(skipList, dir);
-    verifyChildrenList(arrayList, skipList, dir, Collections.emptyList());
-  }
-
-
-  @Test
-  public void testAddFirst() throws Exception {
-    testAddFirst(NUM_SNAPSHOTS);
-  }
-
-  static void testAddFirst(int n) throws Exception {
-    final Path root = new Path("/testAddFirst" + n);
-    DirectoryDiffList.LOG.info("run " + root);
-
-    hdfs.mkdirs(root);
-    for (int i = 1; i < n; i++) {
-      final Path child = getChildPath(root, i);
-      hdfs.mkdirs(child);
-    }
-    INodeDirectory dir = fsdir.getINode(root.toString()).asDirectory();
-    SnapshotTestHelper.createSnapshot(hdfs, root, "s0");
-    for (int i = 1; i < n; i++) {
-      final Path child = getChildPath(root, n - i);
-      hdfs.delete(child, false);
-      hdfs.createSnapshot(root, "s" + i);
-    }
-    DiffList<DirectoryDiff> diffs = dir.getDiffs().asList();
-    List<INode> childrenList = ReadOnlyList.Util.asList(dir.getChildrenList(
-        diffs.get(0).getSnapshotId()));
-    final DirectoryDiffList skipList = new DirectoryDiffList(0, 3, 5);
-    final DiffList<DirectoryDiff> arrayList = new DiffListByArrayList<>(0);
-    for (int i = diffs.size() - 1; i >= 0; i--) {
-      final DirectoryDiff d = diffs.get(i);
-      skipList.addFirst(d);
-      arrayList.addFirst(d);
-    }
-    // verify that the both the children list obtained from hdfs and
-    // DirectoryDiffList are same
-    verifyChildrenList(skipList, dir);
-    verifyChildrenList(arrayList, skipList, dir, childrenList);
-  }
-
-  static INodeDirectory addDiff(int n, DiffList skipList, DiffList arrayList,
-      final Path root) throws Exception {
-    hdfs.mkdirs(root);
-    SnapshotTestHelper.createSnapshot(hdfs, root, "s0");
-    for (int i = 1; i < n; i++) {
-      final Path child = getChildPath(root, i);
-      hdfs.mkdirs(child);
-      hdfs.createSnapshot(root, "s" + i);
-    }
-    INodeDirectory dir = fsdir.getINode(root.toString()).asDirectory();
-    DiffList<DirectoryDiff> diffs = dir.getDiffs().asList();
-    for (DirectoryDiff d : diffs) {
-      skipList.addLast(d);
-      arrayList.addLast(d);
-    }
-    return dir;
-  }
-
-  @Test
-  public void testRemoveFromTail() throws Exception {
-    final int n = NUM_SNAPSHOTS;
-    testRemove("FromTail", n, i -> n - 1 - i);
-  }
-
-  @Test
-  public void testReomveFromHead() throws Exception {
-    testRemove("FromHead", NUM_SNAPSHOTS, i -> 0);
-  }
-
-  @Test
-  public void testRemoveRandom() throws Exception {
-    final int n = NUM_SNAPSHOTS;
-    testRemove("Random", n, i -> ThreadLocalRandom.current().nextInt(n - i));
-  }
-
-  static void testRemove(String name, int n, IntFunction<Integer> 
indexFunction)
-      throws Exception {
-    final Path root = new Path("/testRemove" + name + n);
-    DirectoryDiffList.LOG.info("run " + root);
-
-    final DirectoryDiffList skipList = new DirectoryDiffList(0, 3, 5);
-    final DiffList<DirectoryDiff> arrayList = new DiffListByArrayList<>(0);
-    final INodeDirectory dir = addDiff(n, skipList, arrayList, root);
-    Assert.assertEquals(n, arrayList.size());
-    Assert.assertEquals(n, skipList.size());
-
-    for(int i = 0; i < n; i++) {
-      final int index = indexFunction.apply(i);
-      final DirectoryDiff diff = remove(index, skipList, arrayList);
-      hdfs.deleteSnapshot(root, "s" + diff.getSnapshotId());
-      verifyChildrenList(skipList, dir);
-      verifyChildrenList(arrayList, skipList, dir, Collections.emptyList());
-    }
-  }
-
-  static DirectoryDiff remove(int i, DirectoryDiffList skip,
-      DiffList<DirectoryDiff> array) {
-    DirectoryDiffList.LOG.info("remove " + i);
-    final DirectoryDiff expected = array.remove(i);
-    final DirectoryDiff computed = skip.remove(i);
-    assertDirectoryDiff(expected, computed);
-    return expected;
-  }
-
-  static void assertDirectoryDiff(DirectoryDiff expected,
-      DirectoryDiff computed) {
-    Assert.assertEquals(expected.getSnapshotId(), computed.getSnapshotId());
-  }
-}
\ No newline at end of file


---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscr...@hadoop.apache.org
For additional commands, e-mail: common-commits-h...@hadoop.apache.org

Reply via email to