hbase-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From anoopsamj...@apache.org
Subject [2/2] hbase git commit: HBASE-16608 Introducing the ability to merge ImmutableSegments without copy-compaction or SQM usage. (Anastasia)
Date Mon, 24 Oct 2016 17:40:12 GMT
HBASE-16608 Introducing the ability to merge ImmutableSegments without copy-compaction or SQM usage. (Anastasia)


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

Branch: refs/heads/master
Commit: 988d1f9bc955de53b9bf520b69be5bedc51f4f7e
Parents: 1b12a60
Author: anoopsamjohn <anoopsamjohn@gmail.com>
Authored: Mon Oct 24 23:09:48 2016 +0530
Committer: anoopsamjohn <anoopsamjohn@gmail.com>
Committed: Mon Oct 24 23:09:48 2016 +0530

----------------------------------------------------------------------
 .../hbase/regionserver/CompactingMemStore.java  |  11 +-
 .../hbase/regionserver/CompactionPipeline.java  |  25 +-
 .../hbase/regionserver/HeapMemStoreLAB.java     |  12 +-
 .../regionserver/ImmutableMemStoreLAB.java      |  92 ++++
 .../hbase/regionserver/ImmutableSegment.java    |  28 +-
 .../hbase/regionserver/MemStoreCompactor.java   | 272 +++++-----
 .../regionserver/MemStoreCompactorIterator.java | 159 ------
 .../MemStoreCompactorSegmentsIterator.java      | 149 ++++++
 .../MemStoreMergerSegmentsIterator.java         |  68 +++
 .../regionserver/MemStoreSegmentsIterator.java  |  64 +++
 .../hbase/regionserver/SegmentFactory.java      |  42 +-
 .../regionserver/VersionedSegmentsList.java     |   3 +-
 .../regionserver/TestCompactingMemStore.java    |  15 +
 .../TestCompactingToCellArrayMapMemStore.java   |  35 +-
 .../hbase/regionserver/TestMemStoreLAB.java     |   4 +-
 .../regionserver/TestPerColumnFamilyFlush.java  |   2 +-
 .../TestWalAndCompactingMemStoreFlush.java      | 516 ++++++++++++++-----
 17 files changed, 1038 insertions(+), 459 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hbase/blob/988d1f9b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/CompactingMemStore.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/CompactingMemStore.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/CompactingMemStore.java
index 177f222..1ecd868 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/CompactingMemStore.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/CompactingMemStore.java
@@ -195,9 +195,9 @@ public class CompactingMemStore extends AbstractMemStore {
     return list;
   }
 
-  public boolean swapCompactedSegments(VersionedSegmentsList versionedList,
-      ImmutableSegment result) {
-    return pipeline.swap(versionedList, result);
+  public boolean swapCompactedSegments(VersionedSegmentsList versionedList, ImmutableSegment result,
+      boolean merge) {
+    return pipeline.swap(versionedList, result, !merge);
   }
 
   /**
@@ -394,6 +394,11 @@ public class CompactingMemStore extends AbstractMemStore {
     allowCompaction.set(true);
   }
 
+  @VisibleForTesting
+  void initiateType() {
+    compactor.initiateAction();
+  }
+
   /**
    * @param cell Find the row that comes after this one.  If null, we return the
    *             first.

http://git-wip-us.apache.org/repos/asf/hbase/blob/988d1f9b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/CompactionPipeline.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/CompactionPipeline.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/CompactionPipeline.java
index 6a13f43..28c9383 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/CompactionPipeline.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/CompactionPipeline.java
@@ -90,13 +90,16 @@ public class CompactionPipeline {
    * Swapping only if there were no changes to the suffix of the list while it was compacted.
    * @param versionedList tail of the pipeline that was compacted
    * @param segment new compacted segment
+   * @param closeSuffix whether to close the suffix (to release memory), as part of swapping it out
+   *        During index merge op this will be false and for compaction it will be true.
    * @return true iff swapped tail with new compacted segment
    */
-  public boolean swap(VersionedSegmentsList versionedList, ImmutableSegment segment) {
+  public boolean swap(
+      VersionedSegmentsList versionedList, ImmutableSegment segment, boolean closeSuffix) {
     if (versionedList.getVersion() != version) {
       return false;
     }
-    LinkedList<ImmutableSegment> suffix;
+    List<ImmutableSegment> suffix;
     synchronized (pipeline){
       if(versionedList.getVersion() != version) {
         return false;
@@ -108,13 +111,14 @@ public class CompactionPipeline {
             + versionedList.getStoreSegments().size()
             + ", and the number of cells in new segment is:" + segment.getCellsCount());
       }
-      swapSuffix(suffix,segment);
+      swapSuffix(suffix,segment, closeSuffix);
     }
     if (region != null) {
       // update the global memstore size counter
       long suffixSize = getSegmentsKeySize(suffix);
       long newSize = segment.keySize();
       long delta = suffixSize - newSize;
+      assert ( closeSuffix || delta>0 ); // sanity check
       long globalMemstoreSize = region.addAndGetGlobalMemstoreSize(-delta);
       if (LOG.isDebugEnabled()) {
         LOG.debug("Suffix size: " + suffixSize + " compacted item size: " + newSize
@@ -204,10 +208,19 @@ public class CompactionPipeline {
     return pipeline.peekLast().keySize();
   }
 
-  private void swapSuffix(LinkedList<ImmutableSegment> suffix, ImmutableSegment segment) {
+  private void swapSuffix(List<ImmutableSegment> suffix, ImmutableSegment segment,
+      boolean closeSegmentsInSuffix) {
     version++;
-    for (Segment itemInSuffix : suffix) {
-      itemInSuffix.close();
+    // During index merge we won't be closing the segments undergoing the merge. Segment#close()
+    // will release the MSLAB chunks to pool. But in case of index merge there wont be any data copy
+    // from old MSLABs. So the new cells in new segment also refers to same chunks. In case of data
+    // compaction, we would have copied the cells data from old MSLAB chunks into a new chunk
+    // created for the result segment. So we can release the chunks associated with the compacted
+    // segments.
+    if (closeSegmentsInSuffix) {
+      for (Segment itemInSuffix : suffix) {
+        itemInSuffix.close();
+      }
     }
     pipeline.removeAll(suffix);
     pipeline.addLast(segment);

http://git-wip-us.apache.org/repos/asf/hbase/blob/988d1f9b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HeapMemStoreLAB.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HeapMemStoreLAB.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HeapMemStoreLAB.java
index 378601d..99b2bb6 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HeapMemStoreLAB.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HeapMemStoreLAB.java
@@ -69,6 +69,8 @@ public class HeapMemStoreLAB implements MemStoreLAB {
 
   private AtomicReference<Chunk> curChunk = new AtomicReference<Chunk>();
   // A queue of chunks from pool contained by this memstore LAB
+  // TODO: in the future, it would be better to have List implementation instead of Queue,
+  // as FIFO order is not so important here
   @VisibleForTesting
   BlockingQueue<PooledChunk> pooledChunkQueue = null;
   private final int chunkSize;
@@ -101,11 +103,11 @@ public class HeapMemStoreLAB implements MemStoreLAB {
     }
 
     // if we don't exclude allocations >CHUNK_SIZE, we'd infiniteloop on one!
-    Preconditions.checkArgument(
-      maxAlloc <= chunkSize,
-      MAX_ALLOC_KEY + " must be less than " + CHUNK_SIZE_KEY);
+    Preconditions.checkArgument(maxAlloc <= chunkSize,
+        MAX_ALLOC_KEY + " must be less than " + CHUNK_SIZE_KEY);
   }
 
+
   @Override
   public Cell copyCellInto(Cell cell) {
     int size = KeyValueUtil.length(cell);
@@ -236,8 +238,8 @@ public class HeapMemStoreLAB implements MemStoreLAB {
     return this.curChunk.get();
   }
 
-  @VisibleForTesting
-  BlockingQueue<PooledChunk> getChunkQueue() {
+
+  BlockingQueue<PooledChunk> getPooledChunks() {
     return this.pooledChunkQueue;
   }
 }

http://git-wip-us.apache.org/repos/asf/hbase/blob/988d1f9b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/ImmutableMemStoreLAB.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/ImmutableMemStoreLAB.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/ImmutableMemStoreLAB.java
new file mode 100644
index 0000000..430b642
--- /dev/null
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/ImmutableMemStoreLAB.java
@@ -0,0 +1,92 @@
+/**
+ * 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.hbase.regionserver;
+
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+
+/**
+ * A MemStoreLAB implementation which wraps N MemStoreLABs. Its main duty is in proper managing the
+ * close of the individual MemStoreLAB. This is treated as an immutable one and so do not allow to
+ * add any more Cells into it. {@link #copyCellInto(Cell)} throws Exception
+ */
+@InterfaceAudience.Private
+public class ImmutableMemStoreLAB implements MemStoreLAB {
+
+  private final AtomicInteger openScannerCount = new AtomicInteger();
+  private volatile boolean closed = false;
+
+  private final List<MemStoreLAB> mslabs;
+
+  public ImmutableMemStoreLAB(List<MemStoreLAB> mslabs) {
+    this.mslabs = mslabs;
+  }
+
+  @Override
+  public Cell copyCellInto(Cell cell) {
+    throw new IllegalStateException("This is an Immutable MemStoreLAB.");
+  }
+
+  @Override
+  public void close() {
+    // 'openScannerCount' here tracks the scanners opened on segments which directly refer to this
+    // MSLAB. The individual MSLABs this refers also having its own 'openScannerCount'. The usage of
+    // the variable in close() and decScannerCount() is as as that in HeapMemstoreLAB. Here the
+    // close just delegates the call to the individual MSLABs. The actual return of the chunks to
+    // MSLABPool will happen within individual MSLABs only (which is at the leaf level).
+    // Say an ImmutableMemStoreLAB is created over 2 HeapMemStoreLABs at some point and at that time
+    // both of them were referred by ongoing scanners. So they have > 0 'openScannerCount'. Now over
+    // the new Segment some scanners come in and this MSLABs 'openScannerCount' also goes up and
+    // then come down on finish of scanners. Now a close() call comes to this Immutable MSLAB. As
+    // it's 'openScannerCount' is zero it will call close() on both of the Heap MSLABs. Say by that
+    // time the old scanners on one of the MSLAB got over where as on the other, still an old
+    // scanner is going on. The call close() on that MSLAB will not close it immediately but will
+    // just mark it for closure as it's 'openScannerCount' still > 0. Later once the old scan is
+    // over, the decScannerCount() call will do the actual close and return of the chunks.
+    this.closed = true;
+    // When there are still on going scanners over this MSLAB, we will defer the close until all
+    // scanners finish. We will just mark it for closure. See #decScannerCount(). This will be
+    // called at end of every scan. When it is marked for closure and scanner count reached 0, we
+    // will do the actual close then.
+    checkAndCloseMSLABs(openScannerCount.get());
+  }
+
+  private void checkAndCloseMSLABs(int openScanners) {
+    if (openScanners == 0) {
+      for (MemStoreLAB mslab : this.mslabs) {
+        mslab.close();
+      }
+    }
+  }
+
+  @Override
+  public void incScannerCount() {
+    this.openScannerCount.incrementAndGet();
+  }
+
+  @Override
+  public void decScannerCount() {
+    int count = this.openScannerCount.decrementAndGet();
+    if (this.closed) {
+      checkAndCloseMSLABs(count);
+    }
+  }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/hbase/blob/988d1f9b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/ImmutableSegment.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/ImmutableSegment.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/ImmutableSegment.java
index 12b7916..8e79ad5 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/ImmutableSegment.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/ImmutableSegment.java
@@ -85,13 +85,14 @@ public class ImmutableSegment extends Segment {
    * The input parameter "type" exists for future use when more types of flat ImmutableSegments
    * are going to be introduced.
    */
-  protected ImmutableSegment(CellComparator comparator, MemStoreCompactorIterator iterator,
-      MemStoreLAB memStoreLAB, int numOfCells, Type type) {
+  protected ImmutableSegment(CellComparator comparator, MemStoreSegmentsIterator iterator,
+      MemStoreLAB memStoreLAB, int numOfCells, Type type, boolean merge) {
+
     super(null, // initiailize the CellSet with NULL
         comparator, memStoreLAB);
     this.type = type;
     // build the true CellSet based on CellArrayMap
-    CellSet cs = createCellArrayMapSet(numOfCells, iterator);
+    CellSet cs = createCellArrayMapSet(numOfCells, iterator, merge);
 
     this.setCellSet(null, cs);            // update the CellSet of the new Segment
     this.timeRange = this.timeRangeTracker == null ? null : this.timeRangeTracker.toTimeRange();
@@ -102,7 +103,7 @@ public class ImmutableSegment extends Segment {
    * list of older ImmutableSegments.
    * The given iterator returns the Cells that "survived" the compaction.
    */
-  protected ImmutableSegment(CellComparator comparator, MemStoreCompactorIterator iterator,
+  protected ImmutableSegment(CellComparator comparator, MemStoreSegmentsIterator iterator,
       MemStoreLAB memStoreLAB) {
     super(new CellSet(comparator), // initiailize the CellSet with empty CellSet
         comparator, memStoreLAB);
@@ -155,7 +156,7 @@ public class ImmutableSegment extends Segment {
   /**------------------------------------------------------------------------
    * Change the CellSet of this ImmutableSegment from one based on ConcurrentSkipListMap to one
    * based on CellArrayMap.
-   * If this ImmutableSegment is not based on ConcurrentSkipListMap , this is NOP
+   * If this ImmutableSegment is not based on ConcurrentSkipListMap , this is NOOP
    *
    * Synchronization of the CellSet replacement:
    * The reference to the CellSet is AtomicReference and is updated only when ImmutableSegment
@@ -188,19 +189,26 @@ public class ImmutableSegment extends Segment {
   /////////////////////  PRIVATE METHODS  /////////////////////
   /*------------------------------------------------------------------------*/
   // Create CellSet based on CellArrayMap from compacting iterator
-  private CellSet createCellArrayMapSet(int numOfCells, MemStoreCompactorIterator iterator) {
+  private CellSet createCellArrayMapSet(int numOfCells, MemStoreSegmentsIterator iterator,
+      boolean merge) {
 
     Cell[] cells = new Cell[numOfCells];   // build the Cell Array
     int i = 0;
     while (iterator.hasNext()) {
       Cell c = iterator.next();
       // The scanner behind the iterator is doing all the elimination logic
-      // now we just copy it to the new segment (also MSLAB copy)
-      cells[i] = maybeCloneWithAllocator(c);
-      boolean usedMSLAB = (cells[i] != c);
+      if (merge) {
+        // if this is merge we just move the Cell object without copying MSLAB
+        // the sizes still need to be updated in the new segment
+        cells[i] = c;
+      } else {
+        // now we just copy it to the new segment (also MSLAB copy)
+        cells[i] = maybeCloneWithAllocator(c);
+      }
+      boolean useMSLAB = (getMemStoreLAB()!=null);
       // second parameter true, because in compaction addition of the cell to new segment
       // is always successful
-      updateMetaInfo(c, true, usedMSLAB); // updates the size per cell
+      updateMetaInfo(c, true, useMSLAB); // updates the size per cell
       i++;
     }
     // build the immutable CellSet

http://git-wip-us.apache.org/repos/asf/hbase/blob/988d1f9b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreCompactor.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreCompactor.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreCompactor.java
index 714ffe3..0df3674 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreCompactor.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreCompactor.java
@@ -18,6 +18,7 @@
  */
 package org.apache.hadoop.hbase.regionserver;
 
+import com.google.common.annotations.VisibleForTesting;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.hbase.HConstants;
@@ -43,37 +44,30 @@ import java.util.concurrent.atomic.AtomicBoolean;
 public class MemStoreCompactor {
 
   public static final long DEEP_OVERHEAD = ClassSize
-      .align(ClassSize.OBJECT + 4 * ClassSize.REFERENCE + 2 * Bytes.SIZEOF_INT + Bytes.SIZEOF_DOUBLE
-          + ClassSize.ATOMIC_BOOLEAN);
-
-  // Option for external guidance whether flattening is allowed
-  static final String MEMSTORE_COMPACTOR_FLATTENING = "hbase.hregion.compacting.memstore.flatten";
-  static final boolean MEMSTORE_COMPACTOR_FLATTENING_DEFAULT = true;
-
-  // Option for external setting of the compacted structure (SkipList, CellArray, etc.)
+      .align(ClassSize.OBJECT
+          + 4 * ClassSize.REFERENCE
+          // compactingMemStore, versionedList, action, isInterrupted (the reference)
+          // "action" is an enum and thus it is a class with static final constants,
+          // so counting only the size of the reference to it and not the size of the internals
+          + Bytes.SIZEOF_INT            // compactionKVMax
+          + ClassSize.ATOMIC_BOOLEAN    // isInterrupted (the internals)
+      );
+
+  // Configuration options for MemStore compaction
+  static final String INDEX_COMPACTION_CONFIG = "index-compaction";
+  static final String DATA_COMPACTION_CONFIG  = "data-compaction";
+
+  // The external setting of the compacting MemStore behaviour
+  // Compaction of the index without the data is the default
   static final String COMPACTING_MEMSTORE_TYPE_KEY = "hbase.hregion.compacting.memstore.type";
-  static final int COMPACTING_MEMSTORE_TYPE_DEFAULT = 2;  // COMPACT_TO_ARRAY_MAP as default
-
-  // What percentage of the duplications is causing compaction?
-  static final String COMPACTION_THRESHOLD_REMAIN_FRACTION
-      = "hbase.hregion.compacting.memstore.comactPercent";
-  static final double COMPACTION_THRESHOLD_REMAIN_FRACTION_DEFAULT = 0.2;
+  static final String COMPACTING_MEMSTORE_TYPE_DEFAULT = INDEX_COMPACTION_CONFIG;
 
-  // Option for external guidance whether the flattening is allowed
-  static final String MEMSTORE_COMPACTOR_AVOID_SPECULATIVE_SCAN
-      = "hbase.hregion.compacting.memstore.avoidSpeculativeScan";
-  static final boolean MEMSTORE_COMPACTOR_AVOID_SPECULATIVE_SCAN_DEFAULT = false;
+  // The upper bound for the number of segments we store in the pipeline prior to merging.
+  // This constant is subject to further experimentation.
+  private static final int THRESHOLD_PIPELINE_SEGMENTS = 1;
 
   private static final Log LOG = LogFactory.getLog(MemStoreCompactor.class);
 
-  /**
-   * Types of Compaction
-   */
-  private enum Type {
-    COMPACT_TO_SKIPLIST_MAP,
-    COMPACT_TO_ARRAY_MAP
-  }
-
   private CompactingMemStore compactingMemStore;
 
   // a static version of the segment list from the pipeline
@@ -82,22 +76,28 @@ public class MemStoreCompactor {
   // a flag raised when compaction is requested to stop
   private final AtomicBoolean isInterrupted = new AtomicBoolean(false);
 
-  // the limit to the size of the groups to be later provided to MemStoreCompactorIterator
+  // the limit to the size of the groups to be later provided to MemStoreSegmentsIterator
   private final int compactionKVMax;
 
-  double fraction = 0.8;
-
-  int immutCellsNum = 0;  // number of immutable for compaction cells
+  /**
+   * Types of actions to be done on the pipeline upon MemStoreCompaction invocation.
+   * Note that every value covers the previous ones, i.e. if MERGE is the action it implies
+   * that the youngest segment is going to be flatten anyway.
+   */
+  private enum Action {
+    NOOP,
+    FLATTEN,  // flatten the youngest segment in the pipeline
+    MERGE,    // merge all the segments in the pipeline into one
+    COMPACT   // copy-compact the data of all the segments in the pipeline
+  }
 
-  private Type type = Type.COMPACT_TO_ARRAY_MAP;
+  private Action action = Action.FLATTEN;
 
   public MemStoreCompactor(CompactingMemStore compactingMemStore) {
     this.compactingMemStore = compactingMemStore;
-    this.compactionKVMax = compactingMemStore.getConfiguration().getInt(
-        HConstants.COMPACTION_KV_MAX, HConstants.COMPACTION_KV_MAX_DEFAULT);
-    this.fraction = 1 - compactingMemStore.getConfiguration().getDouble(
-        COMPACTION_THRESHOLD_REMAIN_FRACTION,
-        COMPACTION_THRESHOLD_REMAIN_FRACTION_DEFAULT);
+    this.compactionKVMax = compactingMemStore.getConfiguration()
+        .getInt(HConstants.COMPACTION_KV_MAX, HConstants.COMPACTION_KV_MAX_DEFAULT);
+    initiateAction();
   }
 
   /**----------------------------------------------------------------------
@@ -106,26 +106,16 @@ public class MemStoreCompactor {
    * is already an ongoing compaction or no segments to compact.
    */
   public boolean start() throws IOException {
-    if (!compactingMemStore.hasImmutableSegments()) return false;  // no compaction on empty
-
-    int t = compactingMemStore.getConfiguration().getInt(COMPACTING_MEMSTORE_TYPE_KEY,
-        COMPACTING_MEMSTORE_TYPE_DEFAULT);
-
-    switch (t) {
-      case 1: type = Type.COMPACT_TO_SKIPLIST_MAP;
-        break;
-      case 2: type = Type.COMPACT_TO_ARRAY_MAP;
-        break;
-      default: throw new RuntimeException("Unknown type " + type); // sanity check
+    if (!compactingMemStore.hasImmutableSegments()) { // no compaction on empty pipeline
+      return false;
     }
 
     // get a snapshot of the list of the segments from the pipeline,
     // this local copy of the list is marked with specific version
     versionedList = compactingMemStore.getImmutableSegments();
-    immutCellsNum = versionedList.getNumOfCells();
 
     if (LOG.isDebugEnabled()) {
-      LOG.debug("Starting the MemStore In-Memory Shrink of type " + type + " for store "
+      LOG.debug("Starting the In-Memory Compaction for store "
           + compactingMemStore.getStore().getColumnFamilyName());
     }
 
@@ -143,7 +133,14 @@ public class MemStoreCompactor {
   }
 
   /**----------------------------------------------------------------------
-  * Close the scanners and clear the pointers in order to allow good
+   * The interface to check whether user requested the index-compaction
+   */
+  public boolean isIndexCompaction() {
+    return (action == Action.MERGE);
+  }
+
+  /**----------------------------------------------------------------------
+  * Reset the interruption indicator and clear the pointers in order to allow good
   * garbage collection
   */
   private void releaseResources() {
@@ -152,45 +149,35 @@ public class MemStoreCompactor {
   }
 
   /**----------------------------------------------------------------------
-   * Check whether there are some signs to definitely not to flatten,
-   * returns false if we must compact. If this method returns true we
-   * still need to evaluate the compaction.
+   * Decide what to do with the new and old segments in the compaction pipeline.
+   * Implements basic in-memory compaction policy.
    */
-  private boolean shouldFlatten() {
-    boolean userToFlatten =         // the user configurable option to flatten or not to flatten
-        compactingMemStore.getConfiguration().getBoolean(MEMSTORE_COMPACTOR_FLATTENING,
-            MEMSTORE_COMPACTOR_FLATTENING_DEFAULT);
-    if (userToFlatten==false) {
-      LOG.debug("In-Memory shrink is doing compaction, as user asked to avoid flattening");
-      return false;                 // the user doesn't want to flatten
+  private Action policy() {
+
+    if (isInterrupted.get()) {      // if the entire process is interrupted cancel flattening
+      return Action.NOOP;           // the compaction also doesn't start when interrupted
     }
 
+    if (action == Action.COMPACT) { // compact according to the user request
+      LOG.debug("In-Memory Compaction Pipeline for store " + compactingMemStore.getFamilyName()
+          + " is going to be compacted, number of"
+          + " cells before compaction is " + versionedList.getNumOfCells());
+      return Action.COMPACT;
+    }
+
+    // compaction shouldn't happen or doesn't worth it
     // limit the number of the segments in the pipeline
     int numOfSegments = versionedList.getNumOfSegments();
-    if (numOfSegments > 3) {        // hard-coded for now as it is going to move to policy
-      LOG.debug("In-Memory shrink is doing compaction, as there already are " + numOfSegments
-          + " segments in the compaction pipeline");
-      return false;                 // to avoid "too many open files later", compact now
-    }
-    // till here we hvae all the signs that it is possible to flatten, run the speculative scan
-    // (if allowed by the user) to check the efficiency of compaction
-    boolean avoidSpeculativeScan =   // the user configurable option to avoid the speculative scan
-        compactingMemStore.getConfiguration().getBoolean(MEMSTORE_COMPACTOR_AVOID_SPECULATIVE_SCAN,
-            MEMSTORE_COMPACTOR_AVOID_SPECULATIVE_SCAN_DEFAULT);
-    if (avoidSpeculativeScan==true) {
-      LOG.debug("In-Memory shrink is doing flattening, as user asked to avoid compaction "
-          + "evaluation");
-      return true;                  // flatten without checking the compaction expedience
+    if (numOfSegments > THRESHOLD_PIPELINE_SEGMENTS) {
+      LOG.debug("In-Memory Compaction Pipeline for store " + compactingMemStore.getFamilyName()
+          + " is going to be merged, as there are " + numOfSegments + " segments");
+      return Action.MERGE;          // to avoid too many segments, merge now
     }
-    try {
-      immutCellsNum = countCellsForCompaction();
-      if (immutCellsNum > fraction * versionedList.getNumOfCells()) {
-        return true;
-      }
-    } catch(Exception e) {
-      return true;
-    }
-    return false;
+
+    // if nothing of the above, then just flatten the newly joined segment
+    LOG.debug("The youngest segment in the in-Memory Compaction Pipeline for store "
+        + compactingMemStore.getFamilyName() + " is going to be flattened");
+    return Action.FLATTEN;
   }
 
   /**----------------------------------------------------------------------
@@ -201,95 +188,106 @@ public class MemStoreCompactor {
   private void doCompaction() {
     ImmutableSegment result = null;
     boolean resultSwapped = false;
-
+    Action nextStep = null;
     try {
-      // PHASE I: estimate the compaction expedience - EVALUATE COMPACTION
-      if (shouldFlatten()) {
-        // too much cells "survive" the possible compaction, we do not want to compact!
-        LOG.debug("In-Memory compaction does not pay off - storing the flattened segment"
-            + " for store: " + compactingMemStore.getFamilyName());
-        // Looking for Segment in the pipeline with SkipList index, to make it flat
+      nextStep = policy();
+
+      if (nextStep == Action.NOOP) {
+        return;
+      }
+      if (nextStep == Action.FLATTEN) {
+        // Youngest Segment in the pipeline is with SkipList index, make it flat
         compactingMemStore.flattenOneSegment(versionedList.getVersion());
         return;
       }
 
-      // PHASE II: create the new compacted ImmutableSegment - START COPY-COMPACTION
+      // Create one segment representing all segments in the compaction pipeline,
+      // either by compaction or by merge
       if (!isInterrupted.get()) {
-        result = compact(immutCellsNum);
+        result = createSubstitution();
       }
 
-      // Phase III: swap the old compaction pipeline - END COPY-COMPACTION
+      // Substitute the pipeline with one segment
       if (!isInterrupted.get()) {
-        if (resultSwapped = compactingMemStore.swapCompactedSegments(versionedList, result)) {
+        if (resultSwapped = compactingMemStore.swapCompactedSegments(
+            versionedList, result, (action==Action.MERGE))) {
           // update the wal so it can be truncated and not get too long
           compactingMemStore.updateLowestUnflushedSequenceIdInWAL(true); // only if greater
         }
       }
-    } catch (Exception e) {
+    } catch (IOException e) {
       LOG.debug("Interrupting the MemStore in-memory compaction for store "
           + compactingMemStore.getFamilyName());
       Thread.currentThread().interrupt();
     } finally {
-      if ((result != null) && (!resultSwapped)) result.close();
+      // For the MERGE case, if the result was created, but swap didn't happen,
+      // we DON'T need to close the result segment (meaning its MSLAB)!
+      // Because closing the result segment means closing the chunks of all segments
+      // in the compaction pipeline, which still have ongoing scans.
+      if (nextStep != Action.MERGE) {
+        if ((result != null) && (!resultSwapped)) {
+          result.close();
+        }
+      }
       releaseResources();
     }
 
   }
 
   /**----------------------------------------------------------------------
-   * The copy-compaction is the creation of the ImmutableSegment (from the relevant type)
-   * based on the Compactor Iterator. The new ImmutableSegment is returned.
+   * Creation of the ImmutableSegment either by merge or copy-compact of the segments of the
+   * pipeline, based on the Compactor Iterator. The new ImmutableSegment is returned.
    */
-  private ImmutableSegment compact(int numOfCells) throws IOException {
-
-    LOG.debug("In-Memory compaction does pay off - The estimated number of cells "
-        + "after compaction is " + numOfCells + ", while number of cells before is " + versionedList
-        .getNumOfCells() + ". The fraction of remaining cells should be: " + fraction);
+  private ImmutableSegment createSubstitution() throws IOException {
 
     ImmutableSegment result = null;
-    MemStoreCompactorIterator iterator =
-        new MemStoreCompactorIterator(versionedList.getStoreSegments(),
-            compactingMemStore.getComparator(),
-            compactionKVMax, compactingMemStore.getStore());
-    try {
-      switch (type) {
-      case COMPACT_TO_SKIPLIST_MAP:
-        result = SegmentFactory.instance().createImmutableSegment(
-            compactingMemStore.getConfiguration(), compactingMemStore.getComparator(), iterator);
-        break;
-      case COMPACT_TO_ARRAY_MAP:
-        result = SegmentFactory.instance().createImmutableSegment(
-            compactingMemStore.getConfiguration(), compactingMemStore.getComparator(), iterator,
-            numOfCells, ImmutableSegment.Type.ARRAY_MAP_BASED);
-        break;
-      default: throw new RuntimeException("Unknown type " + type); // sanity check
-      }
-    } finally {
+    MemStoreSegmentsIterator iterator = null;
+
+    switch (action) {
+    case COMPACT:
+      iterator =
+          new MemStoreCompactorSegmentsIterator(versionedList.getStoreSegments(),
+              compactingMemStore.getComparator(),
+              compactionKVMax, compactingMemStore.getStore());
+
+      result = SegmentFactory.instance().createImmutableSegmentByCompaction(
+          compactingMemStore.getConfiguration(), compactingMemStore.getComparator(), iterator,
+          versionedList.getNumOfCells(), ImmutableSegment.Type.ARRAY_MAP_BASED);
       iterator.close();
+      break;
+    case MERGE:
+      iterator =
+          new MemStoreMergerSegmentsIterator(versionedList.getStoreSegments(),
+              compactingMemStore.getComparator(),
+              compactionKVMax, compactingMemStore.getStore());
+
+      result = SegmentFactory.instance().createImmutableSegmentByMerge(
+          compactingMemStore.getConfiguration(), compactingMemStore.getComparator(), iterator,
+          versionedList.getNumOfCells(), ImmutableSegment.Type.ARRAY_MAP_BASED,
+          versionedList.getStoreSegments());
+      iterator.close();
+      break;
+    default: throw new RuntimeException("Unknown action " + action); // sanity check
     }
 
     return result;
   }
 
   /**----------------------------------------------------------------------
-   * Count cells to estimate the efficiency of the future compaction
+   * Initiate the action according to user config, after its default is Action.MERGE
    */
-  private int countCellsForCompaction() throws IOException {
-
-    int cnt = 0;
-    MemStoreCompactorIterator iterator =
-        new MemStoreCompactorIterator(
-            versionedList.getStoreSegments(), compactingMemStore.getComparator(),
-            compactionKVMax, compactingMemStore.getStore());
+  @VisibleForTesting
+  void initiateAction() {
+    String memStoreType = compactingMemStore.getConfiguration().get(COMPACTING_MEMSTORE_TYPE_KEY,
+        COMPACTING_MEMSTORE_TYPE_DEFAULT);
 
-    try {
-      while (iterator.next() != null) {
-        cnt++;
-      }
-    } finally {
-      iterator.close();
+    switch (memStoreType) {
+    case INDEX_COMPACTION_CONFIG: action = Action.MERGE;
+      break;
+    case DATA_COMPACTION_CONFIG: action = Action.COMPACT;
+      break;
+    default:
+      throw new RuntimeException("Unknown memstore type " + memStoreType); // sanity check
     }
-
-    return cnt;
   }
 }

http://git-wip-us.apache.org/repos/asf/hbase/blob/988d1f9b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreCompactorIterator.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreCompactorIterator.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreCompactorIterator.java
deleted file mode 100644
index 9798ec2..0000000
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreCompactorIterator.java
+++ /dev/null
@@ -1,159 +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.hbase.regionserver;
-
-import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.CellComparator;
-import org.apache.hadoop.hbase.HConstants;
-import org.apache.hadoop.hbase.classification.InterfaceAudience;
-import org.apache.hadoop.hbase.client.Scan;
-
-import java.io.IOException;
-import java.util.*;
-
-/**
- * The MemStoreCompactorIterator is designed to perform one iteration over given list of segments
- * For another iteration new instance of MemStoreCompactorIterator needs to be created
- * The iterator is not thread-safe and must have only one instance in each period of time
- */
-@InterfaceAudience.Private
-public class MemStoreCompactorIterator implements Iterator<Cell> {
-
-  private List<Cell> kvs = new ArrayList<Cell>();
-
-  // scanner for full or partial pipeline (heap of segment scanners)
-  // we need to keep those scanners in order to close them at the end
-  private KeyValueScanner scanner;
-
-  // scanner on top of pipeline scanner that uses ScanQueryMatcher
-  private StoreScanner compactingScanner;
-
-  private final ScannerContext scannerContext;
-
-  private boolean hasMore;
-  private Iterator<Cell> kvsIterator;
-
-  // C-tor
-  public MemStoreCompactorIterator(List<ImmutableSegment> segments,
-      CellComparator comparator, int compactionKVMax, Store store) throws IOException {
-
-    this.scannerContext = ScannerContext.newBuilder().setBatchLimit(compactionKVMax).build();
-
-    // list of Scanners of segments in the pipeline, when compaction starts
-    List<KeyValueScanner> scanners = new ArrayList<KeyValueScanner>();
-
-    // create the list of scanners with maximally possible read point, meaning that
-    // all KVs are going to be returned by the pipeline traversing
-    for (Segment segment : segments) {
-      scanners.add(segment.getScanner(store.getSmallestReadPoint()));
-    }
-
-    scanner = new MemStoreScanner(comparator, scanners, true);
-
-    // reinitialize the compacting scanner for each instance of iterator
-    compactingScanner = createScanner(store, scanner);
-
-    hasMore = compactingScanner.next(kvs, scannerContext);
-
-    if (!kvs.isEmpty()) {
-      kvsIterator = kvs.iterator();
-    }
-
-  }
-
-  @Override
-  public boolean hasNext() {
-    if (!kvsIterator.hasNext()) {
-      // refillKVS() method should be invoked only if !kvsIterator.hasNext()
-      if (!refillKVS()) {
-        return false;
-      }
-    }
-    return (kvsIterator.hasNext() || hasMore);
-  }
-
-  @Override
-  public Cell next()  {
-    if (!kvsIterator.hasNext()) {
-      // refillKVS() method should be invoked only if !kvsIterator.hasNext()
-      if (!refillKVS())  return null;
-    }
-    return (!hasMore) ? null : kvsIterator.next();
-  }
-
-  public void close() {
-    compactingScanner.close();
-    compactingScanner = null;
-    scanner = null;
-  }
-
-  @Override
-  public void remove() {
-    throw new UnsupportedOperationException();
-  }
-
-  /**
-   * Creates the scanner for compacting the pipeline.
-   *
-   * @return the scanner
-   */
-  private StoreScanner createScanner(Store store, KeyValueScanner scanner)
-      throws IOException {
-
-    Scan scan = new Scan();
-    scan.setMaxVersions();  //Get all available versions
-    StoreScanner internalScanner =
-        new StoreScanner(store, store.getScanInfo(), scan, Collections.singletonList(scanner),
-            ScanType.COMPACT_RETAIN_DELETES, store.getSmallestReadPoint(),
-            HConstants.OLDEST_TIMESTAMP);
-
-    return internalScanner;
-  }
-
-
-
-  private boolean refillKVS() {
-    kvs.clear();          // clear previous KVS, first initiated in the constructor
-    if (!hasMore) {       // if there is nothing expected next in compactingScanner
-      return false;
-    }
-
-    try {                 // try to get next KVS
-      hasMore = compactingScanner.next(kvs, scannerContext);
-    } catch (IOException ie) {
-      throw new IllegalStateException(ie);
-    }
-
-    if (!kvs.isEmpty() ) {// is the new KVS empty ?
-      kvsIterator = kvs.iterator();
-      return true;
-    } else {
-      // KVS is empty, but hasMore still true?
-      if (hasMore) {      // try to move to next row
-        return refillKVS();
-      }
-
-    }
-    return hasMore;
-  }
-
-
-}
-

http://git-wip-us.apache.org/repos/asf/hbase/blob/988d1f9b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreCompactorSegmentsIterator.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreCompactorSegmentsIterator.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreCompactorSegmentsIterator.java
new file mode 100644
index 0000000..f31c973
--- /dev/null
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreCompactorSegmentsIterator.java
@@ -0,0 +1,149 @@
+/**
+ *
+ * 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.hbase.regionserver;
+
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellComparator;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.client.Scan;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+
+/**
+ * The MemStoreCompactorSegmentsIterator extends MemStoreSegmentsIterator
+ * and performs the scan for compaction operation meaning it is based on SQM
+ */
+@InterfaceAudience.Private
+public class MemStoreCompactorSegmentsIterator extends MemStoreSegmentsIterator {
+
+  private List<Cell> kvs = new ArrayList<Cell>();
+  private boolean hasMore;
+  private Iterator<Cell> kvsIterator;
+
+  // scanner on top of pipeline scanner that uses ScanQueryMatcher
+  private StoreScanner compactingScanner;
+
+  // C-tor
+  public MemStoreCompactorSegmentsIterator(
+      List<ImmutableSegment> segments,
+      CellComparator comparator, int compactionKVMax, Store store
+  ) throws IOException {
+    super(segments,comparator,compactionKVMax,store);
+
+    // build the scanner based on Query Matcher
+    // reinitialize the compacting scanner for each instance of iterator
+    compactingScanner = createScanner(store, scanner);
+
+    hasMore = compactingScanner.next(kvs, scannerContext);
+
+    if (!kvs.isEmpty()) {
+      kvsIterator = kvs.iterator();
+    }
+
+  }
+
+  @Override
+  public boolean hasNext() {
+    if (kvsIterator == null)  { // for the case when the result is empty
+      return false;
+    }
+    if (!kvsIterator.hasNext()) {
+      // refillKVS() method should be invoked only if !kvsIterator.hasNext()
+      if (!refillKVS()) {
+        return false;
+      }
+    }
+    return kvsIterator.hasNext();
+  }
+
+  @Override
+  public Cell next()  {
+    if (kvsIterator == null)  { // for the case when the result is empty
+      return null;
+    }
+    if (!kvsIterator.hasNext()) {
+      // refillKVS() method should be invoked only if !kvsIterator.hasNext()
+      if (!refillKVS())  return null;
+    }
+    return (!hasMore) ? null : kvsIterator.next();
+  }
+
+  public void close() {
+    compactingScanner.close();
+    compactingScanner = null;
+    scanner = null;
+  }
+
+  @Override
+  public void remove() {
+    throw new UnsupportedOperationException();
+  }
+
+  /**
+   * Creates the scanner for compacting the pipeline.
+   *
+   * @return the scanner
+   */
+  private StoreScanner createScanner(Store store, KeyValueScanner scanner)
+      throws IOException {
+
+    Scan scan = new Scan();
+    scan.setMaxVersions();  //Get all available versions
+    StoreScanner internalScanner =
+        new StoreScanner(store, store.getScanInfo(), scan, Collections.singletonList(scanner),
+            ScanType.COMPACT_RETAIN_DELETES, store.getSmallestReadPoint(),
+            HConstants.OLDEST_TIMESTAMP);
+
+    return internalScanner;
+  }
+
+
+  /* Refill kev-value set (should be invoked only when KVS is empty)
+   * Returns true if KVS is non-empty */
+  private boolean refillKVS() {
+    kvs.clear();          // clear previous KVS, first initiated in the constructor
+    if (!hasMore) {       // if there is nothing expected next in compactingScanner
+      return false;
+    }
+
+    try {                 // try to get next KVS
+      hasMore = compactingScanner.next(kvs, scannerContext);
+    } catch (IOException ie) {
+      throw new IllegalStateException(ie);
+    }
+
+    if (!kvs.isEmpty() ) {// is the new KVS empty ?
+      kvsIterator = kvs.iterator();
+      return true;
+    } else {
+      // KVS is empty, but hasMore still true?
+      if (hasMore) {      // try to move to next row
+        return refillKVS();
+      }
+
+    }
+    return hasMore;
+  }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/hbase/blob/988d1f9b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreMergerSegmentsIterator.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreMergerSegmentsIterator.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreMergerSegmentsIterator.java
new file mode 100644
index 0000000..625fc76
--- /dev/null
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreMergerSegmentsIterator.java
@@ -0,0 +1,68 @@
+/**
+ *
+ * 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.hbase.regionserver;
+
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellComparator;
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * The MemStoreMergerSegmentsIterator extends MemStoreSegmentsIterator
+ * and performs the scan for simple merge operation meaning it is NOT based on SQM
+ */
+@InterfaceAudience.Private
+public class MemStoreMergerSegmentsIterator extends MemStoreSegmentsIterator {
+
+  // C-tor
+  public MemStoreMergerSegmentsIterator(List<ImmutableSegment> segments, CellComparator comparator,
+      int compactionKVMax, Store store
+  ) throws IOException {
+    super(segments,comparator,compactionKVMax,store);
+  }
+
+  @Override
+  public boolean hasNext() {
+    return (scanner.peek()!=null);
+  }
+
+  @Override
+  public Cell next()  {
+    Cell result = null;
+    try {                 // try to get next
+      result = scanner.next();
+    } catch (IOException ie) {
+      throw new IllegalStateException(ie);
+    }
+    return result;
+  }
+
+  public void close() {
+    scanner.close();
+    scanner = null;
+  }
+
+  @Override
+  public void remove() {
+    throw new UnsupportedOperationException();
+  }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/hbase/blob/988d1f9b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreSegmentsIterator.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreSegmentsIterator.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreSegmentsIterator.java
new file mode 100644
index 0000000..8790bc2
--- /dev/null
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreSegmentsIterator.java
@@ -0,0 +1,64 @@
+/**
+ *
+ * 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.hbase.regionserver;
+
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellComparator;
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+
+import java.io.IOException;
+import java.util.*;
+
+/**
+ * The MemStoreSegmentsIterator is designed to perform one iteration over given list of segments
+ * For another iteration new instance of MemStoreSegmentsIterator needs to be created
+ * The iterator is not thread-safe and must have only one instance per MemStore
+ * in each period of time
+ */
+@InterfaceAudience.Private
+public abstract class MemStoreSegmentsIterator implements Iterator<Cell> {
+
+  // scanner for full or partial pipeline (heap of segment scanners)
+  // we need to keep those scanners in order to close them at the end
+  protected KeyValueScanner scanner;
+
+  protected final ScannerContext scannerContext;
+
+
+  // C-tor
+  public MemStoreSegmentsIterator(List<ImmutableSegment> segments, CellComparator comparator,
+      int compactionKVMax, Store store) throws IOException {
+
+    this.scannerContext = ScannerContext.newBuilder().setBatchLimit(compactionKVMax).build();
+
+    // list of Scanners of segments in the pipeline, when compaction starts
+    List<KeyValueScanner> scanners = new ArrayList<KeyValueScanner>();
+
+    // create the list of scanners with the smallest read point, meaning that
+    // only relevant KVs are going to be returned by the pipeline traversing
+    for (Segment segment : segments) {
+      scanners.add(segment.getScanner(store.getSmallestReadPoint()));
+    }
+
+    scanner = new MemStoreScanner(comparator, scanners, true);
+  }
+
+  public abstract void close();
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/hbase/blob/988d1f9b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/SegmentFactory.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/SegmentFactory.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/SegmentFactory.java
index 510ebbd..4f60976 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/SegmentFactory.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/SegmentFactory.java
@@ -25,6 +25,8 @@ import org.apache.hadoop.hbase.classification.InterfaceAudience;
 import org.apache.hadoop.hbase.util.ReflectionUtils;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
 
 /**
  * A singleton store segment factory.
@@ -46,18 +48,22 @@ public final class SegmentFactory {
 
   // create skip-list-based (non-flat) immutable segment from compacting old immutable segments
   public ImmutableSegment createImmutableSegment(final Configuration conf,
-      final CellComparator comparator, MemStoreCompactorIterator iterator) {
+      final CellComparator comparator, MemStoreSegmentsIterator iterator) {
     return new ImmutableSegment(comparator, iterator, getMemStoreLAB(conf));
   }
 
-  // create new flat immutable segment from compacting old immutable segment
-  public ImmutableSegment createImmutableSegment(final Configuration conf,
-      final CellComparator comparator, MemStoreCompactorIterator iterator, int numOfCells,
-      ImmutableSegment.Type segmentType) throws IOException {
-    Preconditions.checkArgument(segmentType != ImmutableSegment.Type.SKIPLIST_MAP_BASED,
+  // create new flat immutable segment from compacting old immutable segments
+  public ImmutableSegment createImmutableSegmentByCompaction(final Configuration conf,
+      final CellComparator comparator, MemStoreSegmentsIterator iterator, int numOfCells,
+      ImmutableSegment.Type segmentType)
+      throws IOException {
+    Preconditions.checkArgument(segmentType == ImmutableSegment.Type.ARRAY_MAP_BASED,
         "wrong immutable segment type");
-    return new ImmutableSegment(comparator, iterator, getMemStoreLAB(conf), numOfCells,
-        segmentType);
+    MemStoreLAB memStoreLAB = getMemStoreLAB(conf);
+    return
+        // the last parameter "false" means not to merge, but to compact the pipeline
+        // in order to create the new segment
+        new ImmutableSegment(comparator, iterator, memStoreLAB, numOfCells, segmentType, false);
   }
 
   // create empty immutable segment
@@ -77,6 +83,19 @@ public final class SegmentFactory {
     return generateMutableSegment(conf, comparator, memStoreLAB);
   }
 
+  // create new flat immutable segment from merging old immutable segments
+  public ImmutableSegment createImmutableSegmentByMerge(final Configuration conf,
+      final CellComparator comparator, MemStoreSegmentsIterator iterator, int numOfCells,
+      ImmutableSegment.Type segmentType, List<ImmutableSegment> segments)
+      throws IOException {
+    Preconditions.checkArgument(segmentType == ImmutableSegment.Type.ARRAY_MAP_BASED,
+        "wrong immutable segment type");
+    MemStoreLAB memStoreLAB = getMergedMemStoreLAB(conf, segments);
+    return
+        // the last parameter "true" means to merge the compaction pipeline
+        // in order to create the new segment
+        new ImmutableSegment(comparator, iterator, memStoreLAB, numOfCells, segmentType, true);
+  }
   //****** private methods to instantiate concrete store segments **********//
 
   private MutableSegment generateMutableSegment(final Configuration conf, CellComparator comparator,
@@ -96,4 +115,11 @@ public final class SegmentFactory {
     return memStoreLAB;
   }
 
+  private MemStoreLAB getMergedMemStoreLAB(Configuration conf, List<ImmutableSegment> segments) {
+    List<MemStoreLAB> mslabs = new ArrayList<MemStoreLAB>();
+    for (ImmutableSegment segment : segments) {
+      mslabs.add(segment.getMemStoreLAB());
+    }
+    return new ImmutableMemStoreLAB(mslabs);
+  }
 }

http://git-wip-us.apache.org/repos/asf/hbase/blob/988d1f9b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/VersionedSegmentsList.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/VersionedSegmentsList.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/VersionedSegmentsList.java
index 2e8bead..01160bf 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/VersionedSegmentsList.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/VersionedSegmentsList.java
@@ -19,6 +19,7 @@
 package org.apache.hadoop.hbase.regionserver;
 
 import java.util.LinkedList;
+import java.util.List;
 
 import org.apache.hadoop.hbase.classification.InterfaceAudience;
 
@@ -43,7 +44,7 @@ public class VersionedSegmentsList {
     this.version = version;
   }
 
-  public LinkedList<ImmutableSegment> getStoreSegments() {
+  public List<ImmutableSegment> getStoreSegments() {
     return storeSegments;
   }
 

http://git-wip-us.apache.org/repos/asf/hbase/blob/988d1f9b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompactingMemStore.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompactingMemStore.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompactingMemStore.java
index 211a6d8..84efb09 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompactingMemStore.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompactingMemStore.java
@@ -508,6 +508,11 @@ public class TestCompactingMemStore extends TestDefaultMemStore {
   @Test
   public void testPuttingBackChunksWithOpeningPipelineScanner()
       throws IOException {
+
+    // set memstore to do data compaction and not to use the speculative scan
+    memstore.getConfiguration().set("hbase.hregion.compacting.memstore.type", "data-compaction");
+    ((CompactingMemStore)memstore).initiateType();
+
     byte[] row = Bytes.toBytes("testrow");
     byte[] fam = Bytes.toBytes("testfamily");
     byte[] qf1 = Bytes.toBytes("testqualifier1");
@@ -585,6 +590,10 @@ public class TestCompactingMemStore extends TestDefaultMemStore {
   @Test
   public void testCompaction1Bucket() throws IOException {
 
+    // set memstore to do data compaction and not to use the speculative scan
+    memstore.getConfiguration().set("hbase.hregion.compacting.memstore.type", "data-compaction");
+    ((CompactingMemStore)memstore).initiateType();
+
     String[] keys1 = { "A", "A", "B", "C" }; //A1, A2, B3, C4
 
     // test 1 bucket
@@ -609,6 +618,9 @@ public class TestCompactingMemStore extends TestDefaultMemStore {
   @Test
   public void testCompaction2Buckets() throws IOException {
 
+    // set memstore to do data compaction and not to use the speculative scan
+    memstore.getConfiguration().set("hbase.hregion.compacting.memstore.type", "data-compaction");
+    ((CompactingMemStore)memstore).initiateType();
     String[] keys1 = { "A", "A", "B", "C" };
     String[] keys2 = { "A", "B", "D" };
 
@@ -647,6 +659,9 @@ public class TestCompactingMemStore extends TestDefaultMemStore {
   @Test
   public void testCompaction3Buckets() throws IOException {
 
+    // set memstore to do data compaction and not to use the speculative scan
+    memstore.getConfiguration().set("hbase.hregion.compacting.memstore.type", "data-compaction");
+    ((CompactingMemStore)memstore).initiateType();
     String[] keys1 = { "A", "A", "B", "C" };
     String[] keys2 = { "A", "B", "D" };
     String[] keys3 = { "D", "B", "B" };

http://git-wip-us.apache.org/repos/asf/hbase/blob/988d1f9b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompactingToCellArrayMapMemStore.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompactingToCellArrayMapMemStore.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompactingToCellArrayMapMemStore.java
index f89a040..6499251 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompactingToCellArrayMapMemStore.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompactingToCellArrayMapMemStore.java
@@ -25,6 +25,7 @@ import org.apache.hadoop.hbase.*;
 import org.apache.hadoop.hbase.testclassification.MediumTests;
 import org.apache.hadoop.hbase.testclassification.RegionServerTests;
 import org.apache.hadoop.hbase.util.Bytes;
+
 import org.apache.hadoop.hbase.util.Threads;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
@@ -61,7 +62,8 @@ public class TestCompactingToCellArrayMapMemStore extends TestCompactingMemStore
     compactingSetUp();
     Configuration conf = HBaseConfiguration.create();
 
-    conf.setLong("hbase.hregion.compacting.memstore.type", 2); // compact to CellArrayMap
+    // set memstore to do data compaction and not to use the speculative scan
+    conf.set("hbase.hregion.compacting.memstore.type", "data-compaction");
 
     this.memstore =
         new CompactingMemStore(conf, CellComparator.COMPARATOR, store,
@@ -215,18 +217,17 @@ public class TestCompactingToCellArrayMapMemStore extends TestCompactingMemStore
   }
 
   //////////////////////////////////////////////////////////////////////////////
-  // Flattening tests
+  // Merging tests
   //////////////////////////////////////////////////////////////////////////////
   @Test
-  public void testFlattening() throws IOException {
+  public void testMerging() throws IOException {
 
     String[] keys1 = { "A", "A", "B", "C", "F", "H"};
     String[] keys2 = { "A", "B", "D", "G", "I", "J"};
     String[] keys3 = { "D", "B", "B", "E" };
 
-    // set flattening to true
-    memstore.getConfiguration().setBoolean("hbase.hregion.compacting.memstore.flatten", true);
-
+    memstore.getConfiguration().set("hbase.hregion.compacting.memstore.type", "index-compaction");
+    ((CompactingMemStore)memstore).initiateType();
     addRowsByKeys(memstore, keys1);
 
     ((CompactingMemStore) memstore).flushInMemory(); // push keys to pipeline should not compact
@@ -238,13 +239,31 @@ public class TestCompactingToCellArrayMapMemStore extends TestCompactingMemStore
 
     addRowsByKeys(memstore, keys2); // also should only flatten
 
+    int counter2 = 0;
+    for ( Segment s : memstore.getSegments()) {
+      counter2 += s.getCellsCount();
+    }
+    assertEquals(12, counter2);
+
     ((CompactingMemStore) memstore).disableCompaction();
 
     ((CompactingMemStore) memstore).flushInMemory(); // push keys to pipeline without flattening
     assertEquals(0, memstore.getSnapshot().getCellsCount());
 
+    int counter3 = 0;
+    for ( Segment s : memstore.getSegments()) {
+      counter3 += s.getCellsCount();
+    }
+    assertEquals(12, counter3);
+
     addRowsByKeys(memstore, keys3);
 
+    int counter4 = 0;
+    for ( Segment s : memstore.getSegments()) {
+      counter4 += s.getCellsCount();
+    }
+    assertEquals(16, counter4);
+
     ((CompactingMemStore) memstore).enableCompaction();
 
 
@@ -258,7 +277,7 @@ public class TestCompactingToCellArrayMapMemStore extends TestCompactingMemStore
     for ( Segment s : memstore.getSegments()) {
       counter += s.getCellsCount();
     }
-    assertEquals(10,counter);
+    assertEquals(16,counter);
 
     MemStoreSnapshot snapshot = memstore.snapshot(); // push keys to snapshot
     ImmutableSegment s = memstore.getSnapshot();
@@ -295,7 +314,7 @@ public class TestCompactingToCellArrayMapMemStore extends TestCompactingMemStore
       Threads.sleep(10);
     }
     // Just doing the cnt operation here
-    MemStoreCompactorIterator itr = new MemStoreCompactorIterator(
+    MemStoreSegmentsIterator itr = new MemStoreMergerSegmentsIterator(
         ((CompactingMemStore) memstore).getImmutableSegments().getStoreSegments(),
         CellComparator.COMPARATOR, 10, ((CompactingMemStore) memstore).getStore());
     int cnt = 0;

http://git-wip-us.apache.org/repos/asf/hbase/blob/988d1f9b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestMemStoreLAB.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestMemStoreLAB.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestMemStoreLAB.java
index 1ea5112..fdd6b2c 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestMemStoreLAB.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestMemStoreLAB.java
@@ -172,7 +172,7 @@ public class TestMemStoreLAB {
   public void testLABChunkQueue() throws Exception {
     HeapMemStoreLAB mslab = new HeapMemStoreLAB();
     // by default setting, there should be no chunk queue initialized
-    assertNull(mslab.getChunkQueue());
+    assertNull(mslab.getPooledChunks());
     // reset mslab with chunk pool
     Configuration conf = HBaseConfiguration.create();
     conf.setDouble(MemStoreChunkPool.CHUNK_POOL_MAXSIZE_KEY, 0.1);
@@ -209,7 +209,7 @@ public class TestMemStoreLAB {
     // close the mslab
     mslab.close();
     // make sure all chunks reclaimed or removed from chunk queue
-    int queueLength = mslab.getChunkQueue().size();
+    int queueLength = mslab.getPooledChunks().size();
     assertTrue("All chunks in chunk queue should be reclaimed or removed"
         + " after mslab closed but actually: " + queueLength, queueLength == 0);
   }

http://git-wip-us.apache.org/repos/asf/hbase/blob/988d1f9b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestPerColumnFamilyFlush.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestPerColumnFamilyFlush.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestPerColumnFamilyFlush.java
index 6bfaa59..f0f8c39 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestPerColumnFamilyFlush.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestPerColumnFamilyFlush.java
@@ -130,7 +130,7 @@ public class TestPerColumnFamilyFlush {
     conf.setLong(FlushLargeStoresPolicy.HREGION_COLUMNFAMILY_FLUSH_SIZE_LOWER_BOUND_MIN,
       100 * 1024);
     // Intialize the region
-    Region region = initHRegion("testSelectiveFlushWhenEnabled", conf);
+    Region region = initHRegion("testSelectiveFlushWithDataCompaction", conf);
     // Add 1200 entries for CF1, 100 for CF2 and 50 for CF3
     for (int i = 1; i <= 1200; i++) {
       region.put(createPut(1, i));


Mime
View raw message