accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ktur...@apache.org
Subject [1/5] git commit: ACCUMULO-2827: HeapIterator optimization. Assumes it's more probable that the next entry in a merge comes from the current iterator. Also switching from PriorityBuffer to PriorityQueue<> to avoid unsafe casts
Date Thu, 26 Jun 2014 15:31:39 GMT
Repository: accumulo
Updated Branches:
  refs/heads/1.6.1-SNAPSHOT 7e34ea923 -> f573a14db


ACCUMULO-2827: HeapIterator optimization. Assumes it's more probable that the next entry in
a merge comes from the current iterator. Also switching from PriorityBuffer to PriorityQueue<>
to avoid unsafe casts

Signed-off-by: Keith Turner <kturner@apache.org>


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

Branch: refs/heads/1.6.1-SNAPSHOT
Commit: 9801fe976ab5b86b3ed337daff586b96d2cb7e4d
Parents: 886cd19
Author: Jonathan Park <park@sqrrl.com>
Authored: Mon Jun 23 15:07:27 2014 -0400
Committer: Keith Turner <kturner@apache.org>
Committed: Thu Jun 26 10:45:51 2014 -0400

----------------------------------------------------------------------
 .../core/iterators/system/HeapIterator.java     | 132 ++++++++++---------
 1 file changed, 73 insertions(+), 59 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/accumulo/blob/9801fe97/core/src/main/java/org/apache/accumulo/core/iterators/system/HeapIterator.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/iterators/system/HeapIterator.java
b/core/src/main/java/org/apache/accumulo/core/iterators/system/HeapIterator.java
index e54f37c..4b26dcc 100644
--- a/core/src/main/java/org/apache/accumulo/core/iterators/system/HeapIterator.java
+++ b/core/src/main/java/org/apache/accumulo/core/iterators/system/HeapIterator.java
@@ -17,99 +17,113 @@
 package org.apache.accumulo.core.iterators.system;
 
 import java.io.IOException;
+import java.util.Comparator;
+import java.util.PriorityQueue;
 
 import org.apache.accumulo.core.data.Key;
 import org.apache.accumulo.core.data.Value;
 import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
-import org.apache.commons.collections.buffer.PriorityBuffer;
 
 public abstract class HeapIterator implements SortedKeyValueIterator<Key,Value> {
-  private PriorityBuffer heap;
-  private SortedKeyValueIterator<Key,Value> currentIter;
-  
-  private static class Index implements Comparable<Index> {
-    SortedKeyValueIterator<Key,Value> iter;
-    
-    public Index(SortedKeyValueIterator<Key,Value> iter) {
-      this.iter = iter;
-    }
-    
-    public int compareTo(Index o) {
-      return iter.getTopKey().compareTo(o.iter.getTopKey());
+  private PriorityQueue<SortedKeyValueIterator<Key,Value>> heap;
+  private SortedKeyValueIterator<Key,Value> topIdx = null;
+  private Key nextKey;
+
+  private static class SKVIComparator implements Comparator<SortedKeyValueIterator<Key,Value>>
{
+
+    @Override
+    public int compare(SortedKeyValueIterator<Key,Value> o1, SortedKeyValueIterator<Key,Value>
o2) {
+      return o1.getTopKey().compareTo(o2.getTopKey());
     }
   }
-  
+
   protected HeapIterator() {
     heap = null;
   }
-  
+
   protected HeapIterator(int maxSize) {
     createHeap(maxSize);
   }
-  
+
   protected void createHeap(int maxSize) {
     if (heap != null)
       throw new IllegalStateException("heap already exist");
-    
-    heap = new PriorityBuffer(maxSize == 0 ? 1 : maxSize);
+
+    heap = new PriorityQueue<SortedKeyValueIterator<Key,Value>>(maxSize == 0
? 1 : maxSize, new SKVIComparator());
   }
-  
+
   @Override
   final public Key getTopKey() {
-    return currentIter.getTopKey();
+    return topIdx.getTopKey();
   }
-  
+
   @Override
   final public Value getTopValue() {
-    return currentIter.getTopValue();
+    return topIdx.getTopValue();
   }
-  
+
   @Override
   final public boolean hasTop() {
-    return heap.size() > 0;
+    return topIdx != null;
   }
-  
+
   @Override
   final public void next() throws IOException {
-    switch (heap.size()) {
-      case 0:
-        throw new IllegalStateException("Called next() when there is no top");
-      case 1:
-        // optimization for case when heap contains one entry,
-        // avoids remove and add
-        currentIter.next();
-        if (!currentIter.hasTop()) {
-          heap.remove();
-          currentIter = null;
-        }
-        break;
-      default:
-        Index idx = (Index) heap.remove();
-        idx.iter.next();
-        if (idx.iter.hasTop()) {
-          heap.add(idx);
-        }
-        // to get to the default case heap has at least
-        // two entries, therefore there must be at least
-        // one entry when get() is called below
-        currentIter = ((Index) heap.get()).iter;
+    if (topIdx == null) {
+      throw new IllegalStateException("Called next() when there is no top");
+    }
+
+    topIdx.next();
+    if (!topIdx.hasTop()) {
+      if (nextKey == null) {
+        // No iterators left
+        topIdx = null;
+        return;
+      }
+
+      pullReferencesFromHeap();
+    } else {
+      if (nextKey == null) {
+        // topIdx is the only iterator
+        return;
+      }
+
+      if (nextKey.compareTo(topIdx.getTopKey()) < 0) {
+        // Grab the next top iterator and put the current top iterator back on the heap
+        // This updating of references is special-cased to save on percolation on edge cases
+        // since the current top is guaranteed to not be the minimum
+        SortedKeyValueIterator<Key,Value> nextTopIdx = heap.remove();
+        heap.add(topIdx);
+
+        topIdx = nextTopIdx;
+        nextKey = heap.peek().getTopKey();
+      }
     }
   }
-  
+
+  private void pullReferencesFromHeap() {
+    topIdx = heap.remove();
+    if (!heap.isEmpty()) {
+      nextKey = heap.peek().getTopKey();
+    } else {
+      nextKey = null;
+    }
+  }
+
   final protected void clear() {
     heap.clear();
-    currentIter = null;
+    topIdx = null;
+    nextKey = null;
   }
-  
+
   final protected void addSource(SortedKeyValueIterator<Key,Value> source) {
-    
-    if (source.hasTop())
-      heap.add(new Index(source));
-    
-    if (heap.size() > 0)
-      currentIter = ((Index) heap.get()).iter;
-    else
-      currentIter = null;
+    if (source.hasTop()) {
+      heap.add(source);
+      if (topIdx != null) {
+        heap.add(topIdx);
+      }
+
+      pullReferencesFromHeap();
+    }
   }
-  
 }


Mime
View raw message