You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by an...@apache.org on 2016/10/24 17:40:12 UTC
[2/2] hbase git commit: HBASE-16608 Introducing the ability to merge
ImmutableSegments without copy-compaction or SQM usage. (Anastasia)
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 <an...@gmail.com>
Authored: Mon Oct 24 23:09:48 2016 +0530
Committer: anoopsamjohn <an...@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));